Backport #3552: fix(core): replay ordering - preserve adjacent hook delivery order

.changeset/order-adjacent-hook-deliveries.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Preserve event-log order when adjacent hook payloads resume independent workflow branches.

packages/core/src/delivery-barrier-coverage.test.ts

@@ -3,7 +3,7 @@
  * `CORRUPTED_EVENT_LOG` shapes) and `step-delivery-hop-count.test.ts` (a step
  * result must not overtake the wait/hook branch the log ordered first, however
  * many hops that branch needs). Both of those cover a step result deferring
- * behind an earlier delivery. This file covers the two cases where the
+ * behind an earlier delivery. This file covers the three cases where the
  * delivery-barrier registry did not yet reach:
  *
  *  1. STEP behind STEP, across drain windows. `DEFER_BEHIND.step` used to
@@ -20,11 +20,15 @@
  *     its barrier. They then skipped both the gate and
  *     `awaitEarlierDeliveries`' macrotask yield.
  *
+ *  3. HOOK behind HOOK. Adjacent payloads for independently awaited hooks
+ *     must reach their consumers in log order even when the earlier consumer
+ *     takes more microtask hops to act on its value.
+ *
  * (Upstream this file also covers abort deliveries, which participate in the
  * registry too. The workflow `AbortController` does not exist on this branch,
  * so that case is not portable here.)
  *
- * Cases 1-2 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`.
  *
@@ -394,7 +398,75 @@ describe('hook payload delivery ordering against an earlier step result', () =>
   }
 });
 
-// ─── 3. registry scan cost ─────────────────────────────────────────────────
+// ─── 3. hook behind hook, across different consumer shapes ─────────────────
+describe('hook payload delivery ordering against an earlier hook payload', () => {
+  it('keeps the recorded ULID allocation when the earlier hook uses an async iterator', async () => {
+    const ops: Promise<unknown>[] = [];
+    const [firstPayload, secondPayload] = await Promise.all([
+      dehydrateStepReturnValue({ v: 1 }, 'wrun_test', undefined, ops),
+      dehydrateStepReturnValue({ v: 2 }, 'wrun_test', undefined, ops),
+    ]);
+
+    const events: Event[] = [
+      event('evnt_0', 'hook_created', `hook_${ULIDS[0]}`, {
+        token: 'first',
+        isWebhook: false,
+      }),
+      event('evnt_1', 'hook_created', `hook_${ULIDS[1]}`, {
+        token: 'second',
+        isWebhook: false,
+      }),
+      event('evnt_2', 'hook_received', `hook_${ULIDS[0]}`, {
+        token: 'first',
+        payload: firstPayload,
+      }),
+      event('evnt_3', 'hook_received', `hook_${ULIDS[1]}`, {
+        token: 'second',
+        payload: secondPayload,
+      }),
+      event('evnt_4', 'step_created', `step_${ULIDS[2]}`, {
+        stepName: 'afterFirst',
+      }),
+      event('evnt_5', 'step_created', `step_${ULIDS[3]}`, {
+        stepName: 'afterSecond',
+      }),
+    ];
+
+    const ctx = setupWorkflowContext(events);
+    const useStep = createUseStep(ctx);
+    const createHook = createCreateHook(ctx);
+
+    const error = await replay(ctx, async () => {
+      const first = createHook<{ v: number }>({ token: 'first' });
+      const second = createHook<{ v: number }>({ token: 'second' });
+      const afterFirst = useStep('afterFirst');
+      const afterSecond = useStep('afterSecond');
+
+      await Promise.all([
+        (async () => {
+          for await (const payload of first) {
+            void payload;
+            // Model a layered hook consumer such as an async inbox merger:
+            // the delivery order must survive its continuation depth.
+            for (let i = 0; i < 16; i++) {
+              await Promise.resolve();
+            }
+            await afterFirst();
+            break;
+          }
+        })(),
+        (async () => {
+          await second;
+          await afterSecond();
+        })(),
+      ]);
+    });
+
+    expectSuspendedWithPendingSteps(ctx, error, ['afterFirst', 'afterSecond']);
+  });
+});
+
+// ─── 4. registry scan cost ─────────────────────────────────────────────────
 //
 // `resolvesOnItsOwn` walks the registry recursively: an armed hook re-checks
 // every earlier wait and step, an armed wait every earlier hook and step, and
@@ -432,7 +504,7 @@ describe('delivery-barrier registry scan cost', () => {
   });
 });
 
-// ─── 4. suspension timing: idle must wait out parked deliveries ────────────
+// ─── 5. suspension timing: idle must wait out parked deliveries ────────────
 //
 // Field shape from vercel/workflow#3183: a fire-and-forget `sleep()` (a
 // watchdog — never awaited, never completing in-run) plus a parallel batch of

packages/core/src/private.ts

@@ -235,8 +235,7 @@ interface DeliveryBarrierEntry {
  * Which earlier kinds each delivery kind defers behind. Chosen so that no kind
  * blocks on a peer it does not need to:
  *
- *  - a hook defers behind earlier WAITS and STEPS — not earlier hooks, which
- *    are sequential same-entity payloads and must not block one another;
+ *  - a hook defers behind earlier HOOKS, WAITS and STEPS;
  *  - a wait defers behind earlier HOOKS and STEPS — not earlier waits, since a
  *    wait never needs to queue behind another wait;
  *  - a step defers behind earlier WAITS, HOOKS and STEPS.
@@ -259,7 +258,7 @@ interface DeliveryBarrierEntry {
  * wait-for graph can never contain a cycle.
  */
 const DEFER_BEHIND: Record<DeliveryKind, readonly DeliveryKind[]> = {
-  hook: ['wait', 'step'],
+  hook: ['hook', 'wait', 'step'],
   wait: ['hook', 'step'],
   step: ['wait', 'hook', 'step'],
 };