fix(tables): decouple stale job cleanup

Author: icecrasher321

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

@@ -1,7 +1,7 @@
 /**
  * @vitest-environment node
  */
-import { asyncJobs, workflowExecutionLogs } from '@sim/db/schema'
+import { asyncJobs, tableJobs, workflowExecutionLogs } from '@sim/db/schema'
 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'
@@ -141,6 +141,45 @@ describe('stale execution cleanup deadline grace', () => {
     )
   })
 
+  it('keeps table-job heartbeat cleanup independent from workflow timeout policy', async () => {
+    vi.useFakeTimers()
+    vi.setSystemTime(new Date('2026-08-03T12:00:00.000Z'))
+
+    try {
+      const response = await GET(createRequest())
+
+      expect(response.status).toBe(200)
+      const expectedThreshold = new Date('2026-08-03T10:25:00.000Z')
+      const tableJobComparisons = dbChainMockFns.where.mock.calls
+        .flatMap(([condition]) => flattenConditions(condition))
+        .filter(
+          (condition) =>
+            condition.type === 'lt' &&
+            condition.left === tableJobs.updatedAt &&
+            condition.right instanceof Date &&
+            condition.right.getTime() === expectedThreshold.getTime()
+        )
+
+      expect(tableJobComparisons).toHaveLength(2)
+      expect(tableJobComparisons.map(({ right }) => right)).toEqual([
+        expectedThreshold,
+        expectedThreshold,
+      ])
+
+      const tableJobUpdateIndex = dbChainMockFns.update.mock.calls.findIndex(
+        ([table]) => table === tableJobs
+      )
+      const update = dbChainMockFns.set.mock.calls[tableJobUpdateIndex]?.[0] as {
+        error: string
+      }
+      expect(update.error).toBe(
+        'Job terminated: no progress for more than 95 minutes (worker timeout or crash)'
+      )
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('caps every bulk mutation and returns only scalar export cleanup fields', async () => {
     const stateBatch = Array.from({ length: 1000 }, (_, index) => ({ id: `state-${index}` }))
     const retentionBatch = Array.from({ length: 2000 }, (_, index) => ({

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

@@ -33,6 +33,11 @@ const STALE_THRESHOLD_MINUTES = Math.ceil(STALE_THRESHOLD_MS / 60000)
 const GENERIC_STALE_PROCESSING_ERROR = `Job terminated: stuck in processing for more than ${STALE_THRESHOLD_MINUTES} minutes`
 const EXECUTION_DEADLINE_ERROR = getTimeoutErrorMessage(undefined)
 const MAX_INT32 = 2_147_483_647
+/**
+ * Table jobs run as detached workers with progress heartbeats, independently of workflow timeout
+ * policy. Preserve their historical 90-minute task window plus five-minute cleanup grace.
+ */
+const TABLE_JOB_STALE_THRESHOLD_MINUTES = 95
 /** Terminal table-jobs older than this are pruned; only the latest job per table is ever read. */
 const TABLE_JOB_RETENTION_HOURS = 24
 /**
@@ -105,6 +110,9 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     const stalePendingThreshold = new Date(
       now.getTime() - JOB_PENDING_RETENTION_HOURS * 60 * 60 * 1000
     )
+    const staleTableJobThreshold = new Date(
+      now.getTime() - TABLE_JOB_STALE_THRESHOLD_MINUTES * 60 * 1000
+    )
 
     const staleExecutions = await db
       .select({
@@ -269,10 +277,9 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     // doesn't grow unbounded (the latest job per table is what list/detail reads surface).
     let staleTableJobsMarkedFailed = 0
     try {
-      const now = new Date()
       const staleTableJobPredicate = and(
         eq(tableJobs.status, 'running'),
-        lt(tableJobs.updatedAt, staleThreshold)
+        lt(tableJobs.updatedAt, staleTableJobThreshold)
       )
       const staleTableJobResult = await runBatchedMutation({
         batchSize: STATE_MUTATION_BATCH_SIZE,
@@ -288,7 +295,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
             .update(tableJobs)
             .set({
               status: 'failed',
-              error: `Job terminated: no progress for more than ${STALE_THRESHOLD_MINUTES} minutes (worker timeout or crash)`,
+              error: `Job terminated: no progress for more than ${TABLE_JOB_STALE_THRESHOLD_MINUTES} minutes (worker timeout or crash)`,
               completedAt: now,
               updatedAt: now,
             })