[core] Gate the unconsumed-event check on delivery idleness

.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

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> = {}): 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<Event>();
+      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);
+    });
+  });
 });

packages/core/src/events-consumer.ts

@@ -65,6 +65,16 @@ export interface EventsConsumerOptions {
    * deserialization delays the resolve() that triggers the next subscribe().
    */
   getPromiseQueue: () => Promise<void>;
+  /**
+   * 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<void>;
+  private isDeliveryIdle: () => boolean;
   private pendingUnconsumedCheck: Promise<void> | null = null;
   private pendingUnconsumedTimeout: ReturnType<typeof setTimeout> | 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();
+  }
 }

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,19 +601,33 @@ 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
  * {@link hasParkedCommittedDelivery} covers the detached gap between a slot
  * 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.