fix(schedules): stop reconciliation rewriting healthy carriers and index its scan

Follow-up hardening on the schedule recovery reconciliation.

Reconciliation settled every carrier it examined, including ones that had
already reached a terminal status and recorded their own outcome. Because the
normal completion path never stamps the reconciled marker, each successful
run's carrier was picked up on the next tick and rewritten: `completedAt`
bumped, `error` nulled and `output` replaced with a recovery stub, all of which
`GET /api/jobs/[jobId]` surfaces. A completed carrier whose execution log had
aged out was additionally flipped to failed. Settle only carriers still in
flight; a terminal one is owed schedule accounting and the marker, nothing more.

The recovery scan matched no index. There was none on `async_jobs.updated_at`,
and the unreconciled-terminal branch tested a jsonb extraction, so the whole
OR fell back to a sequential scan and sort of `async_jobs` on every tick. Add
the partial index, and spell the branch's status list and metadata key as SQL
literals: Postgres cannot prove a parameterised predicate implies a literal
index predicate, so a bound key would have left the new index unused.

Irrecoverable carrier tombstones were exempted from retention with no secondary
expiry, so the one class of row that can never reconcile grew without bound.
Give them a longer bounded window instead.

`WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN` had become a per-status budget when the
stale-execution sweep gained its `redacting` pass, silently doubling the
per-run cap. Share the budget across both passes, and add the matching partial
indexes so the second pass keeps an index rather than seq-scanning.

Also consolidates the carrier metadata keys, their predicates and the jsonb
merge into one module -- the two routes were the writer and reader of the same
keys with no shared symbol, so a rename type-checked clean while silently
breaking retention.

Author: waleedlatif1

apps/sim/app/api/cron/cleanup-stale-executions/route.test.ts

@@ -5,7 +5,16 @@ import { asyncJobs, tableJobs, workflowExecutionLogs } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { createMockRequest, dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
-import { MAX_JOB_DURATION_SECONDS, MIN_JOB_DURATION_SECONDS } from '@/lib/core/async-jobs'
+import {
+  JOB_RETENTION_HOURS,
+  MAX_JOB_DURATION_SECONDS,
+  MIN_JOB_DURATION_SECONDS,
+} from '@/lib/core/async-jobs'
+import {
+  SCHEDULE_CARRIER_IRRECOVERABLE_METADATA_KEY,
+  SCHEDULE_CARRIER_IRRECOVERABLE_RETENTION_HOURS,
+  SCHEDULE_CARRIER_RECONCILED_METADATA_KEY,
+} from '@/lib/workflows/schedules/carrier-metadata'
 
 const { mockDeleteFile, mockVerifyCronAuth } = vi.hoisted(() => ({
   mockDeleteFile: vi.fn().mockResolvedValue(undefined),
@@ -62,6 +71,33 @@ function createRequest() {
   )
 }
 
+/** Recursively renders a mocked drizzle fragment, expanding nested fragments. */
+function renderSql(fragment: unknown): string {
+  if (fragment === null || fragment === undefined) return ''
+  if (typeof fragment !== 'object') return String(fragment)
+  const candidate = fragment as { rawSql?: string; strings?: string[]; values?: unknown[] }
+  if (typeof candidate.rawSql === 'string') return candidate.rawSql
+  if (!candidate.strings) return ''
+  return candidate.strings
+    .map((part, index) =>
+      index < (candidate.values?.length ?? 0)
+        ? `${part}${renderSql(candidate.values?.[index])}`
+        : part
+    )
+    .join('')
+}
+
+/** Recursively collects the non-fragment bind values of a mocked fragment. */
+function collectSqlParams(fragment: unknown): unknown[] {
+  if (!fragment || typeof fragment !== 'object') return []
+  const candidate = fragment as { values?: unknown[] }
+  if (!candidate.values) return []
+  return candidate.values.flatMap((value) => {
+    const nested = collectSqlParams(value)
+    return nested.length > 0 ? nested : [value]
+  })
+}
+
 describe('stale execution cleanup deadline grace', () => {
   beforeEach(() => {
     vi.clearAllMocks()
@@ -237,10 +273,10 @@ describe('stale execution cleanup deadline grace', () => {
       flattenConditions(condition)
     )
     const reconciliationMarker = retentionConditions.find((condition) =>
-      condition.toSQL?.().sql.includes('scheduleReconciled')
+      renderSql(condition).includes(SCHEDULE_CARRIER_RECONCILED_METADATA_KEY)
     )
 
-    expect(reconciliationMarker?.toSQL?.().params).toContain(asyncJobs.metadata)
+    expect(collectSqlParams(reconciliationMarker)).toContain(asyncJobs.metadata)
     expect(
       retentionConditions.some(
         (condition) =>
@@ -251,19 +287,47 @@ describe('stale execution cleanup deadline grace', () => {
     ).toBe(true)
   })
 
-  it('retains irrecoverable schedule carrier tombstones indefinitely', async () => {
+  it('spells carrier metadata keys as SQL literals so the partial index matches', async () => {
+    const response = await GET(createRequest())
+
+    expect(response.status).toBe(200)
+    const reconciliationMarker = dbChainMockFns.where.mock.calls
+      .flatMap(([condition]) => flattenConditions(condition))
+      .find((condition) => renderSql(condition).includes(SCHEDULE_CARRIER_RECONCILED_METADATA_KEY))
+
+    expect(renderSql(reconciliationMarker)).toContain(
+      `'${SCHEDULE_CARRIER_RECONCILED_METADATA_KEY}'`
+    )
+    expect(collectSqlParams(reconciliationMarker)).not.toContain(
+      SCHEDULE_CARRIER_RECONCILED_METADATA_KEY
+    )
+  })
+
+  it('deletes irrecoverable schedule carrier tombstones once their longer window lapses', async () => {
     const response = await GET(createRequest())
 
     expect(response.status).toBe(200)
     const retentionConditions = dbChainMockFns.where.mock.calls.flatMap(([condition]) =>
       flattenConditions(condition)
     )
     const irrecoverableExclusion = retentionConditions.find((condition) =>
-      condition.toSQL?.().sql.includes('scheduleRecoveryIrrecoverable')
+      renderSql(condition).includes(SCHEDULE_CARRIER_IRRECOVERABLE_METADATA_KEY)
     )
 
-    expect(irrecoverableExclusion?.toSQL?.().sql).toContain("<> 'true'")
-    expect(irrecoverableExclusion?.toSQL?.().params).toContain(asyncJobs.metadata)
+    expect(renderSql(irrecoverableExclusion)).toContain("<> 'true'")
+    expect(collectSqlParams(irrecoverableExclusion)).toContain(asyncJobs.metadata)
+
+    const tombstoneWindow = retentionConditions.filter(
+      (condition) =>
+        condition.type === 'lt' &&
+        condition.left === asyncJobs.completedAt &&
+        condition.right instanceof Date
+    )
+    const oldest = Math.min(...tombstoneWindow.map(({ right }) => (right as Date).getTime()))
+    const newest = Math.max(...tombstoneWindow.map(({ right }) => (right as Date).getTime()))
+    expect(newest - oldest).toBe(
+      (SCHEDULE_CARRIER_IRRECOVERABLE_RETENTION_HOURS - JOB_RETENTION_HOURS) * 60 * 60 * 1000
+    )
   })
 
   it('keeps table-job heartbeat cleanup independent from workflow timeout policy', async () => {
@@ -382,7 +446,7 @@ describe('stale execution cleanup deadline grace', () => {
     expect(mockDeleteFile).toHaveBeenCalledTimes(1000)
 
     const limits = dbChainMockFns.limit.mock.calls.map(([limit]) => limit)
-    expect(limits.filter((limit) => limit === 100)).toHaveLength(21)
+    expect(limits.filter((limit) => limit === 100)).toHaveLength(20)
     expect(limits.filter((limit) => limit === 1000)).toHaveLength(30)
     expect(limits.filter((limit) => limit === 2000)).toHaveLength(11)
 

apps/sim/app/api/cron/cleanup-stale-executions/route.ts

@@ -17,6 +17,7 @@ import {
   JOB_STATUS,
   MAX_JOB_DURATION_SECONDS,
   MIN_JOB_DURATION_SECONDS,
+  TERMINAL_JOB_STATUSES,
 } from '@/lib/core/async-jobs'
 import {
   getExecutionReservationTtlMs,
@@ -26,7 +27,14 @@ import {
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import type { DbTransaction } from '@/lib/db/types'
 import { elapsedDurationMsSql } from '@/lib/logs/execution/duration'
+import { STALE_SWEEPABLE_EXECUTION_STATUSES } from '@/lib/logs/types'
 import { deleteFile } from '@/lib/uploads/core/storage-service'
+import {
+  carrierNotIrrecoverableSql,
+  carrierReconciledSql,
+  SCHEDULE_CARRIER_IRRECOVERABLE_RETENTION_HOURS,
+} from '@/lib/workflows/schedules/carrier-metadata'
+import { SCHEDULE_EXECUTION_QUEUE_NAME } from '@/lib/workflows/schedules/execution-limits'
 
 const logger = createLogger('CleanupStaleExecutions')
 
@@ -152,12 +160,17 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
         EXTRACT(EPOCH FROM (${cleanupTimestamp} - ${workflowExecutionLogs.startedAt})) / 60
       )::integer`
       const totalDurationMs = elapsedDurationMsSql(now)
-      for (const executionStatus of ['running', 'redacting'] as const) {
+      /**
+       * Swept one status at a time so each pass stays on its own partial index;
+       * `status IN (...)` would match neither. The row budget is shared across
+       * both passes so the per-run cap keeps meaning what its name says.
+       */
+      let workflowRowsConsidered = 0
+      for (const executionStatus of STALE_SWEEPABLE_EXECUTION_STATUSES) {
         const staleExecutionPredicate = and(
           eq(workflowExecutionLogs.status, executionStatus),
           staleExecutionTimePredicate
         )
-        let workflowRowsConsidered = 0
         while (workflowRowsConsidered < WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN) {
           const limit = Math.min(
             WORKFLOW_EXECUTION_MUTATION_BATCH_SIZE,
@@ -216,16 +229,12 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
           workflowRowsConsidered += candidates.length
           if (candidates.length < limit) break
         }
+      }
 
-        if (workflowRowsConsidered >= WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN) {
-          logger.info(
-            'Deferred remaining stale workflow executions after reaching the per-run cap',
-            {
-              status: executionStatus,
-              maxRowsPerRun: WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN,
-            }
-          )
-        }
+      if (workflowRowsConsidered >= WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN) {
+        logger.info('Deferred remaining stale workflow executions after reaching the per-run cap', {
+          maxRowsPerRun: WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN,
+        })
       }
     } catch (error) {
       logger.error('Failed to clean up stale workflow executions:', {
@@ -260,7 +269,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       END`
       const staleProcessingPredicate = and(
         eq(asyncJobs.status, JOB_STATUS.PROCESSING),
-        ne(asyncJobs.type, 'schedule-execution'),
+        ne(asyncJobs.type, SCHEDULE_EXECUTION_QUEUE_NAME),
         staleProcessingDurationPredicate
       )
       const staleProcessingResult = await runBatchedMutation({
@@ -403,7 +412,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     try {
       const stalePendingPredicate = and(
         eq(asyncJobs.status, JOB_STATUS.PENDING),
-        ne(asyncJobs.type, 'schedule-execution'),
+        ne(asyncJobs.type, SCHEDULE_EXECUTION_QUEUE_NAME),
         lt(asyncJobs.createdAt, stalePendingThreshold)
       )
       const stalePendingResult = await runBatchedMutation({
@@ -445,16 +454,27 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     }
 
     const retentionThreshold = new Date(Date.now() - JOB_RETENTION_HOURS * 60 * 60 * 1000)
+    const irrecoverableCarrierRetentionThreshold = new Date(
+      Date.now() - SCHEDULE_CARRIER_IRRECOVERABLE_RETENTION_HOURS * 60 * 60 * 1000
+    )
     let asyncJobsDeleted = 0
 
     try {
       const retainedJobPredicate = and(
-        inArray(asyncJobs.status, [JOB_STATUS.COMPLETED, JOB_STATUS.FAILED, JOB_STATUS.CANCELLED]),
+        inArray(asyncJobs.status, TERMINAL_JOB_STATUSES),
         or(
-          ne(asyncJobs.type, 'schedule-execution'),
+          ne(asyncJobs.type, SCHEDULE_EXECUTION_QUEUE_NAME),
+          /**
+           * Schedule recovery owns a carrier until it stamps the reconciled
+           * marker, so retention waits for it rather than deleting an
+           * occurrence that has not been accounted for yet.
+           */
           and(
-            sql`${asyncJobs.metadata}->>'scheduleReconciled' = 'true'`,
-            sql`COALESCE(${asyncJobs.metadata}->>'scheduleRecoveryIrrecoverable', 'false') <> 'true'`
+            carrierReconciledSql(asyncJobs.metadata),
+            or(
+              carrierNotIrrecoverableSql(asyncJobs.metadata),
+              lt(asyncJobs.completedAt, irrecoverableCarrierRetentionThreshold)
+            )
           )
         ),
         lt(asyncJobs.completedAt, retentionThreshold)

apps/sim/app/api/schedules/execute/route.test.ts

@@ -620,8 +620,11 @@ describe('Scheduled Workflow Execution API Route', () => {
 
     await runScheduleTick('test-request-id')
 
+    expect(dbChainMockFns.set).not.toHaveBeenCalledWith(
+      expect.objectContaining({ status: expect.anything() })
+    )
     expect(dbChainMockFns.set).toHaveBeenCalledWith(
-      expect.objectContaining({ status: 'cancelled', error: 'Cancelled' })
+      expect.objectContaining({ metadata: expect.anything() })
     )
     expect(mockApplyScheduleCancellationUpdate).toHaveBeenCalledOnce()
     expect(mockExecuteScheduleJob).not.toHaveBeenCalled()
@@ -651,8 +654,8 @@ describe('Scheduled Workflow Execution API Route', () => {
 
     await runScheduleTick('test-request-id')
 
-    expect(dbChainMockFns.set).toHaveBeenCalledWith(
-      expect.objectContaining({ status: 'cancelled', error: 'Cancelled' })
+    expect(dbChainMockFns.set).not.toHaveBeenCalledWith(
+      expect.objectContaining({ status: expect.anything() })
     )
     expect(mockApplyScheduleSuccessUpdate).toHaveBeenCalledOnce()
     expect(mockApplyScheduleCancellationUpdate).not.toHaveBeenCalled()
@@ -800,6 +803,30 @@ describe('Scheduled Workflow Execution API Route', () => {
 
     await runScheduleTick('test-request-id')
 
+    expect(dbChainMockFns.set).not.toHaveBeenCalledWith(
+      expect.objectContaining({ status: expect.anything() })
+    )
+    expect(dbChainMockFns.set).toHaveBeenCalledWith(
+      expect.objectContaining({ metadata: expect.anything(), updatedAt: expect.any(Date) })
+    )
+    expect(mockApplyScheduleFailureUpdate).not.toHaveBeenCalled()
+    expect(mockExecuteScheduleJob).not.toHaveBeenCalled()
+  })
+
+  it('settles a malformed in-flight carrier before marking it irrecoverable', async () => {
+    mockShouldExecuteInline.mockReturnValue(true)
+    mockProcessingCounts(0, 0)
+    orderByLimitMock.mockResolvedValueOnce([
+      {
+        id: 'malformed-job-id',
+        status: 'processing',
+        payload: { workflowId: 'workflow-1' },
+      },
+    ])
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'malformed-job-id' }])
+
+    await runScheduleTick('test-request-id')
+
     expect(dbChainMockFns.set).toHaveBeenCalledWith(
       expect.objectContaining({ status: 'failed', completedAt: expect.any(Date) })
     )