fix(execution): align timeout cleanup semantics

Author: icecrasher321

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

@@ -4,6 +4,7 @@
 import { asyncJobs, 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'
 
 const { mockDeleteFile, mockVerifyCronAuth } = vi.hoisted(() => ({
   mockDeleteFile: vi.fn().mockResolvedValue(undefined),
@@ -20,6 +21,7 @@ interface MockCondition {
   conditions?: unknown[]
   left?: unknown
   right?: unknown
+  toSQL?: () => { sql: string; params: unknown[] }
 }
 
 function flattenConditions(condition: unknown): MockCondition[] {
@@ -77,12 +79,28 @@ describe('stale execution cleanup deadline grace', () => {
         expectedThreshold,
         expectedThreshold,
       ])
+
+      const executionUpdateIndex = dbChainMockFns.update.mock.calls.findIndex(
+        ([table]) => table === workflowExecutionLogs
+      )
+      const update = dbChainMockFns.set.mock.calls[executionUpdateIndex]?.[0] as {
+        executionData: { toSQL: () => { sql: string; params: unknown[] } }
+      }
+      const errorExpression = update.executionData.toSQL()
+
+      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'
+      )
     } finally {
       vi.useRealTimers()
     }
   })
 
-  it('reports a configured job duration cap while preserving the generic stale fallback', async () => {
+  it('reports a worker cleanup deadline while preserving the generic stale fallback', async () => {
     const response = await GET(createRequest())
 
     expect(response.status).toBe(200)
@@ -95,12 +113,29 @@ describe('stale execution cleanup deadline grace', () => {
       error: { toSQL: () => { sql: string; params: unknown[] } }
     }
     const errorExpression = update.error.toSQL()
+    const maxDurationGuard = errorExpression.params.find(
+      (value): value is { toSQL: () => { sql: string; params: unknown[] } } =>
+        typeof value === 'object' && value !== null && 'toSQL' in value
+    )
+    const durationPredicate = dbChainMockFns.where.mock.calls
+      .flatMap(([condition]) => flattenConditions(condition))
+      .find((condition) => condition.toSQL?.().sql.includes("interval '1 second'"))
+      ?.toSQL?.()
 
     expect(errorExpression.sql).toContain("->>'maxDurationSeconds'")
-    expect(errorExpression.sql).toContain(
-      "'Job terminated: exceeded configured maximum duration of '"
-    )
-    expect(errorExpression.sql).toContain("|| ' seconds'")
+    const guardExpression = maxDurationGuard?.toSQL()
+    expect(guardExpression?.sql).toContain("jsonb_typeof(?->'maxDurationSeconds') = 'number'")
+    expect(guardExpression?.sql).toContain('>=')
+    expect(guardExpression?.sql).toContain('trunc(')
+    expect(guardExpression?.sql).toContain('<=')
+    expect(guardExpression?.params).toContain(MAX_JOB_DURATION_SECONDS)
+    expect(guardExpression?.params).toContain(MIN_JOB_DURATION_SECONDS)
+    expect(durationPredicate?.sql).toContain('CASE')
+    expect(durationPredicate?.sql).toContain('ELSE')
+    expect(durationPredicate?.sql).toContain('::double precision')
+    expect(errorExpression.sql).toContain("'Job terminated: stuck in processing for more than '")
+    expect(errorExpression.sql).toContain("|| ' seconds (worker cleanup deadline)'")
+    expect(errorExpression.sql).not.toContain('configured maximum duration')
     expect(errorExpression.params).toContainEqual(
       expect.stringMatching(/^Job terminated: stuck in processing for more than \d+ minutes$/)
     )

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

@@ -11,9 +11,16 @@ import { and, eq, exists, gt, inArray, isNull, lt, or, sql } from 'drizzle-orm'
 import { alias } from 'drizzle-orm/pg-core'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
-import { JOB_PENDING_RETENTION_HOURS, JOB_RETENTION_HOURS, JOB_STATUS } from '@/lib/core/async-jobs'
+import {
+  JOB_PENDING_RETENTION_HOURS,
+  JOB_RETENTION_HOURS,
+  JOB_STATUS,
+  MAX_JOB_DURATION_SECONDS,
+  MIN_JOB_DURATION_SECONDS,
+} from '@/lib/core/async-jobs'
 import {
   getExecutionReservationTtlMs,
+  getTimeoutErrorMessage,
   RESERVATION_TTL_BUFFER_MS,
 } from '@/lib/core/execution-limits'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
@@ -24,6 +31,7 @@ const logger = createLogger('CleanupStaleExecutions')
 const STALE_THRESHOLD_MS = getExecutionReservationTtlMs()
 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
 /** Terminal table-jobs older than this are pruned; only the latest job per table is ever read. */
 const TABLE_JOB_RETENTION_HOURS = 24
@@ -142,7 +150,13 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
             executionData: sql`jsonb_set(
               COALESCE(execution_data, '{}'::jsonb),
               ARRAY['error'],
-              to_jsonb(${`Execution terminated: worker timeout or crash after ${staleDurationMinutes} minutes`}::text)
+              to_jsonb(
+                CASE
+                  WHEN ${workflowExecutionLogs.executionDeadlineAt} IS NOT NULL
+                    THEN ${EXECUTION_DEADLINE_ERROR}::text
+                  ELSE ${`Execution terminated: worker timeout or crash after ${staleDurationMinutes} minutes`}::text
+                END
+              )
             )`,
           })
           .where(
@@ -187,18 +201,22 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     let asyncJobsMarkedFailed = 0
 
     try {
+      const hasPositiveMaxDuration = sql<boolean>`CASE
+        WHEN jsonb_typeof(${asyncJobs.metadata}->'maxDurationSeconds') = 'number'
+          THEN (${asyncJobs.metadata}->>'maxDurationSeconds')::numeric >= ${MIN_JOB_DURATION_SECONDS}
+            AND (${asyncJobs.metadata}->>'maxDurationSeconds')::numeric <= ${MAX_JOB_DURATION_SECONDS}
+            AND trunc((${asyncJobs.metadata}->>'maxDurationSeconds')::numeric)
+              = (${asyncJobs.metadata}->>'maxDurationSeconds')::numeric
+        ELSE FALSE
+      END`
+      const staleProcessingDurationPredicate = sql<boolean>`CASE
+        WHEN ${hasPositiveMaxDuration}
+          THEN ${asyncJobs.startedAt} + ((${asyncJobs.metadata}->>'maxDurationSeconds')::double precision * interval '1 second') < ${now}
+        ELSE ${asyncJobs.startedAt} < ${staleThreshold}
+      END`
       const staleProcessingPredicate = and(
         eq(asyncJobs.status, JOB_STATUS.PROCESSING),
-        or(
-          and(
-            sql`jsonb_typeof(${asyncJobs.metadata}->'maxDurationSeconds') = 'number'`,
-            sql`${asyncJobs.startedAt} + ((${asyncJobs.metadata}->>'maxDurationSeconds')::double precision * interval '1 second') < ${now}`
-          ),
-          and(
-            sql`jsonb_typeof(${asyncJobs.metadata}->'maxDurationSeconds') IS DISTINCT FROM 'number'`,
-            lt(asyncJobs.startedAt, staleThreshold)
-          )
-        )
+        staleProcessingDurationPredicate
       )
       const staleProcessingResult = await runBatchedMutation({
         batchSize: STATE_MUTATION_BATCH_SIZE,
@@ -216,10 +234,10 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
               status: JOB_STATUS.FAILED,
               completedAt: new Date(),
               error: sql<string>`CASE
-                WHEN jsonb_typeof(${asyncJobs.metadata}->'maxDurationSeconds') = 'number'
-                  THEN 'Job terminated: exceeded configured maximum duration of '
+                WHEN ${hasPositiveMaxDuration}
+                  THEN 'Job terminated: stuck in processing for more than '
                     || (${asyncJobs.metadata}->>'maxDurationSeconds')
-                    || ' seconds'
+                    || ' seconds (worker cleanup deadline)'
                 ELSE ${GENERIC_STALE_PROCESSING_ERROR}
               END`,
               updatedAt: new Date(),

apps/sim/executor/handlers/workflow/workflow-handler.test.ts

@@ -6,6 +6,7 @@ import {
   resetEnvironmentUtilsMock,
 } from '@sim/testing'
 import { afterAll, beforeAll, beforeEach, describe, expect, it, type Mock, vi } from 'vitest'
+import { createTimeoutAbortController, getExecutionDeadlineAt } from '@/lib/core/execution-limits'
 import { getBlock } from '@/blocks/registry'
 import { BlockType } from '@/executor/constants'
 import { BoundarySafeError } from '@/executor/errors/boundary'
@@ -39,6 +40,7 @@ const {
   mockSafeCompleteWithError,
   mockSafeCompleteWithCancellation,
   mockSetResolvedSecretTraceRegistry,
+  mockSetExecutionDeadlineAt,
   mockSetTraceLargeValueAccess,
   mockDispose,
   executorOptions,
@@ -57,6 +59,7 @@ const {
   mockSafeCompleteWithError: vi.fn(),
   mockSafeCompleteWithCancellation: vi.fn(),
   mockSetResolvedSecretTraceRegistry: vi.fn(),
+  mockSetExecutionDeadlineAt: vi.fn(),
   mockSetTraceLargeValueAccess: vi.fn(),
   mockDispose: vi.fn(),
   executorOptions: [] as Array<Record<string, any>>,
@@ -72,6 +75,7 @@ vi.mock('@/lib/logs/execution/logging-session', () => ({
     safeComplete = mockSafeComplete
     safeCompleteWithError = mockSafeCompleteWithError
     safeCompleteWithCancellation = mockSafeCompleteWithCancellation
+    setExecutionDeadlineAt = mockSetExecutionDeadlineAt
     setResolvedSecretTraceRegistry = mockSetResolvedSecretTraceRegistry
     setTraceLargeValueAccess = mockSetTraceLargeValueAccess
     onBlockStart = vi.fn()
@@ -1162,6 +1166,27 @@ describe('WorkflowBlockHandler', () => {
       })
     })
 
+    it('persists the parent deadline before starting the child session', async () => {
+      const timeoutController = createTimeoutAbortController(60_000)
+
+      try {
+        await handler.execute(
+          customBlockContext({ abortSignal: timeoutController.signal }),
+          customBlock(),
+          {}
+        )
+
+        expect(mockSetExecutionDeadlineAt).toHaveBeenCalledWith(
+          getExecutionDeadlineAt(timeoutController.signal)
+        )
+        expect(mockSetExecutionDeadlineAt.mock.invocationCallOrder[0]).toBeLessThan(
+          mockSafeStart.mock.invocationCallOrder[0]
+        )
+      } finally {
+        timeoutController.cleanup()
+      }
+    })
+
     it('admits against the source payer before executing', async () => {
       await handler.execute(customBlockContext(), customBlock(), {})
 

apps/sim/executor/handlers/workflow/workflow-handler.ts

@@ -3,6 +3,7 @@ import { findCause, getErrorMessage, toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { isRecordLike } from '@sim/utils/object'
 import { resolveBillingAttribution } from '@/lib/billing/core/billing-attribution'
+import { getExecutionDeadlineAt } from '@/lib/core/execution-limits'
 import { getPersonalAndWorkspaceEnv } from '@/lib/environment/utils'
 import { buildNextCallChain, validateCallChain } from '@/lib/execution/call-chain'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
@@ -480,6 +481,7 @@ export class WorkflowBlockHandler implements BlockHandler {
           // child is part of that same logical run and must not add a second.
           { baseExecutionCharge: 0 }
         )
+        childSession.setExecutionDeadlineAt(getExecutionDeadlineAt(ctx.abortSignal))
         childSession.setResolvedSecretTraceRegistry(childResolvedSecretTraceRegistry)
         const correlation = buildCustomBlockCorrelation({
           invokerExecutionId: ctx.executionId,

apps/sim/lib/core/async-jobs/backends/database.test.ts

@@ -25,7 +25,11 @@ vi.mock('@sim/db', () => ({
 }))
 
 import { DatabaseJobQueue } from '@/lib/core/async-jobs/backends/database'
-import { AsyncJobEnqueueError } from '@/lib/core/async-jobs/types'
+import {
+  AsyncJobEnqueueError,
+  MAX_JOB_DURATION_SECONDS,
+  MIN_JOB_DURATION_SECONDS,
+} from '@/lib/core/async-jobs/types'
 
 const EXISTING_JOB = {
   id: 'workflow:1',
@@ -112,6 +116,52 @@ describe('DatabaseJobQueue enqueue', () => {
       })
     )
   })
+
+  it.each([0, -1, 1.5, 4, MAX_JOB_DURATION_SECONDS + 1, Number.POSITIVE_INFINITY])(
+    'rejects invalid maxDurationSeconds %s before inserting',
+    async (maxDurationSeconds) => {
+      const queue = new DatabaseJobQueue()
+
+      const error = await queue
+        .enqueue('workflow-execution', {}, { maxDurationSeconds })
+        .catch((cause: unknown) => cause)
+
+      expect(error).toMatchObject({ acceptance: 'rejected', retryable: false })
+      expect(dbChainMockFns.insert).not.toHaveBeenCalled()
+    }
+  )
+
+  it('accepts the five-second minimum duration', async () => {
+    const queue = new DatabaseJobQueue()
+
+    await queue.enqueue(
+      'workflow-execution',
+      {},
+      {
+        maxDurationSeconds: MIN_JOB_DURATION_SECONDS,
+      }
+    )
+
+    expect(dbChainMockFns.values).toHaveBeenCalledWith(
+      expect.objectContaining({
+        metadata: { maxDurationSeconds: MIN_JOB_DURATION_SECONDS },
+      })
+    )
+  })
+
+  it('does not allow caller metadata to impersonate the reserved duration field', async () => {
+    const queue = new DatabaseJobQueue()
+
+    await queue.enqueue(
+      'workflow-execution',
+      {},
+      {
+        metadata: { maxDurationSeconds: -1 },
+      }
+    )
+
+    expect(dbChainMockFns.values).toHaveBeenCalledWith(expect.objectContaining({ metadata: {} }))
+  })
 })
 
 describe('DatabaseJobQueue batchEnqueueAndWait', () => {
@@ -146,6 +196,20 @@ describe('DatabaseJobQueue batchEnqueueAndWait', () => {
     expect(maxInFlight).toBe(2)
   })
 
+  it('rejects every invalid duration before starting any runner', async () => {
+    const queue = new DatabaseJobQueue()
+    const runner = vi.fn()
+
+    await expect(
+      queue.batchEnqueueAndWait('workflow-execution', [
+        { payload: {}, options: { maxDurationSeconds: 60, runner } },
+        { payload: {}, options: { maxDurationSeconds: 1.5, runner } },
+      ])
+    ).rejects.toMatchObject({ acceptance: 'rejected', retryable: false })
+
+    expect(runner).not.toHaveBeenCalled()
+  })
+
   it('aborts an in-process runner by execution ID', async () => {
     const queue = new DatabaseJobQueue()
     let resolveStarted: (() => void) | undefined
@@ -258,6 +322,26 @@ describe('DatabaseJobQueue batchEnqueueAndWait', () => {
   })
 })
 
+describe('DatabaseJobQueue batchEnqueue', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('validates every row before inserting the batch', async () => {
+    const queue = new DatabaseJobQueue()
+
+    await expect(
+      queue.batchEnqueue('workflow-execution', [
+        { payload: {}, options: { maxDurationSeconds: 60 } },
+        { payload: {}, options: { maxDurationSeconds: 1.5 } },
+      ])
+    ).rejects.toMatchObject({ acceptance: 'rejected', retryable: false })
+
+    expect(dbChainMockFns.insert).not.toHaveBeenCalled()
+  })
+})
+
 describe('DatabaseJobQueue inline claims', () => {
   beforeEach(() => {
     vi.clearAllMocks()