From 426fcc181d5ed84da63eaec48bc11fbffd79c37c Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Tue, 13 Oct 2026 20:45:51 +0000 Subject: [PATCH 1/2] [core] Add failing regression test for setAttributes() delivery ordering (#4770) Co-Authored-By: Claude Signed-off-by: Pranay Prakash <1797812+pranaygp@users.noreply.github.com> Co-Authored-By: Pranay Prakash <1797812+pranaygp@users.noreply.github.com> --- .../src/set-attributes-delivery-order.test.ts | 255 ++++++++++++++++++ 1 file changed, 255 insertions(+) create mode 100644 packages/core/src/set-attributes-delivery-order.test.ts diff --git a/packages/core/src/set-attributes-delivery-order.test.ts b/packages/core/src/set-attributes-delivery-order.test.ts new file mode 100644 index 0000000000..2a72dea532 --- /dev/null +++ b/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 { + 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 { + 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(); + 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' }); + } + }); + }); +}); From c005102f0044c67e88542a65e5bdbedd7a83c3cc Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Tue, 13 Oct 2026 20:54:07 +0000 Subject: [PATCH 2/2] [core] Order setAttributes() resolution with the delivery barriers (#4770) Register the attr_set acknowledgement as an 'attribute' delivery barrier at its log index and resolve setAttributes() only after every earlier delivery it defers behind has been handed to the workflow. Every other kind defers behind an earlier attribute too. The clock is left untouched, as before. Co-Authored-By: Claude Signed-off-by: Pranay Prakash <1797812+pranaygp@users.noreply.github.com> Co-Authored-By: Pranay Prakash <1797812+pranaygp@users.noreply.github.com> --- .changeset/attr-set-delivery-order.md | 5 +++ packages/core/README.md | 4 +- packages/core/src/private.ts | 40 ++++++++++++------- .../core/src/workflow/attribute-dispatcher.ts | 32 ++++++++++++++- 4 files changed, 64 insertions(+), 17 deletions(-) create mode 100644 .changeset/attr-set-delivery-order.md diff --git a/.changeset/attr-set-delivery-order.md b/.changeset/attr-set-delivery-order.md new file mode 100644 index 0000000000..830bdb89b3 --- /dev/null +++ b/.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 diff --git a/packages/core/README.md b/packages/core/README.md index e063bebc21..81bff30d45 100644 --- a/packages/core/README.md +++ b/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. diff --git a/packages/core/src/private.ts b/packages/core/src/private.ts index 21c0186d1c..5835f2bbfd 100644 --- a/packages/core/src/private.ts +++ b/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 = { - 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; }; diff --git a/packages/core/src/workflow/attribute-dispatcher.ts b/packages/core/src/workflow/attribute-dispatcher.ts index 1207e06151..c5f34a63d3 100644 --- a/packages/core/src/workflow/attribute-dispatcher.ts +++ b/packages/core/src/workflow/attribute-dispatcher.ts @@ -8,6 +8,8 @@ import { import { EventConsumerResult } from '../events-consumer.js'; import type { AttributeInvocationQueueItem } from '../global.js'; import { + awaitEarlierDeliveries, + registerDeliveryBarrier, scheduleWorkflowSuspension, type WorkflowOrchestratorContext, } from '../private.js'; @@ -77,9 +79,35 @@ export function createSetAttributes(ctx: WorkflowOrchestratorContext) { } ctx.invocationsQueue.delete(correlationId); - ctx.promiseQueue = ctx.promiseQueue.then(() => { - resolve(); + // Resolving setAttributes() resumes a branch that may draw the next + // correlation ID, so it is branch-deciding like a step result: register + // it with the delivery barriers at its log index and resolve only once + // every earlier delivery it defers behind has been handed to the + // workflow. Otherwise it resolves on microtasks while an earlier step + // result is still parked behind its barrier, and the two branches draw + // their next IDs in an order that differs between replays (#4770). + // The ordering is captured here, at consumption time, for the same + // reason as in step.ts and sleep.ts. + const eventIndex = ctx.eventsConsumer.eventIndex; + const barrier = registerDeliveryBarrier(ctx, eventIndex, 'attribute', { + // Attribute writes have never moved the workflow clock. + deliveredAt: null, }); + const earlierDelivered = awaitEarlierDeliveries( + ctx, + eventIndex, + 'attribute' + ); + // Detached from the serial `promiseQueue` (anchored to its tail so + // queued hydration still runs in log order): an earlier delivery may + // need the queue to drain before it can resolve, so blocking a queue + // slot on it would deadlock. + void ctx.promiseQueue + .then(() => earlierDelivered) + .then(() => { + barrier.markDelivered(); + resolve(); + }); return EventConsumerResult.Finished; });