fix(execution): drain stale workflow backlog

icecrasher321 · icecrasher321 · commit e3f33d12c23d · 2026-08-04T03:45:17.000-07:00
diff --git a/apps/sim/app/api/cron/cleanup-stale-executions/route.test.ts b/apps/sim/app/api/cron/cleanup-stale-executions/route.test.ts
@@ -49,15 +49,7 @@ describe('stale execution cleanup deadline grace', () => {
   it('waits five minutes past a workflow execution deadline in both cleanup predicates', async () => {
     vi.useFakeTimers()
     vi.setSystemTime(new Date('2026-08-03T12:10:00.000Z'))
-    queueTableRows(workflowExecutionLogs, [
-      {
-        id: 'log-1',
-        executionId: 'execution-1',
-        workflowId: 'workflow-1',
-        startedAt: new Date('2026-08-03T11:00:00.000Z'),
-        executionDeadlineAt: new Date('2026-08-03T12:00:00.000Z'),
-      },
-    ])
+    queueTableRows(workflowExecutionLogs, [{ id: 'log-1' }])
     dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'log-1' }])
 
     try {
@@ -84,17 +76,39 @@ describe('stale execution cleanup deadline grace', () => {
         ([table]) => table === workflowExecutionLogs
       )
       const update = dbChainMockFns.set.mock.calls[executionUpdateIndex]?.[0] as {
+        endedAt: Date
+        totalDurationMs: { toSQL: () => { sql: string; params: unknown[] } }
         executionData: { toSQL: () => { sql: string; params: unknown[] } }
       }
       const errorExpression = update.executionData.toSQL()
+      const staleDurationExpression = errorExpression.params.find(
+        (value): value is { toSQL: () => { sql: string; params: unknown[] } } =>
+          typeof value === 'object' &&
+          value !== null &&
+          'toSQL' in value &&
+          value.toSQL().sql.includes('EXTRACT(EPOCH')
+      )
+      const totalDurationExpression = update.totalDurationMs.toSQL()
+      const cleanupTimestamp = totalDurationExpression.params.find(
+        (value): value is { toSQL: () => { sql: string; params: unknown[] } } =>
+          typeof value === 'object' && value !== null && 'toSQL' in value
+      )
 
       expect(errorExpression.sql).toContain('CASE')
       expect(errorExpression.sql).toContain('IS NOT NULL')
       expect(errorExpression.params).toContain(workflowExecutionLogs.executionDeadlineAt)
       expect(errorExpression.params).toContain('Execution timed out')
       expect(errorExpression.params).toContain(
-        'Execution terminated: worker timeout or crash after 70 minutes'
+        'Execution terminated: worker timeout or crash after '
       )
+      expect(staleDurationExpression?.toSQL().sql).toContain('ROUND')
+      expect(staleDurationExpression?.toSQL().params).toContain(workflowExecutionLogs.startedAt)
+      expect(totalDurationExpression.sql).toContain('LEAST')
+      expect(totalDurationExpression.sql).toContain('ROUND')
+      expect(totalDurationExpression.params).toContain(2_147_483_647)
+      expect(totalDurationExpression.params).toContain(workflowExecutionLogs.startedAt)
+      expect(cleanupTimestamp?.toSQL().params).toEqual([new Date('2026-08-03T12:10:00.000Z')])
+      expect(update.endedAt).toEqual(new Date('2026-08-03T12:10:00.000Z'))
     } finally {
       vi.useRealTimers()
     }
@@ -190,6 +204,13 @@ describe('stale execution cleanup deadline grace', () => {
       resultKey: `workspace/workspace-1/exports/table-1/job-${index}/export.csv`,
     }))
 
+    for (let batch = 0; batch < 10; batch++) {
+      const workflowBatch = Array.from({ length: 100 }, (_, index) => ({
+        id: `workflow-state-${batch}-${index}`,
+      }))
+      queueTableRows(workflowExecutionLogs, workflowBatch)
+      dbChainMockFns.returning.mockResolvedValueOnce(workflowBatch)
+    }
     for (let batch = 0; batch < 10; batch++) {
       dbChainMockFns.returning.mockResolvedValueOnce(stateBatch)
     }
@@ -211,6 +232,11 @@ describe('stale execution cleanup deadline grace', () => {
 
     expect(response.status).toBe(200)
     await expect(response.json()).resolves.toMatchObject({
+      executions: {
+        found: 1000,
+        cleaned: 1000,
+        failed: 0,
+      },
       asyncJobs: {
         staleProcessingMarkedFailed: 10_000,
         stalePendingMarkedFailed: 10_000,
@@ -223,14 +249,98 @@ 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(11)
+    expect(limits.filter((limit) => limit === 100)).toHaveLength(20)
     expect(limits.filter((limit) => limit === 1000)).toHaveLength(30)
     expect(limits.filter((limit) => limit === 2000)).toHaveLength(11)
 
+    const workflowUpdates = dbChainMockFns.update.mock.calls.filter(
+      ([table]) => table === workflowExecutionLogs
+    )
+    expect(workflowUpdates).toHaveLength(10)
+
     const returningShapes = dbChainMockFns.returning.mock.calls
       .map(([shape]) => shape)
       .filter((shape): shape is Record<string, unknown> => Boolean(shape))
     expect(returningShapes.some((shape) => 'payload' in shape)).toBe(false)
     expect(returningShapes.some((shape) => 'type' in shape && 'resultKey' in shape)).toBe(true)
   })
+
+  it('drains more than the legacy 100-row workflow cap in one bounded run', async () => {
+    const firstBatch = Array.from({ length: 100 }, (_, index) => ({
+      id: `execution-${index}`,
+    }))
+    const secondBatch = [{ id: 'execution-100' }]
+    queueTableRows(workflowExecutionLogs, firstBatch)
+    queueTableRows(workflowExecutionLogs, secondBatch)
+    dbChainMockFns.returning.mockResolvedValueOnce(firstBatch).mockResolvedValueOnce(secondBatch)
+
+    const response = await GET(createRequest())
+
+    expect(response.status).toBe(200)
+    await expect(response.json()).resolves.toMatchObject({
+      executions: {
+        found: 101,
+        cleaned: 101,
+        failed: 0,
+      },
+    })
+    expect(
+      dbChainMockFns.update.mock.calls.filter(([table]) => table === workflowExecutionLogs)
+    ).toHaveLength(2)
+    expect(dbChainMockFns.limit).toHaveBeenCalledWith(100)
+  })
+
+  it('preserves committed workflow cleanup counts when a later batch fails', async () => {
+    const firstBatch = Array.from({ length: 100 }, (_, index) => ({
+      id: `execution-${index}`,
+    }))
+    const failedBatch = Array.from({ length: 37 }, (_, index) => ({
+      id: `failed-execution-${index}`,
+    }))
+    queueTableRows(workflowExecutionLogs, firstBatch)
+    queueTableRows(workflowExecutionLogs, failedBatch)
+    dbChainMockFns.returning
+      .mockResolvedValueOnce(firstBatch)
+      .mockRejectedValueOnce(new Error('database unavailable'))
+
+    const response = await GET(createRequest())
+
+    expect(response.status).toBe(200)
+    await expect(response.json()).resolves.toMatchObject({
+      executions: {
+        found: 137,
+        cleaned: 100,
+        failed: 37,
+      },
+    })
+    expect(
+      dbChainMockFns.update.mock.calls.filter(([table]) => table === workflowExecutionLogs)
+    ).toHaveLength(2)
+    expect(dbChainMockFns.update.mock.calls.some(([table]) => table === asyncJobs)).toBe(true)
+  })
+
+  it('continues draining when an atomic race updates fewer rows than were selected', async () => {
+    const firstCandidates = Array.from({ length: 100 }, (_, index) => ({
+      id: `execution-${index}`,
+    }))
+    const firstUpdated = firstCandidates.slice(0, 99)
+    const secondBatch = [{ id: 'execution-100' }]
+    queueTableRows(workflowExecutionLogs, firstCandidates)
+    queueTableRows(workflowExecutionLogs, secondBatch)
+    dbChainMockFns.returning.mockResolvedValueOnce(firstUpdated).mockResolvedValueOnce(secondBatch)
+
+    const response = await GET(createRequest())
+
+    expect(response.status).toBe(200)
+    await expect(response.json()).resolves.toMatchObject({
+      executions: {
+        found: 101,
+        cleaned: 100,
+        failed: 0,
+      },
+    })
+    expect(
+      dbChainMockFns.update.mock.calls.filter(([table]) => table === workflowExecutionLogs)
+    ).toHaveLength(2)
+  })
 })
diff --git a/apps/sim/app/api/cron/cleanup-stale-executions/route.ts b/apps/sim/app/api/cron/cleanup-stale-executions/route.ts
@@ -48,6 +48,8 @@ const TABLE_JOB_RETENTION_HOURS = 24
 const DEPLOYMENT_OPERATION_RETENTION_DAYS = 30
 const DEPLOYMENT_OPERATION_PRUNE_BATCH_SIZE = 2000
 const DEPLOYMENT_OPERATION_PRUNE_MAX_BATCHES = 10
+const WORKFLOW_EXECUTION_MUTATION_BATCH_SIZE = 100
+const WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN = 1000
 const STATE_MUTATION_BATCH_SIZE = 1000
 const STATE_MUTATION_MAX_ROWS_PER_RUN = 10_000
 const RETENTION_DELETE_BATCH_SIZE = 2000
@@ -114,45 +116,51 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       now.getTime() - TABLE_JOB_STALE_THRESHOLD_MINUTES * 60 * 1000
     )
 
-    const staleExecutions = await db
-      .select({
-        id: workflowExecutionLogs.id,
-        executionId: workflowExecutionLogs.executionId,
-        workflowId: workflowExecutionLogs.workflowId,
-        startedAt: workflowExecutionLogs.startedAt,
-        executionDeadlineAt: workflowExecutionLogs.executionDeadlineAt,
-      })
-      .from(workflowExecutionLogs)
-      .where(
-        and(
-          eq(workflowExecutionLogs.status, 'running'),
-          or(
-            lt(workflowExecutionLogs.executionDeadlineAt, staleDeadlineThreshold),
-            and(
-              isNull(workflowExecutionLogs.executionDeadlineAt),
-              lt(workflowExecutionLogs.startedAt, staleThreshold)
-            )
-          )
-        )
-      )
-      .limit(100)
-
-    logger.info(`Found ${staleExecutions.length} stale executions to clean up`)
-
+    let staleExecutionsFound = 0
     let cleaned = 0
     let failed = 0
+    let currentWorkflowBatchSize = 0
 
-    for (const execution of staleExecutions) {
-      try {
-        const staleDurationMs = Date.now() - new Date(execution.startedAt).getTime()
-        const staleDurationMinutes = Math.round(staleDurationMs / 60000)
-        const totalDurationMs = Math.min(staleDurationMs, MAX_INT32)
-
-        const [updatedExecution] = await db
+    try {
+      const staleExecutionPredicate = and(
+        eq(workflowExecutionLogs.status, 'running'),
+        or(
+          lt(workflowExecutionLogs.executionDeadlineAt, staleDeadlineThreshold),
+          and(
+            isNull(workflowExecutionLogs.executionDeadlineAt),
+            lt(workflowExecutionLogs.startedAt, staleThreshold)
+          )
+        )
+      )
+      const cleanupTimestamp = sql.param(now, workflowExecutionLogs.startedAt)
+      const staleDurationMinutes = sql<number>`ROUND(
+        EXTRACT(EPOCH FROM (${cleanupTimestamp} - ${workflowExecutionLogs.startedAt})) / 60
+      )::integer`
+      const totalDurationMs = sql<number>`LEAST(
+        ${MAX_INT32},
+        ROUND(EXTRACT(EPOCH FROM (${cleanupTimestamp} - ${workflowExecutionLogs.startedAt})) * 1000)
+      )::integer`
+      let workflowRowsConsidered = 0
+      while (workflowRowsConsidered < WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN) {
+        const limit = Math.min(
+          WORKFLOW_EXECUTION_MUTATION_BATCH_SIZE,
+          WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN - workflowRowsConsidered
+        )
+        currentWorkflowBatchSize = 0
+        const candidates = await db
+          .select({ id: workflowExecutionLogs.id })
+          .from(workflowExecutionLogs)
+          .where(staleExecutionPredicate)
+          .limit(limit)
+        currentWorkflowBatchSize = candidates.length
+        staleExecutionsFound += candidates.length
+        if (candidates.length === 0) break
+
+        const updatedExecutions = await db
           .update(workflowExecutionLogs)
           .set({
             status: 'failed',
-            endedAt: new Date(),
+            endedAt: now,
             executionDeadlineAt: null,
             totalDurationMs,
             executionData: sql`jsonb_set(
@@ -162,45 +170,39 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
                 CASE
                   WHEN ${workflowExecutionLogs.executionDeadlineAt} IS NOT NULL
                     THEN ${EXECUTION_DEADLINE_ERROR}::text
-                  ELSE ${`Execution terminated: worker timeout or crash after ${staleDurationMinutes} minutes`}::text
+                  ELSE ${'Execution terminated: worker timeout or crash after '}::text
+                    || ${staleDurationMinutes}::text
+                    || ' minutes'
                 END
               )
             )`,
           })
           .where(
             and(
-              eq(workflowExecutionLogs.id, execution.id),
-              eq(workflowExecutionLogs.status, 'running'),
-              or(
-                lt(workflowExecutionLogs.executionDeadlineAt, staleDeadlineThreshold),
-                and(
-                  isNull(workflowExecutionLogs.executionDeadlineAt),
-                  lt(workflowExecutionLogs.startedAt, staleThreshold)
-                )
+              staleExecutionPredicate,
+              inArray(
+                workflowExecutionLogs.id,
+                candidates.map(({ id }) => id)
               )
             )
           )
           .returning({ id: workflowExecutionLogs.id })
 
-        if (!updatedExecution) {
-          logger.debug('Skipped stale execution whose state changed during cleanup', {
-            executionId: execution.executionId,
-          })
-          continue
-        }
-
-        logger.info(`Cleaned up stale execution ${execution.executionId}`, {
-          workflowId: execution.workflowId,
-          staleDurationMinutes,
-        })
+        cleaned += updatedExecutions.length
+        workflowRowsConsidered += candidates.length
+        if (candidates.length < limit) break
+      }
 
-        cleaned++
-      } catch (error) {
-        logger.error(`Failed to clean up execution ${execution.executionId}:`, {
-          error: toError(error).message,
+      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,
         })
-        failed++
       }
+    } catch (error) {
+      logger.error('Failed to clean up stale workflow executions:', {
+        error: toError(error).message,
+      })
+      failed += currentWorkflowBatchSize
     }
 
     logger.info(`Stale execution cleanup completed. Cleaned: ${cleaned}, Failed: ${failed}`)
@@ -510,7 +512,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     return NextResponse.json({
       success: true,
       executions: {
-        found: staleExecutions.length,
+        found: staleExecutionsFound,
         cleaned,
         failed,
         thresholdMinutes: STALE_THRESHOLD_MINUTES,
