fix(trigger): complete async workflow runs whose failure is already r… · simstudioai/sim@7561e0f · GitHub
Skip to content

Commit 7561e0f

Browse files
waleedlatif1waleedlatif1
authored
fix(trigger): complete async workflow runs whose failure is already recorded instead of failing them (#9139)
* 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. * 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. * chore(audits): shrink the unused-exports baseline for PreprocessExecutionError * 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. * refactor(trigger): share the recorded-failure completion and skip the redundant error write on a timed-out recorded failure
1 parent 854dc6a commit 7561e0f

11 files changed

Lines changed: 397 additions & 56 deletions

‎apps/sim/background/async-preprocessing-correlation.test.ts‎

Lines changed: 122 additions & 31 deletions

‎apps/sim/background/schedule-execution.ts‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -34,11 +34,9 @@ import { preprocessExecution } from '@/lib/execution/preprocessing'
3434
import { LoggingSession } from '@/lib/logs/execution/logging-session'
3535
import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
3636
import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server'
37-
import {
38-
executeWorkflowCore,
39-
wasExecutionFinalizedByCore,
40-
} from '@/lib/workflows/executor/execution-core'
37+
import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
4138
import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence'
39+
import { isRecordedWorkflowFailure } from '@/lib/workflows/executor/recorded-failure'
4240
import { loadDeployedWorkflowState } from '@/lib/workflows/persistence/utils'
4341
import { notifyScheduleAutoDisabled } from '@/lib/workflows/schedules/disable-notifications'
4442
import type { ScheduleDisableReason } from '@/lib/workflows/schedules/disable-reasons'
@@ -730,7 +728,7 @@ async function runWorkflowExecution({
730728
}
731729
}
732730

733-
if (wasExecutionFinalizedByCore(error, executionId)) {
731+
if (await isRecordedWorkflowFailure(error, executionId, loggingSession)) {
734732
throw error
735733
}
736734

‎apps/sim/background/webhook-execution.ts‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -74,9 +74,9 @@ import { readSlackStreamResponseConfig } from '@/lib/webhooks/slack-stream-confi
7474
import {
7575
executeWorkflowCore,
7676
type PreloadedExecutionEnvironment,
77-
wasExecutionFinalizedByCore,
7877
} from '@/lib/workflows/executor/execution-core'
7978
import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence'
79+
import { isRecordedWorkflowFailure } from '@/lib/workflows/executor/recorded-failure'
8080
import {
8181
loadDeployedWorkflowState,
8282
loadWorkflowDeploymentVersionState,
@@ -1277,15 +1277,11 @@ async function executeWebhookJobInternal(
12771277
executionId,
12781278
})
12791279

1280-
// The finalized flag is set inside a fire-and-forget post-execution promise; await it so the
1281-
// signal is reliable and the failure is fully persisted before we decide fault vs error.
1282-
await loggingSession.waitForPostExecution()
1283-
12841280
// A failure inside workflow execution (block error, provider 4xx, missing required field, etc.)
12851281
// is finalized by core and already recorded in the execution logs. That is a user/workflow error,
12861282
// not a trigger.dev job fault — complete the run normally so we don't fire a false alert. Errors
12871283
// that were not finalized came from the webhook pipeline itself, so we re-throw to fault below.
1288-
if (wasExecutionFinalizedByCore(error, executionId)) {
1284+
if (await isRecordedWorkflowFailure(error, executionId, loggingSession)) {
12891285
return {
12901286
success: false,
12911287
workflowId: payload.workflowId,

‎apps/sim/background/workflow-execution.ts‎

Lines changed: 43 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import { trace } from '@opentelemetry/api'
12
import {
23
parsePrincipal,
34
type SerializedPrincipalV1,
@@ -15,6 +16,7 @@ import {
1516
assertBillingAttributionSnapshot,
1617
type BillingAttributionSnapshot,
1718
} from '@/lib/billing/core/attribution'
19+
import { buildRecordedJobFailureResult } from '@/lib/core/async-jobs/recorded-failure'
1820
import type { AsyncExecutionCorrelation } from '@/lib/core/async-jobs/types'
1921
import { logFailureOnce } from '@/lib/core/errors/failure-log'
2022
import {
@@ -27,15 +29,13 @@ import {
2729
RESERVATION_TTL_BUFFER_MS,
2830
} from '@/lib/core/execution-limits'
2931
import type { RequestAttribution } from '@/lib/core/utils/request-attribution'
30-
import { preprocessExecution } from '@/lib/execution/preprocessing'
32+
import { type PreprocessExecutionError, preprocessExecution } from '@/lib/execution/preprocessing'
3133
import { LoggingSession } from '@/lib/logs/execution/logging-session'
3234
import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
3335
import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server'
34-
import {
35-
executeWorkflowCore,
36-
wasExecutionFinalizedByCore,
37-
} from '@/lib/workflows/executor/execution-core'
36+
import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
3837
import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence'
38+
import { isRecordedWorkflowFailure } from '@/lib/workflows/executor/recorded-failure'
3939
import { WORKFLOW_EXECUTION_CONCURRENCY_LIMIT } from '@/background/concurrency-limits'
4040
import { ExecutionSnapshot } from '@/executor/execution/snapshot'
4141
import type { ExecutionMetadata } from '@/executor/execution/types'
@@ -104,6 +104,28 @@ type LegacyWorkflowExecutionPayload = Omit<WorkflowExecutionPayload, 'principal'
104104

105105
type WorkflowExecutionJobPayload = WorkflowExecutionPayload | LegacyWorkflowExecutionPayload
106106

107+
/**
108+
* Preprocessing rejections that are an outcome of the workflow's or account's
109+
* state — deleted or undeployed workflow, suspended account, exhausted usage
110+
* or rate quota — rather than a fault. A retryable rejection (concurrency
111+
* capacity, an unreadable usage ledger, reservation infrastructure) and every
112+
* 5xx still fail the job.
113+
*/
114+
const NON_FAULT_PREPROCESSING_STATUS_CODES: ReadonlySet<number> = new Set([402, 403, 404, 429])
115+
116+
function isNonFaultPreprocessingRejection(rejection: PreprocessExecutionError): boolean {
117+
return (
118+
NON_FAULT_PREPROCESSING_STATUS_CODES.has(rejection.statusCode) && rejection.retryable !== true
119+
)
120+
}
121+
122+
/** Completes the job with a failure already in the execution log, keeping it on the run's span. */
123+
function completeWithRecordedFailure(workflowId: string, executionId: string, error: unknown) {
124+
const failure = toError(error)
125+
trace.getActiveSpan()?.recordException(failure)
126+
return buildRecordedJobFailureResult({ workflowId, executionId, error: failure.message })
127+
}
128+
107129
function requireLegacyWorkflowJobString(value: unknown, field: string): string {
108130
if (typeof value !== 'string' || !value.trim()) {
109131
throw new Error(`Legacy workflow job ${field} must be a non-empty string`)
@@ -194,6 +216,7 @@ export async function executeWorkflowJob(
194216
loggingSession.setTrustedExecutionCorrelation(correlation)
195217
}
196218
loggingSession.setExecutionDeadlineAt(getExecutionDeadlineAt(timeoutController.signal))
219+
let preprocessingRejection: PreprocessExecutionError | undefined
197220

198221
try {
199222
const preprocessResult = await preprocessExecution({
@@ -217,6 +240,7 @@ export async function executeWorkflowJob(
217240
workflowId,
218241
statusCode: preprocessResult.error?.statusCode,
219242
})
243+
preprocessingRejection = preprocessResult.error
220244

221245
throw new Error(preprocessResult.error?.message || 'Preprocessing failed')
222246
}
@@ -314,8 +338,11 @@ export async function executeWorkflowJob(
314338

315339
if (error instanceof ExecutionTimeoutError) throw error
316340

317-
if (wasExecutionFinalizedByCore(error, executionId)) {
318-
throw error
341+
// A failure core already recorded is the workflow's outcome, not a job fault; a run past
342+
// its deadline (checked after the await, which can outlast it) still fails the job.
343+
if (await isRecordedWorkflowFailure(error, executionId, loggingSession)) {
344+
if (timeoutController.isTimedOut()) throw error
345+
return completeWithRecordedFailure(workflowId, executionId, error)
319346
}
320347

321348
const executionResult = hasExecutionResult(error) ? error.executionResult : undefined
@@ -332,6 +359,15 @@ export async function executeWorkflowJob(
332359
executionState: executionResult?.executionState,
333360
})
334361

362+
// Complete only when the rejection is durably logged; an unsaved one must still alert.
363+
if (
364+
preprocessingRejection &&
365+
isNonFaultPreprocessingRejection(preprocessingRejection) &&
366+
loggingSession.hasCompleted()
367+
) {
368+
return completeWithRecordedFailure(workflowId, executionId, error)
369+
}
370+
335371
throw error
336372
} finally {
337373
void cleanupExecutionBase64Cache(executionId)

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

Lines changed: 53 additions & 0 deletions

0 commit comments

Comments
 (0)