From b9f2ca445b03ab64b807d5af8541764a41326ff6 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 12:24:34 -0700 Subject: [PATCH 01/14] Re-dispatch steps that were created but never started A pending step is handed to the queue under an idempotency key equal to its correlation ID. Queues dedupe a key for the lifetime of the message sent under it, so once a dispatch has been accepted every later replay's re-send is absorbed. That is what keeps concurrent wake replays from multiplying the dispatch, but it also means a step whose message never produces a step_started can never be dispatched again: the run replays forever with one pending step nothing will execute, reaching no terminal state and raising no error. Give those dispatches an epoch derived from the step's durable step_created timestamp. Every replay computes the same epoch, so fan-out stays capped at one message per epoch, while a step still unstarted a full watchdog interval later gets a key the queue has not seen and is dispatched again. A suspension also arms a timer on the soonest boundary, since a run whose only outstanding work is the lost dispatch would otherwise never replay. Scope is steps awaiting their FIRST step_started. A started step is either running (no client-visible completion deadline) or inline-owned, which the ownership lease and its backstop already cover. Both VM engines dispatch pending steps, so both take the epoch key and the boundary wake. --- .changeset/wild-pandas-jam.md | 5 + .../docs/v5/configuration/runtime-tuning.mdx | 8 + packages/core/src/global.ts | 8 + packages/core/src/runtime.ts | 50 ++++- packages/core/src/runtime/constants.ts | 51 +++++ .../core/src/runtime/quickjs-entrypoint.ts | 89 +++++++- .../core/src/runtime/step-dispatch.test.ts | 208 ++++++++++++++++++ packages/core/src/runtime/step-dispatch.ts | 169 ++++++++++++++ packages/core/src/step.ts | 4 + 9 files changed, 587 insertions(+), 5 deletions(-) create mode 100644 .changeset/wild-pandas-jam.md create mode 100644 packages/core/src/runtime/step-dispatch.test.ts create mode 100644 packages/core/src/runtime/step-dispatch.ts diff --git a/.changeset/wild-pandas-jam.md b/.changeset/wild-pandas-jam.md new file mode 100644 index 0000000000..64cfb0259f --- /dev/null +++ b/.changeset/wild-pandas-jam.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Re-dispatch a step that was created but never started, so a lost queue message no longer strands a run diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index 53a59e7759..8e9269d5e2 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -158,6 +158,14 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL - How long after an inline step's latest `step_started` other invocations assume its owner may still be executing the body. Within the lease they defer the step's backstop message; past it they enqueue immediately. - Raise this on self-hosted multi-instance deployments whose inline steps run longer than the default (the default is sized for Vercel's function duration ceiling). +### `WORKFLOW_STEP_DISPATCH_WATCHDOG_SECONDS` + +- Default: `60` +- Clamp: `10` to `900` +- How long a step may sit created but never started before the runtime re-dispatches it. A queued step message is deduplicated on its correlation ID, so if that message is lost the step would otherwise never be delivered again and the run would keep replaying with nothing to execute. Past this interval the re-dispatch uses a key the queue has not seen, and the runtime arms a timer on the boundary so the check happens even when nothing else wakes the run. +- Re-dispatch is at-least-once: if the original message was only slow, both deliveries execute the step body and the loser's result is discarded. Raise this if your queue's dispatch-to-start latency approaches the default. +- Only steps that have never started are covered. A step that has started is left to the inline ownership lease above. + ## Workflow VM engine ### `WORKFLOW_VM` diff --git a/packages/core/src/global.ts b/packages/core/src/global.ts index d54e365a5c..adc3668fe1 100644 --- a/packages/core/src/global.ts +++ b/packages/core/src/global.ts @@ -11,6 +11,14 @@ export interface StepInvocationQueueItem { closureVars?: Record; thisVal?: Serializable; hasCreatedEvent?: boolean; + /** + * `createdAt` (ms) of the step's durable `step_created`, when the replay + * observed one. Anchors the re-dispatch watchdog for a step that never + * started (runtime/step-dispatch.ts). Undefined for a step this suspension + * is creating right now, and on worlds whose events carry no usable + * timestamp — both keep the plain correlation-ID dispatch key. + */ + createdEventAt?: number; /** * Inline step ownership, derived from the step's LATEST `step_started` * during replay: the queue message ID stamped by the invocation running diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 5b34574433..dc79edf7dd 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -91,6 +91,10 @@ import { } from './runtime/replay-budget.js'; import { ReplayRecoveryReporter } from './runtime/replay-recovery-reporter.js'; import { runIdCreatedAt } from './runtime/run-id-time.js'; +import { + getStepDispatchWake, + stepDispatchIdempotencyKey, +} from './runtime/step-dispatch.js'; import { DEFAULT_STEP_MAX_RETRIES, executeStep, @@ -3138,6 +3142,9 @@ export function workflowEntrypoint( const dispatchNowMs = Date.now(); const ownedRecoverySteps: StepInvocationQueueItem[] = []; + // Steps dispatched as queue messages this suspension, + // for the re-dispatch watchdog's boundary wake below. + const dispatchedSteps: StepInvocationQueueItem[] = []; let backstopWakesArmed = 0; for (const step of pendingSteps) { if (inlineCorrelationIds.has(step.correlationId)) { @@ -3192,6 +3199,7 @@ export function workflowEntrypoint( ); continue; } + dispatchedSteps.push(step); dispatches.push( queueMessage( world, @@ -3204,7 +3212,47 @@ export function workflowEntrypoint( requestedAt: new Date(), }, { - idempotencyKey: step.correlationId, + // Epoch-scoped once the step has gone a full + // watchdog interval without ever starting: the + // bare correlation ID is deduped by the queue + // for the lifetime of the message dispatched + // under it, so a dispatch that never produced + // a step_started can only be retried under a + // key the queue has not seen. Within an epoch + // every replay derives the same key, so + // concurrent wake replays still collapse to + // one message. See runtime/step-dispatch.ts. + idempotencyKey: stepDispatchIdempotencyKey( + step, + dispatchNowMs + ), + } + ) + ); + } + // Nothing else wakes a run whose only outstanding work + // is a step dispatch that was lost, so the watchdog + // needs its own timer: one delayed run continuation at + // the earliest boundary among the steps still awaiting + // a first start. Deduped on that boundary, so repeated + // suspensions within an interval arm it once. + const dispatchWake = getStepDispatchWake( + dispatchedSteps, + dispatchNowMs + ); + if (dispatchWake) { + dispatches.push( + queueMessage( + world, + getWorkflowQueueName(workflowName, namespace), + { + runId, + traceCarrier, + requestedAt: new Date(), + }, + { + delaySeconds: dispatchWake.delaySeconds, + idempotencyKey: dispatchWake.idempotencyKey, } ) ); diff --git a/packages/core/src/runtime/constants.ts b/packages/core/src/runtime/constants.ts index 37a82d1d82..38c517b950 100644 --- a/packages/core/src/runtime/constants.ts +++ b/packages/core/src/runtime/constants.ts @@ -408,6 +408,57 @@ export function getInlineOwnershipLeaseSeconds(): number { ); } +/** + * Default re-dispatch watchdog for a step that was created but never started, + * in seconds. A pending step's queue message is keyed by its correlation ID, + * and queues dedupe that key for the message's lifetime, so a dispatch that + * never yields a `step_started` can never be re-sent under it. After this long + * without a start, replays assume the dispatch is gone and enqueue the step + * under a fresh, epoch-scoped key (see runtime/step-dispatch.ts). + * + * The trade is at-least-once execution for the affected step: a dispatch that + * was merely slow rather than lost still delivers, and both executions run the + * body. So the default sits far above normal dispatch-to-start latency, which + * is queue delivery plus a cold start — order of a second, and measured under + * a heavy concurrent-replay storm at under 2.5s for every step in the run. + * A minute of silence is not a slow queue. + * + * Override via `WORKFLOW_STEP_DISPATCH_WATCHDOG_SECONDS` (clamped to + * `MIN_STEP_DISPATCH_WATCHDOG_SECONDS`..`MAX_STEP_DISPATCH_WATCHDOG_SECONDS`): + * raise it on deployments where step messages routinely queue behind a + * concurrency limit for minutes, lower it to recover faster. + */ +export const STEP_DISPATCH_WATCHDOG_SECONDS = 60; + +/** + * Lower bound for the watchdog override. Below this the watchdog starts + * competing with ordinary dispatch latency and duplicates healthy steps. + */ +export const MIN_STEP_DISPATCH_WATCHDOG_SECONDS = 10; + +/** + * Upper bound for the watchdog override. 900s is the queue's maximum + * per-message delay (SQS cap), and the boundary wake is a delayed message of + * at most one interval. + */ +export const MAX_STEP_DISPATCH_WATCHDOG_SECONDS = 900; + +/** + * Effective re-dispatch watchdog for unstarted steps. Override via + * `WORKFLOW_STEP_DISPATCH_WATCHDOG_SECONDS`. + */ +export function getStepDispatchWatchdogSeconds(): number { + return envNumber( + 'WORKFLOW_STEP_DISPATCH_WATCHDOG_SECONDS', + STEP_DISPATCH_WATCHDOG_SECONDS, + { + integer: true, + min: MIN_STEP_DISPATCH_WATCHDOG_SECONDS, + max: MAX_STEP_DISPATCH_WATCHDOG_SECONDS, + } + ); +} + // A replay-consumer mismatch can be caused by a transient divergent replay // rather than an invalid persisted history. Queue bounded recovery replays // before recording terminal corruption for a run that cannot replay. diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index f725cf9ee3..bdf35454f5 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -61,6 +61,11 @@ import { startQuickJSWorkflow, } from './quickjs-runtime.js'; import { ReplayBudget } from './replay-budget.js'; +import { + getStepDispatchWake, + type StepDispatchState, + stepDispatchIdempotencyKey, +} from './step-dispatch.js'; import { executeStep, type StepExecutionResult } from './step-executor.js'; import { runStepSingleFlight } from './step-single-flight.js'; import { getWaitContinuationDispatch } from './wait-continuation.js'; @@ -122,7 +127,10 @@ async function queueStepMessage(params: { * publish — see wait-continuation.ts for the same hazard on wait * keys. `dispatch` is the plain background handoff (overflow / crash * recovery) and keeps the bare correlationId so it stays mutually - * exclusive with the node engine's dispatch of the same step; + * exclusive with the node engine's dispatch of the same step, except + * once the step has gone a full watchdog interval without ever + * starting, where both engines move to the same epoch-scoped key so a + * lost dispatch can be re-sent (see step-dispatch.ts); * `backstop:` covers delayed crash backstops, scoped to the * ownership epoch so a refreshed lease re-arms a NEW backstop instead * of being absorbed by the in-flight one; `retry:` covers delayed @@ -130,6 +138,11 @@ async function queueStepMessage(params: { * hop is enqueueable. */ purpose: 'dispatch' | `backstop:${string}` | `retry:${number}`; + /** + * Replay-derived watchdog state for a `dispatch` publish. Absent (or + * out of watchdog scope) keeps the bare correlationId. + */ + dispatchState?: StepDispatchState; wfdiag: (checkpoint: string, fields: Record) => void; }): Promise { const { @@ -141,6 +154,7 @@ async function queueStepMessage(params: { namespace, nextTraceCarrier, purpose, + dispatchState, wfdiag, } = params; const traceCarrier = await nextTraceCarrier(); @@ -157,7 +171,9 @@ async function queueStepMessage(params: { { idempotencyKey: purpose === 'dispatch' - ? step.correlationId + ? dispatchState + ? stepDispatchIdempotencyKey(dispatchState, Date.now()) + : step.correlationId : `${step.correlationId}:${purpose}`, ...(delaySeconds && delaySeconds > 0 ? { delaySeconds } : {}), } @@ -886,12 +902,29 @@ export async function runWorkflowWithQuickJS(params: { // stamp; a step_retrying lapses ownership permanently for the id. const stepOwnership = new Map< string, - { owner?: string; startedAtMs?: number; sawRetrying: boolean } + { + owner?: string; + startedAtMs?: number; + sawRetrying: boolean; + /** + * `createdAt` (ms) of the step's `step_created`. Anchors the + * re-dispatch watchdog for a step that never started + * (step-dispatch.ts). + */ + createdAtMs?: number; + } >(); const observeEventsForOwnership = (observed: Event[]): void => { for (const e of observed) { if (e.correlationId === undefined) continue; - if (e.eventType === 'step_started') { + if (e.eventType === 'step_created') { + const prior = stepOwnership.get(e.correlationId); + stepOwnership.set(e.correlationId, { + sawRetrying: false, + ...(prior ?? {}), + createdAtMs: e.createdAt ? +new Date(e.createdAt) : undefined, + }); + } else if (e.eventType === 'step_started') { const owner = 'eventData' in e && e.eventData && @@ -904,6 +937,7 @@ export async function runWorkflowWithQuickJS(params: { owner, startedAtMs: e.createdAt ? +new Date(e.createdAt) : undefined, sawRetrying: prior?.sawRetrying ?? false, + createdAtMs: prior?.createdAtMs, }); } else if (e.eventType === 'step_retrying') { const prior = stepOwnership.get(e.correlationId); @@ -915,6 +949,22 @@ export async function runWorkflowWithQuickJS(params: { } }; observeEventsForOwnership(events); + /** + * The re-dispatch watchdog's view of a pending step, assembled from the + * ownership map. A step this invocation is creating right now carries no + * observed `step_created` yet, so it is out of scope and keeps the bare + * dispatch key. + */ + const dispatchStateOf = (step: PendingStep): StepDispatchState => { + const ownership = stepOwnership.get(step.correlationId); + return { + correlationId: step.correlationId, + hasCreatedEvent: step.hasCreatedEvent, + createdEventAt: ownership?.createdAtMs, + lastStartedAt: ownership?.startedAtMs, + sawRetrying: ownership?.sawRetrying, + }; + }; const scheduledWaitContinuations = new Set(); const maxInlineSteps = getMaxInlineSteps(); const budget = new ReplayBudget(); @@ -1050,6 +1100,7 @@ export async function runWorkflowWithQuickJS(params: { namespace, nextTraceCarrier, purpose: 'dispatch', + dispatchState: dispatchStateOf(step), wfdiag, }); } @@ -1182,6 +1233,7 @@ export async function runWorkflowWithQuickJS(params: { namespace, nextTraceCarrier, purpose: 'dispatch', + dispatchState: dispatchStateOf(step), wfdiag, }); } @@ -1581,9 +1633,38 @@ export async function runWorkflowWithQuickJS(params: { return; } + // Nothing else will wake a run whose only outstanding work is a step + // dispatch the queue lost, so arm a timer on the soonest watchdog + // boundary among the steps still awaiting their first start. Same + // scheme as the node engine's suspension wake (step-dispatch.ts): the + // key is derived from the durable creation timestamp, so concurrent + // invocations arm one timer, not one each. + const dispatchWake = getStepDispatchWake( + pendingOperations + .filter((op): op is PendingStep => op.type === 'step') + .map(dispatchStateOf), + Date.now() + ); + if (dispatchWake) { + await queueMessage( + world, + getWorkflowQueueName(workflowRun.workflowName, namespace), + { + runId, + traceCarrier: await nextTraceCarrier(), + requestedAt: new Date(), + }, + { + delaySeconds: dispatchWake.delaySeconds, + idempotencyKey: dispatchWake.idempotencyKey, + } + ); + } + wfdiag('exit_suspended', { action: 'awaiting_external', pendingOpsCount: pendingOperations.length, + dispatchWakeSeconds: dispatchWake?.delaySeconds, }); } else if (result.failed) { // Workflow failed — remap stack trace using inline source maps. diff --git a/packages/core/src/runtime/step-dispatch.test.ts b/packages/core/src/runtime/step-dispatch.test.ts new file mode 100644 index 0000000000..449d82a8da --- /dev/null +++ b/packages/core/src/runtime/step-dispatch.test.ts @@ -0,0 +1,208 @@ +import { afterEach, describe, expect, it } from 'vitest'; +import type { StepInvocationQueueItem } from '../global.js'; +import { getStepDispatchWatchdogSeconds } from './constants.js'; +import { + getStepDispatchWake, + isStepAwaitingFirstStart, + nextStepDispatchBoundaryMs, + stepDispatchEpoch, + stepDispatchIdempotencyKey, +} from './step-dispatch.js'; + +const WATCHDOG_ENV = 'WORKFLOW_STEP_DISPATCH_WATCHDOG_SECONDS'; + +const CREATED_AT = 1_000_000; + +function makeStep( + overrides: Partial = {} +): StepInvocationQueueItem { + return { + type: 'step', + correlationId: 'step_01ABC', + stepName: 'someStep', + args: [], + hasCreatedEvent: true, + createdEventAt: CREATED_AT, + ...overrides, + }; +} + +/** `nowMs` exactly `intervals` watchdog intervals after the step's creation. */ +function atIntervals(intervals: number): number { + return CREATED_AT + intervals * getStepDispatchWatchdogSeconds() * 1000; +} + +afterEach(() => { + delete process.env[WATCHDOG_ENV]; +}); + +describe('isStepAwaitingFirstStart', () => { + it('is true for a created step with no observed start', () => { + expect(isStepAwaitingFirstStart(makeStep())).toBe(true); + }); + + it('is false before step_created is observed', () => { + expect(isStepAwaitingFirstStart(makeStep({ hasCreatedEvent: false }))).toBe( + false + ); + }); + + it('is false once the step has started', () => { + expect( + isStepAwaitingFirstStart(makeStep({ lastStartedAt: CREATED_AT + 10 })) + ).toBe(false); + }); + + it('is false after step_retrying, which implies a prior start', () => { + expect(isStepAwaitingFirstStart(makeStep({ sawRetrying: true }))).toBe( + false + ); + }); +}); + +describe('stepDispatchIdempotencyKey', () => { + it('uses the bare correlation ID within the first watchdog interval', () => { + const step = makeStep(); + expect(stepDispatchIdempotencyKey(step, CREATED_AT)).toBe( + step.correlationId + ); + expect(stepDispatchIdempotencyKey(step, atIntervals(1) - 1)).toBe( + step.correlationId + ); + }); + + it('moves to a fresh key once an interval has passed without a start', () => { + const step = makeStep(); + expect(stepDispatchIdempotencyKey(step, atIntervals(1))).toBe( + `${step.correlationId}:dispatch:1` + ); + expect(stepDispatchIdempotencyKey(step, atIntervals(2.5))).toBe( + `${step.correlationId}:dispatch:2` + ); + }); + + it('is stable across replays within one epoch', () => { + const step = makeStep(); + const early = stepDispatchIdempotencyKey(step, atIntervals(1.01)); + const late = stepDispatchIdempotencyKey(step, atIntervals(1.99)); + expect(early).toBe(late); + }); + + it('keeps the bare key for a step that has already started', () => { + // A started step may be mid-body for far longer than the watchdog; only + // the ownership lease and its backstop may act on it. + const step = makeStep({ lastStartedAt: CREATED_AT + 1 }); + expect(stepDispatchIdempotencyKey(step, atIntervals(10))).toBe( + step.correlationId + ); + }); + + it('keeps the bare key when the world reports no creation timestamp', () => { + const step = makeStep({ createdEventAt: undefined }); + expect(stepDispatchIdempotencyKey(step, atIntervals(10))).toBe( + step.correlationId + ); + }); + + it('keeps the bare key when the creation timestamp is in the future', () => { + const step = makeStep(); + expect(stepDispatchIdempotencyKey(step, CREATED_AT - 60_000)).toBe( + step.correlationId + ); + }); + + it('honours the watchdog override', () => { + process.env[WATCHDOG_ENV] = '10'; + const step = makeStep(); + expect(stepDispatchEpoch(step, CREATED_AT + 10_000)).toBe(1); + expect(stepDispatchEpoch(step, CREATED_AT + 9_999)).toBe(0); + }); + + it('clamps an override below the supported minimum', () => { + process.env[WATCHDOG_ENV] = '1'; + // Clamped up, so one nominal second past creation is still epoch 0. + expect(stepDispatchEpoch(makeStep(), CREATED_AT + 1_000)).toBe(0); + expect(getStepDispatchWatchdogSeconds()).toBeGreaterThan(1); + }); +}); + +describe('nextStepDispatchBoundaryMs', () => { + it('is one interval after creation while inside the first interval', () => { + expect(nextStepDispatchBoundaryMs(makeStep(), CREATED_AT)).toBe( + atIntervals(1) + ); + }); + + it('advances with the epoch', () => { + expect(nextStepDispatchBoundaryMs(makeStep(), atIntervals(1.5))).toBe( + atIntervals(2) + ); + }); + + it('is undefined for a step out of watchdog scope', () => { + expect( + nextStepDispatchBoundaryMs( + makeStep({ lastStartedAt: CREATED_AT }), + atIntervals(3) + ) + ).toBeUndefined(); + }); +}); + +describe('getStepDispatchWake', () => { + it('is undefined when no step is awaiting a first start', () => { + expect( + getStepDispatchWake([makeStep({ lastStartedAt: CREATED_AT })], CREATED_AT) + ).toBeUndefined(); + expect(getStepDispatchWake([], CREATED_AT)).toBeUndefined(); + }); + + it('lands just past the boundary so the wake does not re-arm its own key', () => { + const wake = getStepDispatchWake([makeStep()], CREATED_AT); + expect(wake).toBeDefined(); + const boundary = atIntervals(1); + expect(CREATED_AT + wake!.delaySeconds * 1000).toBeGreaterThan(boundary); + expect(wake?.idempotencyKey).toBe(`step_01ABC:dispatch-wake:${boundary}`); + }); + + it('picks the earliest boundary among several pending steps', () => { + const older = makeStep({ + correlationId: 'step_older', + createdEventAt: CREATED_AT, + }); + const newer = makeStep({ + correlationId: 'step_newer', + createdEventAt: CREATED_AT + 5_000, + }); + const wake = getStepDispatchWake([newer, older], CREATED_AT + 5_000); + expect(wake?.idempotencyKey).toBe( + `step_older:dispatch-wake:${atIntervals(1)}` + ); + }); + + it('breaks boundary ties on correlation ID so replays agree', () => { + const a = makeStep({ correlationId: 'step_a' }); + const b = makeStep({ correlationId: 'step_b' }); + expect(getStepDispatchWake([a, b], CREATED_AT)?.idempotencyKey).toBe( + getStepDispatchWake([b, a], CREATED_AT)?.idempotencyKey + ); + expect(getStepDispatchWake([b, a], CREATED_AT)?.idempotencyKey).toContain( + 'step_a' + ); + }); + + it('keeps the delay inside the queue-supported range at the maximum watchdog', () => { + process.env[WATCHDOG_ENV] = String(Number.MAX_SAFE_INTEGER); + const clamped = getStepDispatchWatchdogSeconds(); + const wake = getStepDispatchWake([makeStep()], CREATED_AT); + expect(wake?.delaySeconds).toBe(clamped); + }); + + it('still asks for a positive delay when the boundary is already past', () => { + // Clock skew or a long-running replay can put "now" past the boundary the + // epoch was computed from. + const step = makeStep(); + const wake = getStepDispatchWake([step], atIntervals(1) - 1); + expect(wake?.delaySeconds).toBeGreaterThanOrEqual(1); + }); +}); diff --git a/packages/core/src/runtime/step-dispatch.ts b/packages/core/src/runtime/step-dispatch.ts new file mode 100644 index 0000000000..45c2094c17 --- /dev/null +++ b/packages/core/src/runtime/step-dispatch.ts @@ -0,0 +1,169 @@ +import { + getStepDispatchWatchdogSeconds, + MAX_STEP_DISPATCH_WATCHDOG_SECONDS, +} from './constants.js'; + +/** + * Re-dispatch watchdog for a step that was created but never started. + * + * A pending step is dispatched as a queue message keyed by its correlation + * ID. Queues dedupe an idempotency key for the lifetime of the original + * message, so that key is effectively permanent: once a message has been + * accepted under it, every later replay's re-send is silently absorbed. That + * is the intended behaviour while the message is doing its job — concurrent + * wake replays must not multiply the dispatch — but it also means a step + * whose message never produces a `step_started` can never be dispatched + * again. The run then keeps replaying forever with one pending step that + * nothing will ever execute: no divergence, no error, no terminal state. + * + * The watchdog gives those dispatches an epoch. Every replay derives the + * epoch from the step's durable `step_created` timestamp, so concurrent + * replays agree on the key and fan-out stays capped at one message per epoch, + * while a step still unstarted a full watchdog interval later gets a key the + * queue has never seen and is dispatched again. + * + * Scope: only steps awaiting their FIRST `step_started`. A step that has + * started is either running (its body has no completion deadline the client + * can see, so re-dispatching on a timer would duplicate healthy long-running + * work) or inline-owned (covered by the ownership lease and its backstop in + * step-ownership.ts). + * + * Re-dispatch is at-least-once: if the original message was merely slow + * rather than lost, both deliveries execute the step body and the loser's + * terminal write is rejected as a conflict. The default interval is set well + * above normal dispatch-to-start latency so that trade only happens for + * dispatches that really are gone. + */ + +/** + * The replay-derived facts about one pending step the watchdog needs. + * `StepInvocationQueueItem` (the node engine's queue item) satisfies it + * structurally; the quickjs engine builds one from its own per-step + * ownership map. + */ +export interface StepDispatchState { + correlationId: string; + /** Whether a durable `step_created` was observed for this step. */ + hasCreatedEvent?: boolean; + /** `createdAt` (ms) of that `step_created`, when the world reports one. */ + createdEventAt?: number; + /** `createdAt` (ms) of the latest observed `step_started`, if any. */ + lastStartedAt?: number; + /** Whether a `step_retrying` was observed, which implies a prior start. */ + sawRetrying?: boolean; +} + +/** + * Whether a pending step is waiting for its first `step_started`: created + * durably, no start observed, and no `step_retrying` (which implies a prior + * start). Steps outside this state keep the bare correlation-ID key. + */ +export function isStepAwaitingFirstStart(step: StepDispatchState): boolean { + return ( + step.hasCreatedEvent === true && + step.lastStartedAt === undefined && + step.sawRetrying !== true + ); +} + +/** + * How many full watchdog intervals have elapsed since the step's + * `step_created`. 0 means the current dispatch is still within its first + * interval, or the step is out of scope / has no usable creation timestamp + * (worlds whose events lack them) — in both cases dispatch keeps today's + * behaviour exactly. + */ +export function stepDispatchEpoch( + step: StepDispatchState, + nowMs: number +): number { + if (!isStepAwaitingFirstStart(step)) return 0; + if (step.createdEventAt === undefined) return 0; + const elapsedMs = nowMs - step.createdEventAt; + if (!Number.isFinite(elapsedMs) || elapsedMs <= 0) return 0; + return Math.floor(elapsedMs / (getStepDispatchWatchdogSeconds() * 1000)); +} + +/** + * Idempotency key for a pending step's dispatch message. Epoch 0 is the bare + * correlation ID, which is also the key the owner's retry handoff enqueues + * under — the two must agree so a re-dispatch never races a queued retry. + * Later epochs suffix the key so the queue sees a message it has not + * deduped. + */ +export function stepDispatchIdempotencyKey( + step: StepDispatchState, + nowMs: number +): string { + const epoch = stepDispatchEpoch(step, nowMs); + return epoch === 0 + ? step.correlationId + : `${step.correlationId}:dispatch:${epoch}`; +} + +/** + * When the step's next watchdog boundary falls, as an absolute epoch-ms + * timestamp, or undefined when the step is out of scope. Derived from the + * durable creation timestamp so every replay computes the same boundary, + * which is what lets the wake armed for it dedupe across replays. + */ +export function nextStepDispatchBoundaryMs( + step: StepDispatchState, + nowMs: number +): number | undefined { + if (!isStepAwaitingFirstStart(step)) return undefined; + if (step.createdEventAt === undefined) return undefined; + const watchdogMs = getStepDispatchWatchdogSeconds() * 1000; + return ( + step.createdEventAt + (stepDispatchEpoch(step, nowMs) + 1) * watchdogMs + ); +} + +/** + * The wake a suspension arms so an unstarted step's next watchdog boundary is + * observed even when nothing else would wake the run. Without it the watchdog + * only helps runs that happen to keep receiving hooks or wait timers; a run + * whose sole outstanding work is the lost dispatch would never replay again. + * + * At most one wake per suspension: the earliest boundary among the pending + * unstarted steps, since a replay at that point re-evaluates all of them. Ties + * break on correlation ID so concurrent replays pick the same step and + * therefore the same key. + */ +export function getStepDispatchWake( + steps: StepDispatchState[], + nowMs: number +): { delaySeconds: number; idempotencyKey: string } | undefined { + let earliest: { boundaryMs: number; correlationId: string } | undefined; + for (const step of steps) { + const boundaryMs = nextStepDispatchBoundaryMs(step, nowMs); + if (boundaryMs === undefined) continue; + if ( + !earliest || + boundaryMs < earliest.boundaryMs || + (boundaryMs === earliest.boundaryMs && + step.correlationId < earliest.correlationId) + ) { + earliest = { boundaryMs, correlationId: step.correlationId }; + } + } + if (!earliest) return undefined; + // A second of slack past the boundary, because a wake that lands even + // marginally early computes the same epoch, re-arms the same key, and is + // deduped away — leaving the run with no timer at all. Clamped to the + // watchdog interval so a creation timestamp in the future (clock skew) + // cannot ask for a delay above the queue's per-message maximum, and floored + // at 1s because a boundary already past still has to be a valid delay. + const maxDelaySeconds = Math.min( + getStepDispatchWatchdogSeconds() + 1, + MAX_STEP_DISPATCH_WATCHDOG_SECONDS + ); + const delaySeconds = Math.min( + maxDelaySeconds, + Math.max(1, Math.ceil((earliest.boundaryMs - nowMs) / 1000) + 1) + ); + return { + delaySeconds, + idempotencyKey: `${earliest.correlationId}:dispatch-wake:${earliest.boundaryMs}`, + }; +} diff --git a/packages/core/src/step.ts b/packages/core/src/step.ts index 1fcc4c11df..ff341bdd59 100644 --- a/packages/core/src/step.ts +++ b/packages/core/src/step.ts @@ -123,6 +123,10 @@ export function createUseStep(ctx: WorkflowOrchestratorContext) { return EventConsumerResult.Finished; } queueItem.hasCreatedEvent = true; + // Anchors the re-dispatch watchdog: how long this step has existed + // without ever starting is what tells a replay its dispatch message + // is gone (runtime/step-dispatch.ts). + queueItem.createdEventAt = +event.createdAt; // Continue waiting for step_started/step_completed/step_failed events return EventConsumerResult.Consumed; } From a073e1fb982482deb225fe128fc99bb610b53fce Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 12:34:16 -0700 Subject: [PATCH 02/14] Cover a dispatch lost on a step's first hand-off The watchdog keyed off the replay-observed step_created, so the suspension that creates a step dispatched under the bare key with no boundary wake armed. Both engines now stamp the creation timestamp from the write itself, putting the step in scope from its first hand-off. --- .../core/src/runtime/quickjs-entrypoint.ts | 50 ++++++++++++++----- .../core/src/runtime/step-dispatch.test.ts | 15 +++++- packages/core/src/runtime/step-dispatch.ts | 35 +++++++------ .../core/src/runtime/suspension-handler.ts | 10 +++- 4 files changed, 80 insertions(+), 30 deletions(-) diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index bdf35454f5..57413a32a3 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -225,6 +225,14 @@ async function dispatchPendingOps(params: { }): Promise<{ createdAttributeEvent: boolean; createdGetConflictHook: boolean; + /** + * `createdAt` (ms) of every `step_created` written by this call, keyed by + * correlation ID. The overflow hand-off queues those steps in the same turn, + * before any feed observes their events, so this is the only way their + * dispatch can be watchdog-scoped from the first hand-off on + * (step-dispatch.ts). + */ + stepCreatedAtMs: Map; }> { const { world, @@ -246,6 +254,7 @@ async function dispatchPendingOps(params: { // setAttributes() promise), so the entrypoint requeues immediately — // same pattern as an elapsed wait. let createdAttributeEvent = false; + const stepCreatedAtMs = new Map(); const opsPromises: Promise[] = []; const processHookOp = async (hook: PendingHook): Promise => { @@ -459,7 +468,7 @@ async function dispatchPendingOps(params: { // on the host side — matching what // `dehydrateStepArguments` does in the node:vm engine. try { - await world.events.create(runId, { + const created = await world.events.create(runId, { eventType: 'step_created', specVersion: SPEC_VERSION_CURRENT, correlationId: step.correlationId, @@ -468,6 +477,12 @@ async function dispatchPendingOps(params: { input: await encryptSerializedData(step.input, encryptionKey), }, }); + if (created.event?.createdAt) { + stepCreatedAtMs.set( + step.correlationId, + +new Date(created.event.createdAt) + ); + } } catch (err) { if (EntityConflictError.is(err)) return; throw err; @@ -532,7 +547,7 @@ async function dispatchPendingOps(params: { // Per-op dispatch runs in parallel. await Promise.all(opsPromises); - return { createdAttributeEvent, createdGetConflictHook }; + return { createdAttributeEvent, createdGetConflictHook, stepCreatedAtMs }; } /** @@ -914,16 +929,25 @@ export async function runWorkflowWithQuickJS(params: { createdAtMs?: number; } >(); + const recordStepCreatedAt = ( + correlationId: string, + createdAtMs: number | undefined + ): void => { + const prior = stepOwnership.get(correlationId); + stepOwnership.set(correlationId, { + sawRetrying: false, + ...(prior ?? {}), + createdAtMs, + }); + }; const observeEventsForOwnership = (observed: Event[]): void => { for (const e of observed) { if (e.correlationId === undefined) continue; if (e.eventType === 'step_created') { - const prior = stepOwnership.get(e.correlationId); - stepOwnership.set(e.correlationId, { - sawRetrying: false, - ...(prior ?? {}), - createdAtMs: e.createdAt ? +new Date(e.createdAt) : undefined, - }); + recordStepCreatedAt( + e.correlationId, + e.createdAt ? +new Date(e.createdAt) : undefined + ); } else if (e.eventType === 'step_started') { const owner = 'eventData' in e && @@ -951,15 +975,14 @@ export async function runWorkflowWithQuickJS(params: { observeEventsForOwnership(events); /** * The re-dispatch watchdog's view of a pending step, assembled from the - * ownership map. A step this invocation is creating right now carries no - * observed `step_created` yet, so it is out of scope and keeps the bare - * dispatch key. + * ownership map. Creation timestamps come from the log for steps created by + * an earlier invocation and from the write itself for steps created in this + * turn, so a step is in watchdog scope from its very first hand-off. */ const dispatchStateOf = (step: PendingStep): StepDispatchState => { const ownership = stepOwnership.get(step.correlationId); return { correlationId: step.correlationId, - hasCreatedEvent: step.hasCreatedEvent, createdEventAt: ownership?.createdAtMs, lastStartedAt: ownership?.startedAtMs, sawRetrying: ownership?.sawRetrying, @@ -1079,6 +1102,9 @@ export async function runWorkflowWithQuickJS(params: { ) { pendingRequeueSignal = true; } + for (const [correlationId, createdAtMs] of dispatched.stepCreatedAtMs) { + recordStepCreatedAt(correlationId, createdAtMs); + } // Hand steps beyond the inline cap to the queue NOW — in the same // turn their step_created was written by the dispatch above. This diff --git a/packages/core/src/runtime/step-dispatch.test.ts b/packages/core/src/runtime/step-dispatch.test.ts index 449d82a8da..861b1096ac 100644 --- a/packages/core/src/runtime/step-dispatch.test.ts +++ b/packages/core/src/runtime/step-dispatch.test.ts @@ -41,9 +41,20 @@ describe('isStepAwaitingFirstStart', () => { expect(isStepAwaitingFirstStart(makeStep())).toBe(true); }); - it('is false before step_created is observed', () => { + it('is false when no durable creation timestamp is known', () => { + expect( + isStepAwaitingFirstStart( + makeStep({ hasCreatedEvent: false, createdEventAt: undefined }) + ) + ).toBe(false); + }); + + it('is true for a step whose creation this invocation just wrote', () => { + // The suspension that writes step_created stamps the timestamp onto the + // queue item it is about to dispatch, so a dispatch lost on the very + // first hand-off is still covered. expect(isStepAwaitingFirstStart(makeStep({ hasCreatedEvent: false }))).toBe( - false + true ); }); diff --git a/packages/core/src/runtime/step-dispatch.ts b/packages/core/src/runtime/step-dispatch.ts index 45c2094c17..4625e8d2dd 100644 --- a/packages/core/src/runtime/step-dispatch.ts +++ b/packages/core/src/runtime/step-dispatch.ts @@ -43,9 +43,13 @@ import { */ export interface StepDispatchState { correlationId: string; - /** Whether a durable `step_created` was observed for this step. */ - hasCreatedEvent?: boolean; - /** `createdAt` (ms) of that `step_created`, when the world reports one. */ + /** + * `createdAt` (ms) of the step's durable `step_created`: stamped by the + * suspension that wrote it, and re-derived from the log by every later + * replay. Undefined means no durable creation is known to this + * invocation (a lazily-created inline step, or a world whose events carry + * no usable timestamp), which leaves the step out of watchdog scope. + */ createdEventAt?: number; /** `createdAt` (ms) of the latest observed `step_started`, if any. */ lastStartedAt?: number; @@ -54,13 +58,14 @@ export interface StepDispatchState { } /** - * Whether a pending step is waiting for its first `step_started`: created - * durably, no start observed, and no `step_retrying` (which implies a prior - * start). Steps outside this state keep the bare correlation-ID key. + * Whether a pending step is waiting for its first `step_started`: a durable + * creation timestamp is known, no start has been observed, and no + * `step_retrying` (which implies a prior start). Steps outside this state keep + * the bare correlation-ID key. */ export function isStepAwaitingFirstStart(step: StepDispatchState): boolean { return ( - step.hasCreatedEvent === true && + step.createdEventAt !== undefined && step.lastStartedAt === undefined && step.sawRetrying !== true ); @@ -77,9 +82,9 @@ export function stepDispatchEpoch( step: StepDispatchState, nowMs: number ): number { - if (!isStepAwaitingFirstStart(step)) return 0; - if (step.createdEventAt === undefined) return 0; - const elapsedMs = nowMs - step.createdEventAt; + const createdEventAt = step.createdEventAt; + if (createdEventAt === undefined || !isStepAwaitingFirstStart(step)) return 0; + const elapsedMs = nowMs - createdEventAt; if (!Number.isFinite(elapsedMs) || elapsedMs <= 0) return 0; return Math.floor(elapsedMs / (getStepDispatchWatchdogSeconds() * 1000)); } @@ -111,12 +116,12 @@ export function nextStepDispatchBoundaryMs( step: StepDispatchState, nowMs: number ): number | undefined { - if (!isStepAwaitingFirstStart(step)) return undefined; - if (step.createdEventAt === undefined) return undefined; + const createdEventAt = step.createdEventAt; + if (createdEventAt === undefined || !isStepAwaitingFirstStart(step)) { + return undefined; + } const watchdogMs = getStepDispatchWatchdogSeconds() * 1000; - return ( - step.createdEventAt + (stepDispatchEpoch(step, nowMs) + 1) * watchdogMs - ); + return createdEventAt + (stepDispatchEpoch(step, nowMs) + 1) * watchdogMs; } /** diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index b02826df4c..78ff954528 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -687,7 +687,15 @@ export async function handleSuspension({ }; try { await ensureRunReady(); - await createGuarded(stepEvent, { requestId }); + const created = await createGuarded(stepEvent, { requestId }); + // Anchors the re-dispatch watchdog from this suspension onward, + // so a dispatch lost on the step's very first hand-off is still + // retried. Later replays read the same timestamp off the event + // log, so every invocation derives the same epoch and the same + // keys (runtime/step-dispatch.ts). + if (created.event?.createdAt) { + queueItem.createdEventAt = +new Date(created.event.createdAt); + } createdStepCorrelationIds.add(queueItem.correlationId); } catch (err) { if (EntityConflictError.is(err)) { From 3d5d03d774befd1936a6ce89c1e110655b40133e Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 12:45:25 -0700 Subject: [PATCH 03/14] Simulate a lost step dispatch in world-sim Adds a dropQueued primitive that takes a message out of the queue without settling it, so the queue keeps its idempotency key claimed for the rest of the scenario. That is what makes a lost message worth simulating: a re-send under the same key is absorbed, and only a writer that changes the key can get the work dispatched again. The lost-step-dispatch scenario uses it against a four-way fan-out that overflows the inline cap, so exactly one step is handed to the queue. With the re-dispatch watchdog the run recovers at the first watchdog boundary; without it the run has no outstanding work at all and stays running forever. Co-Authored-By: Claude Opus 5 --- packages/world-sim/README.md | 14 +++++++ packages/world-sim/src/scenario.ts | 16 ++++++++ packages/world-sim/src/types.ts | 16 ++++++++ workbench/sim-world/scenarios/index.ts | 2 + .../sim-world/scenarios/lost-step-dispatch.ts | 32 ++++++++++++++++ workbench/sim-world/workflows/index.ts | 37 +++++++++++++++++++ 6 files changed, 117 insertions(+) create mode 100644 workbench/sim-world/scenarios/lost-step-dispatch.ts diff --git a/packages/world-sim/README.md b/packages/world-sim/README.md index 6646c36578..8755e9f4b6 100644 --- a/packages/world-sim/README.md +++ b/packages/world-sim/README.md @@ -448,6 +448,7 @@ be behind, never wrong — and an overtaken hook re-takes the tail on `commit()` | `cancelRun(reason?)` | Cancel the run under test | | `advanceTime(ms)` | Jump the virtual clock | | `deliverQueued(select?)` | Deliver one queued message now, concurrently with a held writer | +| `dropQueued(select?)` | Take one queued message out and never deliver it, keeping its idempotency key claimed | | `note(msg)` / `check(name, cond)` | Record a marker / an assertion in the trace; a false check fails the scenario | | `world` | Read-only snapshot: runs, events, steps, hooks, waits, pending messages, rejected calls | | `appendOnlyLog` | Which log this run is playing against — for *phrasing* a check, never for branching the tempo | @@ -493,6 +494,19 @@ Note the missing `await` — awaiting it here would wait for the delivery to wake, fire it, await the hold, and the two are now interleaved. Await the returned promise at the end to assert it found something. +#### `dropQueued`, and what a lost message costs + +`dropQueued` is the same take without the delivery. The message is gone and it +is never settled, so the queue keeps its idempotency key for the rest of the +scenario: every later re-send under that key is absorbed, and the only writer +that can get the work dispatched again is one that changes the key. + +That second half is what makes the fault worth simulating. A queue that forgot +the key on drop would recover on the next replay for free, and nothing about the +runtime's key scheme would be under test. `lost-step-dispatch` is the scenario +that asks the question: a step created, its dispatch dropped, and a run whose +only remaining work is a message that will never arrive. + ## Extending the simulator Changing the instrument itself, routed by task — adding a *scenario* needs none diff --git a/packages/world-sim/src/scenario.ts b/packages/world-sim/src/scenario.ts index 10a841ccf2..68b2dfb8c5 100644 --- a/packages/world-sim/src/scenario.ts +++ b/packages/world-sim/src/scenario.ts @@ -370,6 +370,22 @@ export async function runScenario( await deliver(world, message); return true; }, + dropQueued(select) { + const pending = world.simQueue.view(); + const chosen = select ? select(pending) : pending[0]?.messageId; + if (!chosen) return false; + // Taken but never settled: the queue keeps the idempotency key claimed, + // which is what a lost message looks like to every later writer. + const message = world.simQueue.takeById(chosen); + if (!message) return false; + world.pushTrace({ + kind: 'note', + message: `dropped queue message ${message.messageId}${ + message.idempotencyKey ? ` (key ${message.idempotencyKey})` : '' + }`, + }); + return true; + }, note(message) { world.pushTrace({ kind: 'note', message }); }, diff --git a/packages/world-sim/src/types.ts b/packages/world-sim/src/types.ts index 9fd5a8ddb5..9eede1b767 100644 --- a/packages/world-sim/src/types.ts +++ b/packages/world-sim/src/types.ts @@ -416,6 +416,22 @@ export interface ScenarioApi { deliverQueued( select?: (pending: PendingMessageView[]) => string | undefined ): Promise; + /** + * Take one pending queue message out of the queue and never deliver it, + * modelling a message the queue accepted and lost. + * + * The message is removed but never settled, so its idempotency key stays + * claimed for the rest of the scenario. That is the part that makes the + * fault interesting: a re-send under the same key is silently absorbed, so + * only a writer that changes the key can get the work dispatched again. + * + * `select` receives the pending messages in the loop's own order and returns + * a `messageId`; the default drops the first. Resolves `false` when nothing + * matched. + */ + dropQueued( + select?: (pending: PendingMessageView[]) => string | undefined + ): boolean; /** * Hide the next event this scenario commits from the following `reads` * event-log reads, modelling one concurrent writer the reader missed. diff --git a/workbench/sim-world/scenarios/index.ts b/workbench/sim-world/scenarios/index.ts index a265a957c0..3888791bee 100644 --- a/workbench/sim-world/scenarios/index.ts +++ b/workbench/sim-world/scenarios/index.ts @@ -38,6 +38,7 @@ import { scenario as inFlightAfterDecision } from './in-flight-after-decision.ts import { scenario as inFlightBeforeDecision } from './in-flight-before-decision.ts'; import { scenario as inFlightBeforeDecisionCounted } from './in-flight-before-decision-counted.ts'; import { scenario as longSleep } from './long-sleep.ts'; +import { scenario as lostStepDispatch } from './lost-step-dispatch.ts'; import { scenario as parallelSteps } from './parallel-steps.ts'; import { scenario as peekHookAfterBranch } from './peek-hook-after-branch.ts'; import { scenario as peekHookAtRegistration } from './peek-hook-at-registration.ts'; @@ -94,6 +95,7 @@ export const scenarios: ScenarioSpec[] = [ // ------------------------------------------------------------------------- stepRetriesTwice, parallelSteps, + lostStepDispatch, hookOnExecutionState, // ------------------------------------------------------------------------- diff --git a/workbench/sim-world/scenarios/lost-step-dispatch.ts b/workbench/sim-world/scenarios/lost-step-dispatch.ts new file mode 100644 index 0000000000..1f20e76a7d --- /dev/null +++ b/workbench/sim-world/scenarios/lost-step-dispatch.ts @@ -0,0 +1,32 @@ +import type { ScenarioSpec } from '@workflow/world-sim'; + +export const scenario: ScenarioSpec = { + id: 'lost-step-dispatch', + name: 'a queued step dispatch is accepted and lost', + description: + 'Four steps suspend together against an inline cap of three, so the ' + + 'fourth is handed to the queue. The queue takes the message, keeps its ' + + 'idempotency key, and never delivers it. Every later replay re-sends the ' + + 'dispatch and the queue absorbs it as a duplicate, so the step is created ' + + 'and never started and nothing else can move the run: the fourth ' + + 'Promise.all leg never settles. Recovery is the re-dispatch watchdog — ' + + 'past one watchdog interval the key carries an epoch the queue has not ' + + 'seen, and the wake armed at that boundary is what brings a replay back ' + + 'to notice. Without it the run has no outstanding work at all and stays ' + + 'running forever.', + workflow: 'fanOutStepsWorkflow', + input: ['x'], + script: async (sim) => { + // Hold an inline body pre-commit so the invocation is parked at a point + // where the overflow step's dispatch is already enqueued. + const held = await sim.writer + .step('fanA') + .runToEventProduced('step_completed'); + const dropped = sim.dropQueued( + (pending) => pending.find((m) => m.stepId !== undefined)?.messageId + ); + sim.check('a step dispatch was queued and dropped', dropped); + await held.release(); + }, + expect: { status: 'completed', output: 'a:x|b:x|c:x|d:x' }, +}; diff --git a/workbench/sim-world/workflows/index.ts b/workbench/sim-world/workflows/index.ts index 3b876e06ac..ca7d54ce4d 100644 --- a/workbench/sim-world/workflows/index.ts +++ b/workbench/sim-world/workflows/index.ts @@ -114,6 +114,43 @@ export async function parallelStepsWorkflow(input: string) { return `${a}|${b}`; } +async function fanA(input: string) { + 'use step'; + return `a:${input}`; +} + +async function fanB(input: string) { + 'use step'; + return `b:${input}`; +} + +async function fanC(input: string) { + 'use step'; + return `c:${input}`; +} + +async function fanD(input: string) { + 'use step'; + return `d:${input}`; +} + +/** + * Four steps suspend together, one more than the inline cap, so the last one + * is handed to the queue instead of running in the invocation. That single + * queued dispatch is the one a scenario can take away with `dropQueued` to ask + * what happens when the queue accepts a step message and loses it. + */ +export async function fanOutStepsWorkflow(input: string) { + 'use workflow'; + const parts = await Promise.all([ + fanA(input), + fanB(input), + fanC(input), + fanD(input), + ]); + return parts.join('|'); +} + // --------------------------------------------------------------------------- // 3. Approval — a hook and a step suspended together // --------------------------------------------------------------------------- From 51489b1fcea0f3b6bd0f7371307f0cf88b1a7f38 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 13:10:26 -0700 Subject: [PATCH 04/14] Recover a step whose invocation disappeared mid-body A pending step's queue dispatch is keyed by its correlation ID, and a queue dedupes that key for the lifetime of the message sent under it. So once a dispatch stops making progress, every later replay's re-send is absorbed and the run replays forever with one step nothing will finish. The watchdog already covered a step that was never delivered. It did not cover the other shape: the message is delivered, the step writes step_started, and the invocation running the body disappears before writing a terminal event. Inline ownership arms a backstop wake at the lease for exactly that, but the re-dispatch that wake triggers carried the bare correlation ID the queue had already claimed, so the recovery was deduped away and the run went silent permanently. Measured on the race repro: 432 replays in 57s, the last one 0.6s past the lease boundary, then nothing for the remaining 25 minutes of the run. Both shapes now share one deadline, dispatchLostAtMs: a watchdog interval after step_created for an unstarted step, the end of the ownership lease for a started one. Past it the key is epoch-scoped and a boundary wake is armed, and the epoch advances once per watchdog interval so a lost recovery is itself retried. Anchoring the started case on the lease rather than on a watchdog interval is what keeps healthy long-running bodies from being duplicated. A step in step_retrying stays out of scope: its retry is queued under the bare key with a backoff that can legitimately exceed either deadline. The repro harness called a run stuck after 4 minutes, well inside the runtime's own longest recovery deadline, so a run on its way back was reported as permanently stranded. Its run timeout is now derived from the lease plus a watchdog interval instead of being a second copy of a number the runtime owns. --- .changeset/wild-pandas-jam.md | 2 +- .github/workflows/event-log-race-repro.yml | 13 +- .../docs/v5/configuration/runtime-tuning.mdx | 3 +- .../core/e2e/event-log-race-repro.test.ts | 33 ++++- packages/core/src/runtime.ts | 26 ++-- .../core/src/runtime/step-dispatch.test.ts | 110 ++++++++++++++-- packages/core/src/runtime/step-dispatch.ts | 122 ++++++++++++------ 7 files changed, 234 insertions(+), 75 deletions(-) diff --git a/.changeset/wild-pandas-jam.md b/.changeset/wild-pandas-jam.md index 64cfb0259f..8efd1ed55b 100644 --- a/.changeset/wild-pandas-jam.md +++ b/.changeset/wild-pandas-jam.md @@ -2,4 +2,4 @@ '@workflow/core': patch --- -Re-dispatch a step that was created but never started, so a lost queue message no longer strands a run +Re-dispatch a pending step whose queue message is gone, so a lost dispatch or an invocation that disappears mid-step no longer strands a run diff --git a/.github/workflows/event-log-race-repro.yml b/.github/workflows/event-log-race-repro.yml index 6f78986f2a..b945de8c56 100644 --- a/.github/workflows/event-log-race-repro.yml +++ b/.github/workflows/event-log-race-repro.yml @@ -62,11 +62,14 @@ jobs: name: Event Log Race Repro runs-on: ubuntu-latest # Sized for the harness' default scale: its own test timeout is - # `budget_ms + run_timeout_ms + 60s` (~17 min at the defaults), and the rest is - # checkout, build, the deployment wait, and rendering the summary. A soak - # dispatch that raises `budget_ms` has to raise this too, or the runner kills - # the job before the summary is written. - timeout-minutes: 25 + # `budget_ms + run_timeout_ms + 60s` (~30 min at the defaults, where + # `run_timeout_ms` is derived from the runtime's inline-ownership lease so a + # recovering run is not misreported as stuck), and the rest is checkout, + # build, the deployment wait, and rendering the summary. Only a run that + # actually needs recovery spends that deadline; healthy attempts finish in + # tens of seconds. A soak dispatch that raises `budget_ms` has to raise this + # too, or the runner kills the job before the summary is written. + timeout-minutes: 40 if: ${{ github.event_name == 'workflow_dispatch' || contains(github.event.pull_request.labels.*.name, 'event-log-race-repro') }} permissions: contents: read diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index 8e9269d5e2..8b40493b05 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -164,7 +164,8 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL - Clamp: `10` to `900` - How long a step may sit created but never started before the runtime re-dispatches it. A queued step message is deduplicated on its correlation ID, so if that message is lost the step would otherwise never be delivered again and the run would keep replaying with nothing to execute. Past this interval the re-dispatch uses a key the queue has not seen, and the runtime arms a timer on the boundary so the check happens even when nothing else wakes the run. - Re-dispatch is at-least-once: if the original message was only slow, both deliveries execute the step body and the loser's result is discarded. Raise this if your queue's dispatch-to-start latency approaches the default. -- Only steps that have never started are covered. A step that has started is left to the inline ownership lease above. +- A step that has already started gets the same treatment on a later deadline: it is presumed alive for the inline ownership lease above, and only past the lease does its re-dispatch move to a fresh key. This is what recovers a step whose invocation disappeared mid-body. Once past that point, the interval here is also how often the re-dispatch is retried. +- Steps waiting on a scheduled retry (`step_retrying`) are left alone, since that retry is already queued with its own backoff. ## Workflow VM engine diff --git a/packages/core/e2e/event-log-race-repro.test.ts b/packages/core/e2e/event-log-race-repro.test.ts index b577df7b5c..4e7a0478c4 100644 --- a/packages/core/e2e/event-log-race-repro.test.ts +++ b/packages/core/e2e/event-log-race-repro.test.ts @@ -10,6 +10,10 @@ import { start as rawStart, resumeHook, } from '../src/runtime'; +import { + getInlineOwnershipLeaseSeconds, + getStepDispatchWatchdogSeconds, +} from '../src/runtime/constants'; import { getWorkflowMetadata, setupWorld, trackRun } from './utils'; /** @@ -158,6 +162,30 @@ function envBoolean(name: string, fallback: boolean) { // `workflow_dispatch` inputs straight through and `envNumber` treats an unset or // empty variable as absent, so a blank input lands on the value below rather // than on a second default maintained in YAML. +/** + * How long a run may take before the harness calls it `stuck`. + * + * A healthy attempt finishes in tens of seconds, so this is not a latency + * budget: it is the point past which a run is declared unrecoverable. That + * makes it meaningless to set it below the runtime's own longest recovery + * deadline. A step whose owning invocation disappears mid-body is presumed + * alive for the inline-ownership lease, and its re-dispatch reaches the queue + * one watchdog interval later at worst, so anything shorter reports a run that + * is on its way back as permanently stranded. + * + * Both deadlines are read from this process's environment, which matches the + * deployment under test only when neither side overrides them. Overriding the + * lease or the watchdog on the deployment means setting + * `EVENT_LOG_RACE_REPRO_RUN_TIMEOUT_MS` here to match. + */ +function defaultRunTimeoutMs(): number { + const recoveryMs = + (getInlineOwnershipLeaseSeconds() + getStepDispatchWatchdogSeconds()) * + 1000; + // Slack for the recovered dispatch to be delivered and the step to run. + return recoveryMs + 60_000; +} + const config: ReproConfig = { stepStormAttempts: envNumber('EVENT_LOG_RACE_REPRO_STEP_STORM_ATTEMPTS', 6), hookStormAttempts: envNumber('EVENT_LOG_RACE_REPRO_HOOK_STORM_ATTEMPTS', 6), @@ -173,7 +201,10 @@ const config: ReproConfig = { // absorb the worst case of one in-flight attempt draining its full // `runTimeoutMs` after the budget ends (see `testTimeoutMs` below). budgetMs: envNumber('EVENT_LOG_RACE_REPRO_BUDGET_MS', 12 * 60_000), - runTimeoutMs: envNumber('EVENT_LOG_RACE_REPRO_RUN_TIMEOUT_MS', 240_000), + runTimeoutMs: envNumber( + 'EVENT_LOG_RACE_REPRO_RUN_TIMEOUT_MS', + defaultRunTimeoutMs() + ), hookTimeoutMs: envNumber('EVENT_LOG_RACE_REPRO_HOOK_TIMEOUT_MS', 60_000), rounds: envNumber('EVENT_LOG_RACE_REPRO_ROUNDS', 6), width: envNumber('EVENT_LOG_RACE_REPRO_WIDTH', 8), diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index dc79edf7dd..98bfb16caa 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -3212,16 +3212,18 @@ export function workflowEntrypoint( requestedAt: new Date(), }, { - // Epoch-scoped once the step has gone a full - // watchdog interval without ever starting: the - // bare correlation ID is deduped by the queue - // for the lifetime of the message dispatched - // under it, so a dispatch that never produced - // a step_started can only be retried under a - // key the queue has not seen. Within an epoch - // every replay derives the same key, so - // concurrent wake replays still collapse to - // one message. See runtime/step-dispatch.ts. + // Epoch-scoped once the step's dispatch is + // presumed lost — a watchdog interval without + // a first start, or an expired ownership lease + // with no terminal event. The bare correlation + // ID is deduped by the queue for the lifetime + // of the message dispatched under it, so a + // dispatch that stopped making progress can + // only be retried under a key the queue has + // not seen. Within an epoch every replay + // derives the same key, so concurrent wake + // replays still collapse to one message. See + // runtime/step-dispatch.ts. idempotencyKey: stepDispatchIdempotencyKey( step, dispatchNowMs @@ -3233,8 +3235,8 @@ export function workflowEntrypoint( // Nothing else wakes a run whose only outstanding work // is a step dispatch that was lost, so the watchdog // needs its own timer: one delayed run continuation at - // the earliest boundary among the steps still awaiting - // a first start. Deduped on that boundary, so repeated + // the earliest boundary among the steps dispatched + // here. Deduped on that boundary, so repeated // suspensions within an interval arm it once. const dispatchWake = getStepDispatchWake( dispatchedSteps, diff --git a/packages/core/src/runtime/step-dispatch.test.ts b/packages/core/src/runtime/step-dispatch.test.ts index 861b1096ac..c7a1f08b0b 100644 --- a/packages/core/src/runtime/step-dispatch.test.ts +++ b/packages/core/src/runtime/step-dispatch.test.ts @@ -1,7 +1,11 @@ import { afterEach, describe, expect, it } from 'vitest'; import type { StepInvocationQueueItem } from '../global.js'; -import { getStepDispatchWatchdogSeconds } from './constants.js'; import { + getInlineOwnershipLeaseSeconds, + getStepDispatchWatchdogSeconds, +} from './constants.js'; +import { + dispatchLostAtMs, getStepDispatchWake, isStepAwaitingFirstStart, nextStepDispatchBoundaryMs, @@ -12,6 +16,19 @@ import { const WATCHDOG_ENV = 'WORKFLOW_STEP_DISPATCH_WATCHDOG_SECONDS'; const CREATED_AT = 1_000_000; +const STARTED_AT = CREATED_AT + 1_000; + +/** A step mid-body, its ownership lease still live at `atLeaseOffset(-1)`. */ +function startedStep( + overrides: Partial = {} +): StepInvocationQueueItem { + return makeStep({ lastStartedAt: STARTED_AT, ...overrides }); +} + +/** `nowMs` relative to the end of a started step's ownership lease. */ +function atLeaseOffset(offsetMs: number): number { + return STARTED_AT + getInlineOwnershipLeaseSeconds() * 1000 + offsetMs; +} function makeStep( overrides: Partial = {} @@ -32,6 +49,11 @@ function atIntervals(intervals: number): number { return CREATED_AT + intervals * getStepDispatchWatchdogSeconds() * 1000; } +/** One interval, in ms, at whatever the watchdog is currently configured to. */ +function intervalMs(): number { + return getStepDispatchWatchdogSeconds() * 1000; +} + afterEach(() => { delete process.env[WATCHDOG_ENV]; }); @@ -71,20 +93,40 @@ describe('isStepAwaitingFirstStart', () => { }); }); +describe('dispatchLostAtMs', () => { + it('is one watchdog interval after creation for an unstarted step', () => { + expect(dispatchLostAtMs(makeStep())).toBe(atIntervals(1)); + }); + + it('is the end of the ownership lease for a started step', () => { + expect(dispatchLostAtMs(startedStep())).toBe(atLeaseOffset(0)); + }); + + it('is undefined for a step whose retry is already queued', () => { + expect(dispatchLostAtMs(startedStep({ sawRetrying: true }))).toBeUndefined(); + }); + + it('is undefined when the world reports no creation timestamp', () => { + expect(dispatchLostAtMs(makeStep({ createdEventAt: undefined }))).toBe( + undefined + ); + }); +}); + describe('stepDispatchIdempotencyKey', () => { it('uses the bare correlation ID within the first watchdog interval', () => { const step = makeStep(); expect(stepDispatchIdempotencyKey(step, CREATED_AT)).toBe( step.correlationId ); - expect(stepDispatchIdempotencyKey(step, atIntervals(1) - 1)).toBe( + expect(stepDispatchIdempotencyKey(step, atIntervals(1))).toBe( step.correlationId ); }); it('moves to a fresh key once an interval has passed without a start', () => { const step = makeStep(); - expect(stepDispatchIdempotencyKey(step, atIntervals(1))).toBe( + expect(stepDispatchIdempotencyKey(step, atIntervals(1) + 1)).toBe( `${step.correlationId}:dispatch:1` ); expect(stepDispatchIdempotencyKey(step, atIntervals(2.5))).toBe( @@ -99,11 +141,33 @@ describe('stepDispatchIdempotencyKey', () => { expect(early).toBe(late); }); - it('keeps the bare key for a step that has already started', () => { - // A started step may be mid-body for far longer than the watchdog; only - // the ownership lease and its backstop may act on it. - const step = makeStep({ lastStartedAt: CREATED_AT + 1 }); - expect(stepDispatchIdempotencyKey(step, atIntervals(10))).toBe( + it('keeps the bare key for a started step while its lease is live', () => { + // A step mid-body may take far longer than the watchdog interval, so the + // ownership lease is the only deadline that may act on it. + const step = startedStep(); + expect(stepDispatchIdempotencyKey(step, atLeaseOffset(-1))).toBe( + step.correlationId + ); + }); + + it('moves to a fresh key once a started step outlives its lease', () => { + // The invocation that wrote step_started is gone: past the lease the + // inline backstop wake brings a replay back, and this is the key that + // makes its re-dispatch reach the queue rather than being deduped away. + const step = startedStep(); + expect(stepDispatchIdempotencyKey(step, atLeaseOffset(1))).toBe( + `${step.correlationId}:dispatch:1` + ); + expect(stepDispatchIdempotencyKey(step, atLeaseOffset(intervalMs()))).toBe( + `${step.correlationId}:dispatch:2` + ); + }); + + it('keeps the bare key for a step whose retry is already queued', () => { + // step_retrying means the owner scheduled the next attempt under the bare + // key, with a backoff that may legitimately exceed any deadline here. + const step = startedStep({ sawRetrying: true }); + expect(stepDispatchIdempotencyKey(step, atLeaseOffset(intervalMs()))).toBe( step.correlationId ); }); @@ -125,8 +189,8 @@ describe('stepDispatchIdempotencyKey', () => { it('honours the watchdog override', () => { process.env[WATCHDOG_ENV] = '10'; const step = makeStep(); - expect(stepDispatchEpoch(step, CREATED_AT + 10_000)).toBe(1); - expect(stepDispatchEpoch(step, CREATED_AT + 9_999)).toBe(0); + expect(stepDispatchEpoch(step, CREATED_AT + 10_001)).toBe(1); + expect(stepDispatchEpoch(step, CREATED_AT + 10_000)).toBe(0); }); it('clamps an override below the supported minimum', () => { @@ -150,10 +214,16 @@ describe('nextStepDispatchBoundaryMs', () => { ); }); + it('is the lease boundary for a started step still inside its lease', () => { + expect(nextStepDispatchBoundaryMs(startedStep(), STARTED_AT)).toBe( + atLeaseOffset(0) + ); + }); + it('is undefined for a step out of watchdog scope', () => { expect( nextStepDispatchBoundaryMs( - makeStep({ lastStartedAt: CREATED_AT }), + startedStep({ sawRetrying: true }), atIntervals(3) ) ).toBeUndefined(); @@ -161,13 +231,27 @@ describe('nextStepDispatchBoundaryMs', () => { }); describe('getStepDispatchWake', () => { - it('is undefined when no step is awaiting a first start', () => { + it('is undefined when no pending step has a lost-dispatch deadline', () => { expect( - getStepDispatchWake([makeStep({ lastStartedAt: CREATED_AT })], CREATED_AT) + getStepDispatchWake([startedStep({ sawRetrying: true })], CREATED_AT) ).toBeUndefined(); expect(getStepDispatchWake([], CREATED_AT)).toBeUndefined(); }); + it('re-arms itself past the lease so a lost recovery is retried', () => { + // The dispatch at the lease boundary computes epoch 0 and is deduped, so + // the wake it arms is what carries the run to the first fresh key. + const step = startedStep(); + const wake = getStepDispatchWake([step], atLeaseOffset(0)); + expect(wake?.idempotencyKey).toBe( + `${step.correlationId}:dispatch-wake:${atLeaseOffset(0)}` + ); + const next = getStepDispatchWake([step], atLeaseOffset(1)); + expect(next?.idempotencyKey).toBe( + `${step.correlationId}:dispatch-wake:${atLeaseOffset(intervalMs())}` + ); + }); + it('lands just past the boundary so the wake does not re-arm its own key', () => { const wake = getStepDispatchWake([makeStep()], CREATED_AT); expect(wake).toBeDefined(); diff --git a/packages/core/src/runtime/step-dispatch.ts b/packages/core/src/runtime/step-dispatch.ts index 4625e8d2dd..5b15226a00 100644 --- a/packages/core/src/runtime/step-dispatch.ts +++ b/packages/core/src/runtime/step-dispatch.ts @@ -1,38 +1,50 @@ import { + getInlineOwnershipLeaseSeconds, getStepDispatchWatchdogSeconds, MAX_STEP_DISPATCH_WATCHDOG_SECONDS, } from './constants.js'; /** - * Re-dispatch watchdog for a step that was created but never started. + * Re-dispatch watchdog for a pending step whose dispatch is gone. * * A pending step is dispatched as a queue message keyed by its correlation * ID. Queues dedupe an idempotency key for the lifetime of the original * message, so that key is effectively permanent: once a message has been * accepted under it, every later replay's re-send is silently absorbed. That * is the intended behaviour while the message is doing its job — concurrent - * wake replays must not multiply the dispatch — but it also means a step - * whose message never produces a `step_started` can never be dispatched - * again. The run then keeps replaying forever with one pending step that - * nothing will ever execute: no divergence, no error, no terminal state. + * wake replays must not multiply the dispatch — but it also means a step whose + * message stops making progress can never be dispatched again. The run then + * keeps replaying forever with one pending step that nothing will ever + * execute: no divergence, no error, no terminal state. * - * The watchdog gives those dispatches an epoch. Every replay derives the - * epoch from the step's durable `step_created` timestamp, so concurrent - * replays agree on the key and fan-out stays capped at one message per epoch, - * while a step still unstarted a full watchdog interval later gets a key the - * queue has never seen and is dispatched again. + * Two ways a dispatch stops making progress, both observed on the race repro: * - * Scope: only steps awaiting their FIRST `step_started`. A step that has - * started is either running (its body has no completion deadline the client - * can see, so re-dispatching on a timer would duplicate healthy long-running - * work) or inline-owned (covered by the ownership lease and its backstop in - * step-ownership.ts). + * 1. The message is accepted and never delivered, so no `step_started` is + * ever written. + * 2. The message is delivered, the step writes `step_started`, and the + * invocation executing the body disappears before writing a terminal + * event. Inline-owned steps have a backstop wake at their ownership lease + * for exactly this, but the wake only brings a replay back: the re-dispatch + * it then attempts carried the bare correlation ID, which the queue had + * already claimed, so the recovery was absorbed and the run went silent for + * good. + * + * The watchdog gives those dispatches an epoch, counted from the point the + * dispatch is presumed lost (see `dispatchLostAtMs`). Every replay derives the + * epoch from durable event timestamps, so concurrent replays agree on the key + * and fan-out stays capped at one message per epoch, while a step still + * pending an interval later gets a key the queue has never seen and is + * dispatched again. + * + * Out of scope: a step in `step_retrying`. Its retry is already queued under + * the bare key with a backoff that can legitimately exceed any interval here, + * and re-dispatching it would duplicate work that is scheduled and healthy. * * Re-dispatch is at-least-once: if the original message was merely slow * rather than lost, both deliveries execute the step body and the loser's - * terminal write is rejected as a conflict. The default interval is set well - * above normal dispatch-to-start latency so that trade only happens for - * dispatches that really are gone. + * terminal write is rejected as a conflict. Both deadlines are set well above + * normal latency so that trade only happens for dispatches that really are + * gone. */ /** @@ -60,8 +72,7 @@ export interface StepDispatchState { /** * Whether a pending step is waiting for its first `step_started`: a durable * creation timestamp is known, no start has been observed, and no - * `step_retrying` (which implies a prior start). Steps outside this state keep - * the bare correlation-ID key. + * `step_retrying` (which implies a prior start). */ export function isStepAwaitingFirstStart(step: StepDispatchState): boolean { return ( @@ -72,21 +83,47 @@ export function isStepAwaitingFirstStart(step: StepDispatchState): boolean { } /** - * How many full watchdog intervals have elapsed since the step's - * `step_created`. 0 means the current dispatch is still within its first - * interval, or the step is out of scope / has no usable creation timestamp - * (worlds whose events lack them) — in both cases dispatch keeps today's - * behaviour exactly. + * The instant a pending step's current dispatch is presumed lost, as an + * absolute epoch-ms timestamp, or undefined when the step is out of scope. + * + * Unstarted steps are presumed lost one watchdog interval after their durable + * `step_created`: nothing else bounds how long a dispatch may sit before it + * produces a start. + * + * Started steps are presumed lost at the end of their ownership lease. That is + * the same deadline `stepLeaseRemainingSeconds` uses to schedule the inline + * backstop wake, so the wake and the key it re-dispatches under move together. + * Anchoring on the lease rather than on a watchdog interval is what keeps + * healthy long-running step bodies from being duplicated: the lease is the + * runtime's existing statement of how long an executing step may be presumed + * alive. + */ +export function dispatchLostAtMs(step: StepDispatchState): number | undefined { + if (step.sawRetrying === true) return undefined; + if (step.lastStartedAt !== undefined) { + return step.lastStartedAt + getInlineOwnershipLeaseSeconds() * 1000; + } + if (step.createdEventAt === undefined) return undefined; + return step.createdEventAt + getStepDispatchWatchdogSeconds() * 1000; +} + +/** + * How many re-dispatches the watchdog has reached for this step. 0 means the + * current dispatch is not yet presumed lost, or the step is out of scope (a + * retry in flight, or a world whose events carry no usable timestamp) — in + * both cases dispatch keeps the bare correlation-ID key and today's behaviour + * exactly. Past that the epoch advances once per watchdog interval, so a + * re-dispatch that is itself lost is followed by another. */ export function stepDispatchEpoch( step: StepDispatchState, nowMs: number ): number { - const createdEventAt = step.createdEventAt; - if (createdEventAt === undefined || !isStepAwaitingFirstStart(step)) return 0; - const elapsedMs = nowMs - createdEventAt; + const lostAtMs = dispatchLostAtMs(step); + if (lostAtMs === undefined) return 0; + const elapsedMs = nowMs - lostAtMs; if (!Number.isFinite(elapsedMs) || elapsedMs <= 0) return 0; - return Math.floor(elapsedMs / (getStepDispatchWatchdogSeconds() * 1000)); + return Math.floor(elapsedMs / (getStepDispatchWatchdogSeconds() * 1000)) + 1; } /** @@ -108,32 +145,33 @@ export function stepDispatchIdempotencyKey( /** * When the step's next watchdog boundary falls, as an absolute epoch-ms - * timestamp, or undefined when the step is out of scope. Derived from the - * durable creation timestamp so every replay computes the same boundary, - * which is what lets the wake armed for it dedupe across replays. + * timestamp, or undefined when the step is out of scope. Derived from durable + * event timestamps so every replay computes the same boundary, which is what + * lets the wake armed for it dedupe across replays. */ export function nextStepDispatchBoundaryMs( step: StepDispatchState, nowMs: number ): number | undefined { - const createdEventAt = step.createdEventAt; - if (createdEventAt === undefined || !isStepAwaitingFirstStart(step)) { - return undefined; - } - const watchdogMs = getStepDispatchWatchdogSeconds() * 1000; - return createdEventAt + (stepDispatchEpoch(step, nowMs) + 1) * watchdogMs; + const lostAtMs = dispatchLostAtMs(step); + if (lostAtMs === undefined) return undefined; + const epoch = stepDispatchEpoch(step, nowMs); + // Epoch 0 has not reached the lost-at instant yet, so that instant is itself + // the next boundary. Past it, each epoch lasts one watchdog interval. + if (epoch === 0) return lostAtMs; + return lostAtMs + epoch * getStepDispatchWatchdogSeconds() * 1000; } /** - * The wake a suspension arms so an unstarted step's next watchdog boundary is + * The wake a suspension arms so a pending step's next watchdog boundary is * observed even when nothing else would wake the run. Without it the watchdog * only helps runs that happen to keep receiving hooks or wait timers; a run * whose sole outstanding work is the lost dispatch would never replay again. * * At most one wake per suspension: the earliest boundary among the pending - * unstarted steps, since a replay at that point re-evaluates all of them. Ties - * break on correlation ID so concurrent replays pick the same step and - * therefore the same key. + * steps, since a replay at that point re-evaluates all of them. Ties break on + * correlation ID so concurrent replays pick the same step and therefore the + * same key. */ export function getStepDispatchWake( steps: StepDispatchState[], From c2972221b0dce71795483dd4777a66b87ac5d372 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 13:14:34 -0700 Subject: [PATCH 05/14] Format --- packages/core/src/runtime/step-dispatch.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/core/src/runtime/step-dispatch.test.ts b/packages/core/src/runtime/step-dispatch.test.ts index c7a1f08b0b..d9580c52ea 100644 --- a/packages/core/src/runtime/step-dispatch.test.ts +++ b/packages/core/src/runtime/step-dispatch.test.ts @@ -103,7 +103,9 @@ describe('dispatchLostAtMs', () => { }); it('is undefined for a step whose retry is already queued', () => { - expect(dispatchLostAtMs(startedStep({ sawRetrying: true }))).toBeUndefined(); + expect( + dispatchLostAtMs(startedStep({ sawRetrying: true })) + ).toBeUndefined(); }); it('is undefined when the world reports no creation timestamp', () => { From df221035401c59b90ceffe3c7eeda103ae997b0e Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 13:15:16 -0700 Subject: [PATCH 06/14] Give the repro's stuck deadline slack for a late-losing run --- packages/core/e2e/event-log-race-repro.test.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/packages/core/e2e/event-log-race-repro.test.ts b/packages/core/e2e/event-log-race-repro.test.ts index 4e7a0478c4..873dcf90a1 100644 --- a/packages/core/e2e/event-log-race-repro.test.ts +++ b/packages/core/e2e/event-log-race-repro.test.ts @@ -182,8 +182,11 @@ function defaultRunTimeoutMs(): number { const recoveryMs = (getInlineOwnershipLeaseSeconds() + getStepDispatchWatchdogSeconds()) * 1000; - // Slack for the recovered dispatch to be delivered and the step to run. - return recoveryMs + 60_000; + // The recovery deadline is measured from the lost step's start, not from the + // run's, so the slack has to cover a run that gets that far in before it + // loses a step, plus delivering the recovered dispatch and finishing the + // remaining rounds. + return recoveryMs + 3 * 60_000; } const config: ReproConfig = { From aab960233f9c160b2242983951769267329c2e90 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 13:16:06 -0700 Subject: [PATCH 07/14] Cover a lazily-created inline step in the lost-dispatch deadline --- packages/core/src/runtime/step-dispatch.test.ts | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/packages/core/src/runtime/step-dispatch.test.ts b/packages/core/src/runtime/step-dispatch.test.ts index d9580c52ea..d9d4d3d40f 100644 --- a/packages/core/src/runtime/step-dispatch.test.ts +++ b/packages/core/src/runtime/step-dispatch.test.ts @@ -113,6 +113,15 @@ describe('dispatchLostAtMs', () => { undefined ); }); + + it('covers a started step created lazily by its own inline start', () => { + // A lazily-created inline step has no separate step_created to date, so + // the start it wrote is the only timestamp there is — and it is the one + // the lease is measured from anyway. + expect(dispatchLostAtMs(startedStep({ createdEventAt: undefined }))).toBe( + atLeaseOffset(0) + ); + }); }); describe('stepDispatchIdempotencyKey', () => { From 22f54b9ed0fbf6a95011eba3b9efe6746dce92c2 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 13:37:41 -0700 Subject: [PATCH 08/14] chore: order type import in quickjs entrypoint --- packages/core/src/runtime/quickjs-entrypoint.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index c4adaeead3..132f1c2860 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -70,8 +70,8 @@ import { import { ReplayBudget } from './replay-budget.js'; import { getStepDispatchWake, - stepDispatchEpoch, type StepDispatchState, + stepDispatchEpoch, } from './step-dispatch.js'; import { executeStep, type StepExecutionResult } from './step-executor.js'; import { runStepSingleFlight } from './step-single-flight.js'; From cc33d4c5564d994754d461362bcc87285a1765bb Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 14:16:37 -0700 Subject: [PATCH 09/14] Let the dispatch wake reach a started step's lease boundary The wake's delay was capped at one watchdog interval, but a started step's boundary sits at the end of its ownership lease, ~13 minutes further out. The wake landed early, computed the same epoch, re-armed the boundary-keyed message the queue still held a claim for, and was absorbed, leaving the run with no timer at all. The ceiling now covers the larger of the two deadlines and exists only to bound clock skew. Also corrects the module docs: an unacked queue delivery is redelivered on its own, so the watchdog is not about lost messages. It is about dispatches that ended without a terminal event and have nothing outstanding to retry them. --- packages/core/src/runtime.ts | 10 +-- .../core/src/runtime/step-dispatch.test.ts | 38 +++++++++++- packages/core/src/runtime/step-dispatch.ts | 61 ++++++++++--------- packages/core/src/runtime/step-ownership.ts | 6 +- 4 files changed, 77 insertions(+), 38 deletions(-) diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index cbac62878e..93821e40d5 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -3403,11 +3403,11 @@ export function workflowEntrypoint( // this step's dispatch is presumed lost: a // watchdog interval without a first start, or // an expired ownership lease with no terminal - // event. The queue dedupes a key for the - // lifetime of the message sent under it, so a - // dispatch that stopped making progress can - // only be retried under a key the queue has - // not seen. Within an epoch every replay + // event. A key claim outlives the message sent + // under it, so a dispatch that stopped short of + // a terminal event can only be retried under a + // key the queue has not seen. Within an epoch + // every replay // derives the same key, so concurrent wake // replays still collapse to one message. See // stepDispatchIdempotencyKey and diff --git a/packages/core/src/runtime/step-dispatch.test.ts b/packages/core/src/runtime/step-dispatch.test.ts index b0843f2496..e852be7b81 100644 --- a/packages/core/src/runtime/step-dispatch.test.ts +++ b/packages/core/src/runtime/step-dispatch.test.ts @@ -12,6 +12,7 @@ import { nextStepDispatchBoundaryMs, stepDispatchEpoch, } from './step-dispatch.js'; +import { WAIT_CONTINUATION_MAX_DELAY_SECONDS } from './wait-continuation.js'; const WATCHDOG_ENV = 'WORKFLOW_STEP_DISPATCH_WATCHDOG_SECONDS'; @@ -290,6 +291,21 @@ describe('getStepDispatchWake', () => { expect(wake?.idempotencyKey).toBe(`step_01ABC:dispatch-wake:${boundary}`); }); + it("reaches a started step's boundary, which is a whole lease away", () => { + // A started step's boundary sits at the end of its ownership lease, far + // beyond one watchdog interval. A delay capped at the interval would land + // ~13 minutes early, compute the same epoch, re-arm the same key the queue + // still holds a claim for, and be dropped, leaving the run with no timer + // at all. An early wake is worse than no wake: it also spends the key. + const step = startedStep(); + const nowMs = STARTED_AT; + const wake = getStepDispatchWake([step], nowMs); + expect(wake).toBeDefined(); + const boundary = nextStepDispatchBoundaryMs(step, nowMs); + expect(boundary).toBe(atLeaseOffset(0)); + expect(nowMs + wake!.delaySeconds * 1000).toBeGreaterThan(boundary!); + }); + it('picks the earliest boundary among several pending steps', () => { const older = makeStep({ correlationId: 'step_older', @@ -318,9 +334,27 @@ describe('getStepDispatchWake', () => { it('keeps the delay inside the queue-supported range at the maximum watchdog', () => { process.env[WATCHDOG_ENV] = String(Number.MAX_SAFE_INTEGER); - const clamped = getStepDispatchWatchdogSeconds(); + const boundary = atIntervals(1); const wake = getStepDispatchWake([makeStep()], CREATED_AT); - expect(wake?.delaySeconds).toBe(clamped); + expect(wake).toBeDefined(); + // Still past the boundary, and still a delay a queue message can carry. + expect(CREATED_AT + wake!.delaySeconds * 1000).toBeGreaterThan(boundary); + expect(wake!.delaySeconds).toBeLessThanOrEqual( + WAIT_CONTINUATION_MAX_DELAY_SECONDS + ); + }); + + it('bounds the delay when a step timestamp is stamped in the future', () => { + // The ceiling exists only for clock skew: a step_created far ahead of the + // local clock would otherwise ask for a delay no queue would accept. + const skewed = makeStep({ createdEventAt: CREATED_AT + 10 ** 12 }); + const wake = getStepDispatchWake([skewed], CREATED_AT); + expect(wake!.delaySeconds).toBeLessThanOrEqual( + Math.max( + getStepDispatchWatchdogSeconds(), + getInlineOwnershipLeaseSeconds() + ) + 1 + ); }); it('still asks for a positive delay when the boundary is already past', () => { diff --git a/packages/core/src/runtime/step-dispatch.ts b/packages/core/src/runtime/step-dispatch.ts index c2479c04c7..aa430a8ed8 100644 --- a/packages/core/src/runtime/step-dispatch.ts +++ b/packages/core/src/runtime/step-dispatch.ts @@ -1,33 +1,30 @@ import { getInlineOwnershipLeaseSeconds, getStepDispatchWatchdogSeconds, - MAX_STEP_DISPATCH_WATCHDOG_SECONDS, } from './constants.js'; /** * Re-dispatch watchdog for a pending step whose dispatch is gone. * * A pending step is dispatched as a queue message keyed by the step's identity - * (`stepDispatchIdempotencyKey`). Queues dedupe an idempotency key for the - * lifetime of the original message, so that key is effectively permanent: once - * a message has been accepted under it, every later replay's re-send is - * silently absorbed. That - * is the intended behaviour while the message is doing its job — concurrent - * wake replays must not multiply the dispatch — but it also means a step whose - * message stops making progress can never be dispatched again. The run then - * keeps replaying forever with one pending step that nothing will ever - * execute: no divergence, no error, no terminal state. + * (`stepDispatchIdempotencyKey`). A key claim outlives the message sent under + * it: queues record the claim with a TTL of their own (a day is typical) and + * do not release it when the message is delivered, acked, or exhausted, so a + * later send under the same key produces no message at all. That is the + * intended behaviour while the dispatch is doing its job — concurrent wake + * replays must not multiply it — but it also means a step whose dispatch + * stopped short of a terminal event can never be dispatched again by re-sending + * the same key. The run then keeps replaying with one pending step that nothing + * will execute: no divergence, no error, no terminal state. * - * Two ways a dispatch stops making progress, both observed on the race repro: - * - * 1. The message is accepted and never delivered, so no `step_started` is - * ever written. - * 2. The message is delivered, the step writes `step_started`, and the - * invocation executing the body disappears before writing a terminal - * event. Inline-owned steps have a backstop wake at their ownership lease - * for exactly this, but the wake only brings a replay back: the re-dispatch - * it then attempts carried the same key the queue had already claimed, so - * the recovery was absorbed and the run went silent for good. + * What this does NOT assume is that the queue lost a message. Delivery is + * at-least-once, and an unacked delivery comes back on its own, so a redundant + * send being absorbed is harmless for a dispatch that is still in flight. The + * dispatches this matters for are the ones nothing will retry: a message that + * was acked without the step reaching a terminal event (the step's execution + * ended somewhere the queue considers success), and a send that collapsed into + * an existing claim whose message is already finished. Both leave the step + * pending with nothing outstanding, which is what the epoch below re-opens. * * The watchdog gives those dispatches an epoch, counted from the point the * dispatch is presumed lost (see `dispatchLostAtMs`), which the dispatch key is @@ -178,14 +175,22 @@ export function getStepDispatchWake( if (!earliest) return undefined; // A second of slack past the boundary, because a wake that lands even // marginally early computes the same epoch, re-arms the same key, and is - // deduped away — leaving the run with no timer at all. Clamped to the - // watchdog interval so a creation timestamp in the future (clock skew) - // cannot ask for a delay above the queue's per-message maximum, and floored - // at 1s because a boundary already past still has to be a valid delay. - const maxDelaySeconds = Math.min( - getStepDispatchWatchdogSeconds() + 1, - MAX_STEP_DISPATCH_WATCHDOG_SECONDS - ); + // deduped away — leaving the run with no timer at all. The delay must + // therefore never be clamped BELOW the distance to the boundary it is keyed + // on: an early wake is worse than no wake, since it also burns the key. + // + // The ceiling exists only to bound clock skew (a `step_created` or + // `step_started` stamped in the future would otherwise ask for a delay above + // the queue's per-message maximum). A boundary is at most one watchdog + // interval past an unstarted step's creation or one ownership lease past a + // started step's latest start, so the larger of the two covers every + // legitimate boundary and nothing else. Floored at 1s because a boundary + // already past still has to be a valid delay. + const maxDelaySeconds = + Math.max( + getStepDispatchWatchdogSeconds(), + getInlineOwnershipLeaseSeconds() + ) + 1; const delaySeconds = Math.min( maxDelaySeconds, Math.max(1, Math.ceil((earliest.boundaryMs - nowMs) / 1000) + 1) diff --git a/packages/core/src/runtime/step-ownership.ts b/packages/core/src/runtime/step-ownership.ts index 021511a175..ea173bb6c0 100644 --- a/packages/core/src/runtime/step-ownership.ts +++ b/packages/core/src/runtime/step-ownership.ts @@ -59,10 +59,10 @@ export function stepLeaseRemainingSeconds( * pending backstop per step. But when owner recovery re-stamps the step * (queue redelivery of the owning message → new `step_started` → new * `lastStartedAt`), the key CHANGES. This is load-bearing for liveness: - * queues dedupe an idempotency key for the lifetime of the original - * message — including while a delivery of it is in flight — so a backstop + * a key claim outlives the message sent under it, so a backstop * that fires during a refreshed lease and tries to re-arm under a fixed key - * would dedupe against ITSELF and be dropped, leaving no escape hatch if + * would dedupe against its own spent claim and be dropped, leaving no + * escape hatch if * the recovered owner later dies without further redeliveries. The epoch * suffix gives the re-arm a fresh key. Pending backstops are bounded by the * number of ownership epochs, i.e. the queue's redelivery budget for the From 6d087dbd974142b89bc954db9727cfcd00ca9b5c Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 14:23:25 -0700 Subject: [PATCH 10/14] Correct the watchdog's user-facing description The watchdog does not compensate for a queue dropping messages: an unacked delivery is redelivered on its own. It re-opens a dispatch that ended without a terminal event and has nothing outstanding to retry it. --- .changeset/wild-pandas-jam.md | 2 +- docs/content/docs/v5/configuration/runtime-tuning.mdx | 5 +++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/.changeset/wild-pandas-jam.md b/.changeset/wild-pandas-jam.md index 8efd1ed55b..0a39c38020 100644 --- a/.changeset/wild-pandas-jam.md +++ b/.changeset/wild-pandas-jam.md @@ -2,4 +2,4 @@ '@workflow/core': patch --- -Re-dispatch a pending step whose queue message is gone, so a lost dispatch or an invocation that disappears mid-step no longer strands a run +Re-dispatch a pending step whose dispatch ended without a terminal event, so a run is no longer stranded with a step nothing will execute diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index cb5cfdc606..2bb1c97da4 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -170,9 +170,10 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL - Default: `60` - Clamp: `10` to `900` -- How long a step may sit created but never started before the runtime re-dispatches it. A queued step message is deduplicated on a key derived from the step's identity, so if that message is lost the step would otherwise never be delivered again and the run would keep replaying with nothing to execute. Past this interval the re-dispatch uses a key the queue has not seen, and the runtime arms a timer on the boundary so the check happens even when nothing else wakes the run. +- How long a step may sit created but never started before the runtime re-dispatches it. A queued step message is deduplicated on a key derived from the step's identity, and a queue holds that claim on its own schedule rather than releasing it when the message finishes. So a dispatch that ends without the step reaching a terminal event cannot be revived by sending the same key again, and the run keeps replaying with nothing to execute. Past this interval the re-dispatch uses a key the queue has not seen, and the runtime arms a timer on the boundary so the check happens even when nothing else wakes the run. +- This does not compensate for a queue dropping messages. An undelivered or unacknowledged message is redelivered by the queue itself, and a redundant send being deduplicated is harmless while a dispatch is still in flight. What the watchdog re-opens is a dispatch with nothing outstanding: the message was acknowledged, or collapsed into a claim whose message has already finished, without a terminal event for the step. - Re-dispatch is at-least-once: if the original message was only slow, both deliveries execute the step body and the loser's result is discarded. Raise this if your queue's dispatch-to-start latency approaches the default. -- A step that has already started gets the same treatment on a later deadline: it is presumed alive for the inline ownership lease above, and only past the lease does its re-dispatch move to a fresh key. This is what recovers a step whose invocation disappeared mid-body. Once past that point, the interval here is also how often the re-dispatch is retried. +- A step that has already started gets the same treatment on a later deadline: it is presumed alive for the inline ownership lease above, and only past the lease does its re-dispatch move to a fresh key. Once past that point, the interval here is also how often the re-dispatch is retried. - Steps waiting on a scheduled retry (`step_retrying`) are left alone, since that retry is already queued with its own backoff. ## Workflow VM engine From 8fc86897b30a97923429cf695485c6a1bde25904 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 14:34:26 -0700 Subject: [PATCH 11/14] Arm the quickjs dispatch wake independently of the wait continuation The quickjs suspension returned as soon as it scheduled a delayed wait continuation, so the watchdog timer was skipped whenever the run held both a pending wait and a pending step. The step would then not be re-evaluated until the wait elapsed, which can be hours out. The node engine already sends both messages from one suspension. --- .../core/src/runtime/quickjs-entrypoint.ts | 59 ++++++++++--------- 1 file changed, 31 insertions(+), 28 deletions(-) diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index 132f1c2860..650e0797dd 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -1813,6 +1813,36 @@ export async function runWorkflowWithQuickJS(params: { return; } + // Nothing else wakes a run whose only outstanding work is a step dispatch + // that ended without a terminal event, so arm a timer on the soonest + // watchdog boundary among the pending steps. Armed independently of the + // wait continuation below: a run holding both a pending wait and a lost + // dispatch would otherwise not re-evaluate the step until the wait + // elapsed, which can be hours away. Same scheme as the node engine's + // suspension wake (step-dispatch.ts): the key is derived from the + // boundary, so concurrent invocations arm one timer, not one each. + const dispatchWake = getStepDispatchWake( + pendingOperations + .filter((op): op is PendingStep => op.type === 'step') + .map(dispatchStateOf), + Date.now() + ); + if (dispatchWake) { + await queueMessage( + world, + getWorkflowQueueName(workflowRun.workflowName, namespace), + { + runId, + traceCarrier: await nextTraceCarrier(), + requestedAt: new Date(), + }, + { + delaySeconds: dispatchWake.delaySeconds, + idempotencyKey: dispatchWake.idempotencyKey, + } + ); + } + if (soonestWait) { // Delayed continuation for the soonest pending wait the loop has // not already scheduled. The dispatch helper handles delay @@ -1823,6 +1853,7 @@ export async function runWorkflowWithQuickJS(params: { action: 'schedule_wait_timeout', timeoutSeconds: soonestWait.seconds, waitCorrelationId: soonestWait.correlationId, + dispatchWakeSeconds: dispatchWake?.delaySeconds, }); scheduledWaitContinuations.add(soonestWait.correlationId); await queueMessage( @@ -1841,34 +1872,6 @@ export async function runWorkflowWithQuickJS(params: { return; } - // Nothing else will wake a run whose only outstanding work is a step - // dispatch the queue lost, so arm a timer on the soonest watchdog - // boundary among the pending steps. Same - // scheme as the node engine's suspension wake (step-dispatch.ts): the - // key is derived from the durable creation timestamp, so concurrent - // invocations arm one timer, not one each. - const dispatchWake = getStepDispatchWake( - pendingOperations - .filter((op): op is PendingStep => op.type === 'step') - .map(dispatchStateOf), - Date.now() - ); - if (dispatchWake) { - await queueMessage( - world, - getWorkflowQueueName(workflowRun.workflowName, namespace), - { - runId, - traceCarrier: await nextTraceCarrier(), - requestedAt: new Date(), - }, - { - delaySeconds: dispatchWake.delaySeconds, - idempotencyKey: dispatchWake.idempotencyKey, - } - ); - } - wfdiag('exit_suspended', { action: 'awaiting_external', pendingOpsCount: pendingOperations.length, From ea90d525b04bec04bbea34c6ca3260d9a7414188 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 14:39:47 -0700 Subject: [PATCH 12/14] Reframe the sim scenario's fault model: a spent key, not a dropped message --- packages/world-sim/README.md | 13 +++++++---- .../sim-world/scenarios/lost-step-dispatch.ts | 22 ++++++++++--------- 2 files changed, 21 insertions(+), 14 deletions(-) diff --git a/packages/world-sim/README.md b/packages/world-sim/README.md index 8755e9f4b6..ade3b47f79 100644 --- a/packages/world-sim/README.md +++ b/packages/world-sim/README.md @@ -494,18 +494,23 @@ Note the missing `await` — awaiting it here would wait for the delivery to wake, fire it, await the hold, and the two are now interleaved. Await the returned promise at the end to assert it found something. -#### `dropQueued`, and what a lost message costs +#### `dropQueued`, and what a spent key costs `dropQueued` is the same take without the delivery. The message is gone and it is never settled, so the queue keeps its idempotency key for the rest of the scenario: every later re-send under that key is absorbed, and the only writer that can get the work dispatched again is one that changes the key. -That second half is what makes the fault worth simulating. A queue that forgot +This is not a claim that queues drop messages. Delivery is at-least-once, so an +outstanding delivery comes back on its own. What `dropQueued` models is the end +state shared by every dispatch that stopped short of a terminal event: nothing +outstanding to redeliver, and a key already spent. + +That second half is what makes the fault worth simulating. A queue that released the key on drop would recover on the next replay for free, and nothing about the runtime's key scheme would be under test. `lost-step-dispatch` is the scenario -that asks the question: a step created, its dispatch dropped, and a run whose -only remaining work is a message that will never arrive. +that asks the question: a step created, its dispatch gone, and a run whose only +remaining work sits behind a key no re-send can reach. ## Extending the simulator diff --git a/workbench/sim-world/scenarios/lost-step-dispatch.ts b/workbench/sim-world/scenarios/lost-step-dispatch.ts index 1f20e76a7d..0c4ed060e5 100644 --- a/workbench/sim-world/scenarios/lost-step-dispatch.ts +++ b/workbench/sim-world/scenarios/lost-step-dispatch.ts @@ -2,18 +2,20 @@ import type { ScenarioSpec } from '@workflow/world-sim'; export const scenario: ScenarioSpec = { id: 'lost-step-dispatch', - name: 'a queued step dispatch is accepted and lost', + name: 'a queued step dispatch ends without a terminal event', description: 'Four steps suspend together against an inline cap of three, so the ' + - 'fourth is handed to the queue. The queue takes the message, keeps its ' + - 'idempotency key, and never delivers it. Every later replay re-sends the ' + - 'dispatch and the queue absorbs it as a duplicate, so the step is created ' + - 'and never started and nothing else can move the run: the fourth ' + - 'Promise.all leg never settles. Recovery is the re-dispatch watchdog — ' + - 'past one watchdog interval the key carries an epoch the queue has not ' + - 'seen, and the wake armed at that boundary is what brings a replay back ' + - 'to notice. Without it the run has no outstanding work at all and stays ' + - 'running forever.', + 'fourth is handed to the queue. The message is then removed without ' + + 'settling the step, which is how the sim models the end state of any ' + + 'dispatch that stopped short of a terminal event: nothing is outstanding ' + + 'for the queue to redeliver, while the idempotency claim on the key ' + + 'survives. Every later replay re-sends the dispatch and the claim absorbs ' + + 'it, so the step is created and never started and nothing else can move ' + + 'the run: the fourth Promise.all leg never settles. Recovery is the ' + + 're-dispatch watchdog. Past one watchdog interval the key carries an ' + + 'epoch the queue has not seen, and the wake armed at that boundary is ' + + 'what brings a replay back to notice. Without it the run has no ' + + 'outstanding work at all and stays running forever.', workflow: 'fanOutStepsWorkflow', input: ['x'], script: async (sim) => { From 596021a92269f9f719f61d534f782f5d2f756e8e Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 14:43:33 -0700 Subject: [PATCH 13/14] Cover the quickjs suspension arming the dispatch wake alongside a pending wait --- .../quickjs-entrypoint.dispatch-wake.test.ts | 237 ++++++++++++++++++ 1 file changed, 237 insertions(+) create mode 100644 packages/core/src/runtime/quickjs-entrypoint.dispatch-wake.test.ts diff --git a/packages/core/src/runtime/quickjs-entrypoint.dispatch-wake.test.ts b/packages/core/src/runtime/quickjs-entrypoint.dispatch-wake.test.ts new file mode 100644 index 0000000000..98a1aa8200 --- /dev/null +++ b/packages/core/src/runtime/quickjs-entrypoint.dispatch-wake.test.ts @@ -0,0 +1,237 @@ +/** + * Pins the QuickJS engine's suspension tail arming the re-dispatch watchdog + * wake, and arming it independently of the wait continuation. + * + * A pending step whose `step_created` is durable is neither an inline + * candidate nor an overflow dispatch (both require a step with no created + * event), so its dispatch is already out there and the only thing that brings a + * replay back to notice it stopped short of a terminal event is the boundary + * wake. A run that also holds a pending wait must still get that wake: the wait + * can be hours out, and until it elapses nothing else replays the run. + * + * The QuickJS VM itself is mocked (its WASM import chain is irrelevant to what + * the tail queues): `startQuickJSWorkflow` reports the suspension directly. + */ +import { + type CreateEventRequest, + type Event, + SPEC_VERSION_CURRENT, + type WorkflowRun, + type World, +} from '@workflow/world'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { + getStepDispatchWake, + nextStepDispatchBoundaryMs, +} from './step-dispatch.js'; +import { setWorld } from './world.js'; + +vi.mock('@vercel/functions', () => ({ waitUntil: vi.fn() })); +vi.mock('./get-port-lazy.js', () => ({ + getPortLazy: vi.fn().mockResolvedValue(3000), +})); + +const startQuickJSWorkflow = vi.fn(); +vi.mock('./quickjs-runtime.js', () => ({ + startQuickJSWorkflow: (...args: unknown[]) => startQuickJSWorkflow(...args), +})); + +const RUN_ID = 'wrun_quickjs_dispatch_wake'; +const STEP_CID = 'step_lost_dispatch'; +const WAIT_CID = 'wait_far_future'; +const RUN_STARTED_AT = new Date('2026-05-19T12:00:00.000Z'); + +/** + * A durable log holding a created-but-unstarted step and a pending wait. The + * step's creation is dated far enough back that its dispatch is presumed lost + * under any watchdog interval the env allows, and the wait resumes well past + * every watchdog boundary so it can never be the sooner of the two timers. + */ +function makeLog(stepCreatedAt: Date, waitResumeAt: Date): Event[] { + let slot = 0; + const event = (data: Record, createdAt: Date): Event => + ({ + specVersion: SPEC_VERSION_CURRENT, + ...data, + runId: RUN_ID, + eventId: `evnt_${String(++slot).padStart(26, '0')}`, + createdAt, + }) as Event; + return [ + event( + { + eventType: 'run_created', + eventData: { + deploymentId: 'dpl_quickjs_dispatch_wake', + workflowName: 'workflow', + input: [], + }, + }, + RUN_STARTED_AT + ), + event({ eventType: 'run_started' }, RUN_STARTED_AT), + event( + { + eventType: 'step_created', + correlationId: STEP_CID, + eventData: { stepId: 'lostStep' }, + }, + stepCreatedAt + ), + event( + { + eventType: 'wait_created', + correlationId: WAIT_CID, + eventData: { resumeAt: waitResumeAt.toISOString() }, + }, + stepCreatedAt + ), + ]; +} + +async function runSuspendedScenario(options: { + stepCreatedAt: Date; + waitResumeAt?: Date; +}) { + const waitResumeAt = options.waitResumeAt; + const workflowRun: WorkflowRun = { + runId: RUN_ID, + workflowName: 'workflow', + status: 'running', + input: [], + deploymentId: 'dpl_quickjs_dispatch_wake', + specVersion: SPEC_VERSION_CURRENT, + startedAt: RUN_STARTED_AT, + createdAt: RUN_STARTED_AT, + updatedAt: RUN_STARTED_AT, + }; + + const durableEvents = makeLog( + options.stepCreatedAt, + waitResumeAt ?? new Date(0) + ).filter((e) => waitResumeAt !== undefined || e.eventType !== 'wait_created'); + + let listCallCount = 0; + const queued: { options?: Record }[] = []; + + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + capabilities: {}, + events: { + list: vi.fn(async () => { + listCallCount++; + return listCallCount === 1 + ? { + data: [...durableEvents], + cursor: durableEvents.at(-1)?.eventId ?? null, + hasMore: false, + } + : { data: [], cursor: null, hasMore: false }; + }), + create: vi.fn(async (_runId: string, request: CreateEventRequest) => ({ + event: { ...request, runId: RUN_ID, eventId: 'evnt_created' }, + })), + }, + runs: { get: vi.fn(async () => workflowRun) }, + queue: vi.fn(async (_name: string, _payload: unknown, opts?: unknown) => { + queued.push({ options: opts as Record | undefined }); + return { messageId: `msg_${queued.length}` }; + }), + getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), + } as unknown as World); + + startQuickJSWorkflow.mockResolvedValue({ + result: { + suspended: { + pendingOperations: [ + { + type: 'step', + correlationId: STEP_CID, + stepId: 'lostStep', + // Durable step_created: the dispatch pass leaves this step alone, + // so its message is already out there. + hasCreatedEvent: true, + input: [], + }, + ...(waitResumeAt + ? [ + { + type: 'wait', + correlationId: WAIT_CID, + hasCreatedEvent: true, + resumeAt: waitResumeAt.toISOString(), + }, + ] + : []), + ], + }, + }, + continueWithEvents: vi.fn(), + dispose: vi.fn(), + }); + + const nowMs = Date.now(); + const { runWorkflowWithQuickJS } = await import('./quickjs-entrypoint.js'); + await runWorkflowWithQuickJS({ + workflowCode: '// not evaluated: the VM is mocked', + workflowName: 'workflow', + workflowRun, + }); + + const expectedWake = getStepDispatchWake( + [{ correlationId: STEP_CID, createdEventAt: +options.stepCreatedAt }], + nowMs + ); + return { queued, expectedWake }; +} + +describe('QuickJS suspension arms the re-dispatch watchdog wake', () => { + afterEach(() => { + setWorld(undefined); + vi.clearAllMocks(); + }); + + it('arms it for a created-but-unstarted step with nothing else outstanding', async () => { + const stepCreatedAt = new Date(RUN_STARTED_AT); + const { queued, expectedWake } = await runSuspendedScenario({ + stepCreatedAt, + }); + + expect(expectedWake).toBeDefined(); + const keys = queued.map((m) => m.options?.idempotencyKey); + expect(keys).toContain(expectedWake?.idempotencyKey); + }); + + it('arms it alongside a pending wait, whose continuation is far later', async () => { + const stepCreatedAt = new Date(RUN_STARTED_AT); + // A wait resuming a day out: the continuation is scheduled, and if the + // watchdog wake rode on it the step would not be re-evaluated until then. + const waitResumeAt = new Date(Date.now() + 24 * 60 * 60 * 1000); + const { queued, expectedWake } = await runSuspendedScenario({ + stepCreatedAt, + waitResumeAt, + }); + + expect(expectedWake).toBeDefined(); + const wake = queued.find( + (m) => m.options?.idempotencyKey === expectedWake?.idempotencyKey + ); + // The wake is armed in the same suspension as the wait continuation, not + // skipped by it. Both messages, two distinct keys. + expect(wake).toBeDefined(); + expect(queued.length).toBeGreaterThanOrEqual(2); + const otherKeys = queued + .map((m) => m.options?.idempotencyKey) + .filter((k) => k !== expectedWake?.idempotencyKey); + expect(otherKeys.length).toBeGreaterThan(0); + // And it lands at its own boundary, far short of the wait's resume. + const boundaryMs = nextStepDispatchBoundaryMs( + { correlationId: STEP_CID, createdEventAt: +stepCreatedAt }, + Date.now() + ); + expect(boundaryMs).toBeDefined(); + expect((wake?.options?.delaySeconds as number) * 1000).toBeLessThan( + +waitResumeAt - Date.now() + ); + }); +}); From 81d75f7224b4e931dd7f1f47be49f2bc8eddfb09 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Tue, 11 Aug 2026 15:15:00 -0700 Subject: [PATCH 14/14] test(core): interleave repro scenarios so one slow attempt cannot starve the rest The launch budget was positional: scenarios launched as contiguous blocks, so an attempt that spends its full runTimeoutMs holds its block open past the budget and every scenario behind it reports zero runs. Attempts now launch interleaved in one bounded pass, so a truncated run is proportionally short in every scenario. Cross-run concurrency is unchanged. --- .../core/e2e/event-log-race-repro.test.ts | 120 +++++++++++------- 1 file changed, 76 insertions(+), 44 deletions(-) diff --git a/packages/core/e2e/event-log-race-repro.test.ts b/packages/core/e2e/event-log-race-repro.test.ts index 873dcf90a1..ab98ae3b0e 100644 --- a/packages/core/e2e/event-log-race-repro.test.ts +++ b/packages/core/e2e/event-log-race-repro.test.ts @@ -949,7 +949,6 @@ const plannedAttempts = config.hookSleepAttempts; let overallDeadline = Number.POSITIVE_INFINITY; let launchDeadline = Number.POSITIVE_INFINITY; -let remainingPlanned = plannedAttempts; let budgetExhausted = false; let lastCheckpointAt = 0; @@ -985,35 +984,60 @@ function recordResult(result: ReproRunResult) { } } -async function runScenario( - attempts: number, - concurrency: number, - run: (attempt: number) => Promise -) { - if (attempts <= 0) { - return []; +interface ScenarioPlan { + scenario: Scenario; + attempts: number; + run: (attempt: number) => Promise; +} + +/** + * Every planned attempt in one launch order, with each scenario's attempts + * spread across the whole order rather than grouped into a contiguous block. + * + * Blocks make the launch budget positional. One attempt that spends its full + * `runTimeoutMs` holds its block open past the budget, and every scenario + * behind it reports zero runs: `hook-storm` is the production shape and + * `hook-sleep` is the calibration control, so a truncated run loses exactly the + * parts of the result that carry the most meaning. Interleaved, a truncated run + * is proportionally short in every scenario instead. + * + * Cross-run concurrency is unchanged. One `mapLimit` over this order holds + * `config.concurrency` attempts in flight no matter which scenarios they belong + * to, and the race each attempt reproduces is between replays *within* its own + * run, so which scenarios its neighbours are running does not enter into it. + */ +function interleaveAttempts(plans: ScenarioPlan[]) { + const queues = plans + .filter((plan) => plan.attempts > 0) + .map((plan) => ({ plan, issued: 0 })); + const order: { + scenario: Scenario; + attempt: number; + run: () => Promise; + }[] = []; + for (;;) { + // Whichever scenario is furthest behind its share of the order goes next, + // so every prefix of the order tracks the planned proportions. + let next: (typeof queues)[number] | undefined; + for (const queue of queues) { + if (queue.issued >= queue.plan.attempts) { + continue; + } + if ( + next === undefined || + queue.issued / queue.plan.attempts < next.issued / next.plan.attempts + ) { + next = queue; + } + } + if (next === undefined) { + return order; + } + next.issued += 1; + const { scenario, run } = next.plan; + const attempt = next.issued; + order.push({ scenario, attempt, run: () => run(attempt) }); } - // Each scenario gets the share of the remaining budget its remaining planned - // attempts represent, so a truncated run still carries data for every - // scenario — including the `hook-sleep` control, which runs last and would - // otherwise be the first thing a single global deadline dropped. A scenario - // that finishes under its slice hands the surplus to the next one, since the - // slice is recomputed from the wall-clock left until `overallDeadline`. - const now = Date.now(); - launchDeadline = Math.min( - overallDeadline, - now + ((overallDeadline - now) * attempts) / remainingPlanned - ); - remainingPlanned -= attempts; - const attemptNumbers = Array.from( - { length: attempts }, - (_, index) => index + 1 - ); - return await mapLimit(attemptNumbers, concurrency, async (attempt) => { - const result = await run(attempt); - recordResult(result); - return result; - }); } // Derived from the launch budget, not from the attempt count: the budget is @@ -1077,25 +1101,33 @@ describe('event log race repro', () => { { timeout: testTimeoutMs }, async () => { overallDeadline = Date.now() + config.budgetMs; + launchDeadline = overallDeadline; // Written up front so a kill before the first checkpoint still produces a // file the renderer can report against, rather than nothing at all. writeResults(collected, false); - await runScenario( - config.stepStormAttempts, - config.concurrency, - runStepStormAttempt - ); - await runScenario( - config.hookStormAttempts, - config.concurrency, - runHookStormAttempt - ); - await runScenario( - config.hookSleepAttempts, - config.concurrency, - runHookSleepAttempt - ); + const order = interleaveAttempts([ + { + scenario: 'step-storm', + attempts: config.stepStormAttempts, + run: runStepStormAttempt, + }, + { + scenario: 'hook-storm', + attempts: config.hookStormAttempts, + run: runHookStormAttempt, + }, + { + scenario: 'hook-sleep', + attempts: config.hookSleepAttempts, + run: runHookSleepAttempt, + }, + ]); + await mapLimit(order, config.concurrency, async (item) => { + const result = await item.run(); + recordResult(result); + return result; + }); const results = collected; // A budget-exhausted run launched fewer than `plannedAttempts` attempts,