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 index a58e0bfa8ec..a76409e5cc7 100644 --- a/apps/sim/app/api/cron/cleanup-stale-executions/route.test.ts +++ b/apps/sim/app/api/cron/cleanup-stale-executions/route.test.ts @@ -27,6 +27,7 @@ interface MockCondition { conditions?: unknown[] left?: unknown right?: unknown + column?: unknown values?: unknown toSQL?: () => { sql: string; params: unknown[] } } @@ -89,10 +90,11 @@ describe('stale execution cleanup deadline grace', () => { condition.right.getTime() === expectedThreshold.getTime() ) - expect(deadlineComparisons).toHaveLength(2) + expect(deadlineComparisons).toHaveLength(3) expect(deadlineComparisons.map(({ right }) => right)).toEqual([ expectedThreshold, expectedThreshold, + expectedThreshold, ]) const executionUpdateIndex = dbChainMockFns.update.mock.calls.findIndex( @@ -131,6 +133,21 @@ describe('stale execution cleanup deadline grace', () => { } }) + it('terminalizes stale running and redacting execution logs', async () => { + const response = await GET(createRequest()) + + expect(response.status).toBe(200) + const statusPredicates = dbChainMockFns.where.mock.calls + .flatMap(([condition]) => flattenConditions(condition)) + .filter( + (condition) => condition.type === 'eq' && condition.left === workflowExecutionLogs.status + ) + + expect(statusPredicates.map(({ right }) => right)).toEqual( + expect.arrayContaining(['running', 'redacting']) + ) + }) + it('reports a worker cleanup deadline while preserving the generic stale fallback', async () => { queueTableRows(asyncJobs, [{ id: 'async-job-1' }]) dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'async-job-1' }]) @@ -175,6 +192,80 @@ describe('stale execution cleanup deadline grace', () => { ) }) + it('leaves pending and processing schedule jobs to schedule recovery', async () => { + const response = await GET(createRequest()) + + expect(response.status).toBe(200) + const activeAsyncPredicates = dbChainMockFns.where.mock.calls + .map(([condition]) => flattenConditions(condition)) + .filter((conditions) => + conditions.some( + (condition) => + condition.type === 'ne' && + condition.left === asyncJobs.type && + condition.right === 'schedule-execution' + ) + ) + + expect( + activeAsyncPredicates.some((conditions) => + conditions.some( + (condition) => + condition.type === 'eq' && + condition.left === asyncJobs.status && + condition.right === 'processing' + ) + ) + ).toBe(true) + expect( + activeAsyncPredicates.some((conditions) => + conditions.some( + (condition) => + condition.type === 'eq' && + condition.left === asyncJobs.status && + condition.right === 'pending' + ) + ) + ).toBe(true) + }) + + it('retains terminal schedule carriers until reconciliation is recorded', async () => { + const response = await GET(createRequest()) + + expect(response.status).toBe(200) + const retentionConditions = dbChainMockFns.where.mock.calls.flatMap(([condition]) => + flattenConditions(condition) + ) + const reconciliationMarker = retentionConditions.find((condition) => + condition.toSQL?.().sql.includes('scheduleReconciled') + ) + + expect(reconciliationMarker?.toSQL?.().params).toContain(asyncJobs.metadata) + expect( + retentionConditions.some( + (condition) => + condition.type === 'ne' && + condition.left === asyncJobs.type && + condition.right === 'schedule-execution' + ) + ).toBe(true) + }) + + it('retains irrecoverable schedule carrier tombstones indefinitely', 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') + ) + + expect(irrecoverableExclusion?.toSQL?.().sql).toContain("<> 'true'") + expect(irrecoverableExclusion?.toSQL?.().params).toContain(asyncJobs.metadata) + }) + it('keeps table-job heartbeat cleanup independent from workflow timeout policy', async () => { vi.useFakeTimers() vi.setSystemTime(new Date('2026-08-03T12:00:00.000Z')) @@ -220,8 +311,8 @@ describe('stale execution cleanup deadline grace', () => { const response = await GET(createRequest()) expect(response.status).toBe(200) - expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(7) - expect(dbChainMockFns.for).toHaveBeenCalledTimes(7) + expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(8) + expect(dbChainMockFns.for).toHaveBeenCalledTimes(8) for (const [strength, options] of dbChainMockFns.for.mock.calls) { expect(strength).toBe('update') expect(options).toEqual({ skipLocked: true }) @@ -291,7 +382,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(20) + expect(limits.filter((limit) => limit === 100)).toHaveLength(21) expect(limits.filter((limit) => limit === 1000)).toHaveLength(30) expect(limits.filter((limit) => limit === 2000)).toHaveLength(11) diff --git a/apps/sim/app/api/cron/cleanup-stale-executions/route.ts b/apps/sim/app/api/cron/cleanup-stale-executions/route.ts index f77e35a0741..7bd86cf9e1b 100644 --- a/apps/sim/app/api/cron/cleanup-stale-executions/route.ts +++ b/apps/sim/app/api/cron/cleanup-stale-executions/route.ts @@ -7,7 +7,7 @@ import { } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' -import { and, eq, exists, gt, inArray, isNull, lt, or, sql } from 'drizzle-orm' +import { and, eq, exists, gt, inArray, isNull, lt, ne, 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' @@ -140,14 +140,11 @@ export const GET = withRouteHandler(async (request: NextRequest) => { let currentWorkflowBatchSize = 0 try { - const staleExecutionPredicate = and( - eq(workflowExecutionLogs.status, 'running'), - or( - lt(workflowExecutionLogs.executionDeadlineAt, staleDeadlineThreshold), - and( - isNull(workflowExecutionLogs.executionDeadlineAt), - lt(workflowExecutionLogs.startedAt, staleThreshold) - ) + const staleExecutionTimePredicate = or( + lt(workflowExecutionLogs.executionDeadlineAt, staleDeadlineThreshold), + and( + isNull(workflowExecutionLogs.executionDeadlineAt), + lt(workflowExecutionLogs.startedAt, staleThreshold) ) ) const cleanupTimestamp = sql.param(now, workflowExecutionLogs.startedAt) @@ -155,70 +152,80 @@ export const GET = withRouteHandler(async (request: NextRequest) => { EXTRACT(EPOCH FROM (${cleanupTimestamp} - ${workflowExecutionLogs.startedAt})) / 60 )::integer` const totalDurationMs = elapsedDurationMsSql(now) - 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 + for (const executionStatus of ['running', 'redacting'] as const) { + const staleExecutionPredicate = and( + eq(workflowExecutionLogs.status, executionStatus), + staleExecutionTimePredicate ) - currentWorkflowBatchSize = 0 - const { candidates, updatedExecutions } = await db.transaction(async (tx) => { - const candidates = await tx - .select({ id: workflowExecutionLogs.id }) - .from(workflowExecutionLogs) - .where(staleExecutionPredicate) - .limit(limit) - .for('update', { skipLocked: true }) - currentWorkflowBatchSize = candidates.length - if (candidates.length === 0) return { candidates, updatedExecutions: [] } - - const updatedExecutions = await tx - .update(workflowExecutionLogs) - .set({ - status: 'failed', - endedAt: now, - executionDeadlineAt: null, - totalDurationMs, - executionData: sql`jsonb_set( - COALESCE(execution_data, '{}'::jsonb), - ARRAY['error'], - to_jsonb( - CASE - WHEN ${workflowExecutionLogs.executionDeadlineAt} IS NOT NULL - THEN ${EXECUTION_DEADLINE_ERROR}::text - ELSE ${'Execution terminated: worker timeout or crash after '}::text - || ${staleDurationMinutes}::text - || ' minutes' - END - ) - )`, - }) - .where( - and( - staleExecutionPredicate, - inArray( - workflowExecutionLogs.id, - candidates.map(({ id }) => id) + 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, updatedExecutions } = await db.transaction(async (tx) => { + const candidates = await tx + .select({ id: workflowExecutionLogs.id }) + .from(workflowExecutionLogs) + .where(staleExecutionPredicate) + .limit(limit) + .for('update', { skipLocked: true }) + currentWorkflowBatchSize = candidates.length + if (candidates.length === 0) return { candidates, updatedExecutions: [] } + + const updatedExecutions = await tx + .update(workflowExecutionLogs) + .set({ + status: 'failed', + endedAt: now, + executionDeadlineAt: null, + totalDurationMs, + executionData: sql`jsonb_set( + COALESCE(execution_data, '{}'::jsonb), + ARRAY['error'], + to_jsonb( + CASE + WHEN ${workflowExecutionLogs.executionDeadlineAt} IS NOT NULL + THEN ${EXECUTION_DEADLINE_ERROR}::text + ELSE ${'Execution terminated: worker timeout or crash after '}::text + || ${staleDurationMinutes}::text + || ' minutes' + END + ) + )`, + }) + .where( + and( + staleExecutionPredicate, + inArray( + workflowExecutionLogs.id, + candidates.map(({ id }) => id) + ) ) ) - ) - .returning({ id: workflowExecutionLogs.id }) - - return { candidates, updatedExecutions } - }) - currentWorkflowBatchSize = 0 - staleExecutionsFound += candidates.length - if (candidates.length === 0) break - - cleaned += updatedExecutions.length - 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', { - maxRowsPerRun: WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN, - }) + .returning({ id: workflowExecutionLogs.id }) + + return { candidates, updatedExecutions } + }) + currentWorkflowBatchSize = 0 + staleExecutionsFound += candidates.length + if (candidates.length === 0) break + + cleaned += updatedExecutions.length + 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, + } + ) + } } } catch (error) { logger.error('Failed to clean up stale workflow executions:', { @@ -253,6 +260,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => { END` const staleProcessingPredicate = and( eq(asyncJobs.status, JOB_STATUS.PROCESSING), + ne(asyncJobs.type, 'schedule-execution'), staleProcessingDurationPredicate ) const staleProcessingResult = await runBatchedMutation({ @@ -395,6 +403,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => { try { const stalePendingPredicate = and( eq(asyncJobs.status, JOB_STATUS.PENDING), + ne(asyncJobs.type, 'schedule-execution'), lt(asyncJobs.createdAt, stalePendingThreshold) ) const stalePendingResult = await runBatchedMutation({ @@ -441,6 +450,13 @@ export const GET = withRouteHandler(async (request: NextRequest) => { try { const retainedJobPredicate = and( inArray(asyncJobs.status, [JOB_STATUS.COMPLETED, JOB_STATUS.FAILED, JOB_STATUS.CANCELLED]), + or( + ne(asyncJobs.type, 'schedule-execution'), + and( + sql`${asyncJobs.metadata}->>'scheduleReconciled' = 'true'`, + sql`COALESCE(${asyncJobs.metadata}->>'scheduleRecoveryIrrecoverable', 'false') <> 'true'` + ) + ), lt(asyncJobs.completedAt, retentionThreshold) ) const retainedJobResult = await runBatchedMutation({ diff --git a/apps/sim/app/api/schedules/execute/route.test.ts b/apps/sim/app/api/schedules/execute/route.test.ts index 80422e2661f..9ef6e5f8bba 100644 --- a/apps/sim/app/api/schedules/execute/route.test.ts +++ b/apps/sim/app/api/schedules/execute/route.test.ts @@ -7,6 +7,7 @@ import { createMockSql, dbChainMock, dbChainMockFns, + queueTableRows, requestUtilsMockFns, resetDbChainMock, resetEnvFlagsMock, @@ -16,6 +17,7 @@ import { } from '@sim/testing' import { type NextRequest, NextResponse } from 'next/server' import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest' +import { AsyncJobEnqueueError } from '@/lib/core/async-jobs/types' const orderByLimitMock = vi.fn() @@ -32,10 +34,15 @@ const { mockShouldExecuteInline, mockResolveSystemBillingAttribution, mockAssertBillingAttributionSnapshot, + mockApplyScheduleSuccessUpdate, + mockApplyScheduleCancellationUpdate, mockApplyScheduleFailureUpdate, mockNotifyScheduleAutoDisabled, mockRegisterManualExecutionAborter, mockUnregisterManualExecutionAborter, + mockAsyncJobs, + mockWorkflowSchedule, + mockWorkflowExecutionLogs, } = vi.hoisted(() => ({ mockVerifyCronAuth: vi.fn().mockReturnValue(null), mockExecuteScheduleJob: vi.fn().mockResolvedValue(undefined), @@ -49,10 +56,49 @@ const { mockShouldExecuteInline: vi.fn().mockReturnValue(false), mockResolveSystemBillingAttribution: vi.fn(), mockAssertBillingAttributionSnapshot: vi.fn(), + mockApplyScheduleSuccessUpdate: vi.fn().mockResolvedValue(true), + mockApplyScheduleCancellationUpdate: vi.fn().mockResolvedValue(true), mockApplyScheduleFailureUpdate: vi.fn().mockResolvedValue({ updated: true, disabled: false }), mockNotifyScheduleAutoDisabled: vi.fn().mockResolvedValue(undefined), mockRegisterManualExecutionAborter: vi.fn(), mockUnregisterManualExecutionAborter: vi.fn(), + mockAsyncJobs: { + id: 'id', + type: 'type', + payload: 'payload', + status: 'status', + createdAt: 'createdAt', + runAt: 'runAt', + startedAt: 'startedAt', + completedAt: 'completedAt', + attempts: 'attempts', + maxAttempts: 'maxAttempts', + error: 'error', + output: 'output', + metadata: 'metadata', + updatedAt: 'updatedAt', + }, + mockWorkflowSchedule: { + id: 'id', + workflowId: 'workflowId', + blockId: 'blockId', + cronExpression: 'cronExpression', + lastRanAt: 'lastRanAt', + failedCount: 'failedCount', + infraRetryCount: 'infraRetryCount', + status: 'status', + timezone: 'timezone', + nextRunAt: 'nextRunAt', + lastQueuedAt: 'lastQueuedAt', + archivedAt: 'archivedAt', + deploymentVersionId: 'deploymentVersionId', + sourceType: 'sourceType', + }, + mockWorkflowExecutionLogs: { + executionId: 'executionId', + workflowId: 'workflowId', + status: 'status', + }, })) vi.mock('@/lib/auth/internal', () => ({ @@ -67,6 +113,8 @@ vi.mock('@/lib/billing/core/billing-attribution', () => ({ vi.mock('@/background/schedule-execution', () => ({ executeScheduleJob: mockExecuteScheduleJob, releaseScheduleLock: mockReleaseScheduleLock, + applyScheduleSuccessUpdate: mockApplyScheduleSuccessUpdate, + applyScheduleCancellationUpdate: mockApplyScheduleCancellationUpdate, applyScheduleFailureUpdate: mockApplyScheduleFailureUpdate, })) @@ -95,6 +143,7 @@ vi.mock('@/lib/core/async-jobs', () => ({ vi.mock('drizzle-orm', () => ({ and: vi.fn((...conditions: unknown[]) => ({ type: 'and', conditions })), eq: vi.fn((field: unknown, value: unknown) => ({ field, value, type: 'eq' })), + gt: vi.fn((field: unknown, value: unknown) => ({ field, value, type: 'gt' })), ne: vi.fn((field: unknown, value: unknown) => ({ field, value, type: 'ne' })), lte: vi.fn((field: unknown, value: unknown) => ({ field, value, type: 'lte' })), lt: vi.fn((field: unknown, value: unknown) => ({ field, value, type: 'lt' })), @@ -108,21 +157,7 @@ vi.mock('drizzle-orm', () => ({ vi.mock('@sim/db', () => ({ ...dbChainMock, - workflowSchedule: { - id: 'id', - workflowId: 'workflowId', - blockId: 'blockId', - cronExpression: 'cronExpression', - lastRanAt: 'lastRanAt', - failedCount: 'failedCount', - infraRetryCount: 'infraRetryCount', - status: 'status', - timezone: 'timezone', - nextRunAt: 'nextRunAt', - lastQueuedAt: 'lastQueuedAt', - deploymentVersionId: 'deploymentVersionId', - sourceType: 'sourceType', - }, + workflowSchedule: mockWorkflowSchedule, workflowDeploymentVersion: { id: 'id', workflowId: 'workflowId', @@ -133,20 +168,8 @@ vi.mock('@sim/db', () => ({ userId: 'userId', workspaceId: 'workspaceId', }, - asyncJobs: { - id: 'id', - type: 'type', - payload: 'payload', - status: 'status', - createdAt: 'createdAt', - runAt: 'runAt', - startedAt: 'startedAt', - completedAt: 'completedAt', - attempts: 'attempts', - maxAttempts: 'maxAttempts', - error: 'error', - updatedAt: 'updatedAt', - }, + asyncJobs: mockAsyncJobs, + workflowExecutionLogs: mockWorkflowExecutionLogs, })) vi.mock('@sim/utils/id', () => ({ @@ -311,6 +334,12 @@ describe('Scheduled Workflow Execution API Route', () => { mockExecuteScheduleJob.mockResolvedValue(undefined) mockReleaseScheduleLock.mockReset() mockReleaseScheduleLock.mockResolvedValue(undefined) + mockApplyScheduleSuccessUpdate.mockReset() + mockApplyScheduleSuccessUpdate.mockResolvedValue(true) + mockApplyScheduleCancellationUpdate.mockReset() + mockApplyScheduleCancellationUpdate.mockResolvedValue(true) + mockApplyScheduleFailureUpdate.mockReset() + mockApplyScheduleFailureUpdate.mockResolvedValue({ updated: true, disabled: false }) mockAssertBillingAttributionSnapshot.mockReset() mockAssertBillingAttributionSnapshot.mockImplementation((value: unknown) => { if (!value || typeof value !== 'object') { @@ -355,6 +384,18 @@ describe('Scheduled Workflow Execution API Route', () => { expect(result.processedCount).toBe(0) }) + it('rotates deferred recovery carriers behind untouched work', async () => { + mockShouldExecuteInline.mockReturnValue(true) + dbChainMockFns.returning.mockReturnValueOnce([]).mockReturnValueOnce([]) + + await runScheduleTick('test-request-id') + + expect(dbChainMockFns.orderBy).toHaveBeenCalledWith( + { type: 'asc', field: mockAsyncJobs.updatedAt }, + { type: 'asc', field: mockAsyncJobs.id } + ) + }) + it('should execute multiple schedules in parallel', async () => { dbChainMockFns.limit .mockResolvedValueOnce([ @@ -441,6 +482,36 @@ describe('Scheduled Workflow Execution API Route', () => { ) expect(mockUnregisterManualExecutionAborter).toHaveBeenCalledWith('schedule-execution-1') expect(mockCompleteJob).toHaveBeenCalledWith('job-id-1', null) + + const authoritativeStartCondition = dbChainMockFns.where.mock.calls + .map(([condition]) => condition) + .find( + (condition) => + conditionContains( + condition, + (entry) => entry.type === 'eq' && entry.field === mockAsyncJobs.id + ) && + conditionContains( + condition, + (entry) => + entry.type === 'eq' && + entry.field === mockAsyncJobs.type && + entry.value === 'schedule-execution' + ) && + conditionContains( + condition, + (entry) => + entry.type === 'eq' && + entry.field === mockAsyncJobs.status && + entry.value === 'pending' + ) && + conditionContains( + condition, + (entry) => + entry.type === 'eq' && entry.field === mockAsyncJobs.attempts && entry.value === 0 + ) + ) + expect(authoritativeStartCondition).toBeDefined() }) it('forwards database fallback cancellation into the schedule execution signal', async () => { @@ -482,66 +553,305 @@ describe('Scheduled Workflow Execution API Route', () => { expect(mockReleaseScheduleLock).not.toHaveBeenCalled() }) - it('recovers database fallback jobs after their admitted per-job deadline', async () => { + it.each([ + { persistedStatus: 'completed', accounting: 'success', carrierStatus: 'completed' }, + { persistedStatus: 'failed', accounting: 'failure', carrierStatus: 'completed' }, + { persistedStatus: 'cancelled', accounting: 'cancelled', carrierStatus: 'completed' }, + { persistedStatus: 'pending', accounting: 'success', carrierStatus: 'completed' }, + { persistedStatus: 'paused', accounting: 'success', carrierStatus: 'completed' }, + ])( + 'reconciles a stale database job from a $persistedStatus execution log', + async ({ persistedStatus, accounting, carrierStatus }) => { + mockShouldExecuteInline.mockReturnValue(true) + const claimedAt = new Date('2025-01-01T00:00:00.000Z') + const payload = { + scheduleId: 'schedule-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + now: claimedAt.toISOString(), + } + mockProcessingCounts(0, 0) + orderByLimitMock.mockResolvedValueOnce([ + { id: 'claimed-job-id', payload, status: 'processing' }, + ]) + queueTableRows(mockWorkflowExecutionLogs, [ + { executionId: 'execution-1', workflowId: 'workflow-1', status: persistedStatus }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'claimed-job-id' }]) + + await runScheduleTick('test-request-id') + + expect(mockExecuteScheduleJob).not.toHaveBeenCalled() + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ + status: carrierStatus, + completedAt: expect.any(Date), + output: expect.objectContaining({ executionStatus: persistedStatus }), + }) + ) + if (accounting === 'failure') { + expect(mockApplyScheduleFailureUpdate).toHaveBeenCalledOnce() + } else if (accounting === 'cancelled') { + expect(mockApplyScheduleCancellationUpdate).toHaveBeenCalledOnce() + } else { + expect(mockApplyScheduleSuccessUpdate).toHaveBeenCalledOnce() + } + } + ) + + it('preserves an already-cancelled database carrier while reconciling it', async () => { mockShouldExecuteInline.mockReturnValue(true) - const staleStartedAt = new Date(Date.now() - 6 * 60 * 1000) + const claimedAt = new Date('2025-01-01T00:00:00.000Z') mockProcessingCounts(0, 0) - mockGetJob - .mockResolvedValueOnce({ - id: 'job-id-1', - status: 'processing', - startedAt: staleStartedAt, - metadata: { maxDurationSeconds: 300 }, - }) - .mockResolvedValueOnce({ - id: 'job-id-1', - status: 'pending', + orderByLimitMock.mockResolvedValueOnce([ + { + id: 'cancelled-job-id', + status: 'cancelled', payload: { scheduleId: 'schedule-1', workflowId: 'workflow-1', - workspaceId: 'workspace-1', - billingAttribution: createBillingAttribution('workspace-1'), - now: '2025-01-01T00:00:00.000Z', + executionId: 'execution-1', + now: claimedAt.toISOString(), + scheduledFor: claimedAt.toISOString(), }, - }) - orderByLimitMock - .mockResolvedValueOnce([]) - .mockResolvedValueOnce([]) - .mockResolvedValueOnce([ - { - id: 'job-id-1', - payload: { - scheduleId: 'schedule-1', - workflowId: 'workflow-1', - now: '2025-01-01T00:00:00.000Z', - }, - attempts: 0, - maxAttempts: 3, + }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'cancelled-job-id' }]) + + await runScheduleTick('test-request-id') + + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ status: 'cancelled', error: 'Cancelled' }) + ) + expect(mockApplyScheduleCancellationUpdate).toHaveBeenCalledOnce() + expect(mockExecuteScheduleJob).not.toHaveBeenCalled() + }) + + it('preserves a cancelled carrier when its workflow log completed', async () => { + mockShouldExecuteInline.mockReturnValue(true) + const claimedAt = new Date('2025-01-01T00:00:00.000Z') + mockProcessingCounts(0, 0) + orderByLimitMock.mockResolvedValueOnce([ + { + id: 'cancelled-job-id', + status: 'cancelled', + payload: { + scheduleId: 'schedule-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + now: claimedAt.toISOString(), + scheduledFor: claimedAt.toISOString(), }, - ]) - dbChainMockFns.limit - .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS) - .mockResolvedValueOnce([]) + }, + ]) + queueTableRows(mockWorkflowExecutionLogs, [ + { executionId: 'execution-1', workflowId: 'workflow-1', status: 'completed' }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'cancelled-job-id' }]) + + await runScheduleTick('test-request-id') + + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ status: 'cancelled', error: 'Cancelled' }) + ) + expect(mockApplyScheduleSuccessUpdate).toHaveBeenCalledOnce() + expect(mockApplyScheduleCancellationUpdate).not.toHaveBeenCalled() + expect(mockExecuteScheduleJob).not.toHaveBeenCalled() + }) + + it('marks a terminal carrier reconciled when its occurrence already advanced', async () => { + mockShouldExecuteInline.mockReturnValue(true) + const claimedAt = new Date('2025-01-01T00:00:00.000Z') + const nextOccurrence = new Date('2025-01-02T00:00:00.000Z') + mockProcessingCounts(0, 0) + orderByLimitMock.mockResolvedValueOnce([ + { + id: 'completed-job-id', + status: 'completed', + payload: { + scheduleId: 'schedule-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + now: claimedAt.toISOString(), + scheduledFor: claimedAt.toISOString(), + }, + }, + ]) + queueTableRows(mockWorkflowExecutionLogs, [ + { executionId: 'execution-1', workflowId: 'workflow-1', status: 'completed' }, + ]) + queueTableRows(mockWorkflowSchedule, [ + { + archivedAt: null, + lastQueuedAt: null, + nextRunAt: nextOccurrence, + status: 'active', + }, + ]) + mockApplyScheduleSuccessUpdate.mockResolvedValueOnce(false) + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'completed-job-id' }]) + + await runScheduleTick('test-request-id') + + expect(mockApplyScheduleSuccessUpdate).toHaveBeenCalledOnce() + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ metadata: expect.anything(), updatedAt: expect.any(Date) }) + ) + expect(mockExecuteScheduleJob).not.toHaveBeenCalled() + }) + + it('restores a released claim before retrying terminal carrier accounting', async () => { + mockShouldExecuteInline.mockReturnValue(true) + const claimedAt = new Date('2025-01-01T00:00:00.000Z') + mockProcessingCounts(0, 0) + orderByLimitMock.mockResolvedValueOnce([ + { + id: 'completed-job-id', + status: 'completed', + payload: { + scheduleId: 'schedule-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + now: claimedAt.toISOString(), + scheduledFor: claimedAt.toISOString(), + }, + }, + ]) + queueTableRows(mockWorkflowExecutionLogs, [ + { executionId: 'execution-1', workflowId: 'workflow-1', status: 'completed' }, + ]) + queueTableRows(mockWorkflowSchedule, [ + { + archivedAt: null, + lastQueuedAt: null, + nextRunAt: claimedAt, + status: 'active', + }, + ]) + mockApplyScheduleSuccessUpdate.mockResolvedValueOnce(false).mockResolvedValueOnce(true) dbChainMockFns.returning - .mockReturnValueOnce([{ ...SINGLE_SCHEDULE[0], lastQueuedAt: new Date('2025-01-01') }]) - .mockResolvedValueOnce([{ id: 'job-id-1' }]) + .mockResolvedValueOnce([{ id: 'completed-job-id' }]) + .mockResolvedValueOnce([{ id: 'schedule-1' }]) await runScheduleTick('test-request-id') - expect(mockExecuteScheduleJob).toHaveBeenCalledWith( - expect.objectContaining({ scheduleId: 'schedule-1' }), - expect.any(AbortSignal) + + expect(mockApplyScheduleSuccessUpdate).toHaveBeenCalledTimes(2) + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ lastQueuedAt: claimedAt }) ) - expect(mockCompleteJob).toHaveBeenCalledWith( - expect.stringMatching(/^schedule_[0-9a-f]{32}$/), - null + expect(mockExecuteScheduleJob).not.toHaveBeenCalled() + }) + + it('does not overwrite a newer non-null claim while reconciling an old carrier', async () => { + mockShouldExecuteInline.mockReturnValue(true) + const claimedAt = new Date('2025-01-01T00:00:00.000Z') + const newerClaim = new Date('2025-01-01T00:05:00.000Z') + mockProcessingCounts(0, 0) + orderByLimitMock.mockResolvedValueOnce([ + { + id: 'completed-job-id', + status: 'completed', + payload: { + scheduleId: 'schedule-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + now: claimedAt.toISOString(), + scheduledFor: claimedAt.toISOString(), + }, + }, + ]) + queueTableRows(mockWorkflowExecutionLogs, [ + { executionId: 'execution-1', workflowId: 'workflow-1', status: 'completed' }, + ]) + queueTableRows(mockWorkflowSchedule, [ + { + archivedAt: null, + lastQueuedAt: newerClaim, + nextRunAt: claimedAt, + status: 'active', + }, + ]) + mockApplyScheduleSuccessUpdate.mockResolvedValueOnce(false) + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'completed-job-id' }]) + + await runScheduleTick('test-request-id') + + expect(mockApplyScheduleSuccessUpdate).toHaveBeenCalledOnce() + expect(dbChainMockFns.set).not.toHaveBeenCalledWith( + expect.objectContaining({ lastQueuedAt: claimedAt }) ) + expect(dbChainMockFns.set).not.toHaveBeenCalledWith( + expect.objectContaining({ metadata: expect.anything() }) + ) + expect(mockExecuteScheduleJob).not.toHaveBeenCalled() + }) + + it('marks malformed terminal carriers irrecoverable so they leave the recovery batch', async () => { + mockShouldExecuteInline.mockReturnValue(true) + mockProcessingCounts(0, 0) + orderByLimitMock.mockResolvedValueOnce([ + { + id: 'malformed-job-id', + status: 'completed', + 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) }) + ) + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ metadata: expect.anything(), updatedAt: expect.any(Date) }) + ) + expect(mockApplyScheduleFailureUpdate).not.toHaveBeenCalled() + expect(mockExecuteScheduleJob).not.toHaveBeenCalled() + }) + + it.each([ + { name: 'a missing execution log', executionId: 'execution-1', persistedStatus: null }, + { name: 'a missing execution ID', executionId: undefined, persistedStatus: null }, + { name: 'a stale running log', executionId: 'execution-1', persistedStatus: 'running' }, + { name: 'a stale redacting log', executionId: 'execution-1', persistedStatus: 'redacting' }, + ])('fails a claimed database job with $name without rerunning it', async (testCase) => { + mockShouldExecuteInline.mockReturnValue(true) + const claimedAt = new Date('2025-01-01T00:00:00.000Z') + mockProcessingCounts(0, 0) + orderByLimitMock.mockResolvedValueOnce([ + { + id: 'claimed-job-id', + status: 'processing', + payload: { + scheduleId: 'schedule-1', + workflowId: 'workflow-1', + executionId: testCase.executionId, + now: claimedAt.toISOString(), + }, + }, + ]) + if (testCase.persistedStatus) { + queueTableRows(mockWorkflowExecutionLogs, [ + { + executionId: 'execution-1', + workflowId: 'workflow-1', + status: testCase.persistedStatus, + }, + ]) + } + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'claimed-job-id' }]) + + await runScheduleTick('test-request-id') + + expect(mockExecuteScheduleJob).not.toHaveBeenCalled() expect(dbChainMockFns.set).toHaveBeenCalledWith( expect.objectContaining({ - status: 'pending', - startedAt: null, - error: expect.stringContaining('stale schedule execution processing lease'), + status: 'failed', + error: expect.stringContaining('Indeterminate schedule execution outcome'), }) ) + expect(mockApplyScheduleFailureUpdate).toHaveBeenCalledOnce() }) it('resumes pending database fallback jobs without waiting for a stale schedule claim', async () => { @@ -579,6 +889,14 @@ describe('Scheduled Workflow Execution API Route', () => { expect.any(AbortSignal) ) expect(mockCompleteJob).toHaveBeenCalledWith('pending-job-id', null) + expect( + dbChainMockFns.where.mock.calls.some(([condition]) => + conditionContains( + condition, + (entry) => entry.type === 'eq' && entry.field === 'attempts' && entry.value === 0 + ) + ) + ).toBe(true) }) it.each([ @@ -688,7 +1006,7 @@ describe('Scheduled Workflow Execution API Route', () => { ) }) - it('completes stale pending database fallback jobs whose schedule claim was already released', async () => { + it('cancels stale pending database fallback jobs whose schedule claim was already released', async () => { mockShouldExecuteInline.mockReturnValue(true) const claimedAt = new Date('2025-01-01T00:00:00.000Z') mockProcessingCounts(0, 0) @@ -706,57 +1024,54 @@ describe('Scheduled Workflow Execution API Route', () => { .mockResolvedValueOnce([{ lastQueuedAt: null }]) .mockResolvedValueOnce([]) .mockResolvedValueOnce([]) - dbChainMockFns.returning.mockReturnValueOnce([]).mockReturnValueOnce([]) + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'stale-pending-job-id' }]) await runScheduleTick('test-request-id') expect(mockExecuteScheduleJob).not.toHaveBeenCalled() - expect(mockCompleteJob).toHaveBeenCalledWith( - 'stale-pending-job-id', + expect(dbChainMockFns.set).toHaveBeenCalledWith( expect.objectContaining({ - skipped: true, + status: 'cancelled', + error: expect.stringContaining('claim was released'), + metadata: expect.anything(), }) ) + expect(mockCancelJob).not.toHaveBeenCalled() + expect(mockCompleteJob).not.toHaveBeenCalled() }) - it('fails exhausted stale database fallback jobs instead of retrying forever', async () => { + it('reconciles pending database jobs with attempts instead of executing them', async () => { mockShouldExecuteInline.mockReturnValue(true) const claimedAt = new Date('2025-01-01T00:00:00.000Z') mockProcessingCounts(0, 0) orderByLimitMock.mockResolvedValueOnce([ { - id: 'exhausted-job-id', + id: 'claimed-pending-job-id', + status: 'pending', payload: { scheduleId: 'schedule-1', workflowId: 'workflow-1', + executionId: 'execution-1', now: claimedAt.toISOString(), }, - attempts: 3, - maxAttempts: 3, }, ]) - dbChainMockFns.limit - .mockResolvedValueOnce([]) - .mockResolvedValueOnce([]) - .mockResolvedValueOnce([]) + queueTableRows(mockWorkflowExecutionLogs, [ + { executionId: 'execution-1', workflowId: 'workflow-1', status: 'completed' }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'claimed-pending-job-id' }]) await runScheduleTick('test-request-id') + expect(mockExecuteScheduleJob).not.toHaveBeenCalled() expect(dbChainMockFns.set).toHaveBeenCalledWith( expect.objectContaining({ - status: 'failed', - error: expect.stringContaining('exhausted retry attempts'), - }) - ) - expect(mockApplyScheduleFailureUpdate).toHaveBeenCalledWith( - expect.objectContaining({ - scheduleId: 'schedule-1', - expectedLastQueuedAt: claimedAt, - executor: expect.anything(), + status: 'completed', }) ) + expect(mockApplyScheduleSuccessUpdate).toHaveBeenCalledOnce() }) - it('defers schedule claims when retryable lookup infrastructure fails before enqueue', async () => { + it('preserves the occurrence when carrier lookup is uncertain', async () => { const claimedAt = new Date('2025-01-01T00:00:00.000Z') const schedule = { ...SINGLE_SCHEDULE[0], @@ -772,14 +1087,37 @@ describe('Scheduled Workflow Execution API Route', () => { await runScheduleTick('test-request-id') expect(mockEnqueue).not.toHaveBeenCalled() + expect(mockResolveSystemBillingAttribution).not.toHaveBeenCalled() expect(mockReleaseScheduleLock).not.toHaveBeenCalled() - expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect(dbChainMockFns.set).not.toHaveBeenCalledWith( expect.objectContaining({ - lastQueuedAt: null, - nextRunAt: expect.any(Date), infraRetryCount: 1, }) ) + expect(mockApplyScheduleFailureUpdate).not.toHaveBeenCalled() + }) + + it('preserves the occurrence when enqueue acceptance is unknown', async () => { + const claimedAt = new Date('2025-01-01T00:00:00.000Z') + const schedule = { ...SINGLE_SCHEDULE[0], lastQueuedAt: claimedAt } + mockEnqueue.mockRejectedValueOnce( + new AsyncJobEnqueueError('response lost after enqueue', { + acceptance: 'unknown', + retryable: true, + }) + ) + dbChainMockFns.limit + .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS) + .mockResolvedValueOnce([]) + dbChainMockFns.returning.mockReturnValueOnce([schedule]).mockReturnValueOnce([]) + + await runScheduleTick('test-request-id') + + expect(mockEnqueue).toHaveBeenCalledOnce() + expect(mockApplyScheduleFailureUpdate).not.toHaveBeenCalled() + expect(dbChainMockFns.set).not.toHaveBeenCalledWith( + expect.objectContaining({ infraRetryCount: 1 }) + ) }) it('marks schedules failed when non-retryable setup errors happen before enqueue', async () => { @@ -788,7 +1126,7 @@ describe('Scheduled Workflow Execution API Route', () => { ...SINGLE_SCHEDULE[0], lastQueuedAt: claimedAt, } - mockGetJob.mockRejectedValueOnce(new Error('bad setup invariant')) + mockResolveSystemBillingAttribution.mockRejectedValueOnce(new Error('bad setup invariant')) dbChainMockFns.limit .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS) .mockResolvedValueOnce([]) @@ -816,7 +1154,7 @@ describe('Scheduled Workflow Execution API Route', () => { ...SINGLE_SCHEDULE[0], lastQueuedAt: claimedAt, } - mockGetJob.mockRejectedValueOnce(new Error('bad setup invariant')) + mockResolveSystemBillingAttribution.mockRejectedValueOnce(new Error('bad setup invariant')) mockApplyScheduleFailureUpdate.mockResolvedValueOnce({ updated: true, disabled: true }) dbChainMockFns.limit .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS) @@ -836,7 +1174,7 @@ describe('Scheduled Workflow Execution API Route', () => { ...SINGLE_SCHEDULE[0], lastQueuedAt: claimedAt, } - mockGetJob.mockRejectedValueOnce(new Error('bad setup invariant')) + mockResolveSystemBillingAttribution.mockRejectedValueOnce(new Error('bad setup invariant')) mockApplyScheduleFailureUpdate.mockResolvedValueOnce({ updated: true, disabled: false }) dbChainMockFns.limit .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS) @@ -888,6 +1226,7 @@ describe('Scheduled Workflow Execution API Route', () => { await runScheduleTick('test-request-id') expect(mockEnqueue).not.toHaveBeenCalled() + expect(mockResolveSystemBillingAttribution).not.toHaveBeenCalled() expect(mockReleaseScheduleLock).not.toHaveBeenCalled() expect(dbChainMockFns.set).toHaveBeenCalledWith( expect.objectContaining({ @@ -1008,9 +1347,9 @@ describe('Scheduled Workflow Execution API Route', () => { }) it('cancels stale Trigger.dev runs instead of restoring an expired claim forever', async () => { - const originalClaim = new Date(Date.now() - 2 * 60 * 60 * 1000) const startedAt = new Date(Date.now() - 11 * 60 * 1000) const staleReclaim = new Date() + const originalClaim = staleReclaim const schedule = { ...SINGLE_SCHEDULE[0], lastQueuedAt: staleReclaim, @@ -1023,25 +1362,26 @@ describe('Scheduled Workflow Execution API Route', () => { payload: { scheduleId: 'schedule-1', workflowId: 'workflow-1', + executionId: 'execution-1', now: originalClaim.toISOString(), executionTimeoutMs: 5 * 60 * 1000, }, }) dbChainMockFns.limit .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS) + .mockResolvedValueOnce([{ workflowId: 'workflow-1', status: 'completed' }]) .mockResolvedValueOnce([]) dbChainMockFns.returning.mockReturnValueOnce([schedule]).mockReturnValueOnce([]) await runScheduleTick('test-request-id') expect(mockCancelJob).toHaveBeenCalledWith('trigger-run-id') - expect(mockReleaseScheduleLock).toHaveBeenCalledWith( - 'schedule-1', - 'test-request-id', - expect.any(Date), - expect.stringContaining('cancelling stale queued schedule execution job'), - undefined, - { expectedLastQueuedAt: staleReclaim } + expect(mockApplyScheduleSuccessUpdate).toHaveBeenCalledWith( + expect.objectContaining({ + scheduleId: 'schedule-1', + expectedLastQueuedAt: originalClaim, + }) ) + expect(mockExecuteScheduleJob).not.toHaveBeenCalled() }) it('bounds workflow schedule claims to the configured enqueue budget', async () => { @@ -1077,7 +1417,7 @@ describe('Scheduled Workflow Execution API Route', () => { expect(mockEnqueue).toHaveBeenCalledTimes(100) }) - it('guards route-side stale release updates with the claimed occurrence', async () => { + it('reconciles a terminal provider job with the claimed occurrence', async () => { const claimedAt = new Date('2025-01-01T00:00:00.000Z') const schedule = { ...SINGLE_SCHEDULE[0], @@ -1085,18 +1425,94 @@ describe('Scheduled Workflow Execution API Route', () => { } dbChainMockFns.limit .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS) + .mockResolvedValueOnce([{ workflowId: 'workflow-1', status: 'completed' }]) .mockResolvedValueOnce([]) dbChainMockFns.returning.mockReturnValueOnce([schedule]).mockReturnValueOnce([]) - mockGetJob.mockResolvedValueOnce({ id: 'job-id-1', status: 'completed' }) + mockGetJob.mockResolvedValueOnce({ + id: 'job-id-1', + status: 'completed', + payload: { + scheduleId: 'schedule-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + now: claimedAt.toISOString(), + }, + }) await runScheduleTick('test-request-id') - expect(mockReleaseScheduleLock).toHaveBeenCalledWith( - 'schedule-1', - 'test-request-id', - expect.any(Date), - expect.stringContaining('finished job'), - null, - { expectedLastQueuedAt: claimedAt } + expect(mockApplyScheduleSuccessUpdate).toHaveBeenCalledWith( + expect.objectContaining({ + scheduleId: 'schedule-1', + expectedLastQueuedAt: claimedAt, + }) + ) + expect(mockExecuteScheduleJob).not.toHaveBeenCalled() + }) + + it('uses a terminal cancelled carrier when no execution log exists', async () => { + const claimedAt = new Date('2025-01-01T00:00:00.000Z') + const schedule = { + ...SINGLE_SCHEDULE[0], + lastQueuedAt: claimedAt, + } + dbChainMockFns.limit + .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS) + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([]) + dbChainMockFns.returning.mockReturnValueOnce([schedule]).mockReturnValueOnce([]) + mockGetJob.mockResolvedValueOnce({ + id: 'job-id-1', + status: 'cancelled', + payload: { + scheduleId: 'schedule-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + now: claimedAt.toISOString(), + }, + }) + + await runScheduleTick('test-request-id') + + expect(mockApplyScheduleCancellationUpdate).toHaveBeenCalledWith( + expect.objectContaining({ + scheduleId: 'schedule-1', + expectedLastQueuedAt: claimedAt, + }) + ) + expect(mockApplyScheduleFailureUpdate).not.toHaveBeenCalled() + expect(mockResolveSystemBillingAttribution).not.toHaveBeenCalled() + }) + + it('does not change cadence when reconciliation fails after observing a carrier', async () => { + const claimedAt = new Date('2025-01-01T00:00:00.000Z') + const schedule = { + ...SINGLE_SCHEDULE[0], + lastQueuedAt: claimedAt, + } + dbChainMockFns.limit + .mockResolvedValueOnce(SINGLE_CLAIMED_SCHEDULE_ROWS) + .mockResolvedValueOnce([{ workflowId: 'workflow-1', status: 'completed' }]) + .mockResolvedValueOnce([]) + dbChainMockFns.returning.mockReturnValueOnce([schedule]).mockReturnValueOnce([]) + mockGetJob.mockResolvedValueOnce({ + id: 'job-id-1', + status: 'completed', + payload: { + scheduleId: 'schedule-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + now: claimedAt.toISOString(), + }, + }) + mockApplyScheduleSuccessUpdate.mockRejectedValueOnce(new Error('database unavailable')) + + await runScheduleTick('test-request-id') + + expect(mockEnqueue).not.toHaveBeenCalled() + expect(mockResolveSystemBillingAttribution).not.toHaveBeenCalled() + expect(mockApplyScheduleFailureUpdate).not.toHaveBeenCalled() + expect(dbChainMockFns.set).not.toHaveBeenCalledWith( + expect.objectContaining({ infraRetryCount: 1 }) ) }) diff --git a/apps/sim/app/api/schedules/execute/route.ts b/apps/sim/app/api/schedules/execute/route.ts index b845e3bcf34..82d5c58de20 100644 --- a/apps/sim/app/api/schedules/execute/route.ts +++ b/apps/sim/app/api/schedules/execute/route.ts @@ -1,4 +1,11 @@ -import { asyncJobs, db, workflow, workflowDeploymentVersion, workflowSchedule } from '@sim/db' +import { + asyncJobs, + db, + workflow, + workflowDeploymentVersion, + workflowExecutionLogs, + workflowSchedule, +} from '@sim/db' import { createLogger } from '@sim/logger' import { sha256Hex } from '@sim/security/hash' import { toError } from '@sim/utils/errors' @@ -7,7 +14,7 @@ import { generateId } from '@sim/utils/id' import { isRecordLike } from '@sim/utils/object' import { randomInt } from '@sim/utils/random' import { Cron } from 'croner' -import { and, asc, eq, inArray, isNull, lt, lte, or, sql } from 'drizzle-orm' +import { and, asc, eq, gt, inArray, isNull, lt, lte, or, sql } from 'drizzle-orm' import { type NextRequest, NextResponse } from 'next/server' import type { ExecuteSchedulesResponse } from '@/lib/api/contracts/schedules' import { verifyCronAuth } from '@/lib/auth/internal' @@ -21,7 +28,7 @@ import { JOB_PENDING_RETENTION_HOURS, shouldExecuteInline, } from '@/lib/core/async-jobs' -import { JOB_STATUS, type Job } from '@/lib/core/async-jobs/types' +import { isAsyncJobEnqueueError, JOB_STATUS, type Job } from '@/lib/core/async-jobs/types' import { isRetryableInfrastructureError } from '@/lib/core/errors/retryable-infrastructure' import { getExecutionReservationTtlMs, @@ -32,6 +39,7 @@ import { import { runDetached } from '@/lib/core/utils/background' import { generateRequestId } from '@/lib/core/utils/request' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' +import type { DbOrTx } from '@/lib/db/types' import { registerManualExecutionAborter, unregisterManualExecutionAborter, @@ -46,7 +54,9 @@ import { } from '@/lib/workflows/schedules/execution-limits' import { calculateScheduleInfraRetryDelayMs } from '@/lib/workflows/schedules/retry' import { + applyScheduleCancellationUpdate, applyScheduleFailureUpdate, + applyScheduleSuccessUpdate, executeScheduleJob, releaseScheduleLock, type ScheduleExecutionPayload, @@ -67,6 +77,9 @@ const MAX_TICK_DURATION_MS = 3 * 60 * 1000 const STALE_SCHEDULE_CLAIM_MS = getExecutionReservationTtlMs() const STALE_SCHEDULE_RECOVERY_BATCH_SIZE = 100 const DATABASE_SCHEDULE_START_TURN_WAIT_MS = 1_000 +const SCHEDULE_CARRIER_RECONCILED_METADATA_KEY = 'scheduleReconciled' +const SCHEDULE_CARRIER_IRRECOVERABLE_METADATA_KEY = 'scheduleRecoveryIrrecoverable' +const LEGACY_SCHEDULE_CARRIER_RECOVERY_BLOCKED_METADATA_KEY = 'scheduleRecoveryBlocked' type DatabaseScheduleStartResult = 'started' | 'capacity_full' | 'not_pending' let databaseScheduleStartTurn: Promise | null = null @@ -200,8 +213,21 @@ type DatabaseScheduleExecutionTarget = Pick< > type ScheduleRecoveryMetadata = Pick< ScheduleExecutionPayload, - 'scheduleId' | 'workflowId' | 'now' | 'cronExpression' | 'timezone' | 'executionTimeoutMs' + | 'scheduleId' + | 'workflowId' + | 'now' + | 'cronExpression' + | 'timezone' + | 'executionTimeoutMs' + | 'executionId' + | 'scheduledFor' > +type ScheduleRecoveryOutcome = 'success' | 'paused' | 'failure' | 'cancelled' | 'indeterminate' +type ScheduleRecoveryEvidence = { + outcome: ScheduleRecoveryOutcome + executionStatus: string | null + logFound: boolean +} type SchedulePayloadValidation = | { success: true; payload: ScheduleExecutionPayload } | { success: false; error: string } @@ -221,6 +247,10 @@ function getScheduleRecoveryMetadataFromValue(payload: unknown): ScheduleRecover scheduleId: candidate.scheduleId, workflowId: candidate.workflowId, now: candidate.now, + executionId: + typeof candidate.executionId === 'string' && candidate.executionId.length > 0 + ? candidate.executionId + : undefined, cronExpression: typeof candidate.cronExpression === 'string' ? candidate.cronExpression : undefined, timezone: typeof candidate.timezone === 'string' ? candidate.timezone : undefined, @@ -230,6 +260,7 @@ function getScheduleRecoveryMetadataFromValue(payload: unknown): ScheduleRecover candidate.executionTimeoutMs > 0 ? candidate.executionTimeoutMs : undefined, + scheduledFor: typeof candidate.scheduledFor === 'string' ? candidate.scheduledFor : undefined, } } @@ -284,6 +315,10 @@ function getSchedulePayloadValidation(payload: unknown): SchedulePayloadValidati typeof candidate.deploymentVersionId === 'string' ? candidate.deploymentVersionId : undefined, + deploymentOperationId: + typeof candidate.deploymentOperationId === 'string' + ? candidate.deploymentOperationId + : undefined, lastRanAt: typeof candidate.lastRanAt === 'string' ? candidate.lastRanAt : undefined, failedCount: typeof candidate.failedCount === 'number' ? candidate.failedCount : undefined, infraRetryCount: @@ -304,11 +339,12 @@ async function restoreScheduleClaim( requestId: string, currentClaim: Date, activeClaim: Date, - context: string -): Promise { - if (currentClaim.getTime() === activeClaim.getTime()) return + context: string, + executor: DbOrTx = db +): Promise { + if (currentClaim.getTime() === activeClaim.getTime()) return true - const [restored] = await db + const [restored] = await executor .update(workflowSchedule) .set({ lastQueuedAt: activeClaim, updatedAt: new Date() }) .where( @@ -325,14 +361,15 @@ async function restoreScheduleClaim( }) if (!restored) { - const error = new Error(`Schedule claim restore did not update schedule ${scheduleId}`) logger.warn(`[${requestId}] ${context}`, { scheduleId, currentClaim: currentClaim.toISOString(), activeClaim: activeClaim.toISOString(), }) - throw error + return false } + + return true } function getScheduleExecutionLeaseMs( @@ -379,11 +416,27 @@ function activeScheduleExecutionJobsFilter() { function pendingScheduleExecutionJobsFilter(now: Date) { return and( sql`${asyncJobs.type} = 'schedule-execution' AND ${asyncJobs.status} = 'pending'`, - sql`${asyncJobs.attempts} < ${asyncJobs.maxAttempts}`, + eq(asyncJobs.attempts, 0), or(isNull(asyncJobs.runAt), lte(asyncJobs.runAt, now)) ) } +function claimedPendingScheduleExecutionJobsFilter(now: Date) { + return and( + sql`${asyncJobs.type} = 'schedule-execution' AND ${asyncJobs.status} = 'pending'`, + gt(asyncJobs.attempts, 0), + or(isNull(asyncJobs.runAt), lte(asyncJobs.runAt, now)) + ) +} + +function unreconciledTerminalScheduleExecutionJobsFilter() { + return and( + sql`${asyncJobs.type} = 'schedule-execution'`, + inArray(asyncJobs.status, [JOB_STATUS.COMPLETED, JOB_STATUS.FAILED, JOB_STATUS.CANCELLED]), + sql`COALESCE(${asyncJobs.metadata}->>${SCHEDULE_CARRIER_RECONCILED_METADATA_KEY}, 'false') <> 'true'` + ) +} + function staleScheduleExecutionJobsFilter(now: Date) { const legacyMaxDurationSeconds = STALE_SCHEDULE_CLAIM_MS / 1000 const cleanupGraceSeconds = RESERVATION_TTL_BUFFER_MS / 1000 @@ -416,6 +469,160 @@ function getScheduleNextRunAt( ) } +function classifyScheduleRecoveryEvidence( + status: string | null, + logFound = status !== null +): ScheduleRecoveryEvidence { + switch (status) { + case 'completed': + return { outcome: 'success', executionStatus: status, logFound } + case 'pending': + case 'paused': + return { outcome: 'paused', executionStatus: status, logFound } + case 'failed': + return { outcome: 'failure', executionStatus: status, logFound } + case 'cancelled': + return { outcome: 'cancelled', executionStatus: status, logFound } + default: + return { outcome: 'indeterminate', executionStatus: status, logFound } + } +} + +async function getScheduleRecoveryEvidence( + payload: ScheduleRecoveryMetadata | null +): Promise { + if (!payload?.executionId) return classifyScheduleRecoveryEvidence(null, false) + + const [executionLog] = await db + .select({ + workflowId: workflowExecutionLogs.workflowId, + status: workflowExecutionLogs.status, + }) + .from(workflowExecutionLogs) + .where(eq(workflowExecutionLogs.executionId, payload.executionId)) + .limit(1) + + if (!executionLog) return classifyScheduleRecoveryEvidence(null, false) + if (executionLog.workflowId !== payload.workflowId) return classifyScheduleRecoveryEvidence(null) + + return classifyScheduleRecoveryEvidence(executionLog.status) +} + +async function applyScheduleRecoveryAccounting(params: { + payload: ScheduleRecoveryMetadata + evidence: ScheduleRecoveryEvidence + now: Date + requestId: string + executor?: DbOrTx +}): Promise<{ disabled: boolean; updated: boolean }> { + const { payload, evidence, now, requestId, executor } = params + const claimedAt = getSchedulePayloadClaimedAt(payload) + if (!claimedAt) return { disabled: false, updated: false } + + const nextRunAt = getScheduleNextRunAt(payload, now) + const context = `Error reconciling schedule ${payload.scheduleId} after ${evidence.outcome} recovery` + + if (evidence.outcome === 'success' || evidence.outcome === 'paused') { + const updated = await applyScheduleSuccessUpdate({ + scheduleId: payload.scheduleId, + now, + nextRunAt, + expectedLastQueuedAt: claimedAt, + requestId, + context, + executor, + }) + return { disabled: false, updated } + } + + if (evidence.outcome === 'cancelled') { + const updated = await applyScheduleCancellationUpdate({ + scheduleId: payload.scheduleId, + now, + nextRunAt, + expectedLastQueuedAt: claimedAt, + requestId, + context, + executor, + }) + return { disabled: false, updated } + } + + const result = await applyScheduleFailureUpdate({ + scheduleId: payload.scheduleId, + now, + nextRunAt, + expectedLastQueuedAt: claimedAt, + requestId, + context, + executor, + }) + return { disabled: result.disabled, updated: result.updated } +} + +async function reconcileRecoveredScheduleAccounting(params: { + payload: ScheduleRecoveryMetadata + evidence: ScheduleRecoveryEvidence + now: Date + requestId: string + executor: DbOrTx +}): Promise<{ disabled: boolean; reconciled: boolean }> { + const firstAttempt = await applyScheduleRecoveryAccounting(params) + if (firstAttempt.updated) { + return { disabled: firstAttempt.disabled, reconciled: true } + } + + const claimedAt = getSchedulePayloadClaimedAt(params.payload) + const scheduledFor = params.payload.scheduledFor ? new Date(params.payload.scheduledFor) : null + if (!claimedAt || !scheduledFor || Number.isNaN(scheduledFor.getTime())) { + return { disabled: false, reconciled: false } + } + + const [schedule] = await params.executor + .select({ + archivedAt: workflowSchedule.archivedAt, + lastQueuedAt: workflowSchedule.lastQueuedAt, + nextRunAt: workflowSchedule.nextRunAt, + status: workflowSchedule.status, + }) + .from(workflowSchedule) + .where(eq(workflowSchedule.id, params.payload.scheduleId)) + .for('update') + + if ( + !schedule || + schedule.archivedAt || + schedule.status === 'disabled' || + schedule.status === 'completed' || + !schedule.nextRunAt || + schedule.nextRunAt.getTime() !== scheduledFor.getTime() + ) { + return { disabled: false, reconciled: true } + } + + if (schedule.lastQueuedAt) { + return { disabled: false, reconciled: false } + } + + const [restored] = await params.executor + .update(workflowSchedule) + .set({ lastQueuedAt: claimedAt, updatedAt: params.now }) + .where( + and( + eq(workflowSchedule.id, params.payload.scheduleId), + isNull(workflowSchedule.archivedAt), + eq(workflowSchedule.nextRunAt, scheduledFor), + isNull(workflowSchedule.lastQueuedAt) + ) + ) + .returning({ id: workflowSchedule.id }) + + if (!restored) return { disabled: false, reconciled: false } + + const retry = await applyScheduleRecoveryAccounting(params) + return { disabled: retry.disabled, reconciled: retry.updated } +} + async function markClaimedScheduleFailed( schedule: DatabaseScheduleExecutionTarget, requestId: string, @@ -441,6 +648,87 @@ async function markClaimedScheduleFailed( } } +async function reconcileExistingScheduleJob(params: { + job: Job + schedule: DatabaseScheduleExecutionTarget + currentClaim: Date + requestId: string + jobQueue: JobQueue + cancelCarrier: boolean +}): Promise { + const { job, schedule, currentClaim, requestId, jobQueue, cancelCarrier } = params + const metadata = getScheduleRecoveryMetadataFromJob(job) + const metadataClaim = getSchedulePayloadClaimedAt(metadata) + const scheduleWorkflowId = schedule.workflowId + const validMetadata = + metadata && + metadataClaim && + scheduleWorkflowId && + metadata.scheduleId === schedule.id && + metadata.workflowId === scheduleWorkflowId + ? metadata + : null + + if (cancelCarrier) { + await jobQueue.cancelJob(job.id) + } + + if (!scheduleWorkflowId) { + logger.warn(`[${requestId}] Cannot reconcile schedule job without a workflow`, { + scheduleId: schedule.id, + jobId: job.id, + }) + return + } + + if (validMetadata && metadataClaim) { + const restored = await restoreScheduleClaim( + schedule.id, + requestId, + currentClaim, + metadataClaim, + `Failed to restore schedule ${schedule.id} claim for recovery` + ) + if (!restored) { + logger.info(`[${requestId}] Skipped schedule reconciliation after claim changed`, { + scheduleId: schedule.id, + jobId: job.id, + }) + return + } + } + + const recoveryPayload: ScheduleRecoveryMetadata = + validMetadata ?? + ({ + scheduleId: schedule.id, + workflowId: scheduleWorkflowId, + now: currentClaim.toISOString(), + cronExpression: schedule.cronExpression ?? undefined, + timezone: schedule.timezone, + } satisfies ScheduleRecoveryMetadata) + let evidence = validMetadata + ? await getScheduleRecoveryEvidence(validMetadata) + : classifyScheduleRecoveryEvidence(null, false) + if (!evidence.logFound && job.status === JOB_STATUS.CANCELLED) { + evidence = { outcome: 'cancelled', executionStatus: null, logFound: false } + } + const { disabled } = await applyScheduleRecoveryAccounting({ + payload: recoveryPayload, + evidence, + now: new Date(), + requestId, + }) + + if (disabled) { + await notifyScheduleAutoDisabled({ + scheduleId: schedule.id, + reason: 'consecutive_failures', + requestId, + }) + } +} + async function deferClaimedScheduleAfterQueueFailure( schedule: ClaimedSchedule, requestId: string, @@ -500,7 +788,10 @@ async function handleClaimedScheduleSetupFailure( retryContext: string, failureContext: string ): Promise { - if (isRetryableInfrastructureError(error)) { + const retryable = isAsyncJobEnqueueError(error) + ? error.retryable + : isRetryableInfrastructureError(error) + if (retryable) { await deferClaimedScheduleAfterQueueFailure( schedule, requestId, @@ -525,7 +816,7 @@ async function recoverStaleDatabaseScheduleJobs(now: Date): Promise { * would both notify about writes a rollback discards and issue pooled-client * reads while the transaction still holds row locks under the advisory lock. */ - const disabledScheduleIds: string[] = [] + const disabledScheduleIds = new Set() await db.transaction(async (tx) => { const [lock] = await tx.execute<{ acquired: boolean }>( @@ -538,78 +829,141 @@ async function recoverStaleDatabaseScheduleJobs(now: Date): Promise { return } - const staleRows = await tx + const claimedRows = await tx .select({ id: asyncJobs.id, payload: asyncJobs.payload, - attempts: asyncJobs.attempts, - maxAttempts: asyncJobs.maxAttempts, + status: asyncJobs.status, }) .from(asyncJobs) - .where(staleScheduleExecutionJobsFilter(now)) - .orderBy(asc(asyncJobs.startedAt), asc(asyncJobs.id)) + .where( + or( + staleScheduleExecutionJobsFilter(now), + claimedPendingScheduleExecutionJobsFilter(now), + unreconciledTerminalScheduleExecutionJobsFilter() + ) + ) + .for('update', { skipLocked: true }) + .orderBy(asc(asyncJobs.updatedAt), asc(asyncJobs.id)) .limit(STALE_SCHEDULE_RECOVERY_BATCH_SIZE) - const exhaustedRows = staleRows.filter((row) => row.attempts >= row.maxAttempts) - const retryableRows = staleRows.filter((row) => row.attempts < row.maxAttempts) + const payloads = new Map( + claimedRows.map((row) => [row.id, getScheduleRecoveryMetadataFromValue(row.payload)]) + ) + const executionIds = Array.from( + new Set( + claimedRows.flatMap((row) => { + const executionId = payloads.get(row.id)?.executionId + return executionId ? [executionId] : [] + }) + ) + ) + const executionLogs = + executionIds.length > 0 + ? await tx + .select({ + executionId: workflowExecutionLogs.executionId, + workflowId: workflowExecutionLogs.workflowId, + status: workflowExecutionLogs.status, + }) + .from(workflowExecutionLogs) + .where(inArray(workflowExecutionLogs.executionId, executionIds)) + : [] + const executionLogsById = new Map(executionLogs.map((log) => [log.executionId, log])) - if (exhaustedRows.length > 0) { - await tx + for (const row of claimedRows) { + const payload = getScheduleRecoveryMetadataFromValue(row.payload) + const executionLog = payload?.executionId + ? executionLogsById.get(payload.executionId) + : undefined + const evidence = classifyScheduleRecoveryEvidence( + executionLog && executionLog.workflowId === payload?.workflowId + ? executionLog.status + : null, + Boolean(executionLog) + ) + const recoveryEvidence = + !evidence.logFound && row.status === JOB_STATUS.CANCELLED + ? { outcome: 'cancelled' as const, executionStatus: null, logFound: false } + : evidence + + const knownOutcome = recoveryEvidence.outcome !== 'indeterminate' + const recoveredCarrierStatus = + row.status === JOB_STATUS.CANCELLED + ? JOB_STATUS.CANCELLED + : knownOutcome + ? JOB_STATUS.COMPLETED + : JOB_STATUS.FAILED + const [settledJob] = await tx .update(asyncJobs) .set({ - status: JOB_STATUS.FAILED, + status: recoveredCarrierStatus, completedAt: now, - error: 'Stale schedule execution processing lease exhausted retry attempts', + error: + recoveredCarrierStatus === JOB_STATUS.CANCELLED + ? 'Cancelled' + : knownOutcome + ? null + : `Indeterminate schedule execution outcome${recoveryEvidence.executionStatus ? ` (${recoveryEvidence.executionStatus})` : ''}`, + output: knownOutcome + ? { + recovered: true, + executionId: payload?.executionId ?? null, + executionStatus: recoveryEvidence.executionStatus, + } + : null, updatedAt: now, }) - .where( - inArray( - asyncJobs.id, - exhaustedRows.map((row) => row.id) - ) - ) - } - - for (const row of exhaustedRows) { - const payload = getScheduleRecoveryMetadataFromValue(row.payload) - const claimedAt = getSchedulePayloadClaimedAt(payload) - if (!payload || !claimedAt) continue + .where(and(eq(asyncJobs.id, row.id), eq(asyncJobs.status, row.status))) + .returning({ id: asyncJobs.id }) + + if (!settledJob) continue + if (!payload) { + await tx + .update(asyncJobs) + .set({ + metadata: sql`(COALESCE(${asyncJobs.metadata}, '{}'::jsonb) - ${LEGACY_SCHEDULE_CARRIER_RECOVERY_BLOCKED_METADATA_KEY}) || ${JSON.stringify( + { + [SCHEDULE_CARRIER_RECONCILED_METADATA_KEY]: true, + [SCHEDULE_CARRIER_IRRECOVERABLE_METADATA_KEY]: true, + } + )}::jsonb`, + updatedAt: now, + }) + .where(eq(asyncJobs.id, row.id)) + continue + } - const { disabled } = await applyScheduleFailureUpdate({ - scheduleId: payload.scheduleId, + const { disabled, reconciled } = await reconcileRecoveredScheduleAccounting({ + payload, + evidence: recoveryEvidence, now, - nextRunAt: getScheduleNextRunAt(payload, now), - expectedLastQueuedAt: claimedAt, requestId: 'stale-schedule-recovery', - context: `Error updating schedule ${payload.scheduleId} after stale lease recovery`, executor: tx, }) - if (disabled) disabledScheduleIds.push(payload.scheduleId) - } - - if (retryableRows.length > 0) { - await tx - .update(asyncJobs) - .set({ - status: JOB_STATUS.PENDING, - startedAt: null, - error: 'Recovered after stale schedule execution processing lease', - updatedAt: now, - }) - .where( - inArray( - asyncJobs.id, - retryableRows.map((row) => row.id) - ) - ) + if (disabled) disabledScheduleIds.add(payload.scheduleId) + if (reconciled) { + await tx + .update(asyncJobs) + .set({ + metadata: sql`(COALESCE(${asyncJobs.metadata}, '{}'::jsonb) - ${LEGACY_SCHEDULE_CARRIER_RECOVERY_BLOCKED_METADATA_KEY}) || ${JSON.stringify( + { + [SCHEDULE_CARRIER_RECONCILED_METADATA_KEY]: true, + } + )}::jsonb`, + updatedAt: now, + }) + .where(eq(asyncJobs.id, row.id)) + } } }) - const notifiable = disabledScheduleIds.slice(0, STALE_SCHEDULE_RECOVERY_NOTIFY_LIMIT) - if (disabledScheduleIds.length > notifiable.length) { + const disabledScheduleIdList = Array.from(disabledScheduleIds) + const notifiable = disabledScheduleIdList.slice(0, STALE_SCHEDULE_RECOVERY_NOTIFY_LIMIT) + if (disabledScheduleIdList.length > notifiable.length) { logger.warn('Capped schedule auto-disable notifications for stale recovery batch', { - disabled: disabledScheduleIds.length, + disabled: disabledScheduleIdList.length, notified: notifiable.length, }) } @@ -675,13 +1029,48 @@ async function tryStartDatabaseScheduleJob(jobId: string): Promise { + const now = new Date() + const [cancelled] = await db + .update(asyncJobs) + .set({ + status: JOB_STATUS.CANCELLED, + completedAt: now, + error: 'Cancelled after schedule claim was released', + metadata: sql`(COALESCE(${asyncJobs.metadata}, '{}'::jsonb) - ${LEGACY_SCHEDULE_CARRIER_RECOVERY_BLOCKED_METADATA_KEY}) || ${JSON.stringify( + { + [SCHEDULE_CARRIER_RECONCILED_METADATA_KEY]: true, + } + )}::jsonb`, + updatedAt: now, + }) + .where( + and( + eq(asyncJobs.id, jobId), + eq(asyncJobs.type, 'schedule-execution'), + eq(asyncJobs.status, JOB_STATUS.PENDING), + eq(asyncJobs.attempts, 0) + ) + ) + .returning({ id: asyncJobs.id }) + + return Boolean(cancelled) +} + async function executeDatabaseScheduleJob( jobQueue: JobQueue, jobId: string, @@ -807,16 +1196,12 @@ async function resumePendingDatabaseScheduleJobs( const claimState = await getScheduleClaimState(recoveryMetadata, claimedAt) if (claimState === 'released') { - logger.info(`[${requestId}] Completing stale pending schedule execution job`, { + logger.info(`[${requestId}] Cancelling stale pending schedule execution job`, { scheduleId: recoveryMetadata.scheduleId, workflowId: recoveryMetadata.workflowId, jobId: job.id, }) - await jobQueue.completeJob(job.id, { - skipped: true, - reason: 'schedule claim no longer matches pending job occurrence', - }) - return true + return cancelReleasedPendingDatabaseScheduleCarrier(job.id) } if (claimState === 'claimed_by_other') { logger.info(`[${requestId}] Leaving pending schedule execution job for active claimant`, { @@ -893,75 +1278,32 @@ async function processScheduleItem( useDatabaseFallback: boolean ) { const queueTime = schedule.lastQueuedAt ?? queuedAt - const executionId = generateId() - const workspaceId = schedule.workspaceId ?? undefined - let billingAttribution: BillingAttributionSnapshot - try { - if (!workspaceId) { - throw new Error(`Unable to resolve workspace for schedule ${schedule.id}`) - } - billingAttribution = await resolveSystemBillingAttribution(workspaceId) - } catch (error) { - await handleClaimedScheduleSetupFailure( - schedule, - requestId, - queueTime, - error, - `Failed to defer schedule ${schedule.id} after billing attribution failure`, - `Failed to mark schedule ${schedule.id} failed after billing attribution failure` - ) - return - } - const correlation = { - executionId, - requestId, - source: 'schedule' as const, - workflowId: schedule.workflowId!, - scheduleId: schedule.id, - triggerType: 'schedule', - scheduledFor: schedule.nextRunAt?.toISOString(), - } - const executionTimeoutMs = getExecutionTimeout( - billingAttribution.payerSubscription?.plan, - 'async', - billingAttribution.payerSubscription?.enterpriseWorkflowExecutionTimeoutSeconds - ) - - const payload = { - scheduleId: schedule.id, - workflowId: schedule.workflowId!, - executionId, - requestId, - correlation, - blockId: schedule.blockId || undefined, - workspaceId, - billingAttribution, - deploymentVersionId: schedule.deploymentVersionId || undefined, - deploymentOperationId: schedule.deploymentOperationId || undefined, - cronExpression: schedule.cronExpression || undefined, - timezone: schedule.timezone || undefined, - lastRanAt: schedule.lastRanAt?.toISOString(), - failedCount: schedule.failedCount || 0, - infraRetryCount: schedule.infraRetryCount || 0, - now: queueTime.toISOString(), - scheduledFor: schedule.nextRunAt?.toISOString(), - executionTimeoutMs, - } satisfies ScheduleExecutionPayload - + const scheduleJobId = buildScheduleExecutionJobId(schedule) let enqueuedJobId: string | null = null + let carrierObservedOrLookupUncertain = false try { + let existingJob: Job | null + try { + existingJob = await jobQueue.getJob(scheduleJobId) + if (existingJob) carrierObservedOrLookupUncertain = true + } catch (error) { + carrierObservedOrLookupUncertain = true + throw error + } const delayMs = randomInt(0, SCHEDULE_JITTER_MAX_MS) - const scheduleJobId = buildScheduleExecutionJobId(schedule) - const existingJob = await jobQueue.getJob(scheduleJobId) if (existingJob && ['pending', 'processing'].includes(existingJob.status)) { const activeJobPayload = getScheduleRecoveryMetadataFromJob(existingJob) const activeJobClaim = getSchedulePayloadClaimedAt(activeJobPayload) - if (useDatabaseFallback && isStaleDatabaseScheduleJob(existingJob)) { + if ( + useDatabaseFallback && + (isStaleDatabaseScheduleJob(existingJob) || + (existingJob.status === JOB_STATUS.PENDING && existingJob.attempts > 0)) + ) { await recoverStaleDatabaseScheduleJobs(new Date()) - logger.info(`[${requestId}] Recovered stale database schedule execution jobs`, { + logger.info(`[${requestId}] Reconciled claimed database schedule execution jobs`, { scheduleId: schedule.id, jobId: scheduleJobId, }) @@ -982,19 +1324,22 @@ async function processScheduleItem( jobId: existingJob.id, claimedAt: activeJobClaim.toISOString(), }) - await jobQueue.cancelJob(existingJob.id) - await releaseScheduleLock( - schedule.id, + await reconcileExistingScheduleJob({ + job: existingJob, + schedule, + currentClaim: queueTime, requestId, - queuedAt, - `Released stale schedule ${schedule.id} after cancelling stale schedule execution job`, - undefined, - { expectedLastQueuedAt: queueTime } - ) + jobQueue, + cancelCarrier: true, + }) return } - if (useDatabaseFallback && databaseJob?.status === JOB_STATUS.PENDING) { + if ( + useDatabaseFallback && + databaseJob?.status === JOB_STATUS.PENDING && + databaseJob.attempts === 0 + ) { const payloadValidation = getSchedulePayloadValidation(databaseJob.payload) if (!payloadValidation.success) { const error = `Invalid pending schedule execution payload: ${payloadValidation.error}` @@ -1022,13 +1367,14 @@ async function processScheduleItem( jobId: scheduleJobId, }) if (databaseJobClaim) { - await restoreScheduleClaim( + const restored = await restoreScheduleClaim( schedule.id, requestId, queueTime, databaseJobClaim, `Failed to restore schedule ${schedule.id} claim for pending database fallback job` ) + if (!restored) return } enqueuedJobId = scheduleJobId await executeDatabaseScheduleJob( @@ -1053,24 +1399,14 @@ async function processScheduleItem( jobId: scheduleJobId, status: databaseJob.status, }) - if (databaseJob.status === JOB_STATUS.FAILED) { - await markClaimedScheduleFailed( - schedule, - requestId, - queueTime, - `Failed to mark schedule ${schedule.id} failed after terminal database fallback job` - ) - return - } - - await releaseScheduleLock( - schedule.id, + await reconcileExistingScheduleJob({ + job: databaseJob, + schedule, + currentClaim: queueTime, requestId, - queuedAt, - `Released stale schedule ${schedule.id} for terminal database fallback job ${scheduleJobId}`, - getNextRunFromCronExpression(schedule.cronExpression, schedule.timezone), - { expectedLastQueuedAt: queueTime } - ) + jobQueue, + cancelCarrier: false, + }) return } @@ -1097,21 +1433,75 @@ async function processScheduleItem( return } if (existingJob) { - logger.info(`[${requestId}] Releasing stale schedule claim for finished job`, { + logger.info(`[${requestId}] Reconciling schedule claim for finished job`, { scheduleId: schedule.id, jobId: scheduleJobId, status: existingJob.status, }) - await releaseScheduleLock( - schedule.id, + await reconcileExistingScheduleJob({ + job: existingJob, + schedule, + currentClaim: queueTime, requestId, - queuedAt, - `Released stale schedule ${schedule.id} for finished job ${scheduleJobId}`, - getNextRunFromCronExpression(schedule.cronExpression, schedule.timezone), - { expectedLastQueuedAt: queueTime } + jobQueue, + cancelCarrier: false, + }) + return + } + + const executionId = generateId() + const workspaceId = schedule.workspaceId ?? undefined + let billingAttribution: BillingAttributionSnapshot + try { + if (!workspaceId) { + throw new Error(`Unable to resolve workspace for schedule ${schedule.id}`) + } + billingAttribution = await resolveSystemBillingAttribution(workspaceId) + } catch (error) { + await handleClaimedScheduleSetupFailure( + schedule, + requestId, + queueTime, + error, + `Failed to defer schedule ${schedule.id} after billing attribution failure`, + `Failed to mark schedule ${schedule.id} failed after billing attribution failure` ) return } + const correlation = { + executionId, + requestId, + source: 'schedule' as const, + workflowId: schedule.workflowId!, + scheduleId: schedule.id, + triggerType: 'schedule', + scheduledFor: schedule.nextRunAt?.toISOString(), + } + const executionTimeoutMs = getExecutionTimeout( + billingAttribution.payerSubscription?.plan, + 'async', + billingAttribution.payerSubscription?.enterpriseWorkflowExecutionTimeoutSeconds + ) + const payload = { + scheduleId: schedule.id, + workflowId: schedule.workflowId!, + executionId, + requestId, + correlation, + blockId: schedule.blockId || undefined, + workspaceId, + billingAttribution, + deploymentVersionId: schedule.deploymentVersionId || undefined, + deploymentOperationId: schedule.deploymentOperationId || undefined, + cronExpression: schedule.cronExpression || undefined, + timezone: schedule.timezone || undefined, + lastRanAt: schedule.lastRanAt?.toISOString(), + failedCount: schedule.failedCount || 0, + infraRetryCount: schedule.infraRetryCount || 0, + now: queueTime.toISOString(), + scheduledFor: schedule.nextRunAt?.toISOString(), + executionTimeoutMs, + } satisfies ScheduleExecutionPayload let jobId: string try { @@ -1127,10 +1517,17 @@ async function processScheduleItem( }) enqueuedJobId = jobId } catch (error) { + const classifiedError = isAsyncJobEnqueueError(error) ? error : null + const acceptance = classifiedError?.acceptance ?? 'unknown' logger.error( `[${requestId}] Failed to enqueue schedule execution for workflow ${schedule.workflowId}`, - error + error, + { acceptance, jobId: scheduleJobId } ) + if (acceptance !== 'rejected') { + carrierObservedOrLookupUncertain = true + return + } await handleClaimedScheduleSetupFailure( schedule, requestId, @@ -1173,14 +1570,14 @@ async function processScheduleItem( jobId, status: queuedJob.status, }) - await releaseScheduleLock( - schedule.id, + await reconcileExistingScheduleJob({ + job: queuedJob, + schedule, + currentClaim: queueTime, requestId, - queuedAt, - `Released stale schedule ${schedule.id} for finished job ${jobId}`, - getNextRunFromCronExpression(schedule.cronExpression, schedule.timezone), - { expectedLastQueuedAt: queueTime } - ) + jobQueue, + cancelCarrier: false, + }) return } if (queuedJob) { @@ -1194,15 +1591,14 @@ async function processScheduleItem( jobId, claimedAt: queuedJobClaim.toISOString(), }) - await jobQueue.cancelJob(jobId) - await releaseScheduleLock( - schedule.id, + await reconcileExistingScheduleJob({ + job: queuedJob, + schedule, + currentClaim: queueTime, requestId, - queuedAt, - `Released stale schedule ${schedule.id} after cancelling stale queued schedule execution job`, - undefined, - { expectedLastQueuedAt: queueTime } - ) + jobQueue, + cancelCarrier: true, + }) return } @@ -1229,7 +1625,7 @@ async function processScheduleItem( `[${requestId}] Failed after queueing schedule execution for workflow ${schedule.workflowId}`, error ) - if (!enqueuedJobId) { + if (!enqueuedJobId && !carrierObservedOrLookupUncertain) { await handleClaimedScheduleSetupFailure( schedule, requestId, @@ -1238,7 +1634,14 @@ async function processScheduleItem( `Failed to defer schedule ${schedule.id} after pre-enqueue failure`, `Failed to mark schedule ${schedule.id} failed after non-retryable setup failure` ) + return } + + logger.warn(`[${requestId}] Preserved schedule occurrence after carrier uncertainty`, { + scheduleId: schedule.id, + jobId: enqueuedJobId ?? scheduleJobId, + error: toError(error).message, + }) } } diff --git a/apps/sim/background/schedule-execution.ts b/apps/sim/background/schedule-execution.ts index 6fe79518667..19f2046d4e2 100644 --- a/apps/sim/background/schedule-execution.ts +++ b/apps/sim/background/schedule-execution.ts @@ -269,6 +269,59 @@ export async function releaseScheduleLock( return outcome.updated } +/** Applies successful-run accounting only while the caller still owns the claim. */ +export async function applyScheduleSuccessUpdate(params: { + scheduleId: string + now: Date + nextRunAt: Date + expectedLastQueuedAt: Date | null + requestId: string + context: string + executor?: DbOrTx +}): Promise { + const { scheduleId, now, nextRunAt, expectedLastQueuedAt, requestId, context, executor } = params + + const outcome = await applyScheduleUpdate( + scheduleId, + { + lastRanAt: now, + updatedAt: now, + nextRunAt, + failedCount: 0, + lastQueuedAt: null, + ...resetScheduleInfraRetryCount(), + }, + requestId, + context, + { expectedLastQueuedAt, executor } + ) + + return outcome.updated +} + +/** Applies cancelled-run accounting only while the caller still owns the claim. */ +export async function applyScheduleCancellationUpdate(params: { + scheduleId: string + now: Date + nextRunAt: Date + expectedLastQueuedAt: Date | null + requestId: string + context: string + executor?: DbOrTx +}): Promise { + const { scheduleId, now, nextRunAt, expectedLastQueuedAt, requestId, context, executor } = params + + const outcome = await applyScheduleUpdate( + scheduleId, + buildScheduleCancellationUpdate(now, nextRunAt), + requestId, + context, + { expectedLastQueuedAt, executor } + ) + + return outcome.updated +} + /** * Applies {@link buildScheduleFailureUpdate} through the same guarded write the * trigger.dev path uses, and reports whether the row just transitioned to @@ -1089,17 +1142,14 @@ export async function executeScheduleJob( const nextRunAt = calculateNextRunTime(payload, executionResult.blocks) - await updateClaimedSchedule( - { - lastRanAt: now, - updatedAt: now, - nextRunAt, - failedCount: 0, - lastQueuedAt: null, - ...resetScheduleInfraRetryCount(), - }, - `Error updating schedule ${payload.scheduleId} after success` - ) + await applyScheduleSuccessUpdate({ + scheduleId: payload.scheduleId, + now, + nextRunAt, + expectedLastQueuedAt: claimedAt, + requestId, + context: `Error updating schedule ${payload.scheduleId} after success`, + }) return } @@ -1107,10 +1157,14 @@ export async function executeScheduleJob( logger.info(`[${requestId}] Workflow ${payload.workflowId} execution was cancelled`) const nextRunAt = calculateNextRunTime(payload, executionResult.blocks) - await updateClaimedSchedule( - buildScheduleCancellationUpdate(now, nextRunAt), - `Error updating schedule ${payload.scheduleId} after cancellation` - ) + await applyScheduleCancellationUpdate({ + scheduleId: payload.scheduleId, + now, + nextRunAt, + expectedLastQueuedAt: claimedAt, + requestId, + context: `Error updating schedule ${payload.scheduleId} after cancellation`, + }) return }