[core] setAttributes() resolution bypasses the delivery barriers, so a branch awaiting it can resume out of log order and draw another branch's correlation id (replay divergence, or step results delivered to the wrong branch) #4770

Summary

When a workflow branch calls await setAttributes(...) while another branch of the same run is in flight, the replay engine can resume the two branches in a different order than the event log recorded. Every entity a replay creates draws from one monotonic correlation-id sequence, so a changed resumption order assigns the next ids to the wrong entities. Two symptoms, depending on what the other branch draws next:

1. The other branch creates a hook (or a wait): the attribute branch's next step takes the id the log assigned to hook_created. The step consumer cannot consume that event and the run fails with REPLAY_DIVERGENCE after the recovery replays are exhausted.
2. The other branch calls the same step function: the step consumer accepts the other call's step_created (step name matches, arguments are not compared) and both branches receive the other call's return value. The run completes with wrong data and nothing is logged.

The control workflows without the setAttributes() call never fail. Registering the attr_set delivery with the existing delivery barriers (the same change #4215 made for hook registration outcomes) brings both failure counts to zero in every run.

Versions

• workflow 5.1.0
• @workflow/core 5.1.0 (tag @workflow/core@5.1.0, commit 2f9edf31ca2856527712958580ec5f7f8f3f3587); main still has the same code in attribute-dispatcher.ts
• @workflow/vitest 5.0.2, @workflow/world 5.0.2, @workflow/world-local 5.0.2
• vitest 5.0.3

Environment

• macOS (Darwin 25.6.0), Node v26.8.1, npm 11.19.0
• Local world through @workflow/vitest (in-process handlers, .workflow-data on disk)
• Reproduces with the default log-order draws mode and with WORKFLOW_LOG_ORDER_DRAWS=0

Minimal reproduction

Four files. Each test starts 32 runs concurrently and counts how many end wrong. The two *Control workflows are identical except for the setAttributes() line.

package.json

{
  "name": "workflow-setattributes-repro",
  "private": true,
  "type": "module",
  "scripts": { "test": "vitest run" },
  "devDependencies": {
    "@workflow/core": "5.1.0",
    "@workflow/vitest": "5.0.2",
    "vitest": "5.0.3",
    "workflow": "5.1.0"
  }
}

vitest.config.ts

import { workflow } from "@workflow/vitest";
import { defineConfig } from "vitest/config";

export default defineConfig({
  plugins: [...workflow()],
  test: {
    include: ["workflows/**/*.test.ts"],
    environment: "node",
    testTimeout: 120_000,
  },
});

workflows/attr-vs-hook.ts (symptom 1)

import { createHook, setAttributes } from "workflow";

async function quickStep(label: string) {
  "use step";
  return label;
}

async function delayedStep(ms: number) {
  "use step";
  await new Promise((resolve) => setTimeout(resolve, ms));
  return ms;
}

export async function attrVsHook(token: string, delayMs: number, hops: number) {
  "use workflow";
  const attributeBranch = async () => {
    await quickStep("before-attribute");
    await setAttributes({ phase: "attribute-branch" });
    await quickStep("after-attribute");
  };
  const hookBranch = async () => {
    await delayedStep(delayMs);
    for (let hop = 0; hop < hops; hop += 1) await Promise.resolve();
    using hook = createHook<{ done: boolean }>({ token });
    await hook;
  };
  await Promise.all([attributeBranch(), hookBranch()]);
  return "completed";
}

export async function attrVsHookControl(token: string, delayMs: number, hops: number) {
  "use workflow";
  const attributeBranch = async () => {
    await quickStep("before-attribute");
    await quickStep("after-attribute");
  };
  const hookBranch = async () => {
    await delayedStep(delayMs);
    for (let hop = 0; hop < hops; hop += 1) await Promise.resolve();
    using hook = createHook<{ done: boolean }>({ token });
    await hook;
  };
  await Promise.all([attributeBranch(), hookBranch()]);
  return "completed";
}

workflows/attr-swap.ts (symptom 2; the return value must always be after-attribute|other-branch)

import { setAttributes, sleep } from "workflow";

async function quickStep(label: string) {
  "use step";
  return label;
}

async function delayedStep(ms: number) {
  "use step";
  await new Promise((resolve) => setTimeout(resolve, ms));
  return ms;
}

export async function attrSwap(delayMs: number, hops: number) {
  "use workflow";
  const attributeBranch = async () => {
    await quickStep("before-attribute");
    await setAttributes({ phase: "attribute-branch" });
    return await quickStep("after-attribute");
  };
  const sleepBranch = async () => {
    await delayedStep(delayMs);
    for (let hop = 0; hop < hops; hop += 1) 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}`;
}

export async function attrSwapControl(delayMs: number, hops: number) {
  "use workflow";
  const attributeBranch = async () => {
    await quickStep("before-attribute");
    return await quickStep("after-attribute");
  };
  const sleepBranch = async () => {
    await delayedStep(delayMs);
    for (let hop = 0; hop < hops; hop += 1) 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}`;
}

workflows/repro.test.ts

import { waitForHook } from "@workflow/vitest";
import { expect, test } from "vitest";
import { resumeHook, start } from "workflow/api";

import { attrSwap, attrSwapControl } from "./attr-swap";
import { attrVsHook, attrVsHookControl } from "./attr-vs-hook";

const RUNS = 32;

async function hookFailures(workflow: typeof attrVsHook, label: string) {
  const outcomes = await Promise.all(
    Array.from({ length: RUNS }, async (_, index) => {
      const token = `${label}-${Date.now()}-${index}`;
      const run = await start(workflow, [token, index % 4, index * 3]);
      try {
        const hook = await waitForHook(run, { token, timeout: 15_000 });
        await resumeHook(hook.token, { done: true });
        return await run.returnValue;
      } catch (error) {
        return error instanceof Error ? error.message : String(error);
      }
    })
  );
  const failures = outcomes.filter((o) => o !== "completed");
  console.log(label, `${failures.length}/${RUNS} failed`, failures[0] ?? "");
  return failures.length;
}

async function swapCount(workflow: typeof attrSwap, label: string) {
  const outcomes = await Promise.all(
    Array.from({ length: RUNS }, async (_, index) => {
      const run = await start(workflow, [index % 4, index * 3]);
      try {
        return await run.returnValue;
      } catch (error) {
        return error instanceof Error ? error.message : String(error);
      }
    })
  );
  const wrong = outcomes.filter((o) => o !== "after-attribute|other-branch");
  console.log(label, `${wrong.length}/${RUNS} wrong`, [...new Set(wrong)]);
  return wrong.length;
}

test("attrVsHook: setAttributes() racing a hook registration", async () => {
  expect(await hookFailures(attrVsHook, "attrVsHook")).toBe(0);
});

test("attrVsHook control: no setAttributes()", async () => {
  expect(await hookFailures(attrVsHookControl, "attrVsHookControl")).toBe(0);
});

test("attrSwap: step results delivered to the wrong branch", async () => {
  expect(await swapCount(attrSwap, "attrSwap")).toBe(0);
});

test("attrSwap control: no setAttributes()", async () => {
  expect(await swapCount(attrSwapControl, "attrSwapControl")).toBe(0);
});

Run:

npm install
npx vitest run
WORKFLOW_LOG_ORDER_DRAWS=0 npx vitest run

Measured results

Each cell is "wrong runs / started runs" summed over the listed number of suite executions, with the per-execution counts in parentheses. Every execution starts 32 runs per workflow.

"Patched core" is 5.1.0 with the diff from the "Suggested fix" section applied to the built output of @workflow/core inside the reproduction project.

The attrVsHook failure, as reported by the runtime (run id and ULID body shortened):

code   REPLAY_DIVERGENCE
error  Replay could not consume event: eventType=hook_created, correlationId=hook_01M4…42VH,
       eventId=evnt_00000000000000000000000010.
       pending at this id: step step//./workflows/attr-vs-hook//quickStep (step_01M4…42VH).
       consumer: index=9, length=14, parked=0, lastConsumed=evnt_00000000000000000000000009

The ULID body is identical: the attribute branch's quickStep("after-attribute") drew the id that the log had assigned to the hook branch's createHook(). After three recovery replays the run fails with "Workflow replay diverged 4 times after 3 recovery replays".

The event log of a wrong attrSwap run (abbreviated; step inputs decoded):

007 step_completed   …HP  quickStep    "before-attribute"
008 step_completed   …HQ  delayedStep  0
009 attr_set         …HR  [{"key":"phase","value":"attribute-branch"}]
010 step_created     …HS  quickStep    "other-branch"
011 step_created     …HT  quickStep    "after-attribute"
014 step_completed   …HS  "other-branch"
015 step_completed   …HT  "after-attribute"
016 wait_created     …HV
017 wait_completed   …HV
018 run_completed         "other-branch|after-attribute"

The replay that wrote events 010 and 011 resumed the sleep branch first (…HS is other-branch). The final replay, after wait_completed, resumed the attribute branch first: quickStep("after-attribute") drew …HS, consumed step_created/step_completed for …HS and returned "other-branch", and the sleep branch's quickStep("other-branch") drew …HT and returned "after-attribute". Across the 32 runs of one execution, wrong results occurred for both log orders (13 where the log had other-branch first, 7 where it had after-attribute first), and correct results also occurred for both (5 and 7). The control, whose logs show the same mix of orders, is correct in all 32. The resumption order with setAttributes() in the race is therefore not a function of the log.

Root cause

Source references are to tag @workflow/core@5.1.0 under packages/core/src/.

• workflow/attribute-dispatcher.ts lines 79 to 83: when the matching attr_set event is consumed, createSetAttributes settles the promise with ctx.promiseQueue = ctx.promiseQueue.then(() => resolve()). It does not call registerDeliveryBarrier or awaitEarlierDeliveries.
• Every other branch-deciding delivery does. step.ts lines 286 to 331 (step_completed) and 175 to 224 (step_failed), workflow/sleep.ts lines 99 to 111, workflow/hook.ts lines 183 to 200 (registration outcomes, since fix(core): preserve replay order for hook registration outcomes #4215) and 389 onwards (payloads), and workflow/abort-controller.ts lines 205 to 210 all register a barrier at ctx.eventsConsumer.eventIndex when the event is consumed and resolve from a detached continuation after awaitEarlierDeliveries.
• private.ts lines 587 to 634 (awaitEarlierDeliveries): a gated delivery waits for every earlier barrier it defers behind and then yields a macrotask (setTimeout(resolve, 0) at line 633), or runs quiesceEarlierCascades under log-order draws. DEFER_BEHIND at lines 311 to 315 only knows the kinds hook, wait and step; DeliveryKind at line 254 has no attribute kind, so no later delivery can defer behind an attribute resolution either.
• events-consumer.ts lines 332 to 345 (subscribe) and the PARKABLE_EVENT_TYPES set (lines 88 to 94, which includes attr_set): the walk parks attr_set when the body has not made the call yet and keeps consuming the step and hook events after it in the same drain window. When the branch finally calls setAttributes(), the parked event is claimed on process.nextTick and resolved on the next free promiseQueue slot, that is, on microtasks.
• private.ts lines 154 to 159: generateUlid is one monotonic sequence per run, "so an id is an ordinal over the whole run and both replays of a run must draw in the same order".

Putting these together: after the step result that precedes setAttributes() in the branch is delivered, the attribute branch resumes as soon as promiseQueue is free, while a step result that is earlier in the log than attr_set but belongs to the other branch is still parked behind its barrier's macrotask yield. Which branch draws the next correlation id first depends on how long the pending payload hydrations on promiseQueue take relative to a 0 ms timer, not on log position. Different replays of the same prefix therefore draw in different orders. If the two draws are different kinds the replay diverges; if both are steps with the same name, the consumer accepts the other call's events and the branches swap results.

Per-kind correlation-id sequences (#3301) would remove symptom 1 but not symptom 2, where both racing draws are steps. The fix needs to order the attribute delivery itself.

Suggested fix

Treat the attr_set delivery like a hook registration outcome: register a barrier at its log index, add an attribute kind to DEFER_BEHIND, and settle off the serial queue once the earlier deliveries have been handed to the workflow. This is the diff that produced the "Patched core" column above (validated against the built output of 5.1.0; the TypeScript below is the same change applied to the source).

--- a/packages/core/src/private.ts
+++ b/packages/core/src/private.ts
@@ -251,7 +251,7 @@
 }

 /** 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;
@@ -309,9 +309,10 @@
  * 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'],
 };

 /**
--- a/packages/core/src/workflow/attribute-dispatcher.ts
+++ b/packages/core/src/workflow/attribute-dispatcher.ts
@@ -8,6 +8,8 @@
 import { EventConsumerResult } from '../events-consumer.js';
 import type { AttributeInvocationQueueItem } from '../global.js';
 import {
+  awaitEarlierDeliveries,
+  registerDeliveryBarrier,
   scheduleWorkflowSuspension,
   type WorkflowOrchestratorContext,
 } from '../private.js';
@@ -77,8 +79,27 @@
       }

       ctx.invocationsQueue.delete(correlationId);
+      // Resolving setAttributes() resumes a branch that may draw the next
+      // correlation ID, so it is branch-deciding like a step result or a hook
+      // registration outcome. Register it with the delivery barriers at its
+      // log index and settle only after every earlier delivery it defers
+      // behind has been handed to the workflow. The gate is awaited off the
+      // serial queue, like step.ts and hook.ts do, because an earlier delivery
+      // may need that queue to drain before it can resolve.
+      const eventIndex = ctx.eventsConsumer.eventIndex;
+      const barrier = registerDeliveryBarrier(ctx, eventIndex, 'attribute', {
+        deliveredAt: +event.createdAt,
+      });
+      const earlierDelivered = awaitEarlierDeliveries(
+        ctx,
+        eventIndex,
+        'attribute'
+      );
       ctx.promiseQueue = ctx.promiseQueue.then(() => {
-        resolve();
+        void earlierDelivered.then(() => {
+          barrier.markDelivered();
+          resolve();
+        });
       });
       return EventConsumerResult.Finished;
     });

Notes on the change:

• The barrier is armed and its delivery chain is attached at consumption time, so the invariant documented on registerDeliveryBarrier (an armed barrier is always paired with an unconditional delivery chain) holds and the idle safety net is not needed for it.
• eventIndex is read inside the subscriber, so a parked attr_set claimed through drainParked gets its original log position.
• deliveredAt: +event.createdAt advances the workflow clock to the attr_set event's time, as every other delivery does. If attribute writes should not move the clock, the option could be made optional for this kind.
• Making every kind defer behind attribute is what keeps a later step or hook delivery from overtaking a branch that is still resuming from setAttributes(). With only attribute gating on the others, the reverse overtake would remain possible.

Workaround

Do not await setAttributes() while another branch of the same run is in flight. Call it before the fan-out or after the join:

await quickStep("before-attribute");
await setAttributes({ phase: "attribute-branch" });
const [a, b] = await Promise.all([quickStep("after-attribute"), sleepBranch()]);

Measured with the stock 5.1.0 core: 0/64 wrong over two executions of 32 runs, against 76/128 for the original attrSwap shape.