From e3cbbadb4b4f389b5a147ed193a3f9f259188925 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 14:44:26 -0700 Subject: [PATCH 1/5] fix(trigger): complete async workflow runs whose failure is already recorded instead of failing them Every failed workflow-execution run fires the Trigger.dev error alert, and in practice those runs are user/workflow errors (missing start block, Function runtime errors, provider 4xx/5xx) that core already recorded as the run's failed execution log. The worker meant to rethrow only unrecorded faults, but it checked wasExecutionFinalizedByCore without awaiting the post-execution promise that sets the mark, so the check almost never saw it. The webhook worker already awaited it (#4860); workflow-execution was skipped. - isRecordedWorkflowFailure awaits post-execution, then checks the mark. The webhook worker uses it with no behavior change; the schedule worker uses it so it no longer races core's completion (its caller swallows either way). - workflow-execution completes with { success: false, error } for a recorded failure, and for non-retryable 402/403/404/429 preprocessing rejections (deleted or undeployed workflow, suspended account, exhausted quota) after the same log write as before. Timeouts, unrecorded errors, retryable rejections, and 5xx still fail the job. Trigger retries are maxAttempts 1 and the database runner never retries, so no retry semantics change. - Both queue backends' getJob read a completed workflow-execution job carrying that result as failed, with the same error and output (undefined on Trigger, null in the database) the thrown run had, so /api/jobs, the SDKs, and the v2 run-status queue fallback are unchanged. resume-execution is left throwing: executeResumeJob is also called inline by the resume poller and enqueue path, which rely on the throw. --- .../async-preprocessing-correlation.test.ts | 128 +++++++++++++----- apps/sim/background/schedule-execution.ts | 8 +- apps/sim/background/webhook-execution.ts | 8 +- apps/sim/background/workflow-execution.ts | 50 ++++++- .../core/async-jobs/backends/database.test.ts | 53 ++++++++ .../lib/core/async-jobs/backends/database.ts | 12 +- .../async-jobs/backends/trigger-dev.test.ts | 75 ++++++++++ .../core/async-jobs/backends/trigger-dev.ts | 12 +- .../lib/core/async-jobs/recorded-failure.ts | 51 +++++++ .../workflows/executor/recorded-failure.ts | 19 +++ 10 files changed, 361 insertions(+), 55 deletions(-) create mode 100644 apps/sim/lib/core/async-jobs/recorded-failure.ts create mode 100644 apps/sim/lib/workflows/executor/recorded-failure.ts diff --git a/apps/sim/background/async-preprocessing-correlation.test.ts b/apps/sim/background/async-preprocessing-correlation.test.ts index 0a789bc5dd6..a9153d63d1b 100644 --- a/apps/sim/background/async-preprocessing-correlation.test.ts +++ b/apps/sim/background/async-preprocessing-correlation.test.ts @@ -358,46 +358,70 @@ describe('async preprocessing correlation threading', () => { ) }) - it('preserves a core-finalized execution error for task failure semantics', async () => { - const rawError = Object.assign(new Error('Function 1 failed with activated-secret-value'), { - executionResult: { - success: false, - output: { error: 'Function failed' }, - logs: [], - }, - }) - mockPreprocessExecution.mockResolvedValueOnce({ - success: true, - actorUserId: 'actor-1', - workflowRecord: { - id: 'workflow-1', - userId: 'owner-1', - workspaceId: 'workspace-1', - variables: {}, - }, - billingAttribution, - executionTimeout: {}, - }) - mockExecuteWorkflowCore.mockRejectedValueOnce(rawError) - mockHasExecutionResult.mockImplementation((error) => error === rawError) - mockWasExecutionFinalizedByCore.mockReturnValue(true) + describe('a workflow failure core already recorded', () => { + const recordedError = () => + Object.assign(new Error('Function 1 failed: RuntimeError'), { + executionResult: { success: false, output: { error: 'Function failed' }, logs: [] }, + }) - await expect( - executeWorkflowJob({ + /** Core marks the failure finalized from a post-execution promise it does not await. */ + function finalizeWhenPostExecutionSettles() { + let finalized = false + loggingSessionMockFns.mockWaitForPostExecution.mockImplementationOnce(async () => { + await Promise.resolve() + finalized = true + }) + mockWasExecutionFinalizedByCore.mockImplementation(() => finalized) + } + + function runRecordedFailure(executionId: string) { + mockPreprocessExecution.mockResolvedValueOnce({ + success: true, + actorUserId: 'actor-1', + workflowRecord: { + id: 'workflow-1', + userId: 'owner-1', + workspaceId: 'workspace-1', + variables: {}, + }, + billingAttribution, + executionTimeout: {}, + }) + return executeWorkflowJob({ principal, workflowId: 'workflow-1', userId: 'actor-1', workspaceId: 'workspace-1', billingAttribution, triggerType: 'api', - executionId: 'execution-finalized', - requestId: 'request-finalized', + executionId, + requestId: 'request-recorded', }) - ).rejects.toBe(rawError) + } - expect(loggingSessionMockFns.mockWaitForPostExecution).not.toHaveBeenCalled() - expect(mockWasExecutionFinalizedByCore).toHaveBeenCalledWith(rawError, 'execution-finalized') - expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled() + it('completes the job with the failure once core finalization settles', async () => { + const error = recordedError() + mockExecuteWorkflowCore.mockRejectedValueOnce(error) + mockHasExecutionResult.mockImplementation((value) => value === error) + finalizeWhenPostExecutionSettles() + + await expect(runRecordedFailure('execution-recorded')).resolves.toMatchObject({ + success: false, + workflowId: 'workflow-1', + executionId: 'execution-recorded', + error: 'Function 1 failed: RuntimeError', + }) + expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled() + }) + + it('still fails the job when the run passed its deadline', async () => { + const error = recordedError() + mockExecuteWorkflowCore.mockRejectedValueOnce(error) + mockIsWorkflowTimedOut.mockReturnValue(true) + finalizeWhenPostExecutionSettles() + + await expect(runRecordedFailure('execution-recorded-late')).rejects.toBe(error) + }) }) it('fails the backing job after a cooperative workflow timeout is finalized', async () => { @@ -528,6 +552,48 @@ describe('async preprocessing correlation threading', () => { ) }) + describe('a preprocessing rejection', () => { + function runRejected(error: Record) { + mockPreprocessExecution.mockResolvedValueOnce({ success: false, error }) + return executeWorkflowJob({ + principal, + workflowId: 'workflow-1', + userId: 'actor-1', + workspaceId: 'workspace-1', + triggerType: 'api', + executionId: 'execution-rejected', + requestId: 'request-rejected', + billingAttribution, + }) + } + + it.each([ + { statusCode: 404, message: 'Workflow not found' }, + { statusCode: 403, message: 'Workflow is not deployed' }, + { statusCode: 402, message: 'Usage limit exceeded. Please upgrade your plan to continue.' }, + { statusCode: 429, message: 'Rate limit exceeded. Please try again later.' }, + ])('completes the job for a non-retryable $statusCode', async ({ statusCode, message }) => { + await expect(runRejected({ statusCode, message })).resolves.toMatchObject({ + success: false, + workflowId: 'workflow-1', + executionId: 'execution-rejected', + error: message, + }) + expect(loggingSessionMockFns.mockSafeCompleteWithError).toHaveBeenCalledWith( + expect.objectContaining({ error: expect.objectContaining({ message }) }) + ) + }) + + it.each([ + { statusCode: 429, message: 'Too many concurrent executions', retryable: true }, + { statusCode: 402, message: 'Usage is temporarily unavailable', retryable: true }, + { statusCode: 503, message: 'Usage admission is temporarily unavailable', retryable: true }, + { statusCode: 500, message: 'Internal error while fetching workflow' }, + ])('fails the job for $statusCode $message', async (error) => { + await expect(runRejected(error)).rejects.toThrow(error.message) + }) + }) + it('routes retryable reservation concurrency through bounded infrastructure backoff', async () => { mockPreprocessExecution.mockResolvedValueOnce({ success: false, diff --git a/apps/sim/background/schedule-execution.ts b/apps/sim/background/schedule-execution.ts index 02b0fdc003e..43465132a3c 100644 --- a/apps/sim/background/schedule-execution.ts +++ b/apps/sim/background/schedule-execution.ts @@ -34,11 +34,9 @@ import { preprocessExecution } from '@/lib/execution/preprocessing' import { LoggingSession } from '@/lib/logs/execution/logging-session' import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans' import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server' -import { - executeWorkflowCore, - wasExecutionFinalizedByCore, -} from '@/lib/workflows/executor/execution-core' +import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core' import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence' +import { isRecordedWorkflowFailure } from '@/lib/workflows/executor/recorded-failure' import { loadDeployedWorkflowState } from '@/lib/workflows/persistence/utils' import { notifyScheduleAutoDisabled } from '@/lib/workflows/schedules/disable-notifications' import type { ScheduleDisableReason } from '@/lib/workflows/schedules/disable-reasons' @@ -730,7 +728,7 @@ async function runWorkflowExecution({ } } - if (wasExecutionFinalizedByCore(error, executionId)) { + if (await isRecordedWorkflowFailure(error, executionId, loggingSession)) { throw error } diff --git a/apps/sim/background/webhook-execution.ts b/apps/sim/background/webhook-execution.ts index 21854a4ddf5..1e48ffb2e3d 100644 --- a/apps/sim/background/webhook-execution.ts +++ b/apps/sim/background/webhook-execution.ts @@ -74,9 +74,9 @@ import { readSlackStreamResponseConfig } from '@/lib/webhooks/slack-stream-confi import { executeWorkflowCore, type PreloadedExecutionEnvironment, - wasExecutionFinalizedByCore, } from '@/lib/workflows/executor/execution-core' import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence' +import { isRecordedWorkflowFailure } from '@/lib/workflows/executor/recorded-failure' import { loadDeployedWorkflowState, loadWorkflowDeploymentVersionState, @@ -1277,15 +1277,11 @@ async function executeWebhookJobInternal( executionId, }) - // The finalized flag is set inside a fire-and-forget post-execution promise; await it so the - // signal is reliable and the failure is fully persisted before we decide fault vs error. - await loggingSession.waitForPostExecution() - // A failure inside workflow execution (block error, provider 4xx, missing required field, etc.) // is finalized by core and already recorded in the execution logs. That is a user/workflow error, // not a trigger.dev job fault — complete the run normally so we don't fire a false alert. Errors // that were not finalized came from the webhook pipeline itself, so we re-throw to fault below. - if (wasExecutionFinalizedByCore(error, executionId)) { + if (await isRecordedWorkflowFailure(error, executionId, loggingSession)) { return { success: false, workflowId: payload.workflowId, diff --git a/apps/sim/background/workflow-execution.ts b/apps/sim/background/workflow-execution.ts index 36c7b034056..0f82f77a5ef 100644 --- a/apps/sim/background/workflow-execution.ts +++ b/apps/sim/background/workflow-execution.ts @@ -1,3 +1,4 @@ +import { trace } from '@opentelemetry/api' import { parsePrincipal, type SerializedPrincipalV1, @@ -15,6 +16,7 @@ import { assertBillingAttributionSnapshot, type BillingAttributionSnapshot, } from '@/lib/billing/core/attribution' +import { buildRecordedJobFailureResult } from '@/lib/core/async-jobs/recorded-failure' import type { AsyncExecutionCorrelation } from '@/lib/core/async-jobs/types' import { logFailureOnce } from '@/lib/core/errors/failure-log' import { @@ -27,15 +29,13 @@ import { RESERVATION_TTL_BUFFER_MS, } from '@/lib/core/execution-limits' import type { RequestAttribution } from '@/lib/core/utils/request-attribution' -import { preprocessExecution } from '@/lib/execution/preprocessing' +import { type PreprocessExecutionError, preprocessExecution } from '@/lib/execution/preprocessing' import { LoggingSession } from '@/lib/logs/execution/logging-session' import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans' import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server' -import { - executeWorkflowCore, - wasExecutionFinalizedByCore, -} from '@/lib/workflows/executor/execution-core' +import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core' import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence' +import { isRecordedWorkflowFailure } from '@/lib/workflows/executor/recorded-failure' import { WORKFLOW_EXECUTION_CONCURRENCY_LIMIT } from '@/background/concurrency-limits' import { ExecutionSnapshot } from '@/executor/execution/snapshot' import type { ExecutionMetadata } from '@/executor/execution/types' @@ -104,6 +104,21 @@ type LegacyWorkflowExecutionPayload = Omit = new Set([402, 403, 404, 429]) + +function isNonFaultPreprocessingRejection(rejection: PreprocessExecutionError): boolean { + return ( + NON_FAULT_PREPROCESSING_STATUS_CODES.has(rejection.statusCode) && rejection.retryable !== true + ) +} + function requireLegacyWorkflowJobString(value: unknown, field: string): string { if (typeof value !== 'string' || !value.trim()) { throw new Error(`Legacy workflow job ${field} must be a non-empty string`) @@ -194,6 +209,7 @@ export async function executeWorkflowJob( loggingSession.setTrustedExecutionCorrelation(correlation) } loggingSession.setExecutionDeadlineAt(getExecutionDeadlineAt(timeoutController.signal)) + let preprocessingRejection: PreprocessExecutionError | undefined try { const preprocessResult = await preprocessExecution({ @@ -217,6 +233,7 @@ export async function executeWorkflowJob( workflowId, statusCode: preprocessResult.error?.statusCode, }) + preprocessingRejection = preprocessResult.error throw new Error(preprocessResult.error?.message || 'Preprocessing failed') } @@ -314,8 +331,18 @@ export async function executeWorkflowJob( if (error instanceof ExecutionTimeoutError) throw error - if (wasExecutionFinalizedByCore(error, executionId)) { - throw error + // A failure core already recorded is the workflow's outcome, not a job fault; a run past + // its deadline still fails the job so timeouts keep alerting. + if ( + !timeoutController.isTimedOut() && + (await isRecordedWorkflowFailure(error, executionId, loggingSession)) + ) { + trace.getActiveSpan()?.recordException(toError(error)) + return buildRecordedJobFailureResult({ + workflowId, + executionId, + error: toError(error).message, + }) } const executionResult = hasExecutionResult(error) ? error.executionResult : undefined @@ -332,6 +359,15 @@ export async function executeWorkflowJob( executionState: executionResult?.executionState, }) + if (preprocessingRejection && isNonFaultPreprocessingRejection(preprocessingRejection)) { + trace.getActiveSpan()?.recordException(toError(error)) + return buildRecordedJobFailureResult({ + workflowId, + executionId, + error: toError(error).message, + }) + } + throw error } finally { void cleanupExecutionBase64Cache(executionId) diff --git a/apps/sim/lib/core/async-jobs/backends/database.test.ts b/apps/sim/lib/core/async-jobs/backends/database.test.ts index b9c7bc5773c..0046926f940 100644 --- a/apps/sim/lib/core/async-jobs/backends/database.test.ts +++ b/apps/sim/lib/core/async-jobs/backends/database.test.ts @@ -11,6 +11,7 @@ vi.mock('@/lib/core/execution-limits/metrics', () => ({ })) import { DatabaseJobQueue } from '@/lib/core/async-jobs/backends/database' +import { buildRecordedJobFailureResult } from '@/lib/core/async-jobs/recorded-failure' import { AsyncJobEnqueueError, MAX_JOB_DURATION_SECONDS } from '@/lib/core/async-jobs/types' const EXISTING_JOB = { @@ -450,3 +451,55 @@ describe('DatabaseJobQueue cancellation', () => { }) }) }) + +describe('DatabaseJobQueue a workflow job that completed with a recorded failure', () => { + beforeEach(() => { + resetDbChainMock() + }) + + const settledJob = { + ...EXISTING_JOB, + startedAt: new Date('2026-07-10T00:00:01.000Z'), + completedAt: new Date('2026-07-10T00:00:04.000Z'), + attempts: 1, + } + + it('reads exactly as the row markJobFailed wrote before the runner stopped throwing', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { ...settledJob, status: 'failed', error: 'No start block found', output: null }, + ]) + dbChainMockFns.limit.mockResolvedValueOnce([ + { + ...settledJob, + status: 'completed', + error: null, + output: buildRecordedJobFailureResult({ + workflowId: 'workflow-1', + executionId: 'execution-1', + error: 'No start block found', + }), + }, + ]) + const queue = new DatabaseJobQueue() + + const failedJob = await queue.getJob('workflow:1') + const recordedFailure = await queue.getJob('workflow:1') + + expect(failedJob).toMatchObject({ status: 'failed', error: 'No start block found' }) + expect(recordedFailure).toStrictEqual(failedJob) + }) + + it('leaves a completed webhook job that reported an unsuccessful execution completed', async () => { + const output = { success: false, workflowId: 'workflow-1', executionId: 'execution-1' } + dbChainMockFns.limit.mockResolvedValueOnce([ + { ...settledJob, type: 'webhook-execution', status: 'completed', output }, + ]) + const queue = new DatabaseJobQueue() + + await expect(queue.getJob('workflow:1')).resolves.toMatchObject({ + status: 'completed', + error: undefined, + output, + }) + }) +}) diff --git a/apps/sim/lib/core/async-jobs/backends/database.ts b/apps/sim/lib/core/async-jobs/backends/database.ts index 6387c8fa987..05c6b9bb01f 100644 --- a/apps/sim/lib/core/async-jobs/backends/database.ts +++ b/apps/sim/lib/core/async-jobs/backends/database.ts @@ -4,6 +4,7 @@ import { toError } from '@sim/utils/errors' import { generateShortId } from '@sim/utils/id' import { omit } from '@sim/utils/object' import { and, eq, inArray, or, sql } from 'drizzle-orm' +import { getRecordedJobFailureError } from '@/lib/core/async-jobs/recorded-failure' import { AsyncJobEnqueueError, type EnqueueOptions, @@ -26,18 +27,23 @@ type AsyncJobRow = typeof asyncJobs.$inferSelect type Runner = NonNullable function rowToJob(row: AsyncJobRow): Job { + const recordedFailure = + row.status === JOB_STATUS.COMPLETED + ? getRecordedJobFailureError(row.type, row.output) + : undefined return { id: row.id, type: row.type as JobType, payload: row.payload, - status: row.status as JobStatus, + status: recordedFailure === undefined ? (row.status as JobStatus) : JOB_STATUS.FAILED, createdAt: row.createdAt, startedAt: row.startedAt ?? undefined, completedAt: row.completedAt ?? undefined, attempts: row.attempts, maxAttempts: row.maxAttempts, - error: row.error ?? undefined, - output: row.output as unknown, + error: recordedFailure ?? row.error ?? undefined, + // markJobFailed never writes an output, so a failed row reads back null. + output: recordedFailure === undefined ? (row.output as unknown) : null, metadata: (row.metadata ?? {}) as JobMetadata, } } diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts index 98a5d982be7..92df6bc14f0 100644 --- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts +++ b/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts @@ -22,6 +22,7 @@ vi.mock('@/lib/core/execution-limits/metrics', () => ({ vi.mock('@/lib/core/async-jobs/region', () => asyncJobsRegionMock) import { TriggerDevJobQueue } from '@/lib/core/async-jobs/backends/trigger-dev' +import { buildRecordedJobFailureResult } from '@/lib/core/async-jobs/recorded-failure' import { AsyncJobEnqueueError, JOB_PENDING_RETENTION_HOURS } from '@/lib/core/async-jobs/types' const { @@ -282,6 +283,80 @@ describe('TriggerDevJobQueue status mapping', () => { }) }) + describe('a workflow run that completed with a recorded failure', () => { + const runTimes = { + createdAt: new Date('2026-08-05T12:00:00.000Z'), + startedAt: new Date('2026-08-05T12:00:01.000Z'), + finishedAt: new Date('2026-08-05T12:00:04.000Z'), + updatedAt: new Date('2026-08-05T12:00:05.000Z'), + } + const payload = { workflowId: 'workflow-1', userId: 'user-1' } + + it('reads exactly as the failed run it was before the runner stopped throwing', async () => { + mockRetrieve.mockResolvedValueOnce({ + id: 'run_1', + payload, + status: 'FAILED', + taskIdentifier: 'workflow-execution', + error: { message: 'No start block found' }, + ...runTimes, + }) + mockRetrieve.mockResolvedValueOnce({ + id: 'run_1', + payload, + status: 'COMPLETED', + taskIdentifier: 'workflow-execution', + output: buildRecordedJobFailureResult({ + workflowId: 'workflow-1', + executionId: 'execution-1', + error: 'No start block found', + }), + ...runTimes, + }) + const queue = new TriggerDevJobQueue() + + const failedRun = await queue.getJob('run_1') + const recordedFailure = await queue.getJob('run_1') + + expect(failedRun).toMatchObject({ status: 'failed', error: 'No start block found' }) + expect(recordedFailure).toStrictEqual(failedRun) + }) + + it.each([ + { + name: 'a webhook run that reported an unsuccessful execution', + taskIdentifier: 'webhook-execution', + output: { success: false, workflowId: 'workflow-1', executionId: 'execution-1' }, + }, + { + name: 'a workflow run whose execution was cancelled', + taskIdentifier: 'workflow-execution', + output: { success: false, workflowId: 'workflow-1', executionId: 'execution-1' }, + }, + { + name: 'a run of another task whose output carries an error', + taskIdentifier: 'schedule-execution', + output: { success: false, error: 'boom' }, + }, + ])('leaves $name completed', async ({ taskIdentifier, output }) => { + mockRetrieve.mockResolvedValueOnce({ + id: 'run_1', + payload, + status: 'COMPLETED', + taskIdentifier, + output, + ...runTimes, + }) + const queue = new TriggerDevJobQueue() + + await expect(queue.getJob('run_1')).resolves.toMatchObject({ + status: 'completed', + error: undefined, + output, + }) + }) + }) + /** Trigger.dev can only retrieve its own run ids; anything else fails, and slowly. */ function retrieveOnlyRunIds() { mockRetrieve.mockImplementation(async (id: string) => { diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts index 59dd758ef32..3ef4eaf5a61 100644 --- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts +++ b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts @@ -7,6 +7,7 @@ import { isRecordLike } from '@sim/utils/object' import { taskContext } from '@trigger.dev/core/v3' import { ApiError, runs, type TriggerOptions, tasks } from '@trigger.dev/sdk' import { eq } from 'drizzle-orm' +import { getRecordedJobFailureError } from '@/lib/core/async-jobs/recorded-failure' import { resolveTriggerRegion } from '@/lib/core/async-jobs/region' import { AsyncJobEnqueueError, @@ -482,7 +483,12 @@ export class TriggerDevJobQueue implements JobQueueBackend { : undefined, } - const status = mapTriggerDevStatus(run.status) + const runStatus = mapTriggerDevStatus(run.status) + const recordedFailure = + runStatus === JOB_STATUS.COMPLETED + ? getRecordedJobFailureError(run.taskIdentifier, run.output) + : undefined + const status = recordedFailure === undefined ? runStatus : JOB_STATUS.FAILED return { id: run.id, @@ -494,8 +500,8 @@ export class TriggerDevJobQueue implements JobQueueBackend { completedAt: resolveRunCompletedAt(run.finishedAt, run.updatedAt, status), attempts: run.attemptCount ?? 1, maxAttempts: 3, - error: run.error?.message, - output: run.output as unknown, + error: recordedFailure ?? run.error?.message, + output: recordedFailure === undefined ? (run.output as unknown) : undefined, metadata, } } catch (error) { diff --git a/apps/sim/lib/core/async-jobs/recorded-failure.ts b/apps/sim/lib/core/async-jobs/recorded-failure.ts new file mode 100644 index 00000000000..285fd8cb0a3 --- /dev/null +++ b/apps/sim/lib/core/async-jobs/recorded-failure.ts @@ -0,0 +1,51 @@ +import { isRecordLike } from '@sim/utils/object' +import { truncateAtCodePoint } from '@sim/utils/string' +import type { JobType } from '@/lib/core/async-jobs/types' + +/** + * Job types whose runner reports a failure the run already recorded in its + * execution log by completing with a {@link RecordedJobFailureResult} instead + * of throwing, so the queue does not fault (and alert on) a user or workflow + * error. + */ +const RECORDED_FAILURE_JOB_TYPES: ReadonlySet = new Set(['workflow-execution']) + +/** + * Keeps the result far below Trigger.dev's output offload threshold: an + * offloaded output is not returned inline by `runs.retrieve`, so the job would + * read as completed instead of failed. + */ +const RECORDED_FAILURE_ERROR_MAX_LENGTH = 64_000 + +interface RecordedJobFailureResult { + success: false + workflowId: string + executionId: string + error: string + executedAt: string +} + +export function buildRecordedJobFailureResult(params: { + workflowId: string + executionId: string + error: string +}): RecordedJobFailureResult { + return { + success: false, + workflowId: params.workflowId, + executionId: params.executionId, + error: truncateAtCodePoint(params.error, RECORDED_FAILURE_ERROR_MAX_LENGTH), + executedAt: new Date().toISOString(), + } +} + +/** + * The error of a completed job whose output is a recorded failure, or + * `undefined` for any other job. Queue read paths project such a job as failed + * with this error and no output, so pollers see exactly what they saw when the + * runner threw. + */ +export function getRecordedJobFailureError(type: string, output: unknown): string | undefined { + if (!RECORDED_FAILURE_JOB_TYPES.has(type) || !isRecordLike(output)) return undefined + return output.success === false && typeof output.error === 'string' ? output.error : undefined +} diff --git a/apps/sim/lib/workflows/executor/recorded-failure.ts b/apps/sim/lib/workflows/executor/recorded-failure.ts new file mode 100644 index 00000000000..648b188d5eb --- /dev/null +++ b/apps/sim/lib/workflows/executor/recorded-failure.ts @@ -0,0 +1,19 @@ +import type { LoggingSession } from '@/lib/logs/execution/logging-session' +import { wasExecutionFinalizedByCore } from '@/lib/workflows/executor/execution-core' + +/** + * Whether a failure thrown out of `executeWorkflowCore` is already recorded as + * the run's failed execution log — a user or workflow error, not a job fault. + * + * Core records the failure and marks it finalized inside a post-execution + * promise it does not await before rethrowing, so this settles that promise + * first. Checking the mark without it misses nearly every recorded failure. + */ +export async function isRecordedWorkflowFailure( + error: unknown, + executionId: string, + loggingSession: Pick +): Promise { + await loggingSession.waitForPostExecution() + return wasExecutionFinalizedByCore(error, executionId) +} From d4d94856f6e28fa00a1fe6de726978b7c9931a13 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 14:51:48 -0700 Subject: [PATCH 2/5] fix(trigger): bound recorded workflow failure errors in UTF-8 bytes The previous character cap did not bound bytes against Trigger.dev's 128 KiB output offload threshold. Cap the error at 16 KiB of UTF-8, cut at a code point boundary, so even sixfold JSON escaping stays inline; normal-size messages are returned unchanged. Correct the comment: runs.retrieve resolves offloaded outputs itself, and only a failed presigned fetch leaves the output missing. Drop an unconsumed waitForPostExecution mock from the timeout test. --- .../async-preprocessing-correlation.test.ts | 2 +- .../lib/core/async-jobs/recorded-failure.ts | 22 ++++++++++++++----- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/apps/sim/background/async-preprocessing-correlation.test.ts b/apps/sim/background/async-preprocessing-correlation.test.ts index a9153d63d1b..5452e02b00c 100644 --- a/apps/sim/background/async-preprocessing-correlation.test.ts +++ b/apps/sim/background/async-preprocessing-correlation.test.ts @@ -418,7 +418,7 @@ describe('async preprocessing correlation threading', () => { const error = recordedError() mockExecuteWorkflowCore.mockRejectedValueOnce(error) mockIsWorkflowTimedOut.mockReturnValue(true) - finalizeWhenPostExecutionSettles() + mockWasExecutionFinalizedByCore.mockReturnValue(true) await expect(runRecordedFailure('execution-recorded-late')).rejects.toBe(error) }) diff --git a/apps/sim/lib/core/async-jobs/recorded-failure.ts b/apps/sim/lib/core/async-jobs/recorded-failure.ts index 285fd8cb0a3..5f80b7e8fe3 100644 --- a/apps/sim/lib/core/async-jobs/recorded-failure.ts +++ b/apps/sim/lib/core/async-jobs/recorded-failure.ts @@ -1,5 +1,4 @@ import { isRecordLike } from '@sim/utils/object' -import { truncateAtCodePoint } from '@sim/utils/string' import type { JobType } from '@/lib/core/async-jobs/types' /** @@ -11,11 +10,22 @@ import type { JobType } from '@/lib/core/async-jobs/types' const RECORDED_FAILURE_JOB_TYPES: ReadonlySet = new Set(['workflow-execution']) /** - * Keeps the result far below Trigger.dev's output offload threshold: an - * offloaded output is not returned inline by `runs.retrieve`, so the job would - * read as completed instead of failed. + * Bounds the error in UTF-8 bytes so the result stays inline under Trigger.dev's + * 128 KiB output offload threshold even if JSON escaping inflates every byte + * sixfold (`\u0000`). `runs.retrieve` resolves an offloaded output through + * its presigned URL, but if that fetch fails the output reads as missing and + * the job would read as completed instead of failed. */ -const RECORDED_FAILURE_ERROR_MAX_LENGTH = 64_000 +const RECORDED_FAILURE_ERROR_MAX_BYTES = 16 * 1024 + +/** Cuts at a code point boundary so the kept prefix is valid UTF-8. */ +function truncateUtf8(value: string, maxBytes: number): string { + const bytes = Buffer.from(value, 'utf8') + if (bytes.length <= maxBytes) return value + let end = maxBytes + while (end > 0 && (bytes[end] & 0xc0) === 0x80) end-- + return `${bytes.subarray(0, end).toString('utf8')}...` +} interface RecordedJobFailureResult { success: false @@ -34,7 +44,7 @@ export function buildRecordedJobFailureResult(params: { success: false, workflowId: params.workflowId, executionId: params.executionId, - error: truncateAtCodePoint(params.error, RECORDED_FAILURE_ERROR_MAX_LENGTH), + error: truncateUtf8(params.error, RECORDED_FAILURE_ERROR_MAX_BYTES), executedAt: new Date().toISOString(), } } From d913d6a8ac6d7564ac27d2f012942495924bf7ce Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 15:01:48 -0700 Subject: [PATCH 3/5] chore(audits): shrink the unused-exports baseline for PreprocessExecutionError --- scripts/check-unused-exports.baseline.json | 1 - 1 file changed, 1 deletion(-) diff --git a/scripts/check-unused-exports.baseline.json b/scripts/check-unused-exports.baseline.json index 0dcd68ae086..740d330fa97 100644 --- a/scripts/check-unused-exports.baseline.json +++ b/scripts/check-unused-exports.baseline.json @@ -4367,7 +4367,6 @@ "apps/sim/lib/execution/payloads/large-array-manifest.ts#LargeArrayManifestChunk", "apps/sim/lib/execution/payloads/sandbox-file-mount-ref.ts#SANDBOX_FILE_MOUNT_REF_MARKER", "apps/sim/lib/execution/payloads/sandbox-file-mount-ref.ts#SANDBOX_FILE_MOUNT_REF_VERSION", - "apps/sim/lib/execution/preprocessing.ts#PreprocessExecutionError", "apps/sim/lib/execution/preprocessing.ts#PreprocessExecutionFailure", "apps/sim/lib/execution/remote-sandbox/build-errors.ts#BUILD_LOG_TAIL_CHARS", "apps/sim/lib/execution/remote-sandbox/build-errors.ts#SandboxBuildErrorCode", From 6431fc60c80ec14c17a5547a364b8a2edf6d6d12 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 15:24:04 -0700 Subject: [PATCH 4/5] fix(trigger): keep alerting on deadline-passed and unlogged async workflow failures Re-check the deadline after awaiting core's finalization, since that wait can outlast it, so a timed-out run still fails its job. Complete a non-fault preprocessing rejection only when the logging session confirms the failure row was written (hasCompleted, the same signal core uses to mark a failure finalized); otherwise throw as before so an unsaved rejection still alerts. --- .../async-preprocessing-correlation.test.ts | 23 +++++++++++++++++++ apps/sim/background/workflow-execution.ts | 13 +++++++---- 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/apps/sim/background/async-preprocessing-correlation.test.ts b/apps/sim/background/async-preprocessing-correlation.test.ts index 5452e02b00c..a285ba0e502 100644 --- a/apps/sim/background/async-preprocessing-correlation.test.ts +++ b/apps/sim/background/async-preprocessing-correlation.test.ts @@ -414,6 +414,20 @@ describe('async preprocessing correlation threading', () => { expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled() }) + it('still fails the job when the deadline passes while finalization settles', async () => { + const error = recordedError() + mockExecuteWorkflowCore.mockRejectedValueOnce(error) + let timedOut = false + mockIsWorkflowTimedOut.mockImplementation(() => timedOut) + loggingSessionMockFns.mockWaitForPostExecution.mockImplementationOnce(async () => { + await Promise.resolve() + timedOut = true + }) + mockWasExecutionFinalizedByCore.mockReturnValue(true) + + await expect(runRecordedFailure('execution-recorded-deadline')).rejects.toBe(error) + }) + it('still fails the job when the run passed its deadline', async () => { const error = recordedError() mockExecuteWorkflowCore.mockRejectedValueOnce(error) @@ -573,6 +587,7 @@ describe('async preprocessing correlation threading', () => { { statusCode: 402, message: 'Usage limit exceeded. Please upgrade your plan to continue.' }, { statusCode: 429, message: 'Rate limit exceeded. Please try again later.' }, ])('completes the job for a non-retryable $statusCode', async ({ statusCode, message }) => { + loggingSessionMockFns.mockHasCompleted.mockReturnValueOnce(true) await expect(runRejected({ statusCode, message })).resolves.toMatchObject({ success: false, workflowId: 'workflow-1', @@ -584,6 +599,14 @@ describe('async preprocessing correlation threading', () => { ) }) + it('fails the job when the rejection could not be logged', async () => { + loggingSessionMockFns.mockHasCompleted.mockReturnValueOnce(false) + + await expect( + runRejected({ statusCode: 403, message: 'Workflow is not deployed' }) + ).rejects.toThrow('Workflow is not deployed') + }) + it.each([ { statusCode: 429, message: 'Too many concurrent executions', retryable: true }, { statusCode: 402, message: 'Usage is temporarily unavailable', retryable: true }, diff --git a/apps/sim/background/workflow-execution.ts b/apps/sim/background/workflow-execution.ts index 0f82f77a5ef..a1a77bdecae 100644 --- a/apps/sim/background/workflow-execution.ts +++ b/apps/sim/background/workflow-execution.ts @@ -332,10 +332,10 @@ export async function executeWorkflowJob( if (error instanceof ExecutionTimeoutError) throw error // A failure core already recorded is the workflow's outcome, not a job fault; a run past - // its deadline still fails the job so timeouts keep alerting. + // its deadline (checked after the await, which can outlast it) still fails the job. if ( - !timeoutController.isTimedOut() && - (await isRecordedWorkflowFailure(error, executionId, loggingSession)) + (await isRecordedWorkflowFailure(error, executionId, loggingSession)) && + !timeoutController.isTimedOut() ) { trace.getActiveSpan()?.recordException(toError(error)) return buildRecordedJobFailureResult({ @@ -359,7 +359,12 @@ export async function executeWorkflowJob( executionState: executionResult?.executionState, }) - if (preprocessingRejection && isNonFaultPreprocessingRejection(preprocessingRejection)) { + // Complete only when the rejection is durably logged; an unsaved one must still alert. + if ( + preprocessingRejection && + isNonFaultPreprocessingRejection(preprocessingRejection) && + loggingSession.hasCompleted() + ) { trace.getActiveSpan()?.recordException(toError(error)) return buildRecordedJobFailureResult({ workflowId, From d5e6319c5f5b16d714727ef76cd145d46ed1fc1b Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 15:39:10 -0700 Subject: [PATCH 5/5] refactor(trigger): share the recorded-failure completion and skip the redundant error write on a timed-out recorded failure --- .../async-preprocessing-correlation.test.ts | 2 ++ apps/sim/background/workflow-execution.ts | 27 ++++++++----------- .../lib/core/async-jobs/recorded-failure.ts | 7 ++--- 3 files changed, 17 insertions(+), 19 deletions(-) diff --git a/apps/sim/background/async-preprocessing-correlation.test.ts b/apps/sim/background/async-preprocessing-correlation.test.ts index a285ba0e502..453d423716d 100644 --- a/apps/sim/background/async-preprocessing-correlation.test.ts +++ b/apps/sim/background/async-preprocessing-correlation.test.ts @@ -426,6 +426,7 @@ describe('async preprocessing correlation threading', () => { mockWasExecutionFinalizedByCore.mockReturnValue(true) await expect(runRecordedFailure('execution-recorded-deadline')).rejects.toBe(error) + expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled() }) it('still fails the job when the run passed its deadline', async () => { @@ -435,6 +436,7 @@ describe('async preprocessing correlation threading', () => { mockWasExecutionFinalizedByCore.mockReturnValue(true) await expect(runRecordedFailure('execution-recorded-late')).rejects.toBe(error) + expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled() }) }) diff --git a/apps/sim/background/workflow-execution.ts b/apps/sim/background/workflow-execution.ts index a1a77bdecae..b6d75ebbb2b 100644 --- a/apps/sim/background/workflow-execution.ts +++ b/apps/sim/background/workflow-execution.ts @@ -119,6 +119,13 @@ function isNonFaultPreprocessingRejection(rejection: PreprocessExecutionError): ) } +/** Completes the job with a failure already in the execution log, keeping it on the run's span. */ +function completeWithRecordedFailure(workflowId: string, executionId: string, error: unknown) { + const failure = toError(error) + trace.getActiveSpan()?.recordException(failure) + return buildRecordedJobFailureResult({ workflowId, executionId, error: failure.message }) +} + function requireLegacyWorkflowJobString(value: unknown, field: string): string { if (typeof value !== 'string' || !value.trim()) { throw new Error(`Legacy workflow job ${field} must be a non-empty string`) @@ -333,16 +340,9 @@ export async function executeWorkflowJob( // A failure core already recorded is the workflow's outcome, not a job fault; a run past // its deadline (checked after the await, which can outlast it) still fails the job. - if ( - (await isRecordedWorkflowFailure(error, executionId, loggingSession)) && - !timeoutController.isTimedOut() - ) { - trace.getActiveSpan()?.recordException(toError(error)) - return buildRecordedJobFailureResult({ - workflowId, - executionId, - error: toError(error).message, - }) + if (await isRecordedWorkflowFailure(error, executionId, loggingSession)) { + if (timeoutController.isTimedOut()) throw error + return completeWithRecordedFailure(workflowId, executionId, error) } const executionResult = hasExecutionResult(error) ? error.executionResult : undefined @@ -365,12 +365,7 @@ export async function executeWorkflowJob( isNonFaultPreprocessingRejection(preprocessingRejection) && loggingSession.hasCompleted() ) { - trace.getActiveSpan()?.recordException(toError(error)) - return buildRecordedJobFailureResult({ - workflowId, - executionId, - error: toError(error).message, - }) + return completeWithRecordedFailure(workflowId, executionId, error) } throw error diff --git a/apps/sim/lib/core/async-jobs/recorded-failure.ts b/apps/sim/lib/core/async-jobs/recorded-failure.ts index 5f80b7e8fe3..8406a5eb6ee 100644 --- a/apps/sim/lib/core/async-jobs/recorded-failure.ts +++ b/apps/sim/lib/core/async-jobs/recorded-failure.ts @@ -2,12 +2,12 @@ import { isRecordLike } from '@sim/utils/object' import type { JobType } from '@/lib/core/async-jobs/types' /** - * Job types whose runner reports a failure the run already recorded in its + * The job type whose runner reports a failure the run already recorded in its * execution log by completing with a {@link RecordedJobFailureResult} instead * of throwing, so the queue does not fault (and alert on) a user or workflow * error. */ -const RECORDED_FAILURE_JOB_TYPES: ReadonlySet = new Set(['workflow-execution']) +const RECORDED_FAILURE_JOB_TYPE: JobType = 'workflow-execution' /** * Bounds the error in UTF-8 bytes so the result stays inline under Trigger.dev's @@ -35,6 +35,7 @@ interface RecordedJobFailureResult { executedAt: string } +/** The output a runner completes with when its failure is already in the execution log. */ export function buildRecordedJobFailureResult(params: { workflowId: string executionId: string @@ -56,6 +57,6 @@ export function buildRecordedJobFailureResult(params: { * runner threw. */ export function getRecordedJobFailureError(type: string, output: unknown): string | undefined { - if (!RECORDED_FAILURE_JOB_TYPES.has(type) || !isRecordLike(output)) return undefined + if (type !== RECORDED_FAILURE_JOB_TYPE || !isRecordLike(output)) return undefined return output.success === false && typeof output.error === 'string' ? output.error : undefined }