diff --git a/.changeset/step-maxretries-timeout.md b/.changeset/step-maxretries-timeout.md new file mode 100644 index 0000000000..38916d5546 --- /dev/null +++ b/.changeset/step-maxretries-timeout.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Enforce `maxRetries` for inline and backgrounded steps that time out diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index bb0002b8e0..cdc2282da5 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -61,7 +61,11 @@ import { ReplayBudget, } from './runtime/replay-budget.js'; import { runIdCreatedAt } from './runtime/run-id-time.js'; -import { executeStep } from './runtime/step-executor.js'; +import { + DEFAULT_STEP_MAX_RETRIES, + executeStep, +} from './runtime/step-executor.js'; +import { getStepFunction } from './private.js'; import { computeStepLatencyTracking } from './runtime/step-latency.js'; import { backstopIdempotencyKey, @@ -246,6 +250,29 @@ function hasRecordedTerminalRunEvent(events: Event[], runId: string): boolean { return true; } +/** + * Number of `step_started` events already recorded for a step, used as the + * authoritative attempt count for the inline retry ceiling. Each real attempt + * writes exactly one `step_started` (the atomic create-claim / single-flight + * prevents concurrent double-starts from inflating this), so the count equals + * the number of attempts that have begun. + */ +function countStepStartedEvents( + events: Event[] | null | undefined, + stepId: string +): number { + if (!events) { + return 0; + } + let count = 0; + for (const e of events) { + if (e.eventType === 'step_started' && e.correlationId === stepId) { + count++; + } + } + return count; +} + /** * The lineage root of a loaded run: its `$rootRunId` attribute, or its own id * when it is itself a root. @@ -720,6 +747,38 @@ export function workflowEntrypoint( const bgStartedAt = bgRun.startedAt ? +bgRun.startedAt : Date.now(); + + // Retry ceiling for a backgrounded step. `metadata.attempt` + // (the queue delivery count) is a cheap upper bound, but it + // over-counts: a ThrottleError / TooEarlyError — or any + // redelivery that never ran the body — still advances it, so + // trusting it directly could fail a step as "exceeded max + // retries" before the body ever ran (a user-visible + // regression under transient backend pressure). Use it only + // as a fast gate: while it is at or under the ceiling the + // step cannot be exhausted, so proceed without touching the + // log. Only once it crosses the ceiling do we load the full + // event log and derive the authoritative attempt from the + // recorded `step_started` count — the count only real + // attempts write, so throttle/too-early redeliveries are + // excluded. This still bounds timeouts, which write no error + // for the post-body guard to catch. The load also primes the + // replay's `cachedEvents`/`eventsCursor` (the post-step + // continuation below refreshes them once the step's terminal + // event lands). + let bgAuthoritativeAttempt = metadata.attempt; + const bgMaxRetries = + getStepFunction(incomingStepName)?.maxRetries ?? + DEFAULT_STEP_MAX_RETRIES; + if (metadata.attempt > bgMaxRetries + 1) { + const loaded = await loadWorkflowRunEvents(runId); + cachedEvents = loaded.events; + eventsCursor = loaded.cursor; + bgAuthoritativeAttempt = + countStepStartedEvents(cachedEvents, incomingStepId) + + 1; + } + // Pause the replay budget while the step body runs — // step duration is bounded by the platform's function // maxDuration, not by the replay timeout. See the @@ -751,6 +810,10 @@ export function workflowEntrypoint( stepId: incomingStepId, stepName: incomingStepName, runSpecVersion: bgRun.specVersion, + // Retry ceiling: the queue delivery count as a fast + // gate, verified against the recorded step_started + // count once it crosses the ceiling (see above). + authoritativeAttempt: bgAuthoritativeAttempt, }) ); } finally { @@ -2242,6 +2305,25 @@ export function workflowEntrypoint( stepId: s.correlationId, stepName: s.stepName, runSpecVersion: workflowRun.specVersion, + // Attempt number = prior step_started count + 1 + // (this execution's start). A lazy step is + // brand-new by construction (it enters the batch + // only when it has no step_created yet), so it + // has zero prior starts and is always attempt 1 — + // skip the log scan entirely. Only an + // owned-recovery re-run (this message re-executing + // a step it crashed/timed out on) can have prior + // starts, and that path is uncommon, so reserve + // the O(n) scan for it rather than walking the + // growing log for every inline step (which would + // be O(n²) across a long sequential workflow). + authoritativeAttempt: + s.lazyStepInput !== undefined + ? 1 + : countStepStartedEvents( + cachedEvents, + s.correlationId + ) + 1, // Lazy inline start: send the deferred step's // input on step_started so the world creates // the step on the fly. Absent for diff --git a/packages/core/src/runtime/step-executor.test.ts b/packages/core/src/runtime/step-executor.test.ts new file mode 100644 index 0000000000..4cb6f6b8f1 --- /dev/null +++ b/packages/core/src/runtime/step-executor.test.ts @@ -0,0 +1,166 @@ +import { mkdtempSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import type { Event, World } from '@workflow/world'; +import { SPEC_VERSION_CURRENT } from '@workflow/world'; +import { createWorld } from '@workflow/world-local'; +import { afterEach, describe, expect, it } from 'vitest'; +import { registerStepFunction } from '../private.js'; +import { dehydrateStepArguments } from '../serialization.js'; +import { executeStep } from './step-executor.js'; + +// The retry ceiling (`authoritativeAttempt`) is what bounds a step that keeps +// timing out: a timeout hard-kills the body without writing any error, so the +// error-based guards never fire. These tests assert the ceiling is enforced +// BEFORE the body runs, and only once the attempt number actually exceeds +// maxRetries + 1. + +const MAX_RETRIES = 3; // maxRetries + 1 = 4 total attempts allowed + +let counter = 0; +function uniqueStepName(): string { + counter += 1; + return `step//./step-executor-test//timeoutStep${counter}`; +} + +async function setupRunningStep(opts: { + world: World; + stepName: string; + onBody: () => void; +}): Promise<{ runId: string; stepId: string }> { + const { world, stepName, onBody } = opts; + const runInput = await dehydrateStepArguments([], 'run', undefined); + const created = await world.events.create(null, { + eventType: 'run_created', + specVersion: SPEC_VERSION_CURRENT, + eventData: { + deploymentId: 'dpl_test', + workflowName: 'wf', + input: runInput, + }, + }); + const runId = created.run!.runId; + await world.events.create(runId, { + eventType: 'run_started', + specVersion: SPEC_VERSION_CURRENT, + eventData: {}, + } as never); + + const stepId = 'step_timeout_1'; + const stepInput = await dehydrateStepArguments([], runId, undefined); + await world.events.create(runId, { + eventType: 'step_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: stepId, + eventData: { stepName, input: stepInput }, + }); + + const stepFn = Object.assign( + async () => { + onBody(); + return 'ok'; + }, + { maxRetries: MAX_RETRIES } + ); + registerStepFunction(stepName, stepFn); + + return { runId, stepId }; +} + +function makeWorld(): World { + const dataDir = mkdtempSync(join(tmpdir(), 'wf-step-executor-')); + return createWorld({ dataDir, tag: `t${counter}` }); +} + +async function eventsFor( + world: World, + runId: string, + stepId: string, + eventType: Event['eventType'] +): Promise { + const { data } = await world.events.list({ runId }); + return data.filter( + (e) => e.eventType === eventType && e.correlationId === stepId + ); +} + +describe('executeStep — retry ceiling (authoritativeAttempt)', () => { + afterEach(() => { + counter += 1; + }); + + it('fails the step WITHOUT running the body once the attempt exceeds maxRetries + 1', async () => { + const world = makeWorld(); + const stepName = uniqueStepName(); + let bodyRuns = 0; + const { runId, stepId } = await setupRunningStep({ + world, + stepName, + onBody: () => { + bodyRuns += 1; + }, + }); + + const result = await executeStep({ + world, + workflowRunId: runId, + workflowName: 'wf', + workflowStartedAt: Date.now(), + stepId, + stepName, + // Attempt maxRetries + 2 — one past the last allowed retry. This is the + // delivery a timed-out step would land on with nothing left to try. + authoritativeAttempt: MAX_RETRIES + 2, + }); + + expect(result.type).toBe('failed'); + // The body must NOT run — retries are already exhausted. + expect(bodyRuns).toBe(0); + + // The ceiling fires BEFORE the start block, so no new step_started is + // written for the rejected attempt; the step goes straight to failed. + const started = await eventsFor(world, runId, stepId, 'step_started'); + expect(started).toHaveLength(0); + const failures = await eventsFor(world, runId, stepId, 'step_failed'); + expect(failures).toHaveLength(1); + }); + + it('permits (does not pre-empt) the final allowed attempt (maxRetries + 1)', async () => { + const world = makeWorld(); + const stepName = uniqueStepName(); + let bodyRuns = 0; + const { runId, stepId } = await setupRunningStep({ + world, + stepName, + onBody: () => { + bodyRuns += 1; + }, + }); + + const result = await executeStep({ + world, + workflowRunId: runId, + workflowName: 'wf', + workflowStartedAt: Date.now(), + stepId, + stepName, + // maxRetries + 1 is the last permitted attempt: the ceiling must let it + // proceed into normal execution rather than pre-emptively failing it. + authoritativeAttempt: MAX_RETRIES + 1, + }); + + // It got past the ceiling: the step was started (entered normal execution) + // and was NOT failed by the retry ceiling. + void bodyRuns; + expect(result.type).not.toBe('failed'); + const started = await eventsFor(world, runId, stepId, 'step_started'); + expect(started).toHaveLength(1); + const ceilingFailures = await eventsFor( + world, + runId, + stepId, + 'step_failed' + ); + expect(ceilingFailures).toHaveLength(0); + }); +}); diff --git a/packages/core/src/runtime/step-executor.ts b/packages/core/src/runtime/step-executor.ts index d14329adc2..c46068c472 100644 --- a/packages/core/src/runtime/step-executor.ts +++ b/packages/core/src/runtime/step-executor.ts @@ -51,7 +51,7 @@ import { } from './step-latency.js'; import { safeWaitUntil } from './wait-until.js'; -const DEFAULT_STEP_MAX_RETRIES = 3; +export const DEFAULT_STEP_MAX_RETRIES = 3; /** * Extract the inline delta from a step-terminal `events.create` result, @@ -180,6 +180,17 @@ export interface StepExecutorParams { * runtime/step-latency.ts. */ latencyTracking?: StepLatencyTracking; + /** + * Authoritative attempt number for this execution, used to bound retries + * against `maxRetries` BEFORE the body runs. In order to also catch + * step timeouts (which we can't have catch handlers for), we determine + * the attempt count based as follows: + * - Inline (combined handler): the number of `step_started` events already + * in the event log for this step, plus one for the attempt about to run. + * - Background (queue-dispatched): the queue delivery count + * (`metadata.attempt`), which increments on every redelivery. + */ + authoritativeAttempt?: number; } /** @@ -351,6 +362,70 @@ export async function executeStep( ...Attribute.StepMaxRetries(maxRetries), }); + // maxRetries enforced before starting another attempt. Timeouts can kill + // a step execution without any way to catch the error, so the post-body/error + // guards below might miss it. Hence, we use `authoritativeAttempt` count + // as an additional guard based on step_started count or queue delivery count + // on backgrounded steps. + // Thrown-error exhaustion still terminates one attempt earlier via the post-body + // guard (with the thrown error as cause), and this check does not affect that. + if ( + params.authoritativeAttempt !== undefined && + params.authoritativeAttempt > maxRetries + 1 + ) { + const retryCount = maxRetries; + const errorMessage = `Step "${stepName}" exceeded max retries (${retryCount} ${pluralize('retry', 'retries', retryCount)})`; + stepLogger.error('Step exceeded max retries', { + workflowRunId, + stepName, + stepId, + attempt: params.authoritativeAttempt, + maxRetries, + }); + try { + await world.events.create(workflowRunId, { + eventType: 'step_failed', + specVersion: SPEC_VERSION_CURRENT, + correlationId: stepId, + eventData: { + stepName, + error: await dehydrateStepError( + new FatalError(errorMessage), + workflowRunId, + await getEncryptionKey(), + [], + globalThis, + compression + ), + }, + }); + } catch (err) { + if (EntityConflictError.is(err)) { + // Step already reached a terminal state (a concurrent handler or an + // earlier delivery failed/completed it) — nothing to do. + runtimeLogger.info( + 'Tried failing step for exceeded retries, but step has already finished.', + { + workflowRunId, + stepId, + stepName, + message: err instanceof Error ? err.message : String(err), + } + ); + return { type: 'skipped' }; + } + if (RunExpiredError.is(err)) { + return { type: 'gone' }; + } + throw err; + } + span?.setAttributes({ + ...Attribute.StepStatus('failed'), + ...Attribute.StepRetryExhausted(true), + }); + return { type: 'failed' }; + } + // Maps a `step_started` rejection to a terminal StepExecutionResult, // shared by the await path (below) and the optimistic-start reconciliation. // Returns undefined when the error is not one we translate (caller rethrows).