From ec5c121952abb7949681c8e824ca1ab07ae40f0a Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Mon, 10 Aug 2026 16:58:07 -0700 Subject: [PATCH] Gate the unconsumed-event check on delivery idleness The events consumer walks the log synchronously, but 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. The walk therefore routinely sits on an ordered event that nobody has claimed yet while the workflow is mid-flight on its way to claiming it. The deferred unconsumed-event check resolved that with 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 still on the detached path, and the check raises `ReplayDivergenceError` against a log the 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 at a 100ms window, 34 of 42 at 10ms, with the no-parallel-delivery control clean at both. `hasParkedCommittedDelivery` already documents this hazard for the suspension path, where `scheduleWhenIdle` guards it by polling. Export that predicate as `isDeliveryIdle`, thread it into `EventsConsumer`, and poll it before starting the delay timer. Termination is inherited: it counts only deliveries that resolve on their own, so nothing can gate its own retirement, and a genuinely orphaned event reaches the check on the first poll. --- .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