diff --git a/.changeset/log-order-draws.md b/.changeset/log-order-draws.md new file mode 100644 index 0000000000..bd1cddbcd0 --- /dev/null +++ b/.changeset/log-order-draws.md @@ -0,0 +1,6 @@ +--- +'workflow': patch +'@workflow/core': patch +--- + +Pin correlation-ID draw order to event-log order (Node.js VM engine), so two concurrent replays of the same run assign the same IDs even when one loaded a shorter event-log prefix. Set `WORKFLOW_LOG_ORDER_DRAWS=0` to opt back into arrival-order delivery resolution. diff --git a/.gitignore b/.gitignore index ee38f9164c..4b098bf7f1 100644 --- a/.gitignore +++ b/.gitignore @@ -41,3 +41,6 @@ packages/swc-plugin-workflow/build-hash.json workbench/nextjs-*/public/.well-known/workflow workbench/sveltekit/static/.well-known/workflow +# Per-run e2e diagnostics sidecars written to the repo root by the harness +# (writeDiagnosticsSidecar in packages/core/e2e/utils.ts) +e2e-diagnostics-*.json diff --git a/packages/core/src/delivery-barrier-coverage.test.ts b/packages/core/src/delivery-barrier-coverage.test.ts index e056895912..a62563c009 100644 --- a/packages/core/src/delivery-barrier-coverage.test.ts +++ b/packages/core/src/delivery-barrier-coverage.test.ts @@ -74,6 +74,10 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { }); const ulid = monotonicFactory(() => context.globalThis.Math.random()); const workflowStartedAt = context.globalThis.Date.now(); + // Real-session parity: the log-order-draws quiescence fixpoint keys its + // progress metric on `mintCount`; without it the loop degrades to a single + // turn and this suite would only exercise a degraded variant. + let mintCount = 0; const promiseQueueHolder = { current: Promise.resolve() }; const ctxRef: { current?: WorkflowOrchestratorContext } = {}; const ctx: WorkflowOrchestratorContext = { @@ -90,7 +94,13 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { getPromiseQueue: () => promiseQueueHolder.current, }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateUlid: () => { + mintCount += 1; + return ulid(workflowStartedAt); + }, + get mintCount() { + return mintCount; + }, generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), @@ -669,3 +679,88 @@ describe('suspension timing against parked step deliveries', () => { expectSuspensionSnapshotSteps(error, ['followUp']); }); }); + +// ─── step result above a wait parked behind an unclaimed payload ──────────── +// +// The log-order-draws turnstile (`quiesceEarlierCascades`) refuses to resolve +// a delivery while a LOWER-index ARMED barrier is still registered. An armed +// wait can itself be parked behind an unclaimed buffered hook payload — a +// chain only the idle-gated safety net can move (lowest-first retirement). +// This test pins the termination argument for that shape: the spinning step +// delivery must not count as a parked committed delivery (`resolvesOnItsOwn` +// excludes it — it gates on the parked wait), so `canRetireAbandonedBarriers` +// stays reachable, the net retires the payload, the wait delivers, and the +// turnstile opens. A regression that makes the turnstile wait on parked +// chains directly, or counts the spinner as self-resolving, deadlocks this +// replay instead of suspending it. +describe('log-order draws turnstile above a parked chain', () => { + const scenario = async () => { + const resumeAt = new Date(FIXED_TIMESTAMP + 5_000); + const ops: Promise[] = []; + const [payload, stepAResult] = await Promise.all([ + dehydrateStepReturnValue({ poke: 1 }, 'wrun_test', undefined, ops), + dehydrateStepReturnValue('a', 'wrun_test', undefined, ops), + ]); + + const events: Event[] = [ + event('evnt_0', 'hook_created', `hook_${ULIDS[0]}`, { + token: 'parked-token', + isWebhook: false, + }), + event('evnt_1', 'wait_created', `wait_${ULIDS[1]}`, { resumeAt }), + event('evnt_2', 'step_created', `step_${ULIDS[2]}`, { + stepName: 'stepA', + }), + event('evnt_3', 'step_started', `step_${ULIDS[2]}`, { + stepName: 'stepA', + }), + event('evnt_4', 'hook_received', `hook_${ULIDS[0]}`, { payload }), + event('evnt_5', 'wait_completed', `wait_${ULIDS[1]}`, { resumeAt }), + event('evnt_6', 'step_completed', `step_${ULIDS[2]}`, { + stepName: 'stepA', + result: stepAResult, + }), + event('evnt_7', 'step_created', `step_${ULIDS[3]}`, { + stepName: 'afterBoth', + }), + ]; + + const ctx = setupWorkflowContext(events); + const useStep = createUseStep(ctx); + const sleep = createSleep(ctx); + const createHook = createCreateHook(ctx); + + const error = await replay(ctx, async () => { + const stepA = useStep('stepA'); + const afterBoth = useStep('afterBoth'); + // Fire-and-forget hook: its payload (evnt_4) is consumed but never + // claimed, so its barrier stays unarmed and parks the wait behind it. + createHook({ token: 'parked-token' }); + await Promise.all([sleep('5s'), stepA()]); + await afterBoth(); + }); + + expectSuspendedWithPendingSteps(ctx, error, ['afterBoth']); + }; + + it('terminates and suspends with log-order draws on', async () => { + // Pin the flag rather than inherit the ambient environment: a suite-wide + // WORKFLOW_LOG_ORDER_DRAWS=0 sweep would otherwise silently run the off + // path twice and this test would prove nothing about the turnstile. + vi.stubEnv('WORKFLOW_LOG_ORDER_DRAWS', '1'); + try { + await scenario(); + } finally { + vi.unstubAllEnvs(); + } + }); + + it('terminates and suspends with log-order draws off', async () => { + vi.stubEnv('WORKFLOW_LOG_ORDER_DRAWS', '0'); + try { + await scenario(); + } finally { + vi.unstubAllEnvs(); + } + }); +}); diff --git a/packages/core/src/delivery-barrier-dispenser.test.ts b/packages/core/src/delivery-barrier-dispenser.test.ts index 226b07b7cf..d44064892e 100644 --- a/packages/core/src/delivery-barrier-dispenser.test.ts +++ b/packages/core/src/delivery-barrier-dispenser.test.ts @@ -50,6 +50,10 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { }); const ulid = monotonicFactory(() => context.globalThis.Math.random()); const workflowStartedAt = context.globalThis.Date.now(); + // Real-session parity: the log-order-draws quiescence fixpoint keys its + // progress metric on `mintCount`; without it the loop degrades to a single + // turn and this suite would only exercise a degraded variant. + let mintCount = 0; const promiseQueueHolder = { current: Promise.resolve() }; const ctxRef: { current?: WorkflowOrchestratorContext } = {}; const ctx: WorkflowOrchestratorContext = { @@ -70,7 +74,13 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { getPromiseQueue: () => promiseQueueHolder.current, }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateUlid: () => { + mintCount += 1; + return ulid(workflowStartedAt); + }, + get mintCount() { + return mintCount; + }, generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/log-order-draws.test.ts b/packages/core/src/log-order-draws.test.ts new file mode 100644 index 0000000000..1a998e4eb9 --- /dev/null +++ b/packages/core/src/log-order-draws.test.ts @@ -0,0 +1,319 @@ +import type { Event, WorkflowRun } from '@workflow/world'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import type { WorkflowSuspension } from './global.js'; +import { + dehydrateStepReturnValue, + dehydrateWorkflowArguments, +} from './serialization.js'; +import { runWorkflow } from './workflow.js'; + +/** + * Offline regression coverage for `WORKFLOW_LOG_ORDER_DRAWS=1`: correlation-id + * draw order pinned to event-log order, so draw bindings are stable under + * dense-prefix extension and concurrent replays of different-length prefixes + * mint compatible ids. + * + * The shape is the 2026-08-20 production corruption (five runs on + * 5.0.0-beta.43): a fan-out where each branch launches a step, then races a + * hook against a watchdog sleep. A replay whose dense prefix ends just before + * a sibling branch's launch completion sees that branch parked at its `await` + * minting nothing, so the woken branch's finalize takes the ordinal a fresher + * replay gives the sibling's wait. With draws pinned to log order, the + * finalize is minted inside the cascade of the delivery that enabled it, and + * extending the log can only append draws, never renumber them. + */ + +const RUN_ID = 'wrun_log_order_draws'; + +async function makeRun(): Promise { + const ops: Promise[] = []; + const input = await dehydrateWorkflowArguments([], RUN_ID, undefined, ops); + await Promise.all(ops); + return { + runId: RUN_ID, + workflowName: 'workflow', + status: 'running', + input, + createdAt: new Date('2024-01-01T00:00:00.000Z'), + updatedAt: new Date('2024-01-01T00:00:00.000Z'), + startedAt: new Date('2024-01-01T00:00:00.000Z'), + deploymentId: 'test-deployment', + }; +} + +const TRANSFORM = `;globalThis.__private_workflows = new Map(); + globalThis.__private_workflows.set("workflow", workflow);`; + +/** + * The 2026-08-20 production shape (five corrupted runs on 5.0.0-beta.43): a + * fan-out where each branch launches a step, then races a hook against a + * watchdog sleep, and finalizes when the hook wins. A replay whose dense + * prefix ends just before a sibling branch's launch completion sees that + * branch parked at its `await` — the branch mints nothing, not even its + * watchdog wait — so the woken branch's `finalizeTask` draws the ordinal a + * fresher replay gives the sibling's wait. One correlation id then names both + * a step and a wait, and every later replay fails with an unconsumable + * `step_created` (CORRUPTED_EVENT_LOG). + * + * The two branches' pre-race hops differ on purpose: the woken branch reaches + * its mint through the `Promise.race` resolution (two hops after delivery) + * while the unblocked sibling mints its wait one hop after its own delivery, + * which is how the sibling overtakes it on the shared counter. + */ +const BLOCKED_BRANCH_CODE = ` + const useStep = globalThis[Symbol.for("WORKFLOW_USE_STEP")]; + const createHook = globalThis[Symbol.for("WORKFLOW_CREATE_HOOK")]; + const sleep = globalThis[Symbol.for("WORKFLOW_SLEEP")]; + const launchTask = useStep("launchTask"); + const finalizeTask = useStep("finalizeTask"); + const WATCHDOG = "watchdog"; + async function workflow() { + await Promise.all([0, 1, 2].map(async (task) => { + const hook = createHook({ token: "task-done-" + task }); + await launchTask(task); + const winner = await Promise.race([ + hook, + sleep("1h").then(() => WATCHDOG), + ]); + if (winner !== WATCHDOG) { + await finalizeTask(task); + } + })); + }${TRANSFORM}`; + +/** Replays the blocked-branch workflow and returns every pending entity. */ +async function blockedBranchEntities( + events: Event[] +): Promise<{ type: string; correlationId: string; stepName?: string }[]> { + const run = await makeRun(); + try { + await runWorkflow(BLOCKED_BRANCH_CODE, run, events, undefined); + } catch (error) { + const suspension = error as WorkflowSuspension; + if (suspension.name !== 'WorkflowSuspension') { + throw error; + } + return suspension.steps.map((item) => ({ + type: item.type, + correlationId: item.correlationId, + stepName: (item as { stepName?: string }).stepName, + })); + } + throw new Error('expected the replay to suspend'); +} + +/** + * Builds the two dense prefixes of the production log. The shorter one ends + * before the second branch's launch completion; the longer one appends it. + */ +async function blockedBranchPrefixes(): Promise<{ + shorter: Event[]; + longer: Event[]; +}> { + const initial = await blockedBranchEntities([]); + const hooks = initial.filter((item) => item.type === 'hook'); + const launches = initial.filter((item) => item.stepName === 'launchTask'); + expect(hooks).toHaveLength(3); + expect(launches).toHaveLength(3); + + let at = 0; + const stamp = () => new Date(Date.parse('2024-01-01T00:00:00.000Z') + ++at); + const event = ( + eventType: Event['eventType'], + correlationId: string, + eventData: object + ): Event => ({ + eventId: `event-${at + 1}`, + runId: RUN_ID, + eventType, + correlationId, + eventData: eventData as Event['eventData'], + createdAt: stamp(), + }); + + const ops: Promise[] = []; + const launchResult = await dehydrateStepReturnValue( + 'launched', + RUN_ID, + undefined, + ops + ); + const hookPayload = await dehydrateStepReturnValue( + { done: true }, + RUN_ID, + undefined, + ops + ); + await Promise.all(ops); + + // Mirrors the production slot order: every branch's hook and launch created, + // the first two launches completed, both of their hooks received, and the + // third branch's launch completion as the extension event. + const shorter: Event[] = [ + ...hooks.map((hook, task) => + event('hook_created', hook.correlationId, { + token: `task-done-${task}`, + isWebhook: false, + }) + ), + ...launches.map((launch) => + event('step_created', launch.correlationId, { stepName: 'launchTask' }) + ), + event('step_completed', launches[0]!.correlationId, { + stepName: 'launchTask', + result: launchResult, + }), + event('step_completed', launches[1]!.correlationId, { + stepName: 'launchTask', + result: launchResult, + }), + event('hook_received', hooks[0]!.correlationId, { + payload: hookPayload, + }), + event('hook_received', hooks[1]!.correlationId, { + payload: hookPayload, + }), + ]; + const longer: Event[] = [ + ...shorter, + event('step_completed', launches[2]!.correlationId, { + stepName: 'launchTask', + result: launchResult, + }), + ]; + return { shorter, longer }; +} + +function suspensionBindings( + items: { type: string; correlationId: string; stepName?: string }[] +) { + return items + .map((item) => `${item.correlationId}=${item.type}:${item.stepName ?? ''}`) + .sort(); +} + +describe('arrival-order draws (opt-out, WORKFLOW_LOG_ORDER_DRAWS=0)', () => { + const original = process.env.WORKFLOW_LOG_ORDER_DRAWS; + + beforeEach(() => { + process.env.WORKFLOW_LOG_ORDER_DRAWS = '0'; + }); + + afterEach(() => { + if (original === undefined) { + delete process.env.WORKFLOW_LOG_ORDER_DRAWS; + } else { + process.env.WORKFLOW_LOG_ORDER_DRAWS = original; + } + }); + + // Deliberate CONTROL: asserts the arrival-order BUG still reproduces with + // the flag off. If this stops failing-to-bind (i.e. `rebound` comes back + // empty), the control is obsolete, not broken — most likely because + // positional rebinding was fixed independently of draw scheduling, e.g. + // call-site-addressed correlation ids (vercel/workflow#3179) landing, which + // removes the rebinding in BOTH modes. Delete this block then; do not chase + // it as a regression. + it('rebinds an ordinal from a step to a wait under extension (the control)', async () => { + const { shorter, longer } = await blockedBranchPrefixes(); + const stale = await blockedBranchEntities(shorter); + const fresh = await blockedBranchEntities(longer); + const staleFinalizes = stale + .filter((item) => item.stepName === 'finalizeTask') + .map((item) => item.correlationId); + const freshIds = new Set(fresh.map((item) => item.correlationId)); + const rebound = staleFinalizes.filter((id) => !freshIds.has(id)); + expect(rebound.length).toBeGreaterThan(0); + }); +}); + +describe('log-order draws (the default)', () => { + const original = process.env.WORKFLOW_LOG_ORDER_DRAWS; + + beforeEach(() => { + delete process.env.WORKFLOW_LOG_ORDER_DRAWS; + }); + + afterEach(() => { + if (original !== undefined) { + process.env.WORKFLOW_LOG_ORDER_DRAWS = original; + } + }); + + it('keeps every binding of the shorter prefix under extension', async () => { + const { shorter, longer } = await blockedBranchPrefixes(); + const stale = await blockedBranchEntities(shorter); + const fresh = await blockedBranchEntities(longer); + // Every entity the shorter replay would create must exist, under the SAME + // correlation id and kind, in the longer replay: extension appends draws, + // never renumbers them. + // The corruption signature is one correlation id bound to two different + // entities across the two replays. Compare bindings on shared ids: an id + // present in both pending sets must name the same entity. Ids only in the + // shorter set are entities the extension consumed (the sibling's launch); + // ids only in the longer set are the extension's appended draws. + const byId = ( + items: { type: string; correlationId: string; stepName?: string }[] + ) => + new Map( + items.map((item) => [ + item.correlationId, + `${item.type}:${item.stepName ?? ''}`, + ]) + ); + const staleById = byId(stale); + const freshById = byId(fresh); + const rebound: string[] = []; + for (const [id, binding] of staleById) { + const extended = freshById.get(id); + if (extended !== undefined && extended !== binding) { + rebound.push(`${id}: ${binding} -> ${extended}`); + } + } + expect(rebound).toEqual([]); + // And specifically: the woken branches' finalize steps keep their ids. + const staleFinalizes = stale + .filter((item) => item.stepName === 'finalizeTask') + .map((item) => item.correlationId) + .sort(); + const freshFinalizes = fresh + .filter((item) => item.stepName === 'finalizeTask') + .map((item) => item.correlationId) + .sort(); + expect(freshFinalizes).toEqual(staleFinalizes); + }); + + it('is stable across every dense prefix of the log', async () => { + // The pairwise test above targets the production window; this sweeps all + // of them: no shared correlation id may change entity between any two + // consecutive dense prefixes. + const { longer } = await blockedBranchPrefixes(); + let previous: Map | undefined; + for (let length = 1; length <= longer.length; length++) { + const entities = await blockedBranchEntities(longer.slice(0, length)); + const current = new Map( + entities.map((item) => [ + item.correlationId, + `${item.type}:${item.stepName ?? ''}`, + ]) + ); + if (previous) { + for (const [id, binding] of previous) { + const extended = current.get(id); + expect( + extended === undefined || extended === binding, + `prefix ${length}: ${id} rebound ${binding} -> ${extended}` + ).toBe(true); + } + } + previous = current; + } + }); + + it('is deterministic per prefix', async () => { + const { longer } = await blockedBranchPrefixes(); + const a = suspensionBindings(await blockedBranchEntities(longer)); + const b = suspensionBindings(await blockedBranchEntities(longer)); + expect(b).toEqual(a); + }); +}); diff --git a/packages/core/src/private.ts b/packages/core/src/private.ts index 7da24d0a0b..ce8f5acab2 100644 --- a/packages/core/src/private.ts +++ b/packages/core/src/private.ts @@ -137,6 +137,14 @@ export interface WorkflowOrchestratorContext { invocationsQueue: Map; onWorkflowError: (error: Error) => void; generateUlid: () => string; + /** + * Monotone count of correlation-id draws this replay has made. Progress + * metric for {@link quiesceEarlierCascades}: a macrotask turn in which it + * does not move (and no hydration is in flight) means every woken branch has + * run as far as it can without another delivery. Optional so lightweight + * test contexts degrade to the single-yield behavior. + */ + readonly mintCount?: number; generateNanoid: () => string; /** * Sequential promise queue that ensures all event-driven promise resolutions @@ -422,17 +430,127 @@ function computeResolvesOnItsOwn( * loaded. storm-log-replay.test.ts replays a production log corrupted exactly * that way.) */ +/** + * Whether correlation-id draw order is pinned to event-log order. Default ON: + * only the literal string `WORKFLOW_LOG_ORDER_DRAWS=0` opts out — `=false`, + * `=off`, and every other value keep it enabled. Read per call so tests can + * flip it. + * + * Off, a delivery that had to defer yields ONE macrotask after its + * predecessors resolve — enough for short consumers, but a woken branch whose + * path to its next draw crosses more hops (a `Promise.race` resolution, a + * user-level semaphore, an async-iterator read) can still be overtaken by a + * later-in-log delivery's shorter cascade, so the run's draw order — and + * therefore its correlation ids — depends on how much log this replay loaded. + * On, the yield becomes a fixpoint: the delivery resolves only once every + * earlier cascade has quiesced, making the draw sequence a pure function of + * the dense log, stable under prefix extension, and concurrent writers' + * duplicate creates identical (deduped) instead of colliding. + */ +function isLogOrderDrawsEnabled(): boolean { + return process.env.WORKFLOW_LOG_ORDER_DRAWS !== '0'; +} + +/** + * One quiescence turn: lets the entire pending microtask queue drain, then + * yields to the event loop once. `setImmediate` (check phase) is used where + * available because Node clamps `setTimeout(0)` to ~1ms while `setImmediate` + * costs ~20µs — and every branch-deciding delivery pays this turn at least + * once, so on a sequential replay the clamp is the whole cost. Timers still + * run between consecutive turns (the loop re-enters the event loop each + * iteration, passing through the timers phase), so chains parked on the + * safety-net dispenser's `setTimeout` cadence are not starved. + */ +function quiescenceTurn(delayMs: number): Promise { + if (delayMs === 0 && typeof setImmediate === 'function') { + return new Promise((resolve) => setImmediate(resolve)); + } + return new Promise((resolve) => setTimeout(resolve, delayMs)); +} + +/** + * Waits until the workflow can make no further progress without another + * delivery: repeated (promise-queue drain + event-loop turn)s until a full + * turn passes with no new ULID draws and no hydration in flight. + * + * Termination: each extra iteration requires a new ULID draw or a hydration + * started in the previous turn. `mintCount` counts EVERY draw from the run's + * sequence — correlation ids and the serialization-driven draws (stream ids + * minted through the `STABLE_ULID` global while dehydrating) — which is + * conservative in the safe direction: serialization draws only extend the + * wait, and both body progress and the serialization work one cascade can + * schedule are finite between deliveries, so the fixpoint is reached. The + * loop holds this delivery's own barrier registered (its `markDelivered` has + * not run), so `isDeliveryIdle` stays false and no suspension can preempt the + * cascade being waited out. + * + * A rejected `promiseQueue` settles immediately and forever, so looping at + * the normal cadence on a failed run would degenerate into a busy loop (the + * same hazard {@link ensureBarrierSafetyNet} documents). Iterations that + * observe the queue rejected back off to a 50ms tick instead. + */ +async function quiesceEarlierCascades( + ctx: WorkflowOrchestratorContext, + eventIndex: number +): Promise { + for (;;) { + const mintsBefore = ctx.mintCount ?? 0; + const pendingBefore = ctx.pendingDeliveries; + // Settled or rejected, the queue snapshot only orders us behind work + // already chained; a rejection is the run failing elsewhere. + const queueRejected = await ctx.promiseQueue.then( + () => false, + () => true + ); + await quiescenceTurn(queueRejected ? 50 : 0); + if ( + (ctx.mintCount ?? 0) !== mintsBefore || + pendingBefore !== 0 || + ctx.pendingDeliveries !== 0 + ) { + continue; + } + // Quiet is not enough on its own: several delivery chains can be sitting + // in this loop at once, and letting the first quiet observation resolve + // would break the tie by timer arrival — the arrival-order dependence this + // mode removes. A lower-index ARMED barrier is a delivery committed to + // happening that has not happened yet, so this one keeps waiting. Unarmed + // entries (buffered payloads nobody has claimed) do not block, exactly as + // in `gatesOn`: their handover is claim-driven, which is body-position + // determined and therefore already a function of the prefix. + // + // This is deliberately a WIDER waits-for relation than `gatesOn` (which, + // e.g., excludes wait→wait): under log-order draws EVERY branch-deciding + // delivery must resolve in log order, kinds included. The width is safe + // against the dispenser deadlock that `resolvesOnItsOwn` guards, because + // the one edge the wider relation adds — waiting on an armed entry that + // gatesOn does not model — always bottoms out at the same unarmed payload: + // a lower armed WAIT is non-self-resolving only when it is (transitively) + // parked behind an unclaimed buffered payload, and a wait gates on every + // lower hook and step directly, so the spinner here also gates on that + // payload through `gatesOn` and is itself reported non-self-resolving. + // The dispenser therefore stays unblocked and retires the chain head; see + // the parked-chain test in delivery-barrier-coverage.test.ts. + let lowerArmed = false; + for (const [index, entry] of ctx.pendingDeliveryBarriers ?? []) { + if (index < eventIndex && entry.armed) { + lowerArmed = true; + break; + } + } + if (!lowerArmed) { + return; + } + } +} + export async function awaitEarlierDeliveries( ctx: WorkflowOrchestratorContext, eventIndex: number | undefined, kind: DeliveryKind ): Promise { // Defensive: tolerate contexts that predate this field (test harnesses). - if ( - eventIndex === undefined || - !ctx.pendingDeliveryBarriers || - ctx.pendingDeliveryBarriers.size === 0 - ) { + if (eventIndex === undefined || !ctx.pendingDeliveryBarriers) { return; } const barriers = ctx.pendingDeliveryBarriers; @@ -445,6 +563,19 @@ export async function awaitEarlierDeliveries( } if (earlier.length > 0) { await Promise.all(earlier); + } + if (isLogOrderDrawsEnabled()) { + // Unconditional, not just when a barrier was still registered: an earlier + // delivery's barrier deregisters when its resolve() runs, but the branch + // it woke may still be hops away from its next draw. A later delivery + // consumed after that deregistration sees an empty gate set, and without + // this it would resolve mid-cascade and overtake the draw — the exact + // arrival-order dependence this mode exists to remove. Costs one quiet + // macrotask turn when nothing is in flight. + await quiesceEarlierCascades(ctx, eventIndex); + return; + } + if (earlier.length > 0) { // An earlier delivery being "delivered" only means its `resolve()` ran. // The branch it woke may need an arbitrary number of further microtask // hops before it reaches its next `useStep` call and draws a ULID — a diff --git a/packages/core/src/storm-log-replay.test.ts b/packages/core/src/storm-log-replay.test.ts index a1c6990d33..5192086e26 100644 --- a/packages/core/src/storm-log-replay.test.ts +++ b/packages/core/src/storm-log-replay.test.ts @@ -63,6 +63,10 @@ function setupWorkflowContext( const context = createContext({ seed: SEED, fixedTimestamp: FIXED_TS }); const ulid = monotonicFactory(() => context.globalThis.Math.random()); const workflowStartedAt = context.globalThis.Date.now(); + // Real-session parity: the log-order-draws quiescence fixpoint keys its + // progress metric on `mintCount`; without it the loop degrades to a single + // turn and this suite would only exercise a degraded variant. + let mintCount = 0; const promiseQueueHolder = { current: Promise.resolve() }; const ctxRef: { current?: WorkflowOrchestratorContext } = {}; const ctx: WorkflowOrchestratorContext = { @@ -83,7 +87,13 @@ function setupWorkflowContext( getPromiseQueue: () => promiseQueueHolder.current, }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateUlid: () => { + mintCount += 1; + return ulid(workflowStartedAt); + }, + get mintCount() { + return mintCount; + }, generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/storm-log-sweep.test.ts b/packages/core/src/storm-log-sweep.test.ts index 41ef501c2f..0a7bf7b381 100644 --- a/packages/core/src/storm-log-sweep.test.ts +++ b/packages/core/src/storm-log-sweep.test.ts @@ -36,6 +36,10 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { const context = createContext({ seed: SEED, fixedTimestamp: FIXED_TS }); const ulid = monotonicFactory(() => context.globalThis.Math.random()); const workflowStartedAt = context.globalThis.Date.now(); + // Real-session parity: the log-order-draws quiescence fixpoint keys its + // progress metric on `mintCount`; without it the loop degrades to a single + // turn and this suite would only exercise a degraded variant. + let mintCount = 0; const promiseQueueHolder = { current: Promise.resolve() }; const ctxRef: { current?: WorkflowOrchestratorContext } = {}; const ctx: WorkflowOrchestratorContext = { @@ -56,7 +60,13 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { getPromiseQueue: () => promiseQueueHolder.current, }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateUlid: () => { + mintCount += 1; + return ulid(workflowStartedAt); + }, + get mintCount() { + return mintCount; + }, generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts index 417e611c63..5abdd77a15 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -134,6 +134,18 @@ export async function runWorkflow( const workflowDiscontinuation = withResolvers(); const ulid = monotonicFactory(() => vmGlobalThis.Math.random()); + // The draw counter is the progress metric for `quiesceEarlierCascades` + // (WORKFLOW_LOG_ORDER_DRAWS): a quiet turn is one that drew nothing. It + // counts EVERY draw from this sequence, including the serialization draws + // that mint stream ids through the `STABLE_ULID` global below, which is + // deliberate: quiescence must also wait out serialization-driven draws, + // and counting extra draws only extends the wait (see the termination note + // on `quiesceEarlierCascades`). + let mintCount = 0; + const countingUlid = (seedTime?: number) => { + mintCount += 1; + return ulid(seedTime); + }; const generateNanoid = nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * vmGlobalThis.Math.random()) ); @@ -170,8 +182,11 @@ export async function runWorkflow( globalThis: vmGlobalThis, onWorkflowError: workflowDiscontinuation.reject, eventsConsumer, - generateUlid: () => ulid(+startedAt), + generateUlid: () => countingUlid(+startedAt), generateNanoid, + get mintCount() { + return mintCount; + }, invocationsQueue: new Map(), // Use getter/setter so the EventsConsumer's getPromiseQueue() always // sees the latest queue state as it's mutated by step/hook/sleep callbacks. @@ -234,7 +249,7 @@ export async function runWorkflow( // @ts-expect-error - `@types/node` says symbol is not valid, but it does work vmGlobalThis[WORKFLOW_CONTEXT_SYMBOL] = ctx; // @ts-expect-error - `@types/node` says symbol is not valid, but it does work - vmGlobalThis[STABLE_ULID] = ulid; + vmGlobalThis[STABLE_ULID] = countingUlid; // NOTE: Will have a config override to use the custom fetch step. // For now `fetch` must be explicitly imported from `workflow`.