[core] Fix setAttributes() resolution bypassing delivery barriers

.changeset/attr-set-delivery-order.md

@@ -0,0 +1,5 @@
+---
+"@workflow/core": patch
+---
+
+Order `setAttributes()` resolution with the replay delivery barriers so a branch awaiting it can no longer resume out of event-log order, which could fail the run with `REPLAY_DIVERGENCE` or hand concurrent branches each other's step results

packages/core/README.md

@@ -18,7 +18,9 @@ an extra queued continuation unless other background operations remain pending.
 Hook registration acknowledgements and token conflicts participate in replay
 delivery ordering alongside step results, hook payloads, and sleep completions.
 Concurrent branches awaiting `hook.getConflict()` preserve their step correlation
-IDs when replaying an extended event history.
+IDs when replaying an extended event history. `setAttributes()` acknowledgements
+take part in the same ordering, so a branch awaiting `setAttributes()` while
+another branch is in flight keeps its own step results on replay.
 
 Failed-run logs include underlying error causes and codes to help diagnose failures such as socket, DNS, and TLS errors. Unreadable causes are marked without preventing the run from being recorded as failed.
 

packages/core/src/private.ts

@@ -262,7 +262,7 @@ export interface WorkflowOrchestratorContext {
 }
 
 /** The kind of branch-deciding delivery a barrier represents. */
-export type DeliveryKind = 'hook' | 'wait' | 'step';
+export type DeliveryKind = 'hook' | 'wait' | 'step' | 'attribute';
 
 interface DeliveryBarrierEntry {
   kind: DeliveryKind;
@@ -300,7 +300,13 @@ interface DeliveryBarrierEntry {
  *  - 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.
+ *  - a step defers behind earlier WAITS, HOOKS and STEPS;
+ *  - an attribute (the `attr_set` acknowledgement that resolves
+ *    `setAttributes()`) defers behind every earlier kind, and every kind
+ *    defers behind an earlier attribute. Resolving `setAttributes()` resumes a
+ *    branch that may draw the next correlation ID, so it is as branch-deciding
+ *    as a step result; gating only one direction would still let a later
+ *    delivery overtake a branch resuming from an attribute write (#4770).
  *
  * The step-behind-step edge is not redundant with the serial `promiseQueue`.
  * The queue fixes the order in which step results are HYDRATED, but a step no
@@ -320,9 +326,10 @@ interface DeliveryBarrierEntry {
  * wait-for graph can never contain a cycle.
  */
 const DEFER_BEHIND: Record<DeliveryKind, readonly DeliveryKind[]> = {
-  hook: ['hook', 'wait', 'step'],
-  wait: ['hook', 'step'],
-  step: ['wait', 'hook', 'step'],
+  hook: ['hook', 'wait', 'step', 'attribute'],
+  wait: ['hook', 'step', 'attribute'],
+  step: ['wait', 'hook', 'step', 'attribute'],
+  attribute: ['hook', 'wait', 'step', 'attribute'],
 };
 
 /**
@@ -350,11 +357,11 @@ function gatesOn(
   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);
+  // A step or attribute 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' || kind === 'attribute') && !other.armed);
 }
 
 /**
@@ -445,7 +452,8 @@ function computeResolvesOnItsOwn(
  * 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,
+ * One asymmetry: a STEP result (and, for the same reason, an ATTRIBUTE
+ * acknowledgement) 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
  * often requires the step result itself (`await stepX()` before the
@@ -696,9 +704,11 @@ export function registerDeliveryBarrier(
      * so `Date.now()` in the cascade this delivery wakes reads the delivery's
      * own time, whatever the consumer walk has read ahead of it. Required so
      * that a new delivery site cannot forget the clock: a delivery that does
-     * not move it leaves the cascade it wakes reading a stale time.
+     * not move it leaves the cascade it wakes reading a stale time. `null`
+     * opts out explicitly; attribute acknowledgements use it because they
+     * have never moved the clock, and recorded histories may branch on it.
      */
-    deliveredAt: number;
+    deliveredAt: number | null;
   }
 ): DeliveryBarrier {
   // Idempotent like the handle it backs; `advanceClock` never moves backwards,
@@ -709,7 +719,9 @@ export function registerDeliveryBarrier(
       return false;
     }
     deliveredToWorkflow = true;
-    ctx.advanceClock?.(options.deliveredAt);
+    if (options.deliveredAt !== null) {
+      ctx.advanceClock?.(options.deliveredAt);
+    }
     return true;
   };
 

packages/core/src/set-attributes-delivery-order.test.ts

@@ -0,0 +1,255 @@
+import type { CreateEventRequest, Event, WorkflowRun } from '@workflow/world';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { type QueueItem, WorkflowSuspension } from './global.js';
+import { ReplayPayloadCache } from './replay-payload-cache.js';
+import {
+  dehydrateStepArguments,
+  dehydrateStepReturnValue,
+  dehydrateWorkflowArguments,
+  hydrateWorkflowReturnValue,
+} from './serialization.js';
+import { runWorkflow } from './workflow.js';
+
+afterEach(() => vi.unstubAllEnvs());
+
+// Regression for #4770: resolving `setAttributes()` bypassed the delivery
+// barriers, so a branch awaiting it could resume ahead of a step result that
+// is earlier in the log and draw another branch's correlation ID. When both
+// branches then call the same step, the step consumer accepts the swapped
+// events and the branches silently receive each other's results.
+const PRELUDE = `
+  const step = globalThis[Symbol.for('WORKFLOW_USE_STEP')];
+  const quickStep = step('quickStep');
+  const delayedStep = step('delayedStep');
+  const setAttributes = globalThis[Symbol.for('WORKFLOW_SET_ATTRIBUTES')];
+  const sleep = globalThis[Symbol.for('WORKFLOW_SLEEP')];
+  const createHook = globalThis[Symbol.for('WORKFLOW_CREATE_HOOK')];
+`;
+
+const SWAP_WORKFLOW = (hops: number) => `${PRELUDE}
+  globalThis.__private_workflows = new Map([['workflow', async () => {
+    const attributeBranch = async () => {
+      await quickStep('before-attribute');
+      await setAttributes([{ key: 'phase', value: 'attribute-branch' }]);
+      return await quickStep('after-attribute');
+    };
+    const sleepBranch = async () => {
+      await delayedStep(0);
+      for (let hop = 0; hop < ${hops}; hop++) await Promise.resolve();
+      const other = await quickStep('other-branch');
+      await sleep('300ms');
+      return other;
+    };
+    const [a, b] = await Promise.all([attributeBranch(), sleepBranch()]);
+    return a + '|' + b;
+  }]]);
+`;
+
+const HOOK_WORKFLOW = (hops: number) => `${PRELUDE}
+  globalThis.__private_workflows = new Map([['workflow', async () => {
+    const attributeBranch = async () => {
+      await quickStep('before-attribute');
+      await setAttributes([{ key: 'phase', value: 'attribute-branch' }]);
+      await quickStep('after-attribute');
+    };
+    const hookBranch = async () => {
+      await delayedStep(0);
+      for (let hop = 0; hop < ${hops}; hop++) await Promise.resolve();
+      const hook = createHook({ token: 'attr-vs-hook' });
+      await hook;
+    };
+    await Promise.all([attributeBranch(), hookBranch()]);
+    return 'completed';
+  }]]);
+`;
+
+const RUN_ID = 'wrun_set_attributes_delivery_order';
+
+/** The events a world records when the workflow first asks for `item`. */
+async function creationEvents(item: QueueItem): Promise<CreateEventRequest[]> {
+  switch (item.type) {
+    case 'step':
+      return [
+        {
+          eventType: 'step_created',
+          correlationId: item.correlationId,
+          eventData: {
+            stepName: item.stepName,
+            input: await dehydrateStepArguments(item.args, RUN_ID, undefined),
+          },
+        },
+      ];
+    case 'wait':
+      return [
+        {
+          eventType: 'wait_created',
+          correlationId: item.correlationId,
+          eventData: { resumeAt: item.resumeAt },
+        },
+      ];
+    case 'hook':
+      return [
+        {
+          eventType: 'hook_created',
+          correlationId: item.correlationId,
+          eventData: { token: item.token, isWebhook: false },
+        },
+      ];
+    case 'attribute':
+      return [
+        {
+          eventType: 'attr_set',
+          correlationId: item.correlationId,
+          eventData: { changes: item.changes, writer: { type: 'workflow' } },
+        },
+      ];
+  }
+}
+
+/** The events that complete `item`. Each step echoes its first argument. */
+async function completionEvents(
+  item: QueueItem
+): Promise<CreateEventRequest[]> {
+  switch (item.type) {
+    case 'step':
+      return [
+        {
+          eventType: 'step_started',
+          correlationId: item.correlationId,
+          eventData: { stepName: item.stepName },
+        } as CreateEventRequest,
+        {
+          eventType: 'step_completed',
+          correlationId: item.correlationId,
+          eventData: {
+            result: await dehydrateStepReturnValue(
+              item.args[0],
+              RUN_ID,
+              undefined
+            ),
+          },
+        },
+      ];
+    case 'wait':
+      return [
+        {
+          eventType: 'wait_completed',
+          correlationId: item.correlationId,
+          eventData: { resumeAt: item.resumeAt },
+        } as CreateEventRequest,
+      ];
+    case 'hook':
+      return [
+        {
+          eventType: 'hook_received',
+          correlationId: item.correlationId,
+          eventData: {
+            payload: await dehydrateStepReturnValue(
+              { done: true },
+              RUN_ID,
+              undefined
+            ),
+          },
+        } as CreateEventRequest,
+      ];
+    default:
+      return [];
+  }
+}
+
+/**
+ * Drives a workflow to completion against an in-memory event log, the way a
+ * world does: every round replays the full log in a fresh VM, records the
+ * events for whatever the suspension asks for, and completes what is pending.
+ * Returns the final output plus the log, so the caller can replay it again.
+ */
+async function drive(code: string) {
+  const startedAt = new Date('2025-01-01T00:00:00Z');
+  const run: WorkflowRun = {
+    runId: RUN_ID,
+    workflowName: 'workflow',
+    status: 'running',
+    attributes: {},
+    createdAt: startedAt,
+    updatedAt: startedAt,
+    startedAt,
+    deploymentId: 'test-deployment',
+    input: await dehydrateWorkflowArguments([], RUN_ID, undefined),
+  };
+  const events: Event[] = [];
+  const append = (request: CreateEventRequest) => {
+    events.push({
+      ...request,
+      eventId: `evnt_${String(events.length + 1).padStart(26, '0')}`,
+      runId: RUN_ID,
+      createdAt: new Date(+startedAt + (events.length + 1) * 100),
+      specVersion: 3,
+    } as Event);
+  };
+  const replay = async (cache = new ReplayPayloadCache(undefined)) => {
+    try {
+      const output = await runWorkflow(
+        code,
+        run,
+        [...events],
+        undefined,
+        cache
+      );
+      return { output: await hydrateWorkflowReturnValue(output, RUN_ID) };
+    } catch (error) {
+      if (!(error instanceof WorkflowSuspension)) throw error;
+      return { items: error.items };
+    }
+  };
+
+  const recorded = new Set<string>();
+  let pending: QueueItem[] = [];
+  for (let round = 0; round < 20; round++) {
+    const result = await replay();
+    if ('output' in result) return { output: result.output, replay };
+    const fresh = result.items.filter(
+      (item) => !recorded.has(item.correlationId)
+    );
+    for (const item of fresh) {
+      recorded.add(item.correlationId);
+      for (const event of await creationEvents(item)) append(event);
+    }
+    for (const item of pending) {
+      for (const event of await completionEvents(item)) append(event);
+    }
+    pending = fresh.filter((item) => item.type !== 'attribute');
+  }
+  throw new Error('Workflow did not complete');
+}
+
+describe('setAttributes() delivery ordering', () => {
+  describe.each(['1', '0'])('WORKFLOW_LOG_ORDER_DRAWS=%s', (logOrderDraws) => {
+    it.each([
+      0, 3, 12, 48,
+    ])('does not swap step results between branches (hops=%i)', async (hops) => {
+      vi.stubEnv('WORKFLOW_LOG_ORDER_DRAWS', logOrderDraws);
+      const { output, replay } = await drive(SWAP_WORKFLOW(hops));
+      expect(output).toBe('after-attribute|other-branch');
+      // Fresh replays of the committed log, with cold and warm caches, must
+      // keep handing each branch its own step result.
+      const cache = new ReplayPayloadCache(undefined);
+      for (let attempt = 0; attempt < 3; attempt++) {
+        expect(await replay(cache)).toEqual({
+          output: 'after-attribute|other-branch',
+        });
+      }
+    });
+
+    it.each([
+      0, 3, 12, 48,
+    ])('does not diverge when racing a hook registration (hops=%i)', async (hops) => {
+      vi.stubEnv('WORKFLOW_LOG_ORDER_DRAWS', logOrderDraws);
+      const { output, replay } = await drive(HOOK_WORKFLOW(hops));
+      expect(output).toBe('completed');
+      const cache = new ReplayPayloadCache(undefined);
+      for (let attempt = 0; attempt < 3; attempt++) {
+        expect(await replay(cache)).toEqual({ output: 'completed' });
+      }
+    });
+  });
+});