From 6af2c2c7b8da496d0d7cfcd7d601c2f802e85765 Mon Sep 17 00:00:00 2001 From: Nathan Rajlich Date: Fri, 31 Jul 2026 01:51:45 -0700 Subject: [PATCH 1/8] Make QuickJS the default workflow VM engine (WORKFLOW_VM=node opts back into node:vm) - useQuickJSVm defaults to the QuickJS engine when neither the run's stamped executionContext.workflowVm nor WORKFLOW_VM specifies one; WORKFLOW_VM=node is the explicit node:vm opt-in. - CI matrix inverted to match: quickjs legs leave WORKFLOW_VM unset so the default-selection path is exercised end to end; node legs opt in explicitly (labels/artifacts unchanged via MATRIX_VM). - Entrypoint tests that assert node replay-loop internals against mock worlds pin WORKFLOW_VM=node (the quickjs path would instantiate a WASM VM per call). - Docs + changeset updated. Runs keep the engine stamped at start(). --- .changeset/quickjs-default-engine.md | 5 +++ .github/workflows/tests.yml | 44 +++++++++++-------- .../docs/v5/configuration/runtime-tuning.mdx | 10 ++--- packages/core/src/runtime-trace-mode.test.ts | 6 +++ packages/core/src/runtime.test.ts | 6 +++ .../runtime/precondition-guard-replay.test.ts | 6 +++ packages/core/src/runtime/vm-mode.test.ts | 9 +++- packages/core/src/runtime/vm-mode.ts | 9 ++-- .../runtime/wait-completion-replay.test.ts | 6 +++ scripts/create-test-matrix.mjs | 33 ++++++++------ 10 files changed, 91 insertions(+), 43 deletions(-) create mode 100644 .changeset/quickjs-default-engine.md diff --git a/.changeset/quickjs-default-engine.md b/.changeset/quickjs-default-engine.md new file mode 100644 index 0000000000..07d8217f66 --- /dev/null +++ b/.changeset/quickjs-default-engine.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': minor +--- + +The QuickJS WASM VM is now the default workflow engine. Set `WORKFLOW_VM=node` to opt back into the `node:vm` engine. Existing runs keep executing on the engine stamped in their `executionContext` at start. diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 84bb27400e..40008638b1 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -254,9 +254,9 @@ jobs: strategy: fail-fast: false matrix: - # Workflow VM engines: node:vm (default) and the opt-in QuickJS - # WASM engine (WORKFLOW_VM=quickjs). - vm: [node, quickjs] + # Workflow VM engines: QuickJS WASM (the default — WORKFLOW_VM left + # unset) and the explicit node:vm opt-in (WORKFLOW_VM=node). + vm: [quickjs, node] env: TURBO_TOKEN: ${{ secrets.TURBO_TOKEN }} TURBO_TEAM: ${{ vars.TURBO_TEAM }} @@ -278,7 +278,8 @@ jobs: run: pnpm test working-directory: workbench/vitest env: - WORKFLOW_VM: ${{ matrix.vm }} + # quickjs is the engine default — leave WORKFLOW_VM unset for it. + WORKFLOW_VM: ${{ matrix.vm == 'node' && 'node' || '' }} e2e-package-build: name: Build Shared E2E Packages @@ -331,12 +332,14 @@ jobs: strategy: fail-fast: false matrix: - # Workflow VM engines: node:vm (default) and the opt-in QuickJS - # WASM engine. The env var is set on the e2e test runner, which is - # the client that starts runs against the deployed app — start() - # stamps executionContext.workflowVm so the deployed handler - # executes each run on the requested engine. - vm: [node, quickjs] + # Workflow VM engines: QuickJS WASM (the default — WORKFLOW_VM left + # unset so the default-selection path is exercised end to end) and + # the explicit node:vm opt-in (WORKFLOW_VM=node). The env var is set + # on the e2e test runner, which is the client that starts runs + # against the deployed app — start() stamps + # executionContext.workflowVm so the deployed handler executes each + # run on the requested engine. + vm: [quickjs, node] app: - name: "example" project-id: "prj_xWq20Dd860HHAfzMjK2Mb6TPVxMa" @@ -437,13 +440,16 @@ jobs: run: echo "ms=$(($(date +%s) * 1000))" >> "$GITHUB_OUTPUT" - name: Run E2E Tests - run: pnpm run test:e2e --reporter=verbose --reporter=json --reporter=./packages/core/e2e/github-reporter.ts "--outputFile=e2e-vercel-prod-$APP_NAME-$WORKFLOW_VM.json" + run: pnpm run test:e2e --reporter=verbose --reporter=json --reporter=./packages/core/e2e/github-reporter.ts "--outputFile=e2e-vercel-prod-$APP_NAME-$MATRIX_VM.json" env: NODE_OPTIONS: "--enable-source-maps" DEPLOYMENT_URL: ${{ steps.waitForDeployment.outputs.deployment-url || steps.prodDeployment.outputs.deployment-url }} VERCEL_DEPLOYMENT_ID: ${{ steps.waitForDeployment.outputs.deployment-id || steps.prodDeployment.outputs.deployment-id }} APP_NAME: ${{ matrix.app.name }} - WORKFLOW_VM: ${{ matrix.vm }} + # quickjs is the engine default — leave WORKFLOW_VM unset for it + # (MATRIX_VM carries the label for file/job naming). + WORKFLOW_VM: ${{ matrix.vm == 'node' && 'node' || '' }} + MATRIX_VM: ${{ matrix.vm }} # changeset-release PRs test main's production deployment, so they # must be treated as a production run everywhere downstream. WORKFLOW_VERCEL_ENV: ${{ (github.ref == 'refs/heads/main' || startsWith(github.head_ref, 'changeset-release/')) && 'production' || 'preview' }} @@ -482,8 +488,8 @@ jobs: if: always() env: APP_NAME: ${{ matrix.app.name }} - WORKFLOW_VM: ${{ matrix.vm }} - run: node .github/scripts/aggregate-e2e-results.js . --job-name "E2E Vercel Prod ($APP_NAME - $WORKFLOW_VM)" >> $GITHUB_STEP_SUMMARY || true + MATRIX_VM: ${{ matrix.vm }} + run: node .github/scripts/aggregate-e2e-results.js . --job-name "E2E Vercel Prod ($APP_NAME - $MATRIX_VM)" >> $GITHUB_STEP_SUMMARY || true - name: Upload E2E results if: always() @@ -925,9 +931,9 @@ jobs: strategy: fail-fast: false matrix: - # Workflow VM engines: node:vm (default) and the opt-in QuickJS - # WASM engine (WORKFLOW_VM=quickjs). - vm: [node, quickjs] + # Workflow VM engines: QuickJS WASM (the default — WORKFLOW_VM left + # unset) and the explicit node:vm opt-in (WORKFLOW_VM=node). + vm: [quickjs, node] env: TURBO_TOKEN: ${{ secrets.TURBO_TOKEN }} TURBO_TEAM: ${{ vars.TURBO_TEAM }} @@ -1038,7 +1044,9 @@ jobs: DEV_TEST_CONFIG: '{"generatedStepRegistrationPath":"app/.well-known/workflow/v1/flow/__step_registrations.js","generatedWorkflowPath":"app/.well-known/workflow/v1/flow/route.js","apiFilePath":"app/api/chat/route.ts","apiFileImportPath":"../../..","port":3000,"testWorkflowFile":"96_many_steps.ts"}' DEV_SERVER_LOG_PATH: "${{ github.workspace }}/nextjs-server.log" WORKFLOW_DEV_HMR_LOGS: "1" - WORKFLOW_VM: ${{ matrix.vm }} + # quickjs is the engine default — leave WORKFLOW_VM unset for it + # (MATRIX_VM carries the label for reporting). + WORKFLOW_VM: ${{ matrix.vm == 'node' && 'node' || '' }} MATRIX_VM: ${{ matrix.vm }} - name: Print Next.js server logs diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index 85672dfc86..d3ed3a80b9 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -113,12 +113,12 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL ### `WORKFLOW_VM` -- Default: `node` +- Default: `quickjs` - Values: `node` or `quickjs` - Selects the sandboxed VM engine that executes workflow functions (`"use workflow"`). Step functions are unaffected — they always run with full Node.js access. -- `node` (default) runs workflow code in a [`node:vm`](https://nodejs.org/api/vm.html) context. -- `quickjs` (experimental) runs workflow code in a [QuickJS](https://github.com/quickjs-ng/quickjs) VM compiled to WebAssembly (via [`quickjs-wasi`](https://github.com/vercel-labs/quickjs-wasi)). Both engines implement the same event-replay execution model (seeded PRNG, deterministic clock, and correlation-ID sequences are identical), but the **global surface is not identical** — see the differences below before switching an existing deployment. The QuickJS engine is intended for platforms that do not implement `node:vm`, and is the foundation for future VM-memory snapshotting. -- Global-surface differences under `quickjs` (workflow functions only — step functions always have full Node.js): +- `quickjs` (default) runs workflow code in a [QuickJS](https://github.com/quickjs-ng/quickjs) VM compiled to WebAssembly (via [`quickjs-wasi`](https://github.com/vercel-labs/quickjs-wasi)). It works on platforms that do not implement `node:vm` (e.g. edge runtimes), and is the foundation for VM-memory snapshotting. +- `node` runs workflow code in a [`node:vm`](https://nodejs.org/api/vm.html) context instead. Both engines implement the same event-replay execution model (seeded PRNG, deterministic clock, and correlation-ID sequences are identical), but the **global surface is not identical** — review the differences below before flipping an existing deployment either direction. +- Global-surface differences under `quickjs` relative to `node` (workflow functions only — step functions always have full Node.js): - `crypto.getRandomValues()` and `crypto.randomUUID()` are provided and deterministic (seeded like the node engine's). All `crypto.subtle.*` methods throw with guidance to move to a step function — including `digest`, which the node engine supports. - `Intl` is not available (QuickJS has no ICU). The `Intl.*` constructors throw, and `toLocaleString`-family methods (including `localeCompare`) throw when called **with an explicit locale** — calling them without arguments keeps the engine default. Perform locale-sensitive formatting in a step function. - `WebAssembly` and `Atomics` are not available. @@ -130,7 +130,7 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL - Default: `0` (disabled) - Values: non-negative integer -- Only used by the QuickJS engine (`WORKFLOW_VM=quickjs`). +- Only used by the QuickJS engine (the default; see `WORKFLOW_VM`). - When set above `0`, the runtime persists a **VM-memory snapshot** at a suspension once at least this many events have been processed since the last snapshot. Subsequent invocations restore the VM from the snapshot and replay only the events recorded since — instead of re-executing the workflow from the top against the full event log. - Short-lived runs below the threshold never pay the snapshot cost; long-running or unbounded runs stop scaling their resume cost with total event-log length. `1` snapshots at every qualifying suspension. - Snapshots are an optimization, not a source of truth: the event log remains authoritative, and a missing, corrupt, or incompatible snapshot automatically falls back to a full replay. diff --git a/packages/core/src/runtime-trace-mode.test.ts b/packages/core/src/runtime-trace-mode.test.ts index bf119ac4ac..7f7a0d53f5 100644 --- a/packages/core/src/runtime-trace-mode.test.ts +++ b/packages/core/src/runtime-trace-mode.test.ts @@ -33,6 +33,12 @@ import { workflowEntrypoint } from './runtime.js'; import { dehydrateWorkflowArguments } from './serialization.js'; import { getNextTraceCarrier, getWorkflowTraceMode } from './telemetry.js'; +// These tests exercise the node:vm engine's replay loop internals against +// mock worlds. Pin the engine explicitly — QuickJS is the default, and +// dispatching there would instantiate a WASM VM per entrypoint call (and +// bypass the node replay loop these tests assert on). +process.env.WORKFLOW_VM = 'node'; + vi.mock('@vercel/functions', () => ({ waitUntil: vi.fn((p: Promise) => { p.catch(() => {}); diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index 750cef009f..a0876ec69d 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -20,6 +20,12 @@ import { dehydrateWorkflowArguments, } from './serialization.js'; +// These tests exercise the node:vm engine's replay loop internals against +// mock worlds. Pin the engine explicitly — QuickJS is the default, and +// dispatching there would instantiate a WASM VM per entrypoint call (and +// bypass the node replay loop these tests assert on). +process.env.WORKFLOW_VM = 'node'; + // Capture every promise handed to `waitUntil` so tests can assert that // progress-critical sends are never registered on a detached, unconsumed // promise (which would reject → unhandled rejection → process exit 128, and diff --git a/packages/core/src/runtime/precondition-guard-replay.test.ts b/packages/core/src/runtime/precondition-guard-replay.test.ts index fefa5a0bcf..6a1e450152 100644 --- a/packages/core/src/runtime/precondition-guard-replay.test.ts +++ b/packages/core/src/runtime/precondition-guard-replay.test.ts @@ -45,6 +45,12 @@ import { } from './constants.js'; import { setWorld } from './world.js'; +// These tests exercise the node:vm engine's replay loop internals against +// mock worlds. Pin the engine explicitly — QuickJS is the default, and +// dispatching there would instantiate a WASM VM per entrypoint call (and +// bypass the node replay loop these tests assert on). +process.env.WORKFLOW_VM = 'node'; + vi.mock('@vercel/functions', () => ({ waitUntil: vi.fn(), })); diff --git a/packages/core/src/runtime/vm-mode.test.ts b/packages/core/src/runtime/vm-mode.test.ts index 003f3329af..865dcb6772 100644 --- a/packages/core/src/runtime/vm-mode.test.ts +++ b/packages/core/src/runtime/vm-mode.test.ts @@ -78,8 +78,13 @@ describe('useQuickJSVm', () => { delete process.env.WORKFLOW_VM; }); - it('defaults to node:vm (false) when nothing is configured', () => { - expect(useQuickJSVm(makeRun())).toBe(false); + it('defaults to QuickJS (true) when nothing is configured', () => { + expect(useQuickJSVm(makeRun())).toBe(true); + }); + + it('an empty WORKFLOW_VM value also resolves to the QuickJS default', () => { + process.env.WORKFLOW_VM = ''; + expect(useQuickJSVm(makeRun())).toBe(true); }); it('returns true when WORKFLOW_VM=quickjs is set in the environment', () => { diff --git a/packages/core/src/runtime/vm-mode.ts b/packages/core/src/runtime/vm-mode.ts index 8dcecefa13..80e4665136 100644 --- a/packages/core/src/runtime/vm-mode.ts +++ b/packages/core/src/runtime/vm-mode.ts @@ -1,8 +1,8 @@ /** * VM engine selection for workflow execution. * - * The Node.js `node:vm` engine is the default. The QuickJS WASM engine is - * opt-in via the `WORKFLOW_VM` env var or `executionContext.workflowVm`. + * The QuickJS WASM engine is the default. The Node.js `node:vm` engine is + * opt-in via `WORKFLOW_VM=node` or `executionContext.workflowVm`. * * Both engines implement the same event-replay execution model: on every * workflow handler invocation the workflow function is re-executed from the @@ -53,7 +53,8 @@ export function getWorkflowVmFromEnv( * when `WORKFLOW_VM` is set on the client) takes precedence so a run keeps * executing on the engine it started on. When the run doesn't specify an * engine, the `WORKFLOW_VM` env var on the workflow handler decides. - * The default is the `node:vm` engine. + * The default is the QuickJS engine; `WORKFLOW_VM=node` opts back into + * the `node:vm` engine. * * Throws if `WORKFLOW_VM` or `executionContext.workflowVm` is set to an * unknown value. @@ -71,7 +72,7 @@ export function useQuickJSVm(workflowRun: WorkflowRun): boolean { } return vmFromRun === 'quickjs'; } - return getWorkflowVmFromEnv() === 'quickjs'; + return getWorkflowVmFromEnv() !== 'node'; } /** diff --git a/packages/core/src/runtime/wait-completion-replay.test.ts b/packages/core/src/runtime/wait-completion-replay.test.ts index c4ca46982a..e57d47763b 100644 --- a/packages/core/src/runtime/wait-completion-replay.test.ts +++ b/packages/core/src/runtime/wait-completion-replay.test.ts @@ -17,6 +17,12 @@ import { import { createContext } from '../vm/index.js'; import { setWorld } from './world.js'; +// These tests exercise the node:vm engine's replay loop internals against +// mock worlds. Pin the engine explicitly — QuickJS is the default, and +// dispatching there would instantiate a WASM VM per entrypoint call (and +// bypass the node replay loop these tests assert on). +process.env.WORKFLOW_VM = 'node'; + vi.mock('@vercel/functions', () => ({ waitUntil: vi.fn(), })); diff --git a/scripts/create-test-matrix.mjs b/scripts/create-test-matrix.mjs index 27f9456aee..b7285a13be 100644 --- a/scripts/create-test-matrix.mjs +++ b/scripts/create-test-matrix.mjs @@ -165,32 +165,37 @@ matrix.app.push({ }); // Cross-product with the workflow VM engine axis: every app is tested -// against both the default node:vm engine and the opt-in QuickJS WASM -// engine (WORKFLOW_VM=quickjs). Each engine gets its own artifactSuffix -// and runLabel so CI artifacts and job names are unique. The `vm` field -// is surfaced to the workflow dev server via the WORKFLOW_VM env var in -// tests.yml. -const VMS = ['node', 'quickjs']; +// against both the default QuickJS WASM engine and the opt-in node:vm +// engine (WORKFLOW_VM=node) — the QuickJS legs deliberately leave +// WORKFLOW_VM unset so they exercise the default-selection path end to +// end, not just an explicit opt-in. Each engine gets its own +// artifactSuffix and runLabel so CI artifacts and job names are unique. +// The `vm` field is surfaced to the workflow dev server via the +// WORKFLOW_VM env var in tests.yml (empty ⇒ engine default). +const VMS = [ + { vm: '', label: 'quickjs' }, // default engine (QuickJS) + { vm: 'node', label: 'node' }, // explicit node:vm opt-in +]; matrix.app = matrix.app.flatMap((app) => - VMS.map((vm) => ({ + VMS.map(({ vm, label }) => ({ ...app, vm, - runLabel: [app.runLabel, vm].filter(Boolean).join(' '), - artifactSuffix: [app.artifactSuffix, vm].filter(Boolean).join('-'), + runLabel: [app.runLabel, label].filter(Boolean).join(' '), + artifactSuffix: [app.artifactSuffix, label].filter(Boolean).join('-'), })) ); -// QuickJS engine with VM-memory snapshotting at maximum churn -// (WORKFLOW_SNAPSHOT_THRESHOLD=1 snapshots at every qualifying -// suspension) — exercises the save/restore/delete lifecycle and the -// restore + partial-replay determinism on every run. +// QuickJS engine (the default — WORKFLOW_VM left unset) with VM-memory +// snapshotting at maximum churn (WORKFLOW_SNAPSHOT_THRESHOLD=1 snapshots +// at every qualifying suspension) — exercises the save/restore/delete +// lifecycle and the restore + partial-replay determinism on every run. matrix.app.push( createMatrixEntry( 'nextjs-turbopack', 'example-nextjs-workflow-turbopack', DEV_TEST_CONFIGS['nextjs-turbopack'], { - vm: 'quickjs', + vm: '', snapshotThreshold: '1', runLabel: 'quickjs-snapshot', artifactSuffix: 'quickjs-snapshot', From fbebf7104d97219b73b6a51b0e77e42c45cdd99c Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Mon, 10 Aug 2026 12:46:28 -0700 Subject: [PATCH 2/8] [core] Keep step results ordered behind waits parked on unread hook payloads (#3406) --- .changeset/quiet-donkeys-repeat.md | 5 + .../src/delivery-barrier-coverage.test.ts | 141 ++++++++--- packages/core/src/private.ts | 139 +++++++---- .../core/src/step-delivery-ordering.test.ts | 221 +++++++++++++++++- workbench/fastify/public/index.html | 73 ------ 5 files changed, 428 insertions(+), 151 deletions(-) create mode 100644 .changeset/quiet-donkeys-repeat.md delete mode 100644 workbench/fastify/public/index.html diff --git a/.changeset/quiet-donkeys-repeat.md b/.changeset/quiet-donkeys-repeat.md new file mode 100644 index 0000000000..c9292c4a66 --- /dev/null +++ b/.changeset/quiet-donkeys-repeat.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Fix replay divergence when a step result overtook an earlier sleep or hook delivery that was parked behind an unread hook's payload diff --git a/packages/core/src/delivery-barrier-coverage.test.ts b/packages/core/src/delivery-barrier-coverage.test.ts index ef6232c8a7..b37a8f597d 100644 --- a/packages/core/src/delivery-barrier-coverage.test.ts +++ b/packages/core/src/delivery-barrier-coverage.test.ts @@ -25,10 +25,16 @@ * branch-deciding as any other delivery — but it resolved straight off its * `promiseQueue` slot and registered no barrier. * - * Cases 1-4 assert the same thing: the replay allocates its follow-up step + * Cases 1-3 assert the same thing: the replay allocates its follow-up step * ULIDs in the order the committed log recorded. A regression surfaces as the * production `ReplayDivergenceError`. * + * Section 4 asserts the registry's other job directly, over a registry built + * by hand rather than by a replay: which entries an idle check may ignore. Get + * that wrong in one direction and a chain parked on an unclaimed hook payload + * deadlocks; wrong in the other and a suspension preempts a batch of parked + * step results. + * * The final section covers the SUSPENSION side of the registry * (vercel/workflow#3183): an idle check must not observe idle — and raise a * `WorkflowSuspension` — while a delivery that is committed to reaching the @@ -51,6 +57,7 @@ import { WorkflowSuspension } from './global.js'; import { awaitEarlierDeliveries, registerDeliveryBarrier, + scheduleWhenIdle, type WorkflowOrchestratorContext, } from './private.js'; import { ReplayPayloadCache } from './replay-payload-cache.js'; @@ -490,41 +497,121 @@ describe('abort delivery ordering against an earlier step result', () => { }); }); -// ─── 4. registry scan cost ───────────────────────────────────────────────── +// ─── 4. idle reachability over the barrier registry ──────────────────────── +// +// `hasParkedCommittedDelivery` decides whether an idle check may observe idle, +// and it is the only remaining caller of the recursive `resolvesOnItsOwn` +// walk. Two opposite answers are load-bearing, and neither is covered by the +// replay sections above, which exercise the walk only through whichever shape +// their fixture happens to build: // -// `resolvesOnItsOwn` walks the registry recursively: an armed hook re-checks -// every earlier wait and step, an armed wait every earlier hook and step, and -// so on. Unmemoized that is T(n) = Σ T(j) — exponential — and the registry is -// not small by construction: `EventsConsumer` drains consecutively consumable -// events synchronously while barriers only retire on microtask-driven -// deliveries, so a fan-out of `Promise.race([hook, sleep(watchdog)])` branches -// accumulates one barrier per branch per kind (measured: 49 live barriers for -// 24 branches). +// - A PARKED CHAIN must not be counted. An unclaimed buffered hook payload is +// retired by the idle safety net in `registerDeliveryBarrier`, so counting +// it would gate its own retirement — and that extends to the wait parked +// behind it and the step gated on that wait. If any link were counted, idle +// would be unreachable, no net could fire, and the chain would never +// deliver: a deadlock, not a divergence. +// - An ALL-ARMED BATCH must be counted (vercel/workflow#3183). Parallel step +// results parked between their queue slots and their detached `resolve()` +// are invisible to `pendingDeliveries`, and an idle check that observed idle +// there would raise a `WorkflowSuspension` carrying none of the follow-up +// work the batch was about to create. // -// The scan runs synchronously, before `awaitEarlierDeliveries` first awaits, -// so timing the call alone measures it. Unmemoized, 40 alternating armed -// hook/wait barriers is ~10^8 recursive calls — minutes. Memoized it is -// linear. The bound is deliberately loose; this is an order-of-magnitude -// guard, not a benchmark. -describe('delivery-barrier registry scan cost', () => { - it('stays linear in registry size for a step delivery', () => { - const ctx = { +// Asserted through `scheduleWhenIdle`, which is the coupling that matters, and +// which makes both cases unambiguous: nothing in these registries ever +// delivers, so the callback can only fire if the registry was excluded from +// the idle count AND the barriers' own nets then retired it. +// +// This replaces a timing guard that no longer measured anything. It timed +// `awaitEarlierDeliveries(ctx, 40, 'step')` against 40 alternating armed +// hook/wait barriers to catch an unmemoized exponential walk (4.3e8 recursive +// calls, 84s). That call site is gone: a step now tests `armed` directly, so +// the call is a flat loop. The surviving caller cannot reach an exponential +// shape at all — it returns at the first self-resolving entry, so it only +// advances past entries that short-circuit on their first false child +// (measured: 98 recursive calls unmemoized for the worst 40-barrier shape). +// Timing it would assert nothing; see the memo note on `resolvesOnItsOwn`. +describe('delivery-barrier idle reachability', () => { + function emptyCtx(): WorkflowOrchestratorContext { + return { pendingDeliveries: 0, promiseQueue: Promise.resolve(), pendingDeliveryBarriers: new Map(), } as unknown as WorkflowOrchestratorContext; + } + + /** Whether `scheduleWhenIdle` observes idle within `rounds` timer ticks. */ + async function reachesIdle( + ctx: WorkflowOrchestratorContext, + rounds = 10 + ): Promise { + let idle = false; + scheduleWhenIdle(ctx, () => { + idle = true; + }); + for (let round = 0; round < rounds && !idle; round++) { + await ctx.promiseQueue; + await new Promise((resolve) => setTimeout(resolve, 0)); + } + return idle; + } + + it('unwinds a step parked behind a wait parked on an unclaimed payload, in log order', async () => { + const ctx = emptyCtx(); + const order: string[] = []; + let payloadRetiredBeforeWait: boolean | undefined; + + // The shape `step-delivery-ordering.test.ts` replays, as a registry: an + // unread hook's payload at index 0, a wait behind it, a step gated on that + // wait. Only the payload lacks a delivery chain — nothing in the workflow + // ever claims it, so the idle safety net is the only thing that can retire + // it. The wait and the step get the unconditional chain their real call + // sites attach at event-consumption time, as the INVARIANT on + // `registerDeliveryBarrier` requires of any armed barrier. + registerDeliveryBarrier(ctx, 0, 'hook', { armed: false }); + const wait = registerDeliveryBarrier(ctx, 1, 'wait'); + const step = registerDeliveryBarrier(ctx, 2, 'step'); + const chains = [ + awaitEarlierDeliveries(ctx, 1, 'wait').then(() => { + order.push('wait'); + payloadRetiredBeforeWait = !ctx.pendingDeliveryBarriers?.has(0); + wait.markDelivered(); + }), + awaitEarlierDeliveries(ctx, 2, 'step').then(() => { + order.push('step'); + step.markDelivered(); + }), + ]; + expect(ctx.pendingDeliveryBarriers?.size).toBe(3); + + // Neither chain can run yet: the wait gates on the unclaimed payload, and + // the step gates on the wait (it skips the payload directly, but the skip + // is not transitive through the armed wait). + await Promise.resolve(); + expect(order).toEqual([]); + + // Idle must stay reachable for the payload's net to fire at all. If any + // link of the chain were counted against idle, this would hang. + expect(await reachesIdle(ctx)).toBe(true); + await Promise.all(chains); + + expect(payloadRetiredBeforeWait).toBe(true); + expect(order).toEqual(['wait', 'step']); + expect(ctx.pendingDeliveryBarriers?.size).toBe(0); + }); - const BARRIERS = 40; - for (let index = 0; index < BARRIERS; index++) { - registerDeliveryBarrier(ctx, index, index % 2 ? 'hook' : 'wait'); + it('is blocked by an all-armed batch of step results', async () => { + const ctx = emptyCtx(); + // Armed and undelivered is exactly the window #3183 is about: the batch's + // queue slots have released `pendingDeliveries` and their detached + // `resolve()` calls have not run yet. + for (let index = 0; index < 3; index++) { + registerDeliveryBarrier(ctx, index, 'step'); } - expect(ctx.pendingDeliveryBarriers?.size).toBe(BARRIERS); - const startedAt = performance.now(); - // The floating promise never settles (nothing delivers these barriers); - // only the synchronous scan inside the call is under test. - void awaitEarlierDeliveries(ctx, BARRIERS, 'step'); - expect(performance.now() - startedAt).toBeLessThan(1_000); + expect(await reachesIdle(ctx)).toBe(false); + // The nets are idle-gated too, so nothing retires behind our back. + expect(ctx.pendingDeliveryBarriers?.size).toBe(3); }); }); diff --git a/packages/core/src/private.ts b/packages/core/src/private.ts index 9c04c2fc3d..c298bf5e6f 100644 --- a/packages/core/src/private.ts +++ b/packages/core/src/private.ts @@ -275,24 +275,69 @@ const DEFER_BEHIND: Record = { step: ['wait', 'hook', 'step'], }; +/** + * Whether a delivery of `kind` at log index `index` gates on the earlier + * registry entry `other` (at `otherIndex`). + * + * Single source of truth for that question, called by both + * {@link awaitEarlierDeliveries} (which awaits what it gates on) and + * {@link computeResolvesOnItsOwn} (which recurses into what it gates on). + * Those two MUST agree exactly, and the doc block on + * {@link awaitEarlierDeliveries} stakes deadlock-freedom on it, so the + * condition lives here rather than being spelled out twice. + */ +function gatesOn( + kind: DeliveryKind, + index: number, + otherIndex: number, + other: DeliveryBarrierEntry +): boolean { + if (otherIndex >= index || !DEFER_BEHIND[kind].includes(other.kind)) { + return false; + } + // A step skips an UNARMED earlier entry (an unclaimed buffered hook + // payload) — see the asymmetry described on `awaitEarlierDeliveries`. The + // skip is direct, never transitive: armed entries are still gated on, even + // when they are themselves parked behind such a payload. + return !(kind === 'step' && !other.armed); +} + /** * Whether `entry` will resolve on its own — it is armed, and every earlier - * delivery it defers behind will likewise resolve on its own. + * delivery it actually gates on ({@link gatesOn}) will likewise resolve on its + * own. * - * A step delivery is always self-resolving: it skips uncommitted deliveries - * (see {@link awaitEarlierDeliveries}), and the earlier steps it does defer - * behind are self-resolving by the same argument, inducting down on index. + * A step does not gate on an unclaimed buffered payload, so such a payload + * cannot keep it from resolving. A step DOES gate on earlier armed waits and + * hooks, so one parked behind an unclaimed payload makes the step + * non-self-resolving in turn. Disagreeing with {@link awaitEarlierDeliveries} + * here would not be a cosmetic problem: this predicate is what + * {@link hasParkedCommittedDelivery} uses to decide whether idle is reachable, + * and an entry reported self-resolving while it is in fact parked behind a + * payload that only the idle safety net can retire would gate its own + * retirement. * * Recursion terminates because every edge points to a strictly smaller index. - * `memo` is required rather than an optimization: without it the walk is - * exponential in the number of live hook/wait barriers (each armed entry - * re-walks every earlier entry of the opposite kind, T(n) = Σ T(j)), and the - * registry is not small by construction — `EventsConsumer` drains - * consecutively consumable events synchronously while barriers only retire on - * microtask-driven deliveries, so a fan-out of `Promise.race([hook, sleep])` - * branches accumulates one barrier per branch per kind. Memoized, the walk is - * linear in registry size. The memo MUST be per-call: `armed` mutates between + * `memo` keeps the walk linear in registry size, and the registry is not small + * by construction — `EventsConsumer` drains consecutively consumable events + * synchronously while barriers only retire on microtask-driven deliveries, so + * a fan-out of `Promise.race([hook, sleep])` branches accumulates one barrier + * per branch per kind. The memo MUST be per-call: `armed` mutates between * calls as buffered payloads are claimed. + * + * The memo is an optimization, not a correctness requirement. It once was one: + * `awaitEarlierDeliveries` used to run this walk for every earlier entry of a + * step delivery, with no early exit, which unmemoized is T(n) = Σ T(j) — + * measured at 4.3e8 recursive calls (84s) for 40 alternating armed hook/wait + * barriers. That call site is gone; a step now tests `armed` directly. The one + * surviving caller, {@link hasParkedCommittedDelivery}, cannot reach that + * shape: it returns at the FIRST self-resolving entry, so it only ever + * advances past entries that are non-self-resolving, and those short-circuit + * on their first false child. Every entry it evaluates therefore has + * all-false predecessors and returns after one child, degenerating the walk to + * a chain (measured: 98 calls unmemoized for the worst 40-barrier shape, 1 + * call for the registry above). Do not restore an exponential claim here + * without restoring a caller that can produce it. */ function resolvesOnItsOwn( barriers: Map, @@ -318,16 +363,11 @@ function computeResolvesOnItsOwn( if (!entry.armed) { return false; } - if (entry.kind === 'step') { - return true; - } - const deferBehind = DEFER_BEHIND[entry.kind]; for (const [otherIndex, other] of barriers) { - if ( - otherIndex < index && - deferBehind.includes(other.kind) && - !resolvesOnItsOwn(barriers, otherIndex, other, memo) - ) { + if (!gatesOn(entry.kind, index, otherIndex, other)) { + continue; + } + if (!resolvesOnItsOwn(barriers, otherIndex, other, memo)) { return false; } } @@ -347,17 +387,44 @@ function computeResolvesOnItsOwn( * suspension point first; see the comment at that `await` for why ordering the * `resolve()` calls alone is not enough. * - * One asymmetry: a STEP result additionally skips any earlier delivery that - * will not resolve on its own, i.e. one blocked (directly or transitively) on - * a buffered hook payload no consumer has claimed. Such a payload is delivered - * only when the workflow next reads the hook, and reaching that read very - * commonly requires the step result itself (`await stepX()` before the read). - * Gating the step on it would stall the workflow until the barrier's idle - * safety net fires, which then releases every delivery queued behind that + * What counts as "defers behind" is {@link gatesOn}, shared with + * {@link computeResolvesOnItsOwn} so the two cannot drift. + * + * One asymmetry: a STEP result skips any earlier delivery that is UNARMED, + * i.e. a buffered hook payload no consumer has claimed. Such a payload is + * delivered only when the workflow next reads the hook, and reaching that read + * very commonly requires the step result itself (`await stepX()` before the + * read). Gating the step on it would stall the workflow until the barrier's + * idle safety net fires, which then releases every delivery queued behind that * payload at once — losing exactly the race this ordering exists to protect. * Waits and hooks keep gating on unclaimed payloads: for them, waiting for the * claim IS the ordering guarantee (a `wait_completed` must not preempt a * payload the log ordered first). + * + * The skip is direct, never transitive. A step still gates on an earlier ARMED + * wait or hook, including one that is itself parked behind an unclaimed + * payload. Skipping those too would invert log order for the commonest shape + * there is: a workflow that creates a hook it does not read on this branch, + * races `step` against `sleep`, and has the log say the sleep won. The step + * would then overtake the wait, both branches would swap the correlation ids + * they draw next, and replay would diverge — see + * `step-delivery-ordering.test.ts`. Waiting instead is safe because the + * payload's own idle safety net retires it and the whole chain then delivers + * in log order; {@link hasParkedCommittedDelivery} deliberately reports such a + * step as not self-resolving so that idle stays reachable. + * + * "The whole chain then delivers in log order" rests on the PAYLOAD's safety + * net observing idle before the net of the wait parked behind it. If the + * wait's net fired first, the step's gate would open while the wait was still + * parked on the payload barrier and the inversion above would reappear. Within + * one drain window that order is structural, and carried by FIFO of the + * safety-net polls: nets arm via `setTimeout` in log order during synchronous + * consumption, each polling round re-arms through `promiseQueue.then(...)` in + * the order the checks ran, and each net that fires flips + * {@link hasParkedCommittedDelivery} back to true, re-blocking the rest until + * the released delivery completes. Replay — where divergence manifests — + * always consumes the log in one window. Do not "optimize" the net scheduling + * in a way that breaks that per-window FIFO. */ export async function awaitEarlierDeliveries( ctx: WorkflowOrchestratorContext, @@ -373,18 +440,9 @@ export async function awaitEarlierDeliveries( return; } const barriers = ctx.pendingDeliveryBarriers; - const deferBehind = DEFER_BEHIND[kind]; const earlier: Promise[] = []; - // Shared across this call only — see `resolvesOnItsOwn`. - const selfResolving = new Map(); for (const [index, entry] of barriers) { - if (index >= eventIndex || !deferBehind.includes(entry.kind)) { - continue; - } - if ( - kind === 'step' && - !resolvesOnItsOwn(barriers, index, entry, selfResolving) - ) { + if (!gatesOn(kind, eventIndex, index, entry)) { continue; } earlier.push(entry.delivered); @@ -517,7 +575,10 @@ export function registerDeliveryBarrier( * Deliveries that do NOT resolve on their own must be excluded, not for * accuracy but for termination: an unclaimed buffered hook payload is retired * BY the idle safety net in {@link registerDeliveryBarrier}, so counting it - * here would gate its own retirement. Self-resolving deliveries always + * here would gate its own retirement. That reasoning extends to whatever is + * parked behind such a payload — a wait, and a step gating on that wait — for + * the same reason: the whole chain moves only once the net fires, and it + * cannot fire while the chain is counted. Self-resolving deliveries always * deliver from their own chains (see the INVARIANT on * {@link registerDeliveryBarrier}) and never need that net, so waiting on * them is deadlock-free. diff --git a/packages/core/src/step-delivery-ordering.test.ts b/packages/core/src/step-delivery-ordering.test.ts index 64ffeaa276..60213522f8 100644 --- a/packages/core/src/step-delivery-ordering.test.ts +++ b/packages/core/src/step-delivery-ordering.test.ts @@ -150,8 +150,15 @@ const CORR_IDS = [ '01K11TFZ62YS0YYFDQ3E8B9YCW', '01K11TFZ62YS0YYFDQ3E8B9YCX', '01K11TFZ62YS0YYFDQ3E8B9YCY', + '01K11TFZ62YS0YYFDQ3E8B9YCZ', ]; +function pendingStepNames(ctx: WorkflowOrchestratorContext): string[] { + return [...ctx.invocationsQueue.values()] + .filter((item) => item.type === 'step') + .map((item) => (item.type === 'step' ? item.stepName : '')); +} + async function runWithDiscontinuation( ctx: WorkflowOrchestratorContext, workflowFn: () => Promise @@ -311,12 +318,6 @@ describe('step result delivery ordering across replays', () => { }; } - function pendingStepNames(ctx: WorkflowOrchestratorContext): string[] { - return [...ctx.invocationsQueue.values()] - .filter((item) => item.type === 'step') - .map((item) => (item.type === 'step' ? item.stepName : '')); - } - it('delivers the wait before the step result on the first replay, matching the log', async () => { const hydration = delayHydration(); spy = await hydration.install(); @@ -514,12 +515,6 @@ describe('step result delivery ordering across replays', () => { }; } - function pendingStepNames(ctx: WorkflowOrchestratorContext): string[] { - return [...ctx.invocationsQueue.values()] - .filter((item) => item.type === 'step') - .map((item) => (item.type === 'step' ? item.stepName : '')); - } - it('delivers the hook payload before the step result on the first replay, matching the log', async () => { const hydration = delayHydration(); spy = await hydration.install(); @@ -608,4 +603,206 @@ describe('step result delivery ordering across replays', () => { } }); }); + + /** + * Third shape, and the one that survives the ordering fix in #3139: a step + * result overtaking a wait that is itself parked behind an UNCLAIMED hook + * payload. + * + * A hook payload registers its delivery barrier unarmed when no branch is + * waiting on it (`workflow/hook.ts`, `armed: promises.length > 0`), because + * nothing in the workflow will ever resolve it — only the barrier registry's + * idle safety net retires it. Every other delivery that defers behind hooks + * therefore parks behind that payload, waits included. + * + * A step result may skip an unclaimed payload, or it would stall until that + * safety net fires. The bug is that the skip is TRANSITIVE: the step also + * skips the wait that is merely parked behind the payload, even though the + * wait sits earlier in the log and would otherwise gate it. The step wins a + * race the committed log recorded for the wait, the two branches swap + * correlation ids, and replay diverges. + * + * Production shape (o2flow `stepStormReproWorkflow`): the workflow creates a + * poke hook it never reads, so every `hook_received` arrives unclaimed, and + * the watchdog `wait_completed` events that decide each `Promise.race` sit + * behind it. The hook-storm variant of the same workflow consumes its hook + * and has never reproduced the divergence, which is the control below. + * + * Unlike the two shapes above, this one needs no hydration delay and no + * shared payload cache: the inversion is structural, not a latency race, so + * a single replay on the ordinary path is enough to show it. + */ + describe('step_completed behind a wait parked on an unclaimed hook payload', () => { + const resumeAt = new Date('2026-07-27T12:00:05.000Z'); + + async function buildEventLog(): Promise { + const ops: Promise[] = []; + const [hookPayload, stepAResult] = await Promise.all([ + dehydrateStepReturnValue({ kind: 'poke' }, 'wrun_test', undefined, ops), + dehydrateStepReturnValue('ok', 'wrun_test', undefined, ops), + ]); + + return [ + { + eventId: 'evnt_0', + runId: 'wrun_test', + eventType: 'hook_created', + correlationId: `hook_${CORR_IDS[0]}`, + eventData: { token: 'poke-token', isWebhook: false }, + createdAt: new Date(), + }, + { + eventId: 'evnt_1', + runId: 'wrun_test', + eventType: 'step_created', + correlationId: `step_${CORR_IDS[1]}`, + eventData: { stepName: 'stepA' }, + createdAt: new Date(), + }, + { + eventId: 'evnt_2', + runId: 'wrun_test', + eventType: 'wait_created', + correlationId: `wait_${CORR_IDS[2]}`, + eventData: { resumeAt }, + createdAt: new Date(), + }, + { + eventId: 'evnt_3', + runId: 'wrun_test', + eventType: 'step_started', + correlationId: `step_${CORR_IDS[1]}`, + eventData: { stepName: 'stepA' }, + createdAt: new Date(), + }, + // Nothing in the workflow reads this hook, so its barrier registers + // unarmed and every later delivery that defers behind hooks parks on + // it. + { + eventId: 'evnt_4', + runId: 'wrun_test', + eventType: 'hook_received', + correlationId: `hook_${CORR_IDS[0]}`, + eventData: { token: 'poke-token', payload: hookPayload }, + createdAt: new Date(), + }, + // The live invocation delivered the wait BEFORE the step result: the + // sleep branch resumed first and drew the next correlation id. + { + eventId: 'evnt_5', + runId: 'wrun_test', + eventType: 'wait_completed', + correlationId: `wait_${CORR_IDS[2]}`, + eventData: { resumeAt }, + createdAt: new Date(), + }, + { + eventId: 'evnt_6', + runId: 'wrun_test', + eventType: 'step_completed', + correlationId: `step_${CORR_IDS[1]}`, + eventData: { stepName: 'stepA', result: stepAResult }, + createdAt: new Date(), + }, + { + eventId: 'evnt_7', + runId: 'wrun_test', + eventType: 'step_created', + correlationId: `step_${CORR_IDS[3]}`, + eventData: { stepName: 'afterSleep' }, + createdAt: new Date(), + }, + { + eventId: 'evnt_8', + runId: 'wrun_test', + eventType: 'step_created', + correlationId: `step_${CORR_IDS[4]}`, + eventData: { stepName: 'afterStep' }, + createdAt: new Date(), + }, + ]; + } + + /** + * Draw order: `createHook()` takes CORR_IDS[0], `stepA()` CORR_IDS[1], + * `sleep()` CORR_IDS[2]; then whichever branch resumes FIRST takes + * CORR_IDS[3] and the other takes CORR_IDS[4]. + * + * The `awaited` variant adds a third branch that awaits the payload and + * draws nothing, so both variants replay the SAME event log and differ + * only in whether the payload is claimed. + */ + function workflowBody( + ctx: WorkflowOrchestratorContext, + poke: 'unclaimed' | 'awaited' + ) { + const useStep = createUseStep(ctx); + const sleep = createSleep(ctx); + const createHook = createCreateHook(ctx); + + return async () => { + const stepA = useStep('stepA'); + const afterStep = useStep('afterStep'); + const afterSleep = useStep('afterSleep'); + const pokeHook = createHook<{ kind: string }>({ token: 'poke-token' }); + + const branchStep = (async () => { + await stepA(); + await afterStep(); + })(); + const branchSleep = (async () => { + await sleep(resumeAt); + await afterSleep(); + })(); + const branchPoke = (async () => { + if (poke === 'awaited') { + await pokeHook; + } + })(); + + await Promise.all([branchStep, branchSleep, branchPoke]); + }; + } + + it('keeps log order when the hook payload is never claimed', async () => { + const events = await buildEventLog(); + + const ctx = setupWorkflowContext(events); + const { error } = await runWithDiscontinuation( + ctx, + workflowBody(ctx, 'unclaimed') + ); + + expect(error).toBeDefined(); + // FAILS on `main`: the step result skips the wait transitively through + // the unclaimed payload, `afterStep` draws CORR_IDS[3], and replay + // diverges at evnt_7 with the production error shape ("... belongs to + // \"afterSleep\", but the current step consumer is \"afterStep\""). + if (!WorkflowSuspension.is(error)) { + throw error; + } + expect(pendingStepNames(ctx).sort()).toEqual(['afterSleep', 'afterStep']); + expect(ctx.eventsConsumer.eventIndex).toBe(events.length); + }); + + // Control: same event log, but a branch awaits the payload, so the hook + // barrier arms, the wait no longer parks behind it, and the step gates on + // the wait the ordinary way. This passes on `main` and must keep passing. + it('keeps log order when the hook payload is claimed', async () => { + const events = await buildEventLog(); + + const ctx = setupWorkflowContext(events); + const { error } = await runWithDiscontinuation( + ctx, + workflowBody(ctx, 'awaited') + ); + + expect(error).toBeDefined(); + if (!WorkflowSuspension.is(error)) { + throw error; + } + expect(pendingStepNames(ctx).sort()).toEqual(['afterSleep', 'afterStep']); + expect(ctx.eventsConsumer.eventIndex).toBe(events.length); + }); + }); }); diff --git a/workbench/fastify/public/index.html b/workbench/fastify/public/index.html deleted file mode 100644 index 59870afcbd..0000000000 --- a/workbench/fastify/public/index.html +++ /dev/null @@ -1,73 +0,0 @@ - - - - - - Workflow SDK + Nitro Example - - - - -

Workflow SDK + Nitro Example

-
- - - - From 4ec7acaa7196a6f2f5025a65f05d5bdaaf5705ba Mon Sep 17 00:00:00 2001 From: Luca Maraschi <332968+lucamaraschi@users.noreply.github.com> Date: Mon, 10 Aug 2026 13:08:53 -0700 Subject: [PATCH 3/8] feat(builders): observe accepted transforms (#3163) Signed-off-by: Luca Maraschi Co-authored-by: Peter Wielander --- .changeset/tidy-dodos-observe.md | 5 + packages/builders/README.md | 26 ++++ packages/builders/src/base-builder.ts | 3 + packages/builders/src/index.ts | 6 +- .../builders/src/swc-esbuild-plugin.test.ts | 119 ++++++++++++++++++ packages/builders/src/swc-esbuild-plugin.ts | 29 +++++ packages/builders/src/types.ts | 14 +++ 7 files changed, 201 insertions(+), 1 deletion(-) create mode 100644 .changeset/tidy-dodos-observe.md diff --git a/.changeset/tidy-dodos-observe.md b/.changeset/tidy-dodos-observe.md new file mode 100644 index 0000000000..260eae53a8 --- /dev/null +++ b/.changeset/tidy-dodos-observe.md @@ -0,0 +1,5 @@ +--- +'@workflow/builders': minor +--- + +Add an optional observer for accepted workflow SWC transform results. diff --git a/packages/builders/README.md b/packages/builders/README.md index ed0f61fe2a..763be67f89 100644 --- a/packages/builders/README.md +++ b/packages/builders/README.md @@ -30,6 +30,32 @@ class MyBuilder extends BaseBuilder { } ``` +### Observing transforms + +Builder configurations can provide an optional `onAfterTransform` observer for +tooling that derives metadata from the exact SWC output used by a build: + +```typescript +const builder = new MyBuilder({ + // Other builder configuration... + onAfterTransform: async ({ + mode, + filename, + absolutePath, + source, + code, + workflowManifest, + }) => { + // Observe the accepted transform result. + }, +}); +``` + +The observer is awaited after the transform's manifest entries have been +accepted. It cannot replace the generated code, and throwing aborts the build. +A source file may be observed multiple times across transform modes, bundles, +and watch rebuilds, so consumers should deduplicate results when necessary. + ## Architecture The builder system uses: diff --git a/packages/builders/src/base-builder.ts b/packages/builders/src/base-builder.ts index 16009dd401..607df9c109 100644 --- a/packages/builders/src/base-builder.ts +++ b/packages/builders/src/base-builder.ts @@ -1159,6 +1159,7 @@ export const __steps_registered = true; projectRoot: this.transformProjectRoot, moduleSpecifierRoot: this.moduleSpecifierRoot, workflowManifest, + onAfterTransform: this.config.onAfterTransform, bundleTransitiveLocalStepDependencies, rewriteTsExtensions, sideEffectEntries: normalizedSideEffectEntries, @@ -1392,6 +1393,7 @@ export const __steps_registered = true; projectRoot: this.transformProjectRoot, moduleSpecifierRoot: this.moduleSpecifierRoot, workflowManifest, + onAfterTransform: this.config.onAfterTransform, sideEffectEntries: normalizedWorkflowSideEffectEntries, }), // This plugin must run after the swc plugin to ensure dead code elimination @@ -1940,6 +1942,7 @@ ${createWorkflowRouteHandlersCode(`workflowEntrypoint(workflowCode${workflowEntr mode: 'step', projectRoot: this.transformProjectRoot, moduleSpecifierRoot: this.moduleSpecifierRoot, + onAfterTransform: this.config.onAfterTransform, sideEffectEntries: normalizedClientSideEffectEntries, }), ], diff --git a/packages/builders/src/index.ts b/packages/builders/src/index.ts index 0214bcee51..1a50c98217 100644 --- a/packages/builders/src/index.ts +++ b/packages/builders/src/index.ts @@ -45,7 +45,11 @@ export { type SerdeClassCheckResult, } from './serde-checker.js'; export { StandaloneBuilder } from './standalone.js'; -export { createSwcPlugin } from './swc-esbuild-plugin.js'; +export { + createSwcPlugin, + type WorkflowAfterTransformHook, + type WorkflowTransformResult, +} from './swc-esbuild-plugin.js'; export { detectWorkflowPatterns, generatedWorkflowPathPattern, diff --git a/packages/builders/src/swc-esbuild-plugin.test.ts b/packages/builders/src/swc-esbuild-plugin.test.ts index 06b8d477b9..c61a677f3d 100644 --- a/packages/builders/src/swc-esbuild-plugin.test.ts +++ b/packages/builders/src/swc-esbuild-plugin.test.ts @@ -51,10 +51,127 @@ describe('createSwcPlugin externalizeNonSteps', () => { rmSync(testRoot, { recursive: true, force: true }); }); + it('reports authoritative transform results to an optional observer', async () => { + const srcDir = join(testRoot, 'src'); + const stepFile = join(srcDir, 'step.ts'); + const source = 'export const value = 42;'; + const workflowManifest = { + steps: { + 'src/step.ts': { + value: { + stepId: 'step//src/step//value', + }, + }, + }, + }; + const onAfterTransform = vi.fn(); + + writeFile(stepFile, source); + applySwcTransformMock.mockResolvedValue({ + code: `${source}\n/* transformed */`, + workflowManifest, + }); + + await esbuild.build({ + entryPoints: [stepFile], + absWorkingDir: testRoot, + outdir: join(testRoot, 'out'), + bundle: true, + write: false, + plugins: [ + createSwcPlugin({ + mode: 'step', + entriesToBundle: [stepFile], + onAfterTransform, + }), + ], + }); + + expect(onAfterTransform).toHaveBeenCalledOnce(); + expect(onAfterTransform).toHaveBeenCalledWith({ + mode: 'step', + filename: 'src/step.ts', + absolutePath: stepFile, + source, + code: `${source}\n/* transformed */`, + workflowManifest, + }); + }); + + it('awaits asynchronous transform observers', async () => { + const stepFile = join(testRoot, 'src', 'step.ts'); + let markObserverStarted: () => void = () => {}; + let releaseObserver: () => void = () => {}; + const observerStarted = new Promise((resolve) => { + markObserverStarted = resolve; + }); + const observerBlocked = new Promise((resolve) => { + releaseObserver = resolve; + }); + let buildCompleted = false; + + writeFile(stepFile, 'export const value = 42;'); + + const build = esbuild.build({ + entryPoints: [stepFile], + absWorkingDir: testRoot, + outdir: join(testRoot, 'out'), + bundle: true, + write: false, + plugins: [ + createSwcPlugin({ + mode: 'step', + entriesToBundle: [stepFile], + onAfterTransform: async () => { + markObserverStarted(); + await observerBlocked; + }, + }), + ], + }); + void build.then(() => { + buildCompleted = true; + }); + + await observerStarted; + await Promise.resolve(); + expect(buildCompleted).toBe(false); + + releaseObserver(); + await build; + expect(buildCompleted).toBe(true); + }); + + it('fails the build when a transform observer throws', async () => { + const stepFile = join(testRoot, 'src', 'step.ts'); + + writeFile(stepFile, 'export const value = 42;'); + + await expect( + esbuild.build({ + entryPoints: [stepFile], + absWorkingDir: testRoot, + outdir: join(testRoot, 'out'), + bundle: true, + write: false, + plugins: [ + createSwcPlugin({ + mode: 'step', + entriesToBundle: [stepFile], + onAfterTransform: () => { + throw new Error('transform observer failed'); + }, + }), + ], + }) + ).rejects.toThrow(/transform observer failed/); + }); + it('fails the build when two files emit the same step id', async () => { const srcDir = join(testRoot, 'src'); const firstStepFile = join(srcDir, 'confirmation.ts'); const secondStepFile = join(srcDir, 'reschedule.ts'); + const onAfterTransform = vi.fn(); writeFile(firstStepFile, `export const first = true;`); writeFile(secondStepFile, `export const second = true;`); @@ -86,10 +203,12 @@ describe('createSwcPlugin externalizeNonSteps', () => { plugins: [ createSwcPlugin({ mode: 'step', + onAfterTransform, }), ], }) ).rejects.toThrow(/Duplicate workflow step ID/); + expect(onAfterTransform).toHaveBeenCalledOnce(); }); it('fails the build when two files emit the same workflow id', async () => { diff --git a/packages/builders/src/swc-esbuild-plugin.ts b/packages/builders/src/swc-esbuild-plugin.ts index a609bb8e4a..9b18bc3bba 100644 --- a/packages/builders/src/swc-esbuild-plugin.ts +++ b/packages/builders/src/swc-esbuild-plugin.ts @@ -15,6 +15,19 @@ import { import { resolveModuleSpecifier } from './module-specifier.js'; import { resolveWorkflowAliasRelativePath } from './workflow-alias.js'; +export interface WorkflowTransformResult { + readonly mode: 'step' | 'workflow'; + readonly filename: string; + readonly absolutePath: string; + readonly source: string; + readonly code: string; + readonly workflowManifest: WorkflowManifest; +} + +export type WorkflowAfterTransformHook = ( + result: WorkflowTransformResult +) => void | Promise; + export interface SwcPluginOptions { mode: 'step' | 'workflow'; entriesToBundle?: string[]; @@ -22,6 +35,13 @@ export interface SwcPluginOptions { projectRoot?: string; moduleSpecifierRoot?: string; workflowManifest?: WorkflowManifest; + /** + * Optional observer invoked after a transform's manifest entries have been + * accepted. A file may be observed multiple times across modes, bundles, and + * watch rebuilds. The observer is awaited, cannot replace the generated code, + * and aborts the build if it throws. + */ + onAfterTransform?: WorkflowAfterTransformHook; /** * Rewrite TypeScript extensions (.ts, .tsx, .mts, .cts) to their JS * equivalents (.js, .mjs, .cjs) in externalized import paths. @@ -516,6 +536,15 @@ export function createSwcPlugin(options: SwcPluginOptions): Plugin { workflowIdsForCurrentBuild ); + await options.onAfterTransform?.({ + mode: options.mode, + filename: relativeFilepath, + absolutePath: args.path, + source: normalizedSource, + code: transformedCode, + workflowManifest, + }); + return { contents: transformedCode, loader, diff --git a/packages/builders/src/types.ts b/packages/builders/src/types.ts index 8bae74301d..e127cc4ef6 100644 --- a/packages/builders/src/types.ts +++ b/packages/builders/src/types.ts @@ -1,3 +1,5 @@ +import type { WorkflowAfterTransformHook } from './swc-esbuild-plugin.js'; + export const validBuildTargets = [ 'standalone', 'vercel-build-output-api', @@ -49,6 +51,18 @@ interface BaseWorkflowConfig { workflowManifestPath?: string; + /** + * Optional observer invoked after each authoritative SWC transform has been + * accepted into a workflow bundle's manifest. + * + * A source file may be observed multiple times across transform modes, + * bundles, and watch rebuilds. The observer is awaited and cannot replace the + * transformed code. Throwing rejects the build, allowing integrations to + * require their derived artifacts to remain consistent with the emitted + * workflow bundles. + */ + onAfterTransform?: WorkflowAfterTransformHook; + // Optional prefix for debug files (e.g., "_" for Astro to ignore them) debugFilePrefix?: string; From dc61ea1b1313e2c5c165a1d873ac5604ce5d26f3 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Mon, 10 Aug 2026 13:52:01 -0700 Subject: [PATCH 4/8] docs(builders): make the onAfterTransform sample self-contained (#3424) --- .changeset/olive-pugs-repeat.md | 4 ++++ packages/builders/README.md | 26 +++++++++++++------------- 2 files changed, 17 insertions(+), 13 deletions(-) create mode 100644 .changeset/olive-pugs-repeat.md diff --git a/.changeset/olive-pugs-repeat.md b/.changeset/olive-pugs-repeat.md new file mode 100644 index 0000000000..d1b19b720d --- /dev/null +++ b/.changeset/olive-pugs-repeat.md @@ -0,0 +1,4 @@ +--- +--- + +Fix the `onAfterTransform` sample in the builders README so it type-checks on its own. diff --git a/packages/builders/README.md b/packages/builders/README.md index 763be67f89..62b2a11a41 100644 --- a/packages/builders/README.md +++ b/packages/builders/README.md @@ -36,19 +36,19 @@ Builder configurations can provide an optional `onAfterTransform` observer for tooling that derives metadata from the exact SWC output used by a build: ```typescript -const builder = new MyBuilder({ - // Other builder configuration... - onAfterTransform: async ({ - mode, - filename, - absolutePath, - source, - code, - workflowManifest, - }) => { - // Observe the accepted transform result. - }, -}); +import type { WorkflowAfterTransformHook } from '@workflow/builders'; + +// Pass as `onAfterTransform` in the builder configuration. +const onAfterTransform: WorkflowAfterTransformHook = async ({ + mode, + filename, + absolutePath, + source, + code, + workflowManifest, +}) => { + // Observe the accepted transform result. +}; ``` The observer is awaited after the transform's manifest entries have been From 1a64f684723757c5a839abb94189b953dd3ac536 Mon Sep 17 00:00:00 2001 From: Makoto Arata Date: Tue, 11 Aug 2026 07:49:02 +0900 Subject: [PATCH 5/8] fix(core): preserve `new.target` in the deterministic `Date` override so `Date` subclasses work (#3372) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test: add failing test for Date subclassing in workflow VM Co-Authored-By: Claude Fable 5 Signed-off-by: ar_tama * fix(core): preserve `new.target` in the deterministic `Date` override so `Date` subclasses work in workflow functions The VM's `Date` override was a plain function, so `class X extends Date` lost the subclass identity: `super()` returned a fresh plain `Date` that became `this`, dropping the subclass's methods and fields. This silently broke `Date` subclasses like `TZDate` from `@date-fns/tz`. Using `class Date extends Date_` keeps `new.target` intact, and `extends` already wires up the prototype chain and statics, so the manual `prototype` assignment and `Object.setPrototypeOf` fix-ups are no longer needed. Determinism is unchanged: zero-arg construction still returns the fixed timestamp and `Date.now()` is still overridden. Fixes #3371 Co-Authored-By: Claude Fable 5 Signed-off-by: ar_tama * test: add failing test for calling `Date()` without `new` Co-Authored-By: Claude Fable 5 Signed-off-by: ar_tama * fix(core): keep `Date()` callable without `new` Use a plain function that branches on `new.target` and constructs via `Reflect.construct(Date_, args, new.target)` instead of a class: subclassing still works (`new.target` is forwarded), and calling `Date()` without `new` now matches the spec — arguments are ignored and the (fixed) time string is returned, where the previous override returned a `Date` object. Co-Authored-By: Claude Fable 5 Signed-off-by: ar_tama * chore: update changeset to match the final `Reflect.construct` implementation Co-Authored-By: Claude Fable 5 Signed-off-by: ar_tama --------- Signed-off-by: ar_tama Co-authored-by: Claude Fable 5 --- .changeset/date-subclass-vm.md | 5 +++ packages/core/src/vm/index.test.ts | 58 ++++++++++++++++++++++++++++++ packages/core/src/vm/index.ts | 21 ++++++----- 3 files changed, 76 insertions(+), 8 deletions(-) create mode 100644 .changeset/date-subclass-vm.md diff --git a/.changeset/date-subclass-vm.md b/.changeset/date-subclass-vm.md new file mode 100644 index 0000000000..25760a72eb --- /dev/null +++ b/.changeset/date-subclass-vm.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Fix `Date` subclassing inside workflow functions. The deterministic `Date` override in the workflow VM now forwards `new.target` via `Reflect.construct`, so subclasses like `TZDate` from `@date-fns/tz` keep their identity, methods, and fields. Calling `Date()` without `new` now returns the (fixed) time string per spec, instead of a `Date` object. diff --git a/packages/core/src/vm/index.test.ts b/packages/core/src/vm/index.test.ts index 3f2ebbabeb..5cc9bdef50 100644 --- a/packages/core/src/vm/index.test.ts +++ b/packages/core/src/vm/index.test.ts @@ -49,6 +49,64 @@ describe('createContext', () => { expect(result).toEqual(specificTime); }); + it('should support subclassing `Date`', () => { + const { context } = createContext({ seed, fixedTimestamp }); + + const result = vm.runInContext( + ` + class Sub extends Date { + constructor(...args) { + super(...args); + this.tag = 'sub'; + } + label() { + return 'sub'; + } + } + const sub = new Sub(2026, 6, 29); + const defaulted = new Sub(); + ({ + isSub: sub instanceof Sub, + isDate: sub instanceof Date, + keepsMethods: sub.label(), + keepsFields: sub.tag, + argsForwarded: sub.getTime() === new Date(2026, 6, 29).getTime(), + defaultedIsFixed: defaulted.getTime(), + }) + `, + context + ); + + expect(result.isSub).toBe(true); + expect(result.isDate).toBe(true); + expect(result.keepsMethods).toBe('sub'); + expect(result.keepsFields).toBe('sub'); + expect(result.argsForwarded).toBe(true); + expect(result.defaultedIsFixed).toEqual(fixedTimestamp); + }); + + it('should keep `Date()` callable without `new`, returning the fixed time string', () => { + const { context } = createContext({ seed, fixedTimestamp }); + + const result = vm.runInContext('Date()', context); + + expect(result).toBeTypeOf('string'); + expect(result).toEqual(vm.runInContext('new Date().toString()', context)); + // Per spec, `Date()` as a function ignores its arguments + expect(vm.runInContext('Date(2000, 0, 1)', context)).toEqual(result); + }); + + it('should preserve `Date` static methods', () => { + const { context } = createContext({ seed, fixedTimestamp }); + + expect( + vm.runInContext("Date.parse('2000-01-01T00:00:00.000Z')", context) + ).toEqual(946684800000); + expect(vm.runInContext('Date.UTC(2000, 0, 1)', context)).toEqual( + 946684800000 + ); + }); + it('should have deterministic `crypto.getRandomValues()`', () => { const { context } = createContext({ seed, fixedTimestamp }); diff --git a/packages/core/src/vm/index.ts b/packages/core/src/vm/index.ts index 0355abb884..a747075ac9 100644 --- a/packages/core/src/vm/index.ts +++ b/packages/core/src/vm/index.ts @@ -60,17 +60,22 @@ export function createContext(options: CreateContextOptions) { // Deterministic `Math.random()` g.Math.random = rng; - // Override `Date` constructor to return fixed time when called without arguments + // Override `Date` constructor to return fixed time when called without + // arguments. Constructing through `Reflect.construct` with `new.target` + // keeps subclassing intact (e.g. `TZDate` from `@date-fns/tz`), while a + // plain function (rather than a `class`) keeps `Date()` callable without + // `new`, which per spec ignores its arguments and returns the time string. const Date_ = g.Date; // biome-ignore lint/suspicious/noShadowRestrictedNames: We're shadowing the global `Date` property to make it deterministic. - (g as any).Date = function Date( - ...args: Parameters<(typeof globalThis)['Date']>[] - ) { - if (args.length === 0) { - return new Date_(fixedTimestamp); + (g as any).Date = function Date(...args: any[]) { + if (new.target === undefined) { + return new Date_(fixedTimestamp).toString(); } - // @ts-expect-error - Args is `Date` constructor arguments - return new Date_(...args); + return Reflect.construct( + Date_, + args.length === 0 ? [fixedTimestamp] : args, + new.target + ); }; (g as any).Date.prototype = Date_.prototype; // Preserve static methods From 2d5ca54086ac66e0fa3ad66e90c2626acbf25d67 Mon Sep 17 00:00:00 2001 From: Caleb An Date: Mon, 10 Aug 2026 16:07:44 -0700 Subject: [PATCH 6/8] Set vercel approvers to workflow team (#3435) --- .vercel.approvers | 1 + 1 file changed, 1 insertion(+) create mode 100644 .vercel.approvers diff --git a/.vercel.approvers b/.vercel.approvers new file mode 100644 index 0000000000..f8ea5dc538 --- /dev/null +++ b/.vercel.approvers @@ -0,0 +1 @@ +@vercel/workflow From 69c30ff49eb89c0c4c4b2642c37985fdf64fa9fd Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Mon, 10 Aug 2026 17:35:08 -0700 Subject: [PATCH 7/8] Gate the unconsumed-event check on delivery idleness (#3439) --- .changeset/tidy-buttons-swim.md | 5 + packages/core/src/events-consumer.test.ts | 85 +++++++++- packages/core/src/events-consumer.ts | 91 +++++++++-- packages/core/src/private.ts | 32 +++- .../unconsumed-check-delivery-idle.test.ts | 149 ++++++++++++++++++ packages/core/src/workflow.ts | 9 ++ 6 files changed, 347 insertions(+), 24 deletions(-) create mode 100644 .changeset/tidy-buttons-swim.md create mode 100644 packages/core/src/unconsumed-check-delivery-idle.test.ts diff --git a/.changeset/tidy-buttons-swim.md b/.changeset/tidy-buttons-swim.md new file mode 100644 index 0000000000..d07ebeeecc --- /dev/null +++ b/.changeset/tidy-buttons-swim.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Stop reporting replay divergence for an event the workflow is still on its way to consuming, by waiting for in-flight step and hook deliveries instead of a fixed delay diff --git a/packages/core/src/events-consumer.test.ts b/packages/core/src/events-consumer.test.ts index ecf828f728..c85a660ea6 100644 --- a/packages/core/src/events-consumer.test.ts +++ b/packages/core/src/events-consumer.test.ts @@ -1,7 +1,11 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import { describe, expect, it, vi } from 'vitest'; -import { EventConsumerResult, EventsConsumer } from './events-consumer.js'; +import { + DEFERRED_CHECK_DELAY_MS, + EventConsumerResult, + EventsConsumer, +} from './events-consumer.js'; // Helper function to create mock events function createMockEvent(overrides: Partial = {}): Event { @@ -476,4 +480,83 @@ describe('EventsConsumer', () => { expect(onUnconsumedEvent).not.toHaveBeenCalled(); }); }); + + describe('delivery-idle gate', () => { + // An event nobody claims is only evidence of divergence once the workflow + // VM has stopped reacting. While a delivery is in flight the walk is + // simply ahead of the code that would register the consumer, so the check + // has to wait rather than time out. See `isDeliveryIdle` in private.ts. + it('should not fire the unconsumed check while a delivery is in flight', async () => { + const event = createMockEvent(); + const onUnconsumedEvent = vi.fn(); + let idle = false; + const consumer = new EventsConsumer([event], { + onUnconsumedEvent, + getPromiseQueue: () => Promise.resolve(), + isDeliveryIdle: () => idle, + }); + + consumer.subscribe( + vi.fn().mockReturnValue(EventConsumerResult.NotConsumed) + ); + + // Several times the window the check would otherwise have fired in. + await new Promise((resolve) => + setTimeout(resolve, DEFERRED_CHECK_DELAY_MS * 5) + ); + expect(onUnconsumedEvent).not.toHaveBeenCalled(); + + idle = true; + await vi.waitFor(() => { + expect(onUnconsumedEvent).toHaveBeenCalledWith(event); + }); + }); + + it('should let a consumer registered during the wait claim the event', async () => { + const event = createMockEvent(); + const onUnconsumedEvent = vi.fn(); + let idle = false; + const consumer = new EventsConsumer([event], { + onUnconsumedEvent, + getPromiseQueue: () => Promise.resolve(), + isDeliveryIdle: () => idle, + }); + + consumer.subscribe( + vi.fn().mockReturnValue(EventConsumerResult.NotConsumed) + ); + await new Promise((resolve) => + setTimeout(resolve, DEFERRED_CHECK_DELAY_MS * 2) + ); + + // What the in-flight delivery was on its way to doing: resume workflow + // code that subscribes the consumer this event belongs to. + consumer.subscribe(vi.fn().mockReturnValue(EventConsumerResult.Finished)); + idle = true; + + await vi.waitFor(() => { + expect(consumer.eventIndex).toBe(1); + }); + await new Promise((resolve) => + setTimeout(resolve, DEFERRED_CHECK_DELAY_MS * 2) + ); + expect(onUnconsumedEvent).not.toHaveBeenCalled(); + }); + + it('should fire without delay for an event no delivery is waiting on', async () => { + const event = createMockEvent(); + const unconsumedReceived = withResolvers(); + const consumer = new EventsConsumer([event], { + onUnconsumedEvent: unconsumedReceived.resolve, + getPromiseQueue: () => Promise.resolve(), + isDeliveryIdle: () => true, + }); + + consumer.subscribe( + vi.fn().mockReturnValue(EventConsumerResult.NotConsumed) + ); + + expect(await unconsumedReceived.promise).toEqual(event); + }); + }); }); diff --git a/packages/core/src/events-consumer.ts b/packages/core/src/events-consumer.ts index cc05c6f975..2367000eab 100644 --- a/packages/core/src/events-consumer.ts +++ b/packages/core/src/events-consumer.ts @@ -65,6 +65,16 @@ export interface EventsConsumerOptions { * deserialization delays the resolve() that triggers the next subscribe(). */ getPromiseQueue: () => Promise; + /** + * Whether no data delivery is in flight (`isDeliveryIdle` in private.ts). + * The unconsumed-event check waits for this before it fires: a delivery in + * flight means the workflow VM is mid-reaction, and an event it has not + * claimed yet is an event it has not reached yet. + * + * Defaults to always-idle so the tests that drive a consumer with no + * orchestrator context keep the pre-existing timing. + */ + isDeliveryIdle?: () => boolean; } export class EventsConsumer { @@ -74,6 +84,7 @@ export class EventsConsumer { private onConsumedEvent?: (event: Event) => void; private onUnconsumedEvent: (event: Event) => void; private getPromiseQueue: () => Promise; + private isDeliveryIdle: () => boolean; private pendingUnconsumedCheck: Promise | null = null; private pendingUnconsumedTimeout: ReturnType | null = null; private unconsumedCheckVersion = 0; @@ -87,6 +98,7 @@ export class EventsConsumer { this.onConsumedEvent = options.onConsumedEvent; this.onUnconsumedEvent = options.onUnconsumedEvent; this.getPromiseQueue = options.getPromiseQueue; + this.isDeliveryIdle = options.isDeliveryIdle ?? (() => true); } append(events: Event[]): void { @@ -218,22 +230,71 @@ export class EventsConsumer { ) .then(() => this.getPromiseQueue()) .then(() => { - // Use a delayed setTimeout after the queue drains. The delay must be - // long enough for promise chains to propagate across the VM boundary - // (from resolve() in the host context through to the workflow code - // calling subscribe() in the VM context). Node.js does not guarantee - // that setTimeout(0) fires after all cross-context microtasks settle, - // so we use a small but non-zero delay. Any subscribe() call that - // arrives during this window will cancel the check via version - // invalidation + clearTimeout. - this.pendingUnconsumedTimeout = setTimeout(() => { - this.pendingUnconsumedTimeout = null; - if (this.unconsumedCheckVersion === checkVersion) { - this.pendingUnconsumedCheck = null; - this.onUnconsumedEvent(currentEvent); - } - }, getDeferredCheckDelayMs()); + // Wait out any delivery still in flight before starting the timer. + // The queue draining says the host has no hydration work left; it + // does not say the VM has finished reacting to what was hydrated. + this.whenDeliveryIdle(checkVersion, () => { + // Use a delayed setTimeout once deliveries are idle. The delay must + // be long enough for promise chains to propagate across the VM + // boundary (from resolve() in the host context through to the + // workflow code calling subscribe() in the VM context). Node.js + // does not guarantee that setTimeout(0) fires after all + // cross-context microtasks settle, so we use a small but non-zero + // delay. Any subscribe() call that arrives during this window will + // cancel the check via version invalidation + clearTimeout. + this.pendingUnconsumedTimeout = setTimeout(() => { + this.pendingUnconsumedTimeout = null; + if (this.unconsumedCheckVersion === checkVersion) { + this.pendingUnconsumedCheck = null; + this.onUnconsumedEvent(currentEvent); + } + }, getDeferredCheckDelayMs()); + }); }); } } + + /** + * Run `fn` once no data delivery is in flight, polling the way + * `scheduleWhenIdle` does: let the promise queue drain, re-check a timer + * tick later, repeat. + * + * Without this the check is a bet that every delivery the walk is running + * ahead of lands inside a fixed window. Consumption is synchronous while the + * resolution it triggers is not: a step result hydrates in the host, resolves + * from a detached continuation behind `awaitEarlierDeliveries`, and only then + * does VM code run far enough to subscribe the next consumer. Replaying a + * batch of N parallel step results leaves N-1 of them on that detached path + * with the queue already drained, so the walk sits on the ordered event the + * VM is about to draw and the window is the only thing standing between a + * healthy run and `ReplayDivergenceError`. On a backend whose deliveries take + * longer than the window, that bet loses: the local race repro corrupts 34 of + * 42 runs at a 10ms window and 0 of 114 at 100ms, on the same event logs. + * + * Termination is `hasParkedCommittedDelivery`'s: it counts only deliveries + * that resolve on their own, so nothing here can gate its own retirement. A + * genuinely orphaned event has no delivery to wait on and reaches `fn` on the + * first poll. + */ + private whenDeliveryIdle(checkVersion: number, fn: () => void): void { + const poll = () => { + this.pendingUnconsumedTimeout = null; + if (this.unconsumedCheckVersion !== checkVersion) { + return; + } + if (this.isDeliveryIdle()) { + fn(); + return; + } + this.getPromiseQueue().then(() => { + if (this.unconsumedCheckVersion !== checkVersion) { + return; + } + // Held in the same field the fired check uses so subscribe() cancels a + // poll in progress exactly as it cancels the check itself. + this.pendingUnconsumedTimeout = setTimeout(poll, 0); + }); + }; + poll(); + } } diff --git a/packages/core/src/private.ts b/packages/core/src/private.ts index c298bf5e6f..f7bedcd471 100644 --- a/packages/core/src/private.ts +++ b/packages/core/src/private.ts @@ -583,7 +583,9 @@ export function registerDeliveryBarrier( * {@link registerDeliveryBarrier}) and never need that net, so waiting on * them is deadlock-free. */ -function hasParkedCommittedDelivery(ctx: WorkflowOrchestratorContext): boolean { +export function hasParkedCommittedDelivery( + ctx: WorkflowOrchestratorContext +): boolean { const barriers = ctx.pendingDeliveryBarriers; if (!barriers || barriers.size === 0) { return false; @@ -599,12 +601,7 @@ function hasParkedCommittedDelivery(ctx: WorkflowOrchestratorContext): boolean { } /** - * Schedule a callback to fire only after all pending data deliveries - * (step results, hook payloads) and async deserialization have completed. - * Uses a polling loop: setTimeout(0) → check pendingDeliveries and the - * barrier registry → if anything is still in flight, wait for promiseQueue → - * repeat. This handles the multi-round delivery pattern where each hook - * payload delivery cycle appends new async work to the promiseQueue. + * Whether no data delivery (step result, hook payload) is in flight right now. * * "In flight" is two distinct windows, each with its own guard: * `pendingDeliveries > 0` covers hydration inside the serial queue slots, and @@ -612,6 +609,25 @@ function hasParkedCommittedDelivery(ctx: WorkflowOrchestratorContext): boolean { * releasing that counter and the delivery's `resolve()` actually running — * deliberately outside `pendingDeliveries` (see step.ts), and invisible to it. * + * Anything that decides a replay is over, or that a replay went wrong, has to + * consult this first: while it is false the workflow VM is mid-reaction, so + * what it has and has not done yet says nothing about the run. Two callers + * read it, for the two such decisions: {@link scheduleWhenIdle} for the + * suspension, and the events consumer's unconsumed-event check for divergence. + */ +export function isDeliveryIdle(ctx: WorkflowOrchestratorContext): boolean { + return ctx.pendingDeliveries === 0 && !hasParkedCommittedDelivery(ctx); +} + +/** + * Schedule a callback to fire only after all pending data deliveries + * (step results, hook payloads) and async deserialization have completed. + * Uses a polling loop: setTimeout(0) → check pendingDeliveries and the + * barrier registry → if anything is still in flight, wait for promiseQueue → + * repeat. This handles the multi-round delivery pattern where each hook + * payload delivery cycle appends new async work to the promiseQueue. What + * counts as in flight is {@link isDeliveryIdle}. + * * The initial `setTimeout(0)` macrotask is load-bearing and must NOT be * downgraded to a microtask (`queueMicrotask`/`Promise.resolve().then`). * `pendingDeliveries` only guards the host-side hydration window; between a @@ -629,7 +645,7 @@ export function scheduleWhenIdle( fn: () => void ): void { const check = () => { - if (ctx.pendingDeliveries > 0 || hasParkedCommittedDelivery(ctx)) { + if (!isDeliveryIdle(ctx)) { // A delivery is still hydrating, or is committed but parked behind its // deferral (whose resolve runs on a detached timer, not this queue). // Either way: let the queue drain, then re-check a timer tick later. diff --git a/packages/core/src/unconsumed-check-delivery-idle.test.ts b/packages/core/src/unconsumed-check-delivery-idle.test.ts new file mode 100644 index 0000000000..8a13f74dd8 --- /dev/null +++ b/packages/core/src/unconsumed-check-delivery-idle.test.ts @@ -0,0 +1,149 @@ +import { withResolvers } from '@workflow/utils'; +import type { Event } from '@workflow/world'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { EventConsumerResult, EventsConsumer } from './events-consumer.js'; +import type { WorkflowOrchestratorContext } from './private.js'; +import { isDeliveryIdle, registerDeliveryBarrier } from './private.js'; + +/** + * The events consumer walks the log synchronously; the resolutions that walk + * triggers do not resolve synchronously. A step result hydrates in the host, + * resolves from a detached continuation behind `awaitEarlierDeliveries`, and + * only then does VM code run far enough to `subscribe()` the consumer for the + * next event. So the walk routinely sits on an ordered event (`step_created`, + * `wait_created`) that nobody has claimed yet while the workflow is mid-flight + * on its way to claiming it. + * + * The unconsumed-event check used to resolve that by waiting a fixed + * `DEFERRED_CHECK_DELAY_MS` after the promise queue drained, which is a bet + * that every delivery lands inside the window. Replaying a batch of N parallel + * step results loses it: the queue drains with N-1 of them still on the + * detached path, and the check declares `ReplayDivergenceError` against a log + * the very same replay goes on to reproduce exactly. Measured on the event-log + * race repro against world-postgres, on identical event logs: 0 of 114 runs + * corrupted with a 100ms window, 34 of 42 with a 10ms one. + * + * `hasParkedCommittedDelivery` in private.ts already documents this hazard for + * the suspension path (vercel/workflow#3183). These tests pin the same guard + * on the divergence path, using the production predicate rather than a mock: + * a real armed delivery barrier must hold the check off however long it takes, + * and the check must still fire for an event no delivery is waiting on. + */ + +function createEvent(overrides: Partial = {}): Event { + return { + id: 'event-1', + workflow_run_id: 'run-1', + event_type: 'step_created', + event_data: {}, + sequence_number: 1, + created_at: new Date(), + ...overrides, + } as unknown as Event; +} + +/** + * The slice of the orchestrator context that `isDeliveryIdle` and + * `registerDeliveryBarrier` read. Everything else a replay carries is + * irrelevant to whether a delivery is in flight. + */ +function createDeliveryContext(): WorkflowOrchestratorContext { + const promiseQueueHolder = { current: Promise.resolve() }; + return { + pendingDeliveries: 0, + pendingDeliveryBarriers: new Map(), + get promiseQueue() { + return promiseQueueHolder.current; + }, + set promiseQueue(value: Promise) { + promiseQueueHolder.current = value; + }, + } as unknown as WorkflowOrchestratorContext; +} + +afterEach(() => { + vi.unstubAllEnvs(); +}); + +describe('unconsumed-event check against in-flight deliveries', () => { + it('does not declare divergence while a step delivery is outstanding', async () => { + // Far shorter than the delivery below, so the run survives only if the + // check waits for the delivery rather than for the clock. + vi.stubEnv('WORKFLOW_DEFERRED_CHECK_DELAY_MS', '10'); + + const ctx = createDeliveryContext(); + const event = createEvent(); + const onUnconsumedEvent = vi.fn(); + const consumer = new EventsConsumer([event], { + onUnconsumedEvent, + getPromiseQueue: () => ctx.promiseQueue, + isDeliveryIdle: () => isDeliveryIdle(ctx), + }); + + // A step result committed to being delivered, sitting on the detached + // continuation that `pendingDeliveries` deliberately does not cover. + const barrier = registerDeliveryBarrier(ctx, 0, 'step'); + + consumer.subscribe( + vi.fn().mockReturnValue(EventConsumerResult.NotConsumed) + ); + + await new Promise((resolve) => setTimeout(resolve, 250)); + expect(onUnconsumedEvent).not.toHaveBeenCalled(); + + // The delivery lands and the workflow reaches the call this event records. + barrier.markDelivered(); + consumer.subscribe(vi.fn().mockReturnValue(EventConsumerResult.Finished)); + + await vi.waitFor(() => { + expect(consumer.eventIndex).toBe(1); + }); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(onUnconsumedEvent).not.toHaveBeenCalled(); + }); + + it('does not declare divergence while a payload is hydrating', async () => { + vi.stubEnv('WORKFLOW_DEFERRED_CHECK_DELAY_MS', '10'); + + const ctx = createDeliveryContext(); + const event = createEvent(); + const onUnconsumedEvent = vi.fn(); + const consumer = new EventsConsumer([event], { + onUnconsumedEvent, + getPromiseQueue: () => ctx.promiseQueue, + isDeliveryIdle: () => isDeliveryIdle(ctx), + }); + + // The other in-flight window: hydration inside a serial queue slot. + ctx.pendingDeliveries++; + + consumer.subscribe( + vi.fn().mockReturnValue(EventConsumerResult.NotConsumed) + ); + + await new Promise((resolve) => setTimeout(resolve, 250)); + expect(onUnconsumedEvent).not.toHaveBeenCalled(); + + ctx.pendingDeliveries--; + await vi.waitFor(() => { + expect(onUnconsumedEvent).toHaveBeenCalledWith(event); + }); + }); + + it('still declares divergence for an event no delivery is waiting on', async () => { + const ctx = createDeliveryContext(); + const event = createEvent(); + const unconsumedReceived = withResolvers(); + const consumer = new EventsConsumer([event], { + onUnconsumedEvent: unconsumedReceived.resolve, + getPromiseQueue: () => ctx.promiseQueue, + isDeliveryIdle: () => isDeliveryIdle(ctx), + }); + + consumer.subscribe( + vi.fn().mockReturnValue(EventConsumerResult.NotConsumed) + ); + + expect(await unconsumedReceived.promise).toEqual(event); + }); +}); diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts index e3499816fb..a6e1b0672b 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -24,6 +24,7 @@ import type { QueueItem } from './global.js'; import { ENOTSUP, WorkflowSuspension } from './global.js'; import { runtimeLogger } from './logger.js'; import type { WorkflowOrchestratorContext } from './private.js'; +import { isDeliveryIdle } from './private.js'; import { ReplayPayloadCache } from './replay-payload-cache.js'; import { getPortLazy } from './runtime/get-port-lazy.js'; import { runIdCreatedAt } from './runtime/run-id-time.js'; @@ -387,6 +388,11 @@ async function createWorkflowSession({ // by step/hook/sleep callbacks as events are processed. const promiseQueueHolder = { current: Promise.resolve() }; + // Same reason as the queue holder: the consumer is built before the context + // whose delivery state it has to read. Idle until the context exists, which + // is before any delivery can be registered against it. + const deliveryIdleHolder = { current: (): boolean => true }; + const eventsConsumer = new EventsConsumer(events, { onConsumedEvent: (event) => { updateTimestamp(+event.createdAt); @@ -400,6 +406,7 @@ async function createWorkflowSession({ ); }, getPromiseQueue: () => promiseQueueHolder.current, + isDeliveryIdle: () => deliveryIdleHolder.current(), }); const workflowContext: WorkflowOrchestratorContext = { @@ -426,6 +433,8 @@ async function createWorkflowSession({ replayPayloadCache, }; + deliveryIdleHolder.current = () => isDeliveryIdle(workflowContext); + // Consume run lifecycle events - these are structural events that don't // need special handling in the workflow, but must be consumed to advance // past them in the event log From a8bff0285530a9be71e288ab330099f9bc82a515 Mon Sep 17 00:00:00 2001 From: Nathan Rajlich Date: Tue, 11 Aug 2026 00:57:17 -0700 Subject: [PATCH 8/8] core: quickjs engine divergence-detection and write-fencing parity --- .changeset/quickjs-divergence-parity.md | 5 + .../src/runtime/quickjs-divergence.test.ts | 353 ++++++++++++++++++ .../core/src/runtime/quickjs-divergence.ts | 244 ++++++++++++ .../quickjs-entrypoint.fencing.test.ts | 235 ++++++++++++ .../core/src/runtime/quickjs-entrypoint.ts | 321 +++++++++++----- .../core/src/runtime/quickjs-runtime.test.ts | 177 +++++++++ packages/core/src/runtime/quickjs-runtime.ts | 80 ++++ 7 files changed, 1324 insertions(+), 91 deletions(-) create mode 100644 .changeset/quickjs-divergence-parity.md create mode 100644 packages/core/src/runtime/quickjs-divergence.test.ts create mode 100644 packages/core/src/runtime/quickjs-divergence.ts create mode 100644 packages/core/src/runtime/quickjs-entrypoint.fencing.test.ts diff --git a/.changeset/quickjs-divergence-parity.md b/.changeset/quickjs-divergence-parity.md new file mode 100644 index 0000000000..18f8eadc4a --- /dev/null +++ b/.changeset/quickjs-divergence-parity.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +QuickJS engine: divergence-detection and write-fencing parity with the node:vm engine. Replays now arbitrate the event log at each fixed point (orphaned events, stepName/token/resumeAt mismatches) and escalate `ReplayDivergenceError` through the existing recovery machinery instead of silently delivering wrong payloads or surfacing corruption as `USER_ERROR`; all replay-context event writes now carry the optimistic-concurrency precondition snapshot (closing the documented KNOWN GAP), with `run_failed` left unfenced to match the node engine. diff --git a/packages/core/src/runtime/quickjs-divergence.test.ts b/packages/core/src/runtime/quickjs-divergence.test.ts new file mode 100644 index 0000000000..20b6ad7c79 --- /dev/null +++ b/packages/core/src/runtime/quickjs-divergence.test.ts @@ -0,0 +1,353 @@ +import type { Event } from '@workflow/world'; +import { describe, expect, it } from 'vitest'; +import { + findReplayDivergence, + type VmKnownOp, + type VmReplayView, +} from './quickjs-divergence.js'; + +/** + * Unit coverage for the QuickJS engine's replay-divergence arbitration — + * the pure core behind the fixed-point sweep in quickjs-runtime.ts. The + * semantics mirror the node:vm engine's EventsConsumer checks: every + * non-structural event must be claimed by an operation the replay drew, + * with matching identity fields. + */ + +function makeEvent(overrides: Partial>): Event { + return { + eventId: 'evnt_01TEST', + runId: 'wrun_test', + eventType: 'step_created', + correlationId: 'step_01AAAA', + eventData: {}, + createdAt: new Date('2025-01-01T00:00:00Z'), + ...overrides, + } as unknown as Event; +} + +function view(ops: VmKnownOp[], extra?: Partial): VmReplayView { + return { ops, ...extra }; +} + +describe('findReplayDivergence', () => { + describe('structural events (never require a claim)', () => { + it.each([ + ['run_created', {}], + ['run_started', {}], + ['run_completed', {}], + ['run_failed', {}], + ])('skips %s', (eventType) => { + const events = [makeEvent({ eventType, correlationId: 'wrun_test' })]; + expect(findReplayDivergence(events, view([]))).toBeNull(); + }); + + it('skips events without a correlationId', () => { + const events = [ + makeEvent({ eventType: 'step_created', correlationId: undefined }), + ]; + expect(findReplayDivergence(events, view([]))).toBeNull(); + }); + + it('skips attr_set events written by a step', () => { + const events = [ + makeEvent({ + eventType: 'attr_set', + correlationId: 'attr_01AAAA', + eventData: { changes: {}, writer: { type: 'step' } }, + }), + ]; + expect(findReplayDivergence(events, view([]))).toBeNull(); + }); + + it('skips event types outside the tracked families', () => { + const events = [ + makeEvent({ eventType: 'some_future_event', correlationId: 'x_1' }), + ]; + expect(findReplayDivergence(events, view([]))).toBeNull(); + }); + }); + + describe('orphaned events', () => { + it('reports a step event for a correlation id the replay never drew', () => { + const events = [ + makeEvent({ + eventType: 'step_created', + correlationId: 'step_01ORPHAN', + eventId: 'evnt_ORPHAN', + eventData: { stepName: 'step//test//other' }, + }), + ]; + const err = findReplayDivergence( + events, + view([{ correlationId: 'step_01AAAA', type: 'step' }]) + ); + expect(err).not.toBeNull(); + expect(err?.name).toBe('ReplayDivergenceError'); + expect(err?.message).toContain('Replay could not consume event'); + expect(err?.message).toContain('step_01ORPHAN'); + expect(err?.eventId).toBe('evnt_ORPHAN'); + }); + + it('reports an attr_set written by the workflow for an unknown id', () => { + const events = [ + makeEvent({ + eventType: 'attr_set', + correlationId: 'attr_01ORPHAN', + eventData: { changes: {}, writer: { type: 'workflow' } }, + }), + ]; + expect(findReplayDivergence(events, view([]))).not.toBeNull(); + }); + + it('accepts a hook event whose id is only known to the hook machinery', () => { + const events = [ + makeEvent({ + eventType: 'hook_received', + correlationId: 'hook_01AAAA', + eventData: {}, + }), + ]; + expect( + findReplayDivergence(events, view([], { hookCids: ['hook_01AAAA'] })) + ).toBeNull(); + expect( + findReplayDivergence(events, view([], { abortCids: ['hook_01AAAA'] })) + ).toBeNull(); + }); + + it('returns the FIRST divergence in log order', () => { + const events = [ + makeEvent({ + eventType: 'wait_created', + correlationId: 'wait_01FIRST', + eventId: 'evnt_FIRST', + }), + makeEvent({ + eventType: 'step_created', + correlationId: 'step_01SECOND', + eventId: 'evnt_SECOND', + }), + ]; + const err = findReplayDivergence(events, view([])); + expect(err?.eventId).toBe('evnt_FIRST'); + }); + }); + + describe('family mismatches', () => { + it('reports a step event for an id the replay drew as a wait', () => { + const events = [ + makeEvent({ + eventType: 'step_completed', + correlationId: 'wait_01AAAA', + eventData: { result: 1 }, + }), + ]; + const err = findReplayDivergence( + events, + view([{ correlationId: 'wait_01AAAA', type: 'wait' }]) + ); + expect(err?.message).toContain('does not match'); + }); + + it('accepts hook_disposed claimed by a hook_dispose op sharing the hook id', () => { + const events = [ + makeEvent({ + eventType: 'hook_disposed', + correlationId: 'hook_01AAAA', + }), + ]; + expect( + findReplayDivergence( + events, + view([ + { correlationId: 'hook_01AAAA', type: 'hook', token: 't' }, + { correlationId: 'hook_01AAAA', type: 'hook_dispose' }, + ]) + ) + ).toBeNull(); + }); + }); + + describe('identity mismatches', () => { + const stepOp: VmKnownOp = { + correlationId: 'step_01AAAA', + type: 'step', + stepId: 'step//test//releaseStep', + }; + + it('reports a step event recorded for a different step function', () => { + const events = [ + makeEvent({ + eventType: 'step_created', + correlationId: 'step_01AAAA', + eventData: { stepName: 'step//test//recoverStep' }, + }), + ]; + const err = findReplayDivergence(events, view([stepOp])); + expect(err?.message).toContain('belongs to "step//test//recoverStep"'); + expect(err?.message).toContain( + 'current step consumer is "step//test//releaseStep"' + ); + }); + + it('accepts a step event with the matching stepName', () => { + const events = [ + makeEvent({ + eventType: 'step_created', + correlationId: 'step_01AAAA', + eventData: { stepName: 'step//test//releaseStep' }, + }), + ]; + expect(findReplayDivergence(events, view([stepOp]))).toBeNull(); + }); + + it('accepts a step event that does not carry a stepName', () => { + const events = [ + makeEvent({ + eventType: 'step_completed', + correlationId: 'step_01AAAA', + eventData: { result: 42 }, + }), + ]; + expect(findReplayDivergence(events, view([stepOp]))).toBeNull(); + }); + + it('reports a hook event recorded under a different token', () => { + const events = [ + makeEvent({ + eventType: 'hook_created', + correlationId: 'hook_01AAAA', + eventData: { token: 'other-token' }, + }), + ]; + const err = findReplayDivergence( + events, + view([{ correlationId: 'hook_01AAAA', type: 'hook', token: 'mine' }]) + ); + expect(err?.message).toContain('belongs to token "other-token"'); + }); + + it('accepts a hook event with the matching token', () => { + const events = [ + makeEvent({ + eventType: 'hook_created', + correlationId: 'hook_01AAAA', + eventData: { token: 'mine' }, + }), + ]; + expect( + findReplayDivergence( + events, + view([{ correlationId: 'hook_01AAAA', type: 'hook', token: 'mine' }]) + ) + ).toBeNull(); + }); + + it('reports a wait_completed with a different resumeAt', () => { + const events = [ + makeEvent({ + eventType: 'wait_completed', + correlationId: 'wait_01AAAA', + eventData: { resumeAt: '2025-01-01T00:01:00.000Z' }, + }), + ]; + const err = findReplayDivergence( + events, + view([ + { + correlationId: 'wait_01AAAA', + type: 'wait', + resumeAt: '2025-01-01T00:02:00.000Z', + }, + ]) + ); + expect(err?.message).toContain('resumeAt'); + }); + + it('accepts a wait_completed with the matching resumeAt (and one without eventData)', () => { + const ops = [ + { + correlationId: 'wait_01AAAA', + type: 'wait', + resumeAt: '2025-01-01T00:01:00.000Z', + }, + ]; + expect( + findReplayDivergence( + [ + makeEvent({ + eventType: 'wait_completed', + correlationId: 'wait_01AAAA', + eventData: { resumeAt: '2025-01-01T00:01:00.000Z' }, + }), + ], + view(ops) + ) + ).toBeNull(); + // The entrypoint's elapsed-wait pass writes wait_completed without + // eventData — nothing to validate, must be accepted. + expect( + findReplayDivergence( + [ + makeEvent({ + eventType: 'wait_completed', + correlationId: 'wait_01AAAA', + eventData: undefined, + }), + ], + view(ops) + ) + ).toBeNull(); + }); + }); + + it('accepts a fully reproduced log', () => { + const events = [ + makeEvent({ eventType: 'run_created', correlationId: 'wrun_test' }), + makeEvent({ eventType: 'run_started', correlationId: 'wrun_test' }), + makeEvent({ + eventType: 'hook_created', + correlationId: 'hook_01AAAA', + eventData: { token: 't1' }, + }), + makeEvent({ + eventType: 'step_created', + correlationId: 'step_01AAAA', + eventData: { stepName: 'step//test//add' }, + }), + makeEvent({ + eventType: 'step_completed', + correlationId: 'step_01AAAA', + eventData: { result: 17 }, + }), + makeEvent({ + eventType: 'wait_created', + correlationId: 'wait_01AAAA', + }), + makeEvent({ + eventType: 'hook_received', + correlationId: 'hook_01AAAA', + eventData: { payload: 'x' }, + }), + ]; + expect( + findReplayDivergence( + events, + view([ + { correlationId: 'hook_01AAAA', type: 'hook', token: 't1' }, + { + correlationId: 'step_01AAAA', + type: 'step', + stepId: 'step//test//add', + }, + { + correlationId: 'wait_01AAAA', + type: 'wait', + resumeAt: '2025-01-01T00:05:00.000Z', + }, + ]) + ) + ).toBeNull(); + }); +}); diff --git a/packages/core/src/runtime/quickjs-divergence.ts b/packages/core/src/runtime/quickjs-divergence.ts new file mode 100644 index 0000000000..a459d4f500 --- /dev/null +++ b/packages/core/src/runtime/quickjs-divergence.ts @@ -0,0 +1,244 @@ +import { ReplayDivergenceError } from '@workflow/errors'; +import type { Event } from '@workflow/world'; + +/** + * Replay-divergence arbitration for the QuickJS engine. + * + * The node:vm engine detects a corrupted / diverged event log through its + * `EventsConsumer`: every event must be *claimed* by a consumer registered by + * the replaying workflow code, consumers validate identity fields + * (`stepName`, hook `token`, wait `resumeAt`), and an event nobody claims is + * escalated as `ReplayDivergenceError` — which `runtime.ts` turns into + * bounded recovery replays and, past the budget, a terminal + * `CorruptedEventLogError`. + * + * The QuickJS engine has no consumer registry: the host feeds events into VM + * heap structures keyed by correlation id. Before this module existed, a log + * the replay did not reproduce was absorbed silently — a `step_completed` + * for a different step's ordinal resolved the wrong call with the wrong + * payload (no `stepName` check), and events for correlation ids the replay + * never drew sat in dead buffers while `run_completed` was written over the + * unreproduced log. The same corruption class the node engine reports as + * `CORRUPTED_EVENT_LOG` therefore surfaced on QuickJS as `USER_ERROR` + * (e.g. a self-`HookConflictError`) or as a silently wrong completion. + * + * `findReplayDivergence` closes that gap. It runs against the replay's + * *fixed point* — the moment the VM's job queue is fully drained and no more + * progress is possible — which is a stronger position than the node engine + * ever gets to check from: there is no in-flight cross-realm microtask work + * to wait out, so no grace window or delivery-idle heuristic is needed. The + * caller dumps the VM's known operations (every correlation id the replay + * drew, with identity fields) and this function arbitrates the observed + * event log against them. + */ + +/** One operation the replaying VM has drawn a correlation id for. */ +export interface VmKnownOp { + correlationId: string; + /** `step` | `wait` | `hook` | `hook_dispose` | `attribute` (open set). */ + type: string; + /** Step operations: the full step id (`step////`). */ + stepId?: string; + /** Hook operations: the user-supplied (or system-derived) token. */ + token?: string; + /** Wait operations: the ISO resume timestamp the VM computed. */ + resumeAt?: string; +} + +/** + * The replay's view of its own draws at a fixed point, dumped from the VM. + */ +export interface VmReplayView { + ops: VmKnownOp[]; + /** Correlation ids registered in `globalThis.__hooks` (hook machinery). */ + hookCids?: string[]; + /** Correlation ids registered in `globalThis.__abortSignals`. */ + abortCids?: string[]; +} + +/** The operation family an event type must be claimed by. */ +function expectedFamilies(eventType: string): string[] | null { + if (eventType.startsWith('step_')) return ['step']; + if (eventType.startsWith('wait_')) return ['wait']; + if (eventType.startsWith('hook_')) return ['hook', 'hook_dispose']; + if (eventType === 'attr_set') return ['attribute']; + return null; +} + +function eventDataOf(event: Event): Record | undefined { + return 'eventData' in event && event.eventData + ? (event.eventData as Record) + : undefined; +} + +/** + * Events that need no claim from workflow code (parity with the node + * engine's structural lifecycle consumer in workflow.ts): run lifecycle + * events, attribute writes not performed by the workflow body, and anything + * without a correlation id or outside the families the engines track. + */ +function isStructural(event: Event): boolean { + if (!event.correlationId) return true; + if (event.eventType.startsWith('run_')) return true; + if (event.eventType === 'attr_set') { + const writer = eventDataOf(event)?.writer as { type?: string } | undefined; + if (writer?.type !== 'workflow') return true; + } + return expectedFamilies(event.eventType) === null; +} + +/** + * Arbitrate the observed event log against the replay's fixed-point view. + * Returns the first divergence in log order, or `null` when the replay + * reproduced the log. Divergences, in the order they are checked per event: + * + * 1. **Orphaned event** — the log contains a correlation id this replay + * never drew. Mirrors the node engine's unconsumed-event check + * (`onUnconsumedEvent` → workflow.ts). + * 2. **Family mismatch** — the correlation id exists but belongs to a + * different kind of operation (a `step_*` event for a cid the replay + * drew as a wait, etc.). Mirrors the node consumers' unexpected-event + * checks. + * 3. **Identity mismatch** — the correlation id and family match but the + * recorded identity differs from what this replay derived: a step + * event's `stepName` (step.ts), a hook event's `token` (hook.ts), or a + * `wait_completed`'s `resumeAt` (sleep.ts). + */ +export function findReplayDivergence( + events: readonly Event[], + view: VmReplayView +): ReplayDivergenceError | null { + const opsByCid = new Map(); + for (const op of view.ops) { + const list = opsByCid.get(op.correlationId); + if (list) { + list.push(op); + } else { + opsByCid.set(op.correlationId, [op]); + } + } + const auxCids = new Set([ + ...(view.hookCids ?? []), + ...(view.abortCids ?? []), + ]); + + for (const event of events) { + if (isStructural(event)) continue; + const divergence = arbitrateEvent(event, opsByCid, auxCids); + if (divergence) return divergence; + } + + return null; +} + +function arbitrateEvent( + event: Event, + opsByCid: Map, + auxCids: Set +): ReplayDivergenceError | null { + const cid = event.correlationId as string; + const families = expectedFamilies(event.eventType) as string[]; + const ops = opsByCid.get(cid); + + if (ops === undefined || ops.length === 0) { + // Hook machinery can know a cid the pending list does not (defensive; + // the bootstrap pushes a pending op for every draw today). + if (families.includes('hook') && auxCids.has(cid)) return null; + return new ReplayDivergenceError( + `Replay could not consume event: eventType=${event.eventType}, correlationId=${cid}, eventId=${event.eventId}.`, + { eventId: event.eventId } + ); + } + + const familyOps = ops.filter((op) => families.includes(op.type)); + if (familyOps.length === 0) { + return new ReplayDivergenceError( + `Replay divergence: event ${event.eventType} for ${cid} does not match the "${ops[0].type}" operation the replay drew for that correlation id`, + { eventId: event.eventId } + ); + } + + return findIdentityMismatch(event, families, familyOps); +} + +/** + * Identity validation for an event whose correlation id and family both + * matched a VM operation: the recorded identity fields must equal what this + * replay derived. Mirrors the node consumers' checks in step.ts, hook.ts + * and sleep.ts. + */ +function findIdentityMismatch( + event: Event, + families: string[], + familyOps: VmKnownOp[] +): ReplayDivergenceError | null { + if (families.includes('step')) { + return findStepNameMismatch(event, familyOps[0]); + } + if (families.includes('hook')) { + return findHookTokenMismatch(event, familyOps); + } + if (event.eventType === 'wait_completed') { + return findResumeAtMismatch(event, familyOps[0]); + } + return null; +} + +function findStepNameMismatch( + event: Event, + op: VmKnownOp +): ReplayDivergenceError | null { + const eventStepName = eventDataOf(event)?.stepName; + if ( + typeof eventStepName === 'string' && + typeof op.stepId === 'string' && + eventStepName !== op.stepId + ) { + return new ReplayDivergenceError( + `Replay divergence: step event ${event.eventType} for ${event.correlationId} belongs to "${eventStepName}", but the current step consumer is "${op.stepId}"`, + { eventId: event.eventId } + ); + } + return null; +} + +function findHookTokenMismatch( + event: Event, + familyOps: VmKnownOp[] +): ReplayDivergenceError | null { + const eventToken = eventDataOf(event)?.token; + const op = familyOps.find((o) => typeof o.token === 'string'); + if ( + typeof eventToken === 'string' && + op !== undefined && + eventToken !== op.token + ) { + return new ReplayDivergenceError( + `Replay divergence: hook event ${event.eventType} for ${event.correlationId} belongs to token "${eventToken}", but the current hook consumer expects "${op.token}"`, + { eventId: event.eventId } + ); + } + return null; +} + +function findResumeAtMismatch( + event: Event, + op: VmKnownOp +): ReplayDivergenceError | null { + const eventResumeAt = eventDataOf(event)?.resumeAt; + if (eventResumeAt === undefined || typeof op.resumeAt !== 'string') { + return null; + } + const eventMs = new Date(eventResumeAt as string | Date).getTime(); + const expectedMs = new Date(op.resumeAt).getTime(); + if (eventMs !== expectedMs) { + const eventForMessage = Number.isFinite(eventMs) + ? new Date(eventMs).toISOString() + : String(eventResumeAt); + return new ReplayDivergenceError( + `Replay divergence: wait_completed event for ${event.correlationId} has resumeAt "${eventForMessage}", but the current wait consumer expects "${new Date(op.resumeAt).toISOString()}"`, + { eventId: event.eventId } + ); + } + return null; +} diff --git a/packages/core/src/runtime/quickjs-entrypoint.fencing.test.ts b/packages/core/src/runtime/quickjs-entrypoint.fencing.test.ts new file mode 100644 index 0000000000..c7c4672db2 --- /dev/null +++ b/packages/core/src/runtime/quickjs-entrypoint.fencing.test.ts @@ -0,0 +1,235 @@ +/** + * Pins the QuickJS engine's optimistic-concurrency precondition guard: + * every replay-context event write must carry the view snapshot + * (`stateUpdatedAt` / `stateEventCount`) describing the event log the + * invocation derived the write from, so a supporting World can reject a + * stale writer with 412 — parity with the node:vm engine's `createGuarded` + * (suspension-handler.ts) and guarded `run_completed`. The terminal + * `run_failed` is deliberately unfenced (same asymmetry as the node + * engine: a failing run must be able to terminate from a stale view). + * + * The QuickJS VM itself is mocked — the guard lives entirely in the + * entrypoint's write paths. + */ +import { + type CreateEventRequest, + type Event, + SPEC_VERSION_CURRENT, + type WorkflowRun, + type World, +} from '@workflow/world'; +import { monotonicFactory } from 'ulid'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { dehydrateStepReturnValue } from '../serialization.js'; +import { latestEventStateUpdatedAt } from './helpers.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 ulid = monotonicFactory(); + +function makeEvent( + eventType: string, + overrides: Partial> = {} +): Event { + return { + eventId: `evnt_${ulid()}`, + runId: 'wrun_quickjs_fencing', + eventType, + eventData: {}, + createdAt: new Date('2026-05-19T12:00:00.000Z'), + ...overrides, + } as unknown as Event; +} + +async function runScenario(options: { + events: Event[]; + vmResult: Record; +}) { + const runId = 'wrun_quickjs_fencing'; + const startedAt = new Date('2026-05-19T12:00:00.000Z'); + const workflowRun: WorkflowRun = { + runId, + workflowName: 'workflow', + status: 'running', + input: [], + deploymentId: 'dpl_quickjs_fencing', + specVersion: SPEC_VERSION_CURRENT, + startedAt, + createdAt: startedAt, + updatedAt: startedAt, + }; + + const created: { + request: CreateEventRequest; + params: Record | undefined; + }[] = []; + let listCallCount = 0; + const listEvents = vi.fn(async () => { + listCallCount++; + if (listCallCount === 1) { + return { + data: [...options.events], + cursor: options.events.at(-1)?.eventId ?? null, + hasMore: false, + }; + } + return { data: [], cursor: null, hasMore: false }; + }); + const createEvent = vi.fn( + async ( + _runId: string, + request: CreateEventRequest, + params?: Record + ) => { + created.push({ request, params }); + return { event: { ...request, runId, eventId: `evnt_${ulid()}` } }; + } + ); + + setWorld({ + specVersion: SPEC_VERSION_CURRENT, + capabilities: { preconditionGuard: true }, + events: { list: listEvents, create: createEvent }, + runs: { get: vi.fn(async () => workflowRun) }, + queue: vi.fn().mockResolvedValue({ messageId: 'msg_quickjs' }), + getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), + } as unknown as World); + + startQuickJSWorkflow.mockResolvedValue({ + result: options.vmResult, + continueWithEvents: vi.fn(), + dispose: vi.fn(), + }); + + const { runWorkflowWithQuickJS } = await import('./quickjs-entrypoint.js'); + await runWorkflowWithQuickJS({ + workflowCode: '// mocked VM', + workflowName: 'workflow', + workflowRun, + preloadedEvents: options.events, + preloadedEventsComplete: true, + }); + + return { created }; +} + +afterEach(() => { + vi.clearAllMocks(); + vi.unstubAllEnvs(); +}); + +describe('QuickJS entrypoint precondition guard', () => { + const baseLog = () => [ + makeEvent('run_created', { eventData: { input: undefined } }), + makeEvent('run_started'), + ]; + + it('fences run_completed with the view snapshot', async () => { + const events = baseLog(); + const { created } = await runScenario({ + events, + vmResult: { + completed: { + result: await dehydrateStepReturnValue( + 'done', + 'wrun_quickjs_fencing', + undefined + ), + }, + }, + }); + + const runCompleted = created.find( + (c) => c.request.eventType === 'run_completed' + ); + expect(runCompleted).toBeDefined(); + expect(runCompleted?.params).toMatchObject({ + stateUpdatedAt: latestEventStateUpdatedAt(events), + stateEventCount: events.length, + }); + }); + + it('fences suspension writes (hook_created, wait_created) with the view snapshot', async () => { + const events = baseLog(); + const { created } = await runScenario({ + events, + vmResult: { + suspended: { + pendingOperations: [ + { + type: 'hook', + correlationId: 'hook_01FENCE', + token: 'fence-token', + isWebhook: false, + hasCreatedEvent: false, + }, + { + type: 'wait', + correlationId: 'wait_01FENCE', + // Far future so the loop schedules a continuation instead of + // completing the wait. + resumeAt: '2027-01-01T00:00:00.000Z', + hasCreatedEvent: false, + }, + ], + }, + }, + }); + + const expected = { + stateUpdatedAt: latestEventStateUpdatedAt(events), + stateEventCount: events.length, + }; + const hookCreated = created.find( + (c) => c.request.eventType === 'hook_created' + ); + const waitCreated = created.find( + (c) => c.request.eventType === 'wait_created' + ); + expect(hookCreated?.params).toMatchObject(expected); + expect(waitCreated?.params).toMatchObject(expected); + }); + + it('leaves run_failed unfenced (terminal-failure asymmetry)', async () => { + const { created } = await runScenario({ + events: baseLog(), + vmResult: { + failed: { message: 'boom', name: 'Error' }, + }, + }); + + const runFailed = created.find((c) => c.request.eventType === 'run_failed'); + expect(runFailed).toBeDefined(); + expect(runFailed?.params).toBeUndefined(); + }); + + it('sends no snapshot when the guard is disabled', async () => { + vi.stubEnv('WORKFLOW_PRECONDITION_GUARD', '0'); + const { created } = await runScenario({ + events: baseLog(), + vmResult: { + completed: { + result: await dehydrateStepReturnValue( + 'done', + 'wrun_quickjs_fencing', + undefined + ), + }, + }, + }); + + const runCompleted = created.find( + (c) => c.request.eventType === 'run_completed' + ); + expect(runCompleted?.params ?? {}).not.toHaveProperty('stateUpdatedAt'); + }); +}); diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index f725cf9ee3..cf4ea07e36 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -49,7 +49,13 @@ import { getMaxInlineSteps, } from './constants.js'; import { getPortLazy } from './get-port-lazy.js'; -import { getWorkflowQueueName, queueMessage } from './helpers.js'; +import { + getWorkflowQueueName, + isPreconditionGuardEnabled, + latestEventStateUpdatedAt, + type PreconditionSnapshotParams, + queueMessage, +} from './helpers.js'; import { BASELINE_BUNDLE_FILENAME, type PendingAttribute, @@ -205,6 +211,15 @@ async function dispatchPendingOps(params: { * queues. */ nextTraceCarrier: () => Promise>; + /** + * Optimistic-concurrency snapshot describing the event log this batch of + * writes was derived from (see `preconditionSnapshotParams`). Attached to + * every event create so a supporting World rejects writes from a stale + * view with 412 — the rejection propagates to runtime.ts, which restarts + * the replay over a corrected log. Mirrors the node:vm engine's + * `createGuarded` in suspension-handler.ts. + */ + precondition?: PreconditionSnapshotParams; wfdiag: (checkpoint: string, fields: Record) => void; }): Promise<{ createdAttributeEvent: boolean; @@ -218,6 +233,7 @@ async function dispatchPendingOps(params: { pendingOperations, namespace, nextTraceCarrier, + precondition, } = params; const skipStepCreation = params.skipStepCreation; const wfdiag = params.wfdiag; @@ -261,26 +277,30 @@ async function dispatchPendingOps(params: { typeof hook.metadata === 'undefined' ? undefined : await encryptSerializedData(hook.metadata, encryptionKey); - const result = await world.events.create(runId, { - eventType: 'hook_created', - specVersion: SPEC_VERSION_CURRENT, - correlationId: hook.correlationId, - eventData: { - token: hook.token, - tokenRetentionUntil: - hook.tokenRetentionUntil === undefined - ? undefined - : new Date(hook.tokenRetentionUntil), - metadata: encryptedMetadata, - // Always include isWebhook explicitly. Worlds default it to - // `true` when absent, which would break the public webhook - // endpoint's 404 guard for hooks created via createHook(). - isWebhook: hook.isWebhook, - // System hooks (AbortController) are exempt from user - // token namespace conflict checks. - ...(hook.isSystem ? { isSystem: true } : {}), - } as any, - }); + const result = await world.events.create( + runId, + { + eventType: 'hook_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: hook.correlationId, + eventData: { + token: hook.token, + tokenRetentionUntil: + hook.tokenRetentionUntil === undefined + ? undefined + : new Date(hook.tokenRetentionUntil), + metadata: encryptedMetadata, + // Always include isWebhook explicitly. Worlds default it to + // `true` when absent, which would break the public webhook + // endpoint's 404 guard for hooks created via createHook(). + isWebhook: hook.isWebhook, + // System hooks (AbortController) are exempt from user + // token namespace conflict checks. + ...(hook.isSystem ? { isSystem: true } : {}), + } as any, + }, + precondition + ); // If storage detected a real token conflict with another // workflow's hook, re-queue so the workflow handler can @@ -321,15 +341,19 @@ async function dispatchPendingOps(params: { )) as Uint8Array) : undefined; try { - await world.events.create(runId, { - eventType: 'hook_received', - specVersion: SPEC_VERSION_CURRENT, - correlationId: hook.correlationId, - eventData: { - token: hook.token, - payload: abortPayload, - } as any, - }); + await world.events.create( + runId, + { + eventType: 'hook_received', + specVersion: SPEC_VERSION_CURRENT, + correlationId: hook.correlationId, + eventData: { + token: hook.token, + payload: abortPayload, + } as any, + }, + precondition + ); } catch (err) { if (!EntityConflictError.is(err)) throw err; } @@ -363,11 +387,15 @@ async function dispatchPendingOps(params: { op: PendingHookDispose ): Promise => { try { - await world.events.create(runId, { - eventType: 'hook_disposed', - specVersion: SPEC_VERSION_CURRENT, - correlationId: op.correlationId, - }); + await world.events.create( + runId, + { + eventType: 'hook_disposed', + specVersion: SPEC_VERSION_CURRENT, + correlationId: op.correlationId, + }, + precondition + ); } catch (err) { if (EntityConflictError.is(err)) return; // Disposing a hook whose entity no longer (or never) exists is an @@ -443,15 +471,19 @@ async function dispatchPendingOps(params: { // on the host side — matching what // `dehydrateStepArguments` does in the node:vm engine. try { - await world.events.create(runId, { - eventType: 'step_created', - specVersion: SPEC_VERSION_CURRENT, - correlationId: step.correlationId, - eventData: { - stepName: step.stepId, - input: await encryptSerializedData(step.input, encryptionKey), + await world.events.create( + runId, + { + eventType: 'step_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: step.correlationId, + eventData: { + stepName: step.stepId, + input: await encryptSerializedData(step.input, encryptionKey), + }, }, - }); + precondition + ); } catch (err) { if (EntityConflictError.is(err)) return; throw err; @@ -467,18 +499,22 @@ async function dispatchPendingOps(params: { opsPromises.push( (async () => { try { - await world.events.create(runId, { - eventType: 'attr_set', - specVersion: SPEC_VERSION_CURRENT, - correlationId: attr.correlationId, - eventData: { - changes: attr.changes, - writer: { type: 'workflow' }, - ...(attr.allowReservedAttributes - ? { allowReservedAttributes: true } - : {}), - } as any, - }); + await world.events.create( + runId, + { + eventType: 'attr_set', + specVersion: SPEC_VERSION_CURRENT, + correlationId: attr.correlationId, + eventData: { + changes: attr.changes, + writer: { type: 'workflow' }, + ...(attr.allowReservedAttributes + ? { allowReservedAttributes: true } + : {}), + } as any, + }, + precondition + ); createdAttributeEvent = true; } catch (err) { if (EntityConflictError.is(err)) { @@ -496,14 +532,18 @@ async function dispatchPendingOps(params: { opsPromises.push( (async () => { try { - await world.events.create(runId, { - eventType: 'wait_created', - specVersion: SPEC_VERSION_CURRENT, - correlationId: wait.correlationId, - eventData: { - resumeAt: new Date(wait.resumeAt), + await world.events.create( + runId, + { + eventType: 'wait_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: wait.correlationId, + eventData: { + resumeAt: new Date(wait.resumeAt), + }, }, - }); + precondition + ); } catch (err) { if (EntityConflictError.is(err)) return; throw err; @@ -525,15 +565,16 @@ async function dispatchPendingOps(params: { * This replaces the `node:vm` replay path (runWorkflow + EventsConsumer) * with a QuickJS VM invocation that performs the same full event replay. * - * KNOWN GAP — precondition guard: unlike the node:vm path, no event write - * in this file participates in the optimistic-concurrency precondition - * guard (`withPreconditionRetry` + `stateUpdatedAtForCreate`), which - * protects a writer holding a stale event-log snapshot from clobbering a - * concurrent one. The engine currently relies on per-(runId, - * correlationId) event uniqueness (EntityConflictError dedup) alone. This - * is a deliberate simplification while the engine is experimental — wiring - * the guard is tracked follow-up work; anyone adding new write paths here - * should not assume parity with the node engine on this axis. + * Precondition guard: every replay-context event write in this file carries + * the optimistic-concurrency snapshot (`preconditionSnapshotParams` + * semantics — see the view tracker in `runWorkflowWithQuickJS`), so a + * supporting World rejects writes derived from a stale event-log view with + * 412. The rejection propagates out of this entrypoint to the replay loop's + * catch in runtime.ts, which restarts the replay over a corrected log — + * the same recovery the node:vm engine uses. `run_failed` is deliberately + * unfenced, matching the node engine's asymmetry (a terminal failure must + * be recordable even from a stale view). When adding a new write path + * here, thread the snapshot through it. */ export async function runWorkflowWithQuickJS(params: { workflowCode: string; @@ -687,6 +728,7 @@ export async function runWorkflowWithQuickJS(params: { // preload (lazy hook fast path) is trusted the same way. let events: Event[]; let eventsFetchedPages = 0; + let initialFetchCursor: string | null = null; const usePreloaded = (preloadedEventsComplete === true && Array.isArray(preloadedEvents) && @@ -720,8 +762,60 @@ export async function runWorkflowWithQuickJS(params: { } events = allEvents; + initialFetchCursor = cursor; } + // ---- Optimistic-concurrency view tracker ---- + // The precondition snapshot attached to every replay-context write below + // describes the event-log view this invocation derived the write from: + // the maximum event-id ULID time over every event it has observed, the + // count of those events, and the listing cursor. Self-written events are + // observed either inline (when the create returns the event and it is + // spliced into the local log) or on read-back via fetchUnseenEvents — + // watermark and count always describe the same consistent set, which is + // the invariant the backend's marker/count comparison relies on. See + // `preconditionSnapshotParams` in helpers.ts for the field semantics. + // + // Also derives the open-hook state used to suppress optimistic inline + // starts (see the executeStep call in the inline loop): while a hook is + // open, an out-of-band hook_received can make this view stale at any + // moment, so a fenced claim must settle before the step body runs. + let viewMaxEvent: Event | undefined; + let viewEventCount = 0; + let viewCursor: string | null = initialFetchCursor; + const openHookCids = new Set(); + const observeView = (observed: Event[], cursor?: string | null): void => { + for (const e of observed) { + if (!e.eventId) continue; + viewEventCount++; + if (viewMaxEvent === undefined || e.eventId > viewMaxEvent.eventId) { + viewMaxEvent = e; + } + if (e.correlationId) { + if (e.eventType === 'hook_created') { + openHookCids.add(e.correlationId); + } else if (e.eventType === 'hook_disposed') { + openHookCids.delete(e.correlationId); + } + } + } + if (cursor) viewCursor = cursor; + }; + observeView(events); + const preconditionSnapshot = (): PreconditionSnapshotParams => { + if (!isPreconditionGuardEnabled() || viewMaxEvent === undefined) return {}; + const stateUpdatedAt = latestEventStateUpdatedAt([viewMaxEvent]); + if (stateUpdatedAt === undefined) return {}; + return { + stateUpdatedAt, + stateEventCount: viewEventCount, + ...(viewCursor ? { stateCursor: viewCursor } : {}), + }; + }; + const guardEnforced = + isPreconditionGuardEnabled() && + world.capabilities?.preconditionGuard === true; + // Event-limit guard: fail a runaway run once its log reaches the // server-supplied ceiling — same enforcement point as the node:vm // engine's replay loop. @@ -765,12 +859,21 @@ export async function runWorkflowWithQuickJS(params: { const resumeAt = eventData?.resumeAt; if (resumeAt && now >= new Date(resumeAt as string).getTime()) { try { - const result = await world.events.create(runId, { - eventType: 'wait_completed', - specVersion: SPEC_VERSION_CURRENT, - correlationId: event.correlationId, - }); - if (result.event) events.push(result.event); + const result = await world.events.create( + runId, + { + eventType: 'wait_completed', + specVersion: SPEC_VERSION_CURRENT, + correlationId: event.correlationId, + }, + preconditionSnapshot() + ); + if (result.event) { + events.push(result.event); + // Spliced into the local log ahead of the seen-set init, so a + // read-back never observes it — count it into the view here. + observeView([result.event]); + } } catch (err) { if (EntityConflictError.is(err)) continue; throw err; @@ -959,6 +1062,7 @@ export async function runWorkflowWithQuickJS(params: { hasMore = response.data.length > 0 && response.cursor != null; } observeEventsForOwnership(unseen); + observeView(unseen, cursor); return unseen; }; @@ -1021,6 +1125,7 @@ export async function runWorkflowWithQuickJS(params: { nextTraceCarrier, pendingOperations: opsToDispatch, skipStepCreation: inlineClaimCids, + precondition: preconditionSnapshot(), wfdiag, }); if ( @@ -1066,11 +1171,15 @@ export async function runWorkflowWithQuickJS(params: { waitCompletePromises.push( (async () => { try { - await world.events.create(runId, { - eventType: 'wait_completed', - specVersion: SPEC_VERSION_CURRENT, - correlationId: wait.correlationId, - }); + await world.events.create( + runId, + { + eventType: 'wait_completed', + specVersion: SPEC_VERSION_CURRENT, + correlationId: wait.correlationId, + }, + preconditionSnapshot() + ); } catch (err) { if (EntityConflictError.is(err)) return; throw err; @@ -1261,6 +1370,19 @@ export async function runWorkflowWithQuickJS(params: { // own, matching the node:vm engine, where a long sequential // workflow likewise runs step-by-step until the platform reclaims // the invocation and a redelivery resumes from the log. + // Guard the inline claims: the lazy step_started (which creates the + // step) carries the same view snapshot as the durable writes above, + // so a stale replay can never commit a step body's result — the + // node engine's `inlineClaimSnapshot`. While a hook is open, an + // out-of-band hook_received can stale this view at any moment; + // on guard-enforcing Worlds, await the claim before running the + // body so a 412-fenced step never executes user code (mirrors the + // node engine's suppressOptimisticStart). + const inlineClaimSnapshot = preconditionSnapshot(); + const suppressOptimisticStart = + guardEnforced && + (openHookCids.size > 0 || + pendingOperations.some((op) => op.type === 'hook')); budget.pause(); let outcomes: StepExecutionResult[]; try { @@ -1296,6 +1418,10 @@ export async function runWorkflowWithQuickJS(params: { // A lazy step is brand-new by construction — first // attempt. authoritativeAttempt: 1, + // Fence the claim against stale views — see + // inlineClaimSnapshot above. + preconditionSnapshot: inlineClaimSnapshot, + suppressOptimisticStart, }))() ) ) @@ -1405,6 +1531,7 @@ export async function runWorkflowWithQuickJS(params: { namespace, nextTraceCarrier, pendingOperations: result.completed.drainOperations, + precondition: preconditionSnapshot(), wfdiag, }); } catch (err) { @@ -1422,16 +1549,24 @@ export async function runWorkflowWithQuickJS(params: { // events have the same `encr`-prefixed payload shape that the node:vm // engine's `dehydrateWorkflowReturnValue` produces. try { - await world.events.create(runId, { - eventType: 'run_completed', - specVersion: SPEC_VERSION_CURRENT, - eventData: { - output: await encryptSerializedData( - result.completed.result, - encryptionKey - ), + await world.events.create( + runId, + { + eventType: 'run_completed', + specVersion: SPEC_VERSION_CURRENT, + eventData: { + output: await encryptSerializedData( + result.completed.result, + encryptionKey + ), + }, }, - }); + // Fenced: a completion derived from a stale view must not land — + // the 412 propagates to runtime.ts, which restarts the replay + // over the corrected log (same as the node engine's guarded + // run_completed). + preconditionSnapshot() + ); wfdiag('exit_completed', { result: 'run_completed_written' }); } catch (err) { if (EntityConflictError.is(err) || RunExpiredError.is(err)) { @@ -1645,6 +1780,7 @@ export async function runWorkflowWithQuickJS(params: { namespace, nextTraceCarrier, pendingOperations: result.failed.drainOperations, + precondition: preconditionSnapshot(), wfdiag, }); } catch (err) { @@ -1769,6 +1905,9 @@ export async function runWorkflowWithQuickJS(params: { } } try { + // Deliberately UNFENCED (no precondition snapshot), matching the + // node engine's asymmetry: a terminal run_failed must be recordable + // even from a stale view, or a failing run could never terminate. await world.events.create(runId, { eventType: 'run_failed', specVersion: SPEC_VERSION_CURRENT, diff --git a/packages/core/src/runtime/quickjs-runtime.test.ts b/packages/core/src/runtime/quickjs-runtime.test.ts index a5232d1e0f..36da372a60 100644 --- a/packages/core/src/runtime/quickjs-runtime.test.ts +++ b/packages/core/src/runtime/quickjs-runtime.test.ts @@ -5,6 +5,7 @@ import { __peekBaselineEntryForTests, BASELINE_BUNDLE_FILENAME, runQuickJSWorkflow, + startQuickJSWorkflow, } from './quickjs-runtime.js'; /** Helper to deserialize the format-prefixed result bytes */ @@ -1499,3 +1500,179 @@ describe('baseline snapshot startup optimization', () => { __clearBaselineSnapshotCacheForTests(); }); }); + +describe('replay-divergence arbitration', () => { + // Parity with the node:vm engine's EventsConsumer: a log the replay does + // not reproduce must escalate as ReplayDivergenceError (which runtime.ts + // turns into bounded recovery replays and, past the budget, a terminal + // CORRUPTED_EVENT_LOG) — never be absorbed into a wrong completion, + // a wrong-payload delivery, or a spurious user error. + + const stepWorkflow = ` + var add = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("step//test//add"); + async function workflow() { return await add(10, 7); } + workflow.workflowId = "workflow//test//workflow"; + globalThis.__private_workflows.set("workflow//test//workflow", workflow); + `; + + it('rejects a replay whose log contains a correlation id it never drew', async () => { + const run = makeRun(); + await expect( + runQuickJSWorkflow({ + workflowCode: stepWorkflow, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: [ + runCreatedEvent(run), + { + eventId: 'evnt_orphan', + runId: run.runId, + eventType: 'step_created' as const, + correlationId: 'step_01JUNKJUNKJUNKJUNKJUNKJUNK', + eventData: { stepName: 'step//test//other' }, + createdAt: new Date('2025-01-01T00:00:01Z'), + }, + ], + }) + ).rejects.toMatchObject({ + name: 'ReplayDivergenceError', + eventId: 'evnt_orphan', + }); + }); + + it('rejects a replay when a recorded step belongs to a different step function', async () => { + const run = makeRun(); + // Learn the deterministic correlation id the workflow draws. + const first = await runQuickJSWorkflow({ + workflowCode: stepWorkflow, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: [], + }); + const stepCid = first.suspended!.pendingOperations[0].correlationId; + + // Same ordinal, different step function — the racing-writer shape that + // used to resolve the wrong call with the wrong payload silently. + await expect( + runQuickJSWorkflow({ + workflowCode: stepWorkflow, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: [ + runCreatedEvent(run), + { + eventId: 'evnt_wrong_step', + runId: run.runId, + eventType: 'step_created' as const, + correlationId: stepCid, + eventData: { stepName: 'step//test//DIFFERENT' }, + createdAt: new Date('2025-01-01T00:00:01Z'), + }, + { + eventId: 'evnt_wrong_step_done', + runId: run.runId, + eventType: 'step_completed' as const, + correlationId: stepCid, + eventData: { result: 999 }, + createdAt: new Date('2025-01-01T00:00:02Z'), + }, + ], + }) + ).rejects.toMatchObject({ name: 'ReplayDivergenceError' }); + }); + + it('accepts a healthy replay of a fully reproduced log', async () => { + const run = makeRun(); + const first = await runQuickJSWorkflow({ + workflowCode: stepWorkflow, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: [], + }); + const stepCid = first.suspended!.pendingOperations[0].correlationId; + + const replayed = await runQuickJSWorkflow({ + workflowCode: stepWorkflow, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: [ + runCreatedEvent(run), + { + eventId: 'evnt_step_created', + runId: run.runId, + eventType: 'step_created' as const, + correlationId: stepCid, + eventData: { stepName: 'step//test//add' }, + createdAt: new Date('2025-01-01T00:00:01Z'), + }, + { + eventId: 'evnt_step_done', + runId: run.runId, + eventType: 'step_completed' as const, + correlationId: stepCid, + eventData: { result: 17 }, + createdAt: new Date('2025-01-01T00:00:02Z'), + }, + ], + }); + expect(replayed.completed).toBeDefined(); + expect(unwrapResult(replayed.completed!.result)).toBe(17); + }); + + it('records a genuine user failure instead of arbitrating the log', async () => { + // The workflow throws before ever drawing the orphaned id. A genuine + // user failure must be recorded as such — divergence detection only + // arbitrates logs the replay claims to have reproduced. + const run = makeRun(); + const throwingWorkflow = ` + async function workflow() { throw new Error("user boom"); } + workflow.workflowId = "workflow//test//workflow"; + globalThis.__private_workflows.set("workflow//test//workflow", workflow); + `; + const result = await runQuickJSWorkflow({ + workflowCode: throwingWorkflow, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: [ + runCreatedEvent(run), + { + eventId: 'evnt_orphan', + runId: run.runId, + eventType: 'step_created' as const, + correlationId: 'step_01JUNKJUNKJUNKJUNKJUNKJUNK', + eventData: { stepName: 'step//test//other' }, + createdAt: new Date('2025-01-01T00:00:01Z'), + }, + ], + }); + expect(result.failed?.message).toBe('user boom'); + }); + + it('rejects a live continuation fed events the session cannot reproduce', async () => { + const run = makeRun(); + const session = await startQuickJSWorkflow({ + workflowCode: stepWorkflow, + workflowId: 'workflow//test//workflow', + workflowRun: run, + events: [runCreatedEvent(run)], + }); + expect(session.result.suspended).toBeDefined(); + const stepCid = + session.result.suspended!.pendingOperations[0].correlationId; + + await expect( + session.continueWithEvents([ + { + eventId: 'evnt_wrong_step', + runId: run.runId, + eventType: 'step_created' as const, + correlationId: stepCid, + eventData: { stepName: 'step//test//DIFFERENT' }, + createdAt: new Date('2025-01-01T00:00:01Z'), + } as any, + ]) + ).rejects.toMatchObject({ name: 'ReplayDivergenceError' }); + // The session disposed itself on divergence; dispose() must be a no-op. + session.dispose(); + }); +}); diff --git a/packages/core/src/runtime/quickjs-runtime.ts b/packages/core/src/runtime/quickjs-runtime.ts index cf4f02a9c2..9468d35fb1 100644 --- a/packages/core/src/runtime/quickjs-runtime.ts +++ b/packages/core/src/runtime/quickjs-runtime.ts @@ -29,6 +29,7 @@ * `node:vm` engine's replay determinism. */ +import type { ReplayDivergenceError } from '@workflow/errors'; import type { Event, RunInput, @@ -54,6 +55,10 @@ import { isQuickJSBaselineSnapshotEnabled, } from './constants.js'; import { quickjsExtensions, quickjsWasm } from './quickjs-assets.generated.js'; +import { + findReplayDivergence, + type VmReplayView, +} from './quickjs-divergence.js'; import { adoptSerdeRoot, captureSerdeRoot, @@ -1696,12 +1701,30 @@ export async function startQuickJSWorkflow( } } + // ---- Replay-divergence arbitration ---- + // The replay is at a fixed point: every deliverable event has been + // delivered and the VM's job queue is drained. Arbitrate the log + // against the VM's draws before evaluating the workflow state — a + // divergence must escalate through runtime.ts's recovery machinery + // (bounded recovery replays → CorruptedEventLogError), never be + // absorbed into a wrong completion/suspension. Parity with the + // node:vm engine's EventsConsumer checks. + const observedEvents: Event[] = [...events]; + { + const divergence = sweepForReplayDivergence(vm, observedEvents); + if (divergence) { + vm.dispose(); + throw divergence; + } + } + // ---- Check result ---- return makeLiveSession( vm, serde, interruptBudget, advanceClock, + observedEvents, options.encryptionKey ); } @@ -1732,6 +1755,7 @@ function makeLiveSession( serde: QuickJSSerde, interruptBudget: InterruptBudget, advanceClock: (ms: number) => void, + observedEvents: Event[], encryptionKey?: DecryptionKey ): QuickJSWorkflowSession { const result = checkWorkflowState(vm, serde, { keepAliveOnSuspend: true }); @@ -1768,6 +1792,19 @@ function makeLiveSession( } while (batch > 0); } while (madeProgress && --maxIterations > 0); + // Fixed point for this burst — arbitrate the full observed log + // (initial replay + every fed delta) against the VM's draws before + // evaluating state. See the sweep in startQuickJSWorkflow. + observedEvents.push(...newEvents); + { + const divergence = sweepForReplayDivergence(vm, observedEvents); + if (divergence) { + alive = false; + vm.dispose(); + throw divergence; + } + } + const next = checkWorkflowState(vm, serde, { keepAliveOnSuspend: true, }); @@ -2406,6 +2443,49 @@ function markCreated(vm: QuickJS, cidJs: string, opType?: string): void { ).dispose(); } +// ---- Replay-Divergence Arbitration ---- + +/** + * Arbitrate the observed event log against the VM's fixed-point view of its + * own draws. Returns the divergence to escalate, or `null` when the replay + * reproduced the log (or when the workflow already failed — a genuine user + * failure is recorded as such; divergence detection only arbitrates logs the + * replay claims to have reproduced). + * + * Must only be called at a fixed point (event drain loop converged, VM job + * queue empty): that is what makes an unclaimed event evidence of divergence + * rather than of work still in flight. The node:vm engine needs a grace + * window and a delivery-idle gate to reach the same certainty; here the + * host drains the VM's microtask queue synchronously, so the fixed point is + * exact. + */ +function sweepForReplayDivergence( + vm: QuickJS, + observedEvents: readonly Event[] +): ReplayDivergenceError | null { + { + using failed = vm.evalCode('globalThis.__workflowError !== undefined'); + if (vm.dump(failed)) return null; + } + using viewHandle = vm.evalCode(`(function(){ + return { + ops: globalThis.__pending.map(function(p){ + return { + correlationId: p.correlationId, + type: p.type, + stepId: p.stepId, + token: p.token, + resumeAt: p.resumeAt, + }; + }), + hookCids: globalThis.__hooks ? Object.keys(globalThis.__hooks) : [], + abortCids: globalThis.__abortSignals ? Object.keys(globalThis.__abortSignals) : [], + }; + })()`); + const view = vm.dump(viewHandle) as VmReplayView; + return findReplayDivergence(observedEvents, view); +} + // ---- State Checking ---- /**