diff --git a/.changeset/wild-pandas-jam.md b/.changeset/wild-pandas-jam.md new file mode 100644 index 0000000000..0a39c38020 --- /dev/null +++ b/.changeset/wild-pandas-jam.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +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/.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 b3e1c85824..2bb1c97da4 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -166,6 +166,16 @@ 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 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. 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 ### `WORKFLOW_VM` diff --git a/packages/core/e2e/event-log-race-repro.test.ts b/packages/core/e2e/event-log-race-repro.test.ts index b577df7b5c..ab98ae3b0e 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,33 @@ 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; + // 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 = { stepStormAttempts: envNumber('EVENT_LOG_RACE_REPRO_STEP_STORM_ATTEMPTS', 6), hookStormAttempts: envNumber('EVENT_LOG_RACE_REPRO_HOOK_STORM_ATTEMPTS', 6), @@ -173,7 +204,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), @@ -915,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; @@ -951,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 @@ -1043,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, 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 4d19bb2a8b..93821e40d5 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -93,6 +93,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, + stepDispatchEpoch, +} from './runtime/step-dispatch.js'; import { DEFAULT_STEP_MAX_RETRIES, executeStep, @@ -3304,6 +3308,11 @@ export function workflowEntrypoint( const dispatchNowMs = Date.now(); const ownedRecoverySteps: StepInvocationQueueItem[] = []; + // Steps whose step-execution message went out this + // suspension, here or from the suspension handler's + // parallel dispatch, 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)) { @@ -3321,6 +3330,11 @@ export function workflowEntrypoint( step.correlationId ) ) { + // Its message can still be lost, and this is the + // only suspension that will ever see the step + // without something else already waking the run, + // so it still counts for the boundary wake. + dispatchedSteps.push(step); continue; } const ownershipActive = @@ -3372,6 +3386,7 @@ export function workflowEntrypoint( ); continue; } + dispatchedSteps.push(step); dispatches.push( queueMessage( world, @@ -3384,21 +3399,55 @@ export function workflowEntrypoint( requestedAt: new Date(), }, { - // Step-identity-scoped: dedupes against every - // other dispatch of THIS step (concurrent - // handlers, crash-recovery re-dispatch, the - // suspension handler's resilient publish) - // without absorbing a dispatch of a different - // step under a reassigned correlation id — - // see stepDispatchIdempotencyKey. + // Step-identity-scoped, and epoch-scoped once + // this step's dispatch is presumed lost: a + // watchdog interval without a first start, or + // an expired ownership lease with no terminal + // 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 + // runtime/step-dispatch.ts. idempotencyKey: stepDispatchIdempotencyKey( step.correlationId, - step.stepName + step.stepName, + stepDispatchEpoch(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 dispatched + // here. 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, + } + ) + ); + } if (suspensionResult.waitTimeout) { dispatches.push( queueMessage( diff --git a/packages/core/src/runtime/constants.ts b/packages/core/src/runtime/constants.ts index a6b71eb839..a2aa63f73b 100644 --- a/packages/core/src/runtime/constants.ts +++ b/packages/core/src/runtime/constants.ts @@ -440,6 +440,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/helpers.ts b/packages/core/src/runtime/helpers.ts index 7c8d1cfcee..c3cefd96e8 100644 --- a/packages/core/src/runtime/helpers.ts +++ b/packages/core/src/runtime/helpers.ts @@ -1164,12 +1164,23 @@ function fnv1a32Hex(value: string): string { * Every producer of a step-dispatch (or step-retry) message must use this * key. Cross-version mixing is not a concern: queue messages are pinned to * the deployment that produced them, so one run never sees two key schemes. + * + * `epoch` is the re-dispatch watchdog's count of how many times this step's + * dispatch has been presumed lost (`stepDispatchEpoch` in + * runtime/step-dispatch.ts). Identity scoping alone cannot recover a lost + * dispatch: the queue holds the key for the lifetime of the message sent under + * it, so a step whose message will never make progress needs a key the queue + * has not seen. Epoch 0 is the unsuffixed identity key, so producers that have + * no lost dispatch to account for (the suspension handler's parallel publish, a + * delayed retry) keep sharing one key with the dispatch pass. */ export function stepDispatchIdempotencyKey( correlationId: string, - stepName: string + stepName: string, + epoch = 0 ): string { - return `${correlationId}:${fnv1a32Hex(stepName)}`; + const identity = `${correlationId}:${fnv1a32Hex(stepName)}`; + return epoch === 0 ? identity : `${identity}:dispatch:${epoch}`; } /** 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() + ); + }); +}); diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index ce88f86906..650e0797dd 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -68,6 +68,11 @@ import { startQuickJSWorkflow, } from './quickjs-runtime.js'; import { ReplayBudget } from './replay-budget.js'; +import { + getStepDispatchWake, + type StepDispatchState, + stepDispatchEpoch, +} 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'; @@ -129,8 +134,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 uses the step-identity-scoped dispatch key - * (stepDispatchIdempotencyKey) so it stays mutually - * exclusive with the node engine's dispatch of the same step; + * (stepDispatchIdempotencyKey) so it stays mutually exclusive with the + * node engine's dispatch of the same step, and once that dispatch is + * presumed lost both engines suffix it with the same watchdog epoch so + * the 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 @@ -138,6 +145,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 unsuffixed identity key. + */ + dispatchState?: StepDispatchState; /** * Resilient step dispatch: the serialized (possibly encrypted) step input * to carry on the message as `stepInput`, so the consumer can idempotently @@ -157,6 +169,7 @@ async function queueStepMessage(params: { namespace, nextTraceCarrier, purpose, + dispatchState, stepInput, wfdiag, } = params; @@ -176,11 +189,18 @@ async function queueStepMessage(params: { // The 'dispatch' key is step-identity-scoped (correlationId + hashed // step name) — shared with the node engine's dispatch of the same step // so the two stay mutually exclusive, without a revoked resilient - // message absorbing a reassigned correlation id's legitimate dispatch. - // See stepDispatchIdempotencyKey. + // message absorbing a reassigned correlation id's legitimate dispatch — + // and suffixed with the re-dispatch watchdog's epoch once this step's + // dispatch is presumed lost. See stepDispatchIdempotencyKey. idempotencyKey: purpose === 'dispatch' - ? stepDispatchIdempotencyKey(step.correlationId, step.stepId) + ? stepDispatchIdempotencyKey( + step.correlationId, + step.stepId, + dispatchState + ? stepDispatchEpoch(dispatchState, Date.now()) + : undefined + ) : `${step.correlationId}:${purpose}`, ...(delaySeconds && delaySeconds > 0 ? { delaySeconds } : {}), } @@ -248,6 +268,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; /** Step cids already published via resilient dispatch — see above. */ queuedStepCids: Set; }> { @@ -293,6 +321,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 => { @@ -589,11 +618,21 @@ async function dispatchPendingOps(params: { } throw err; } + // Anchors the re-dispatch watchdog for this step. A retryably + // rejected create leaves no timestamp to anchor from, but its + // message did go out (a queue failure above is fatal), and the + // next replay reads the timestamp off the re-ensured event. + if (createResult.status === 'fulfilled') { + const createdAt = createResult.value.event?.createdAt; + if (createdAt) { + stepCreatedAtMs.set(step.correlationId, +new Date(createdAt)); + } + } return; } try { - await world.events.create(runId, { + const created = await world.events.create(runId, { eventType: 'step_created', specVersion: SPEC_VERSION_CURRENT, correlationId: step.correlationId, @@ -602,6 +641,12 @@ async function dispatchPendingOps(params: { input: encryptedInput, }, }); + if (created.event?.createdAt) { + stepCreatedAtMs.set( + step.correlationId, + +new Date(created.event.createdAt) + ); + } } catch (err) { if (EntityConflictError.is(err)) return; throw err; @@ -667,7 +712,12 @@ async function dispatchPendingOps(params: { // Per-op dispatch runs in parallel. await Promise.all(opsPromises); - return { createdAttributeEvent, createdGetConflictHook, queuedStepCids }; + return { + createdAttributeEvent, + createdGetConflictHook, + stepCreatedAtMs, + queuedStepCids, + }; } /** @@ -1037,12 +1087,38 @@ 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 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_started') { + if (e.eventType === 'step_created') { + recordStepCreatedAt( + e.correlationId, + e.createdAt ? +new Date(e.createdAt) : undefined + ); + } else if (e.eventType === 'step_started') { const owner = 'eventData' in e && e.eventData && @@ -1055,6 +1131,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); @@ -1066,6 +1143,21 @@ export async function runWorkflowWithQuickJS(params: { } }; observeEventsForOwnership(events); + /** + * The re-dispatch watchdog's view of a pending step, assembled from the + * 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, + createdEventAt: ownership?.createdAtMs, + lastStartedAt: ownership?.startedAtMs, + sawRetrying: ownership?.sawRetrying, + }; + }; const scheduledWaitContinuations = new Set(); const maxInlineSteps = getMaxInlineSteps(); const budget = new ReplayBudget(); @@ -1195,6 +1287,9 @@ export async function runWorkflowWithQuickJS(params: { ) { pendingRequeueSignal = true; } + for (const [correlationId, createdAtMs] of dispatched.stepCreatedAtMs) { + recordStepCreatedAt(correlationId, createdAtMs); + } for (const cid of dispatched.queuedStepCids) { queuedStepIds.add(cid); @@ -1212,6 +1307,7 @@ export async function runWorkflowWithQuickJS(params: { namespace, nextTraceCarrier, purpose: 'dispatch', + dispatchState: dispatchStateOf(step), wfdiag, }); }) @@ -1345,6 +1441,7 @@ export async function runWorkflowWithQuickJS(params: { namespace, nextTraceCarrier, purpose: 'dispatch', + dispatchState: dispatchStateOf(step), wfdiag, }); } @@ -1716,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 @@ -1726,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( @@ -1747,6 +1875,7 @@ export async function runWorkflowWithQuickJS(params: { 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..e852be7b81 --- /dev/null +++ b/packages/core/src/runtime/step-dispatch.test.ts @@ -0,0 +1,367 @@ +import { afterEach, describe, expect, it } from 'vitest'; +import type { StepInvocationQueueItem } from '../global.js'; +import { + getInlineOwnershipLeaseSeconds, + getStepDispatchWatchdogSeconds, +} from './constants.js'; +import { stepDispatchIdempotencyKey } from './helpers.js'; +import { + dispatchLostAtMs, + getStepDispatchWake, + isStepAwaitingFirstStart, + nextStepDispatchBoundaryMs, + stepDispatchEpoch, +} from './step-dispatch.js'; +import { WAIT_CONTINUATION_MAX_DELAY_SECONDS } from './wait-continuation.js'; + +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 = {} +): 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; +} + +/** One interval, in ms, at whatever the watchdog is currently configured to. */ +function intervalMs(): number { + return 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 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( + true + ); + }); + + 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('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 + ); + }); + + 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) + ); + }); +}); + +/** + * The key a dispatch is actually published under: the step-identity key from + * `helpers.ts`, suffixed with the watchdog epoch. Every producer of a step + * message shares the unsuffixed form, so the two layers are asserted together. + */ +function dispatchKey(step: StepInvocationQueueItem, nowMs: number): string { + return stepDispatchIdempotencyKey( + step.correlationId, + step.stepName, + stepDispatchEpoch(step, nowMs) + ); +} + +/** The unsuffixed identity key, as the other step-message producers derive it. */ +function identityKey(step: StepInvocationQueueItem): string { + return stepDispatchIdempotencyKey(step.correlationId, step.stepName); +} + +describe('dispatch idempotency key', () => { + it('is the identity key within the first watchdog interval', () => { + const step = makeStep(); + expect(dispatchKey(step, CREATED_AT)).toBe(identityKey(step)); + expect(dispatchKey(step, atIntervals(1))).toBe(identityKey(step)); + }); + + it('moves to a fresh key once an interval has passed without a start', () => { + const step = makeStep(); + expect(dispatchKey(step, atIntervals(1) + 1)).toBe( + `${identityKey(step)}:dispatch:1` + ); + expect(dispatchKey(step, atIntervals(2.5))).toBe( + `${identityKey(step)}:dispatch:2` + ); + }); + + it('is stable across replays within one epoch', () => { + const step = makeStep(); + const early = dispatchKey(step, atIntervals(1.01)); + const late = dispatchKey(step, atIntervals(1.99)); + expect(early).toBe(late); + }); + + it('keeps the identity 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(dispatchKey(step, atLeaseOffset(-1))).toBe(identityKey(step)); + }); + + 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(dispatchKey(step, atLeaseOffset(1))).toBe( + `${identityKey(step)}:dispatch:1` + ); + expect(dispatchKey(step, atLeaseOffset(intervalMs()))).toBe( + `${identityKey(step)}:dispatch:2` + ); + }); + + it('keeps the identity key for a step whose retry is already queued', () => { + // step_retrying means the owner scheduled the next attempt under the + // identity key, with a backoff that may legitimately exceed any deadline + // here. + const step = startedStep({ sawRetrying: true }); + expect(dispatchKey(step, atLeaseOffset(intervalMs()))).toBe( + identityKey(step) + ); + }); + + it('keeps the identity key when the world reports no creation timestamp', () => { + const step = makeStep({ createdEventAt: undefined }); + expect(dispatchKey(step, atIntervals(10))).toBe(identityKey(step)); + }); + + it('keeps the identity key when the creation timestamp is in the future', () => { + const step = makeStep(); + expect(dispatchKey(step, CREATED_AT - 60_000)).toBe(identityKey(step)); + }); + + it('separates two steps that share a correlation ID', () => { + // A guard-corrected replay can re-derive one correlation ID for a + // different step; the epoch suffix must not collapse them. + const a = makeStep({ stepName: 'first' }); + const b = makeStep({ stepName: 'second' }); + expect(dispatchKey(a, atIntervals(2))).not.toBe( + dispatchKey(b, atIntervals(2)) + ); + }); + + it('honours the watchdog override', () => { + process.env[WATCHDOG_ENV] = '10'; + const step = makeStep(); + 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', () => { + 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 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( + startedStep({ sawRetrying: true }), + atIntervals(3) + ) + ).toBeUndefined(); + }); +}); + +describe('getStepDispatchWake', () => { + it('is undefined when no pending step has a lost-dispatch deadline', () => { + expect( + 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(); + const boundary = atIntervals(1); + expect(CREATED_AT + wake!.delaySeconds * 1000).toBeGreaterThan(boundary); + 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', + 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 boundary = atIntervals(1); + const wake = getStepDispatchWake([makeStep()], CREATED_AT); + 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', () => { + // 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..aa430a8ed8 --- /dev/null +++ b/packages/core/src/runtime/step-dispatch.ts @@ -0,0 +1,202 @@ +import { + getInlineOwnershipLeaseSeconds, + getStepDispatchWatchdogSeconds, +} 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`). 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. + * + * 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 + * suffixed with. 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 unsuffixed 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. Both deadlines are set well above + * normal 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; + /** + * `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; + /** Whether a `step_retrying` was observed, which implies a prior start. */ + sawRetrying?: boolean; +} + +/** + * 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). + */ +export function isStepAwaitingFirstStart(step: StepDispatchState): boolean { + return ( + step.createdEventAt !== undefined && + step.lastStartedAt === undefined && + step.sawRetrying !== true + ); +} + +/** + * 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 unsuffixed identity key, which every other + * producer of a step message uses. 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 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)) + 1; +} + +/** + * 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 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 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 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 + * 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. 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) + ); + return { + delaySeconds, + idempotencyKey: `${earliest.correlationId}:dispatch-wake:${earliest.boundaryMs}`, + }; +} 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 diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index 13363a196c..d12ef94162 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -860,6 +860,17 @@ export async function handleSuspension({ throw err; } } else { + // Anchors the re-dispatch watchdog from this suspension onward, + // so a dispatch lost on the step's very first hand-off is still + // retried. A retryably-rejected create leaves no timestamp to + // anchor from, but its message did go out (a queue failure above + // is fatal), and the next replay reads the timestamp off the + // re-ensured event. + if (createResult.value.event?.createdAt) { + queueItem.createdEventAt = +new Date( + createResult.value.event.createdAt + ); + } createdStepCorrelationIds.add(queueItem.correlationId); } return; @@ -867,7 +878,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)) { 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; } diff --git a/packages/world-sim/README.md b/packages/world-sim/README.md index 6646c36578..ade3b47f79 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,24 @@ 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 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. + +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 gone, and a run whose only +remaining work sits behind a key no re-send can reach. + ## 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..0c4ed060e5 --- /dev/null +++ b/workbench/sim-world/scenarios/lost-step-dispatch.ts @@ -0,0 +1,34 @@ +import type { ScenarioSpec } from '@workflow/world-sim'; + +export const scenario: ScenarioSpec = { + id: 'lost-step-dispatch', + 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 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) => { + // 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 // ---------------------------------------------------------------------------