From fbebf7104d97219b73b6a51b0e77e42c45cdd99c Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Mon, 10 Aug 2026 12:46:28 -0700 Subject: [PATCH 1/5] [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 2/5] 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 3/5] 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 4/5] 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 5/5] 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