From ced9ee129653b9c46962d5a952214bd087773590 Mon Sep 17 00:00:00 2001 From: Marcus Chandra Date: Thu, 8 Oct 2026 01:02:22 -0700 Subject: [PATCH] fix(schedule): recheck queued carriers before maximum execution lifetime --- .../app/api/schedules/execute/route.test.ts | 101 ++++++++++++++++++ apps/sim/app/api/schedules/execute/route.ts | 7 +- 2 files changed, 107 insertions(+), 1 deletion(-) diff --git a/apps/sim/app/api/schedules/execute/route.test.ts b/apps/sim/app/api/schedules/execute/route.test.ts index ec145236695..26f8e78004c 100644 --- a/apps/sim/app/api/schedules/execute/route.test.ts +++ b/apps/sim/app/api/schedules/execute/route.test.ts @@ -74,6 +74,7 @@ const { timezone: 'timezone', nextRunAt: 'nextRunAt', lastQueuedAt: 'lastQueuedAt', + updatedAt: 'updatedAt', archivedAt: 'archivedAt', deploymentVersionId: 'deploymentVersionId', sourceType: 'sourceType', @@ -1199,6 +1200,106 @@ describe('Scheduled Workflow Execution API Route', () => { expect(mockEnqueue).toHaveBeenCalledTimes(100) }) + it('rechecks queued carriers within minutes while throttling live carrier lookups', async () => { + dbChainMockFns.limit.mockResolvedValue([]) + await runScheduleTick('test-request-id') + + let claimGate: Record | undefined + for (const [condition] of dbChainMockFns.where.mock.calls) { + conditionContains(condition, (entry) => { + if ( + entry.type === 'or' && + Array.isArray(entry.conditions) && + entry.conditions.some( + (child) => child.type === 'isNull' && child.field === 'lastQueuedAt' + ) + ) + claimGate = entry + return false + }) + } + expect(claimGate).toBeDefined() + + const evaluate = ( + entry: Record, + row: Record + ): boolean => { + if (entry.type === 'and' || entry.type === 'or') { + const conditions = entry.conditions as Record[] + return entry.type === 'and' + ? conditions.every((child) => evaluate(child, row)) + : conditions.some((child) => evaluate(child, row)) + } + const left = row[String(entry.field)] + if (entry.type === 'isNull') return left === null + const right = entry.value instanceof Date ? entry.value : row[String(entry.value)] + if (entry.type === 'lt') return !!left && !!right && left < right + throw new Error('Unexpected claim predicate') + } + const ago = (minutes: number) => new Date(Date.now() - minutes * 60_000) + const row = { nextRunAt: ago(121), lastQueuedAt: ago(120), updatedAt: ago(120) } + expect(evaluate(claimGate!, row)).toBe(true) + expect(evaluate(claimGate!, { ...row, updatedAt: ago(2) })).toBe(false) + expect(evaluate(claimGate!, { ...row, lastQueuedAt: ago(2) })).toBe(false) + expect(evaluate(claimGate!, { ...row, lastQueuedAt: null })).toBe(true) + }) + + it('preserves a running provider job with a long execution timeout during recheck', async () => { + const claimedAt = new Date(Date.now() - 20 * 60_000) + dbChainMockFns.limit + .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS) + .mockResolvedValueOnce([]) + dbChainMockFns.returning + .mockReturnValueOnce([{ ...SINGLE_SCHEDULE[0], lastQueuedAt: new Date() }]) + .mockResolvedValueOnce([{ id: 'schedule-1' }]) + mockGetJob.mockResolvedValueOnce({ + id: 'running-provider-job', + status: 'processing', + startedAt: claimedAt, + payload: { + scheduleId: 'schedule-1', + workflowId: 'workflow-1', + now: claimedAt.toISOString(), + executionTimeoutMs: 7 * 24 * 60 * 60_000, + }, + }) + await runScheduleTick('test-request-id') + expect(mockCancelJob).not.toHaveBeenCalled() + expect(mockEnqueue).not.toHaveBeenCalled() + expect(mockApplyScheduleFailureUpdate).not.toHaveBeenCalled() + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ lastQueuedAt: claimedAt, updatedAt: expect.any(Date) }) + ) + }) + + it('reconciles a crashed provider whose execution never finished redacting', async () => { + const claimedAt = new Date(Date.now() - 30 * 60_000) + dbChainMockFns.limit + .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS) + .mockResolvedValueOnce([{ workflowId: 'workflow-1', status: 'redacting' }]) + .mockResolvedValueOnce([]) + dbChainMockFns.returning + .mockReturnValueOnce([{ ...SINGLE_SCHEDULE[0], lastQueuedAt: new Date() }]) + .mockResolvedValueOnce([{ id: 'schedule-1' }]) + mockGetJob.mockResolvedValueOnce({ + id: 'crashed-provider-job', + status: 'failed', + payload: { + scheduleId: 'schedule-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + now: claimedAt.toISOString(), + executionTimeoutMs: 90 * 60_000, + }, + }) + await runScheduleTick('test-request-id') + expect(mockApplyScheduleFailureUpdate).toHaveBeenCalledWith( + expect.objectContaining({ scheduleId: 'schedule-1', expectedLastQueuedAt: claimedAt }) + ) + expect(mockEnqueue).not.toHaveBeenCalled() + expect(mockExecuteScheduleJob).not.toHaveBeenCalled() + }) + it('reconciles a terminal provider job with the claimed occurrence', async () => { const claimedAt = new Date('2025-01-01T00:00:00.000Z') const schedule = { diff --git a/apps/sim/app/api/schedules/execute/route.ts b/apps/sim/app/api/schedules/execute/route.ts index 8e001717d37..b42d697d033 100644 --- a/apps/sim/app/api/schedules/execute/route.ts +++ b/apps/sim/app/api/schedules/execute/route.ts @@ -85,6 +85,8 @@ const WORKFLOW_CHUNK_SIZE = 100 const STALE_SCHEDULE_RECOVERY_NOTIFY_LIMIT = 25 const MAX_TICK_DURATION_MS = 3 * 60 * 1000 const STALE_SCHEDULE_CLAIM_MS = getExecutionReservationTtlMs() +/** Recheck carrier state independently of the maximum supported execution lifetime. */ +const SCHEDULE_CLAIM_RECHECK_MS = 5 * 60_000 const STALE_SCHEDULE_RECOVERY_BATCH_SIZE = 100 const DATABASE_SCHEDULE_START_TURN_WAIT_MS = 1_000 type DatabaseScheduleStartResult = 'started' | 'capacity_full' | 'not_pending' @@ -98,7 +100,10 @@ const dueFilter = (queuedAt: Date) => or( isNull(workflowSchedule.lastQueuedAt), lt(workflowSchedule.lastQueuedAt, workflowSchedule.nextRunAt), - lt(workflowSchedule.lastQueuedAt, new Date(queuedAt.getTime() - STALE_SCHEDULE_CLAIM_MS)) + and( + lt(workflowSchedule.lastQueuedAt, new Date(queuedAt.getTime() - SCHEDULE_CLAIM_RECHECK_MS)), + lt(workflowSchedule.updatedAt, new Date(queuedAt.getTime() - SCHEDULE_CLAIM_RECHECK_MS)) + ) ) )