From bf3f1ff203ab46e04467def5db00cca6c7c08b64 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Mon, 3 Aug 2026 12:11:21 -0700 Subject: [PATCH 1/4] Derive correlation ids from per-kind sequences Correlation ids were the Nth draw of one monotonic ULID sequence per run, shared by steps, waits, hooks, attribute writes, abort controllers and stream ids. Every id was an ordinal over the whole run, so one extra draw of any kind renumbered every entity of every kind after it, and two replays that disagreed about a single sleep minted different ids for every step that followed. Each kind now draws from its own independent sequence, behind WORKFLOW_PER_KIND_CORRELATION_IDS=1. Also routes STABLE_ULID through the same generator. Serialization minted stream ids with no seed time, and monotonicFactory's increment branch returns encodeTime(lastTime), so one such call latched the host wall clock and every later id in the run carried a per-replay timestamp instead of fixedTimestamp. That fix applies in both modes. Co-Authored-By: Claude Opus 5 --- .changeset/per-kind-correlation-ids.md | 6 + .../docs/v5/configuration/runtime-tuning.mdx | 9 + packages/core/src/abort-consistency.test.ts | 11 +- packages/core/src/abort-controller.test.ts | 11 +- .../core/src/abort-replay-ordering.test.ts | 11 +- .../async-deserialization-ordering.test.ts | 11 +- packages/core/src/correlation-id.test.ts | 154 +++++++++++++ packages/core/src/correlation-id.ts | 214 ++++++++++++++++++ .../src/delivery-barrier-coverage.test.ts | 11 +- .../core/src/hook-sleep-interaction.test.ts | 11 +- packages/core/src/private.ts | 8 +- .../core/src/step-delivery-hop-count.test.ts | 11 +- .../core/src/step-delivery-ordering.test.ts | 11 +- .../src/step-hydration-memoization.test.ts | 11 +- packages/core/src/step.test.ts | 11 +- packages/core/src/step.ts | 2 +- packages/core/src/workflow.test.ts | 4 +- packages/core/src/workflow.ts | 28 ++- .../core/src/workflow/abort-controller.ts | 8 +- .../core/src/workflow/attribute-dispatcher.ts | 2 +- packages/core/src/workflow/hook.test.ts | 11 +- packages/core/src/workflow/hook.ts | 2 +- packages/core/src/workflow/sleep.test.ts | 11 +- packages/core/src/workflow/sleep.ts | 2 +- workbench/nextjs-turbopack/vercel.json | 3 +- 25 files changed, 546 insertions(+), 28 deletions(-) create mode 100644 .changeset/per-kind-correlation-ids.md create mode 100644 packages/core/src/correlation-id.test.ts create mode 100644 packages/core/src/correlation-id.ts diff --git a/.changeset/per-kind-correlation-ids.md b/.changeset/per-kind-correlation-ids.md new file mode 100644 index 0000000000..34906fabab --- /dev/null +++ b/.changeset/per-kind-correlation-ids.md @@ -0,0 +1,6 @@ +--- +'@workflow/core': patch +'workflow': patch +--- + +Give each kind of entity a workflow creates its own sequence of correlation IDs, so an extra hook or sleep no longer renumbers every step after it diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index c460380f88..0da5273e47 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -75,6 +75,15 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL - Delay before a re-invocation caused by a rejected event creation. - Unlike an in-process restart, which re-reads immediately, a re-invocation only happens once the in-process budget failed to catch up — so the delay gives the other writers a moment to quiesce. +### `WORKFLOW_PER_KIND_CORRELATION_IDS` + +- Default: disabled +- Experimental. Gives each kind of entity a workflow creates — steps, waits, hooks, attribute writes, abort controllers, stream IDs — its own sequence of correlation IDs. +- With one sequence shared by every kind, an ID is an ordinal over the whole run, so a single extra draw of any kind shifts every ID after it. Two concurrent replays of the same run that disagree about one `sleep()` then assign different IDs to every step that follows, and each writes events the other can neither match nor consume, which fails the run with `CORRUPTED_EVENT_LOG`. Per-kind sequences confine that to the kind that actually differs. +- IDs remain ordered within a kind, so hooks are still listed in creation order. +- Correlation IDs of runs created before the setting changed are not affected: a run keeps replaying on the deployment it started on. On platforms without that guarantee, only enable it while no runs are in flight — a replay that switches schemes mid-run cannot consume its own earlier events. +- Set `1` to enable. + ## Inline execution ### `WORKFLOW_V2_TIMEOUT_MS` diff --git a/packages/core/src/abort-consistency.test.ts b/packages/core/src/abort-consistency.test.ts index fcd5ed09f7..3cf2dd1896 100644 --- a/packages/core/src/abort-consistency.test.ts +++ b/packages/core/src/abort-consistency.test.ts @@ -10,6 +10,10 @@ import type { Event, WorkflowRun } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from './correlation-id.js'; import { describe, expect, it, vi } from 'vitest'; import { EventsConsumer } from './events-consumer.js'; import type { WorkflowSuspension } from './global.js'; @@ -44,7 +48,12 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { getPromiseQueue: () => Promise.resolve(), }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateCorrelationId: createCorrelationIdGenerator({ + seed: 'test', + fixedTimestamp: workflowStartedAt, + positional: () => ulid(workflowStartedAt), + perKind: isPerKindCorrelationIdsEnabled(), + }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/abort-controller.test.ts b/packages/core/src/abort-controller.test.ts index ee839b19d6..0fd39fcd66 100644 --- a/packages/core/src/abort-controller.test.ts +++ b/packages/core/src/abort-controller.test.ts @@ -11,6 +11,10 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from './correlation-id.js'; import { describe, expect, it, vi } from 'vitest'; import { DEFERRED_CHECK_DELAY_MS, EventsConsumer } from './events-consumer.js'; import type { WorkflowOrchestratorContext } from './private.js'; @@ -42,7 +46,12 @@ function setupWorkflowContext( getPromiseQueue: () => ctx.promiseQueue, }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateCorrelationId: createCorrelationIdGenerator({ + seed: 'test', + fixedTimestamp: workflowStartedAt, + positional: () => ulid(workflowStartedAt), + perKind: isPerKindCorrelationIdsEnabled(), + }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/abort-replay-ordering.test.ts b/packages/core/src/abort-replay-ordering.test.ts index 8ddf4aec03..e926eb5adb 100644 --- a/packages/core/src/abort-replay-ordering.test.ts +++ b/packages/core/src/abort-replay-ordering.test.ts @@ -26,6 +26,10 @@ import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from './correlation-id.js'; import { describe, expect, it, vi } from 'vitest'; import { EventsConsumer } from './events-consumer.js'; import { @@ -77,7 +81,12 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { getPromiseQueue: () => ctx.promiseQueue, }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateCorrelationId: createCorrelationIdGenerator({ + seed: 'test', + fixedTimestamp: workflowStartedAt, + positional: () => ulid(workflowStartedAt), + perKind: isPerKindCorrelationIdsEnabled(), + }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/async-deserialization-ordering.test.ts b/packages/core/src/async-deserialization-ordering.test.ts index 7a1ff1d346..da5416c911 100644 --- a/packages/core/src/async-deserialization-ordering.test.ts +++ b/packages/core/src/async-deserialization-ordering.test.ts @@ -2,6 +2,10 @@ import { FatalError } from '@workflow/errors'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from './correlation-id.js'; import { afterEach, beforeAll, describe, expect, it, vi } from 'vitest'; import { registerSerializationClass } from './class-serialization.js'; import { EventsConsumer } from './events-consumer.js'; @@ -57,7 +61,12 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { getPromiseQueue: () => Promise.resolve(), }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateCorrelationId: createCorrelationIdGenerator({ + seed: 'test', + fixedTimestamp: workflowStartedAt, + positional: () => ulid(workflowStartedAt), + perKind: isPerKindCorrelationIdsEnabled(), + }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/correlation-id.test.ts b/packages/core/src/correlation-id.test.ts new file mode 100644 index 0000000000..db95c65df4 --- /dev/null +++ b/packages/core/src/correlation-id.test.ts @@ -0,0 +1,154 @@ +import { decodeTime, monotonicFactory } from 'ulid'; +import { describe, expect, it } from 'vitest'; +import { + CORRELATION_ID_LENGTH, + type CorrelationIdKind, + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from './correlation-id.js'; + +const SEED = 'wrun_abc:myWorkflow:dpl_123'; +const FIXED_TIMESTAMP = 1753481739458; + +function makeGenerator( + overrides: { seed?: string; fixedTimestamp?: number; perKind?: boolean } = {} +) { + // A stand-in for the run's shared sequence. Seeded so the positional mode is + // reproducible across the two generators a replay-stability test builds. + let counter = 0; + const ulid = monotonicFactory(() => { + counter = (counter * 1103515245 + 12345) % 2147483648; + return counter / 2147483648; + }); + const fixedTimestamp = overrides.fixedTimestamp ?? FIXED_TIMESTAMP; + return createCorrelationIdGenerator({ + seed: overrides.seed ?? SEED, + fixedTimestamp, + positional: () => ulid(fixedTimestamp), + perKind: overrides.perKind ?? true, + }); +} + +const KINDS: CorrelationIdKind[] = [ + 'step', + 'wait', + 'hook', + 'attr', + 'abort', + 'abortHook', + 'stream', +]; + +describe('createCorrelationIdGenerator', () => { + it('mints syntactically valid ULIDs carrying fixedTimestamp', () => { + const generate = makeGenerator(); + for (const kind of KINDS) { + const id = generate(kind); + expect(id).toHaveLength(CORRELATION_ID_LENGTH); + expect(id).toMatch(/^[0-9A-HJKMNP-TV-Z]+$/); + expect(decodeTime(id)).toBe(FIXED_TIMESTAMP); + } + }); + + it('is deterministic across replays of the same run', () => { + const first = makeGenerator(); + const second = makeGenerator(); + const draw = (generate: (kind: CorrelationIdKind) => string) => [ + generate('step'), + generate('step'), + generate('wait'), + generate('step'), + generate('hook'), + ]; + expect(draw(first)).toEqual(draw(second)); + }); + + it('mints different ids for different runs', () => { + const first = makeGenerator({ seed: 'wrun_one:w:dpl' }); + const second = makeGenerator({ seed: 'wrun_two:w:dpl' }); + expect(first('step')).not.toBe(second('step')); + }); + + it('gives every kind its own starting point', () => { + const generate = makeGenerator(); + const ids = KINDS.map((kind) => generate(kind)); + expect(new Set(ids).size).toBe(KINDS.length); + }); + + it('increases monotonically within a kind', () => { + const generate = makeGenerator(); + const ids = [generate('hook'), generate('hook'), generate('hook')]; + expect(ids).toEqual([...ids].sort()); + expect(new Set(ids).size).toBe(ids.length); + }); + + it('does not renumber one kind when another draws more often', () => { + // The whole point of per-kind sources: two replays that disagree about how + // many hooks, sleeps or streams were created still agree about which id + // belongs to the Nth step. + const withoutExtras = makeGenerator(); + const withExtras = makeGenerator(); + + const steps = [withoutExtras('step'), withoutExtras('step')]; + + withExtras('hook'); + const interleaved = [withExtras('step')]; + withExtras('wait'); + withExtras('stream'); + withExtras('attr'); + withExtras('abort'); + interleaved.push(withExtras('step')); + + expect(interleaved).toEqual(steps); + }); + + it('keeps abort controllers from renumbering user hooks', () => { + const withoutController = makeGenerator(); + const withController = makeGenerator(); + withController('abort'); + withController('abortHook'); + expect(withController('hook')).toBe(withoutController('hook')); + }); + + it('keeps every id on fixedTimestamp in both modes', () => { + // `monotonicFactory` returns `encodeTime(lastTime)` on its increment branch, + // so a single draw that omits the seed time latches the host wall clock and + // every later id in the run carries a timestamp that differs per replay. + // Stream ids used to be drawn that way. + for (const perKind of [true, false]) { + const generate = makeGenerator({ perKind }); + for (const kind of ['stream', 'stream', 'step', 'hook'] as const) { + expect(decodeTime(generate(kind))).toBe(FIXED_TIMESTAMP); + } + } + }); + + it('ignores the kind when per-kind sources are disabled', () => { + const generate = makeGenerator({ perKind: false }); + const shared = makeGenerator({ perKind: false }); + // Positional mode is one sequence for the whole run, so drawing `wait` + // consumes the ordinal the next `step` would otherwise have had. + expect(generate('step')).toBe(shared('step')); + expect(generate('wait')).toBe(shared('step')); + }); +}); + +describe('isPerKindCorrelationIdsEnabled', () => { + it('reads WORKFLOW_PER_KIND_CORRELATION_IDS, defaulting to disabled', () => { + const original = process.env.WORKFLOW_PER_KIND_CORRELATION_IDS; + try { + delete process.env.WORKFLOW_PER_KIND_CORRELATION_IDS; + expect(isPerKindCorrelationIdsEnabled()).toBe(false); + process.env.WORKFLOW_PER_KIND_CORRELATION_IDS = '1'; + expect(isPerKindCorrelationIdsEnabled()).toBe(true); + process.env.WORKFLOW_PER_KIND_CORRELATION_IDS = '0'; + expect(isPerKindCorrelationIdsEnabled()).toBe(false); + } finally { + if (original === undefined) { + delete process.env.WORKFLOW_PER_KIND_CORRELATION_IDS; + } else { + process.env.WORKFLOW_PER_KIND_CORRELATION_IDS = original; + } + } + }); +}); diff --git a/packages/core/src/correlation-id.ts b/packages/core/src/correlation-id.ts new file mode 100644 index 0000000000..d3844ff5ed --- /dev/null +++ b/packages/core/src/correlation-id.ts @@ -0,0 +1,214 @@ +import { encodeTime, incrementBase32 } from 'ulid'; + +/** + * Correlation-id generation for the entity families a replay can create. + * + * Correlation ids are minted by the workflow VM and are the server's identity + * gate: a conditional create on the id is what makes a duplicate write from a + * second live replay idempotent instead of additive. That only works if two + * replays of the same run mint the same id for the same entity. + * + * Historically every id was the Nth draw of *one* monotonic ULID sequence per + * run, shared by steps, waits, hooks, attribute writes, abort controllers and + * stream ids alike. Every id was therefore an ordinal over the whole run, and a + * single extra draw of any kind renumbered every entity of every kind after it. + * Two replays that agreed about every step but disagreed about one `sleep()` + * would mint different ids for all subsequent steps, so their writes appended + * side by side instead of colliding, and the settled log ended up holding two + * names for one logical step. Only one of them can be consumed on the next + * replay; the other is fatal (`onUnconsumedEvent`). + * + * Per-kind sources narrow that coupling to one kind at a time: each family + * draws from its own independent incrementing sequence, so a disagreement about + * how many hooks or sleeps were created no longer renames steps. + * + * Ids stay syntactically valid ULIDs (10 Crockford characters of + * `fixedTimestamp` plus 16 of body), because correlation ids are validated as + * prefixed 26-char ULIDs by the backend, and they stay monotonic *within* a + * kind, because `hooks.list` is ordered by hook id. + * + * This does not make ids independent of *ordinal position within their own + * kind*: two replays that disagree about how many steps ran still mint + * different ids for the next step. That is a narrower failure than the shared + * sequence's, not an eliminated one. + */ + +/** Entity families that draw correlation ids, each from its own sequence. */ +export type CorrelationIdKind = + /** `step_` ids, one per step invocation. */ + | 'step' + /** `wait_` ids, one per `sleep()`. */ + | 'wait' + /** `hook_` ids for hooks created by workflow code. */ + | 'hook' + /** `attr_` ids, one per attribute write. */ + | 'attr' + /** + * The abort controller's own id, which becomes its stream name and hook + * token. Separate from `hook` so constructing an abort controller does not + * renumber later user hooks. + */ + | 'abort' + /** `hook_` ids for the internal system hook backing an abort controller. */ + | 'abortHook' + /** + * Stream ids minted during serialization (`STABLE_ULID`). Not a correlation + * id, but it drew from the same shared sequence, so a workflow that + * serialized a stream renumbered every entity created after it. + */ + | 'stream'; + +/** Mints the ULID body of a correlation id for one entity family. */ +export type CorrelationIdGenerator = (kind: CorrelationIdKind) => string; + +const CROCKFORD = '0123456789ABCDEFGHJKMNPQRSTVWXYZ'; + +/** Number of Crockford characters in a ULID's random component. */ +const BODY_CHARS = 16; + +/** Number of Crockford characters in a ULID's timestamp component. */ +const TIME_CHARS = 10; + +function mul32(a: number, b: number): number { + return Math.imul(a, b) >>> 0; +} + +function rotl32(value: number, shift: number): number { + return ((value << shift) | (value >>> (32 - shift))) >>> 0; +} + +/** MurmurHash3's 32-bit finalizer. */ +function fmix32(input: number): number { + let h = input >>> 0; + h = (h ^ (h >>> 16)) >>> 0; + h = mul32(h, 0x85ebca6b); + h = (h ^ (h >>> 13)) >>> 0; + h = mul32(h, 0xc2b2ae35); + return (h ^ (h >>> 16)) >>> 0; +} + +/** + * Deterministic 128-bit hash of a string, as four 32-bit lanes. + * + * A MurmurHash3-style mixer over UTF-16 code units, rotating which lane absorbs + * each unit and diffusing across lanes at the end. Only determinism and + * diffusion matter here: this is not a cryptographic hash, claims no + * bit-compatibility with any reference implementation, and must not be used for + * anything that outlives a deployment's replays. + */ +function hash128(input: string): [number, number, number, number] { + const lanes: [number, number, number, number] = [ + 0x9e3779b1, 0x85ebca77, 0xc2b2ae3d, 0x27d4eb2f, + ]; + for (let i = 0; i < input.length; i++) { + let k = input.charCodeAt(i) >>> 0; + k = mul32(k, 0xcc9e2d51); + k = rotl32(k, 15); + k = mul32(k, 0x1b873593); + const lane = i & 3; + let h = (lanes[lane] ^ k) >>> 0; + h = rotl32(h, 13); + lanes[lane] = (mul32(h, 5) + 0xe6546b64) >>> 0; + } + lanes[0] = (lanes[0] ^ input.length) >>> 0; + // Two passes so every lane depends on every other lane. + for (let pass = 0; pass < 2; pass++) { + for (let lane = 0; lane < 4; lane++) { + const previous = lanes[(lane + 3) & 3]; + lanes[lane] = fmix32((lanes[lane] ^ previous) >>> 0); + } + } + return lanes; +} + +/** + * Derives a kind's starting body: 80 bits of a 128-bit hash as 16 Crockford + * characters, most significant first. + * + * The leading character is confined to the alphabet's lower half so the body + * starts below half of the 80-bit space. `incrementBase32` throws on overflow, + * and without this a base that happened to land near `Z…Z` would make overflow + * reachable after few draws rather than after 2^79 of them. + */ +function deriveBody(seed: string, kind: CorrelationIdKind): string { + const lanes = hash128(`${seed} correlation-kind ${kind}`); + const bytes = [ + (lanes[0] >>> 24) & 0xff, + (lanes[0] >>> 16) & 0xff, + (lanes[0] >>> 8) & 0xff, + lanes[0] & 0xff, + (lanes[1] >>> 24) & 0xff, + (lanes[1] >>> 16) & 0xff, + (lanes[1] >>> 8) & 0xff, + lanes[1] & 0xff, + (lanes[2] >>> 24) & 0xff, + (lanes[2] >>> 16) & 0xff, + ]; + let body = ''; + let accumulator = 0; + let bits = 0; + for (const byte of bytes) { + accumulator = ((accumulator << 8) | byte) >>> 0; + bits += 8; + while (bits >= 5) { + const index = (accumulator >>> (bits - 5)) & 31; + body += CROCKFORD[body.length === 0 ? index & 15 : index]; + bits -= 5; + } + } + return body; +} + +/** + * Builds a replay's correlation-id generator. + * + * `perKind: false` returns the run's single shared monotonic sequence and + * ignores the kind entirely, so both schemes go through one call path and the + * flag is the only difference between them. + */ +export function createCorrelationIdGenerator(options: { + /** + * The run's replay-stable seed. Must not vary between replays of one run, and + * must differ between runs, or two runs would mint identical ids. + */ + seed: string; + fixedTimestamp: number; + /** The run's shared monotonic sequence, used as-is when `perKind` is false. */ + positional: () => string; + perKind: boolean; +}): CorrelationIdGenerator { + const { seed, fixedTimestamp, positional, perKind } = options; + + if (!perKind) { + return positional; + } + + const time = encodeTime(fixedTimestamp, TIME_CHARS); + const bodies = new Map(); + + return (kind: CorrelationIdKind) => { + const previous = bodies.get(kind); + const body = + previous === undefined + ? deriveBody(seed, kind) + : incrementBase32(previous); + bodies.set(kind, body); + return `${time}${body}`; + }; +} + +/** Length of a ULID, exported so tests need not restate it. */ +export const CORRELATION_ID_LENGTH = TIME_CHARS + BODY_CHARS; + +/** + * Whether each entity family draws correlation ids from its own sequence rather + * than from one sequence shared by the whole run. + * + * A run cannot change scheme mid-flight: a replay under the other scheme mints + * ids its own earlier events do not carry, so it can consume none of them. On + * Vercel, skew protection keeps a run on the deployment that started it, which + * makes the flag safe to flip there. + */ +export function isPerKindCorrelationIdsEnabled(): boolean { + return process.env.WORKFLOW_PER_KIND_CORRELATION_IDS === '1'; +} diff --git a/packages/core/src/delivery-barrier-coverage.test.ts b/packages/core/src/delivery-barrier-coverage.test.ts index 836d7d3948..5ed905c86e 100644 --- a/packages/core/src/delivery-barrier-coverage.test.ts +++ b/packages/core/src/delivery-barrier-coverage.test.ts @@ -44,6 +44,10 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from './correlation-id.js'; import { describe, expect, it, vi } from 'vitest'; import { EventsConsumer } from './events-consumer.js'; import { WorkflowSuspension } from './global.js'; @@ -85,7 +89,12 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { getPromiseQueue: () => promiseQueueHolder.current, }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateCorrelationId: createCorrelationIdGenerator({ + seed: 'test', + fixedTimestamp: workflowStartedAt, + positional: () => ulid(workflowStartedAt), + perKind: isPerKindCorrelationIdsEnabled(), + }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/hook-sleep-interaction.test.ts b/packages/core/src/hook-sleep-interaction.test.ts index 29d64fcc58..68e0fbca47 100644 --- a/packages/core/src/hook-sleep-interaction.test.ts +++ b/packages/core/src/hook-sleep-interaction.test.ts @@ -3,6 +3,10 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from './correlation-id.js'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { EventsConsumer } from './events-consumer.js'; import { WorkflowSuspension } from './global.js'; @@ -57,7 +61,12 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { getPromiseQueue: () => promiseQueueHolder.current, }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateCorrelationId: createCorrelationIdGenerator({ + seed: 'test', + fixedTimestamp: workflowStartedAt, + positional: () => ulid(workflowStartedAt), + perKind: isPerKindCorrelationIdsEnabled(), + }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/private.ts b/packages/core/src/private.ts index 16fb6a3e4e..d825b343da 100644 --- a/packages/core/src/private.ts +++ b/packages/core/src/private.ts @@ -4,6 +4,7 @@ import { withResolvers } from '@workflow/utils'; import type { WorldCapabilities } from '@workflow/world'; +import type { CorrelationIdGenerator } from './correlation-id.js'; import type { EventsConsumer } from './events-consumer.js'; import type { QueueItem } from './global.js'; import type { ReplayPayloadCache } from './replay-payload-cache.js'; @@ -142,7 +143,12 @@ export interface WorkflowOrchestratorContext { */ invocationsQueue: Map; onWorkflowError: (error: Error) => void; - generateUlid: () => string; + /** + * Mints a correlation id body for one entity family. Every entity a replay + * creates draws from here, and the family is what keeps a disagreement about + * one family's count from renumbering another's. + */ + generateCorrelationId: CorrelationIdGenerator; generateNanoid: () => string; /** * Sequential promise queue that ensures all event-driven promise resolutions diff --git a/packages/core/src/step-delivery-hop-count.test.ts b/packages/core/src/step-delivery-hop-count.test.ts index f7ea3e09b1..9b92fae4e7 100644 --- a/packages/core/src/step-delivery-hop-count.test.ts +++ b/packages/core/src/step-delivery-hop-count.test.ts @@ -28,6 +28,10 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from './correlation-id.js'; import { describe, expect, it, vi } from 'vitest'; import { EventsConsumer } from './events-consumer.js'; import { WorkflowSuspension } from './global.js'; @@ -68,7 +72,12 @@ function setupWorkflowContext( getPromiseQueue: () => promiseQueueHolder.current, }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateCorrelationId: createCorrelationIdGenerator({ + seed: 'test', + fixedTimestamp: workflowStartedAt, + positional: () => ulid(workflowStartedAt), + perKind: isPerKindCorrelationIdsEnabled(), + }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/step-delivery-ordering.test.ts b/packages/core/src/step-delivery-ordering.test.ts index 85933b7e7c..823ac7b757 100644 --- a/packages/core/src/step-delivery-ordering.test.ts +++ b/packages/core/src/step-delivery-ordering.test.ts @@ -3,6 +3,10 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from './correlation-id.js'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { EventsConsumer } from './events-consumer.js'; import { WorkflowSuspension } from './global.js'; @@ -117,7 +121,12 @@ function setupWorkflowContext( getPromiseQueue: () => promiseQueueHolder.current, }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateCorrelationId: createCorrelationIdGenerator({ + seed: 'test', + fixedTimestamp: workflowStartedAt, + positional: () => ulid(workflowStartedAt), + perKind: isPerKindCorrelationIdsEnabled(), + }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/step-hydration-memoization.test.ts b/packages/core/src/step-hydration-memoization.test.ts index 7df9abdf38..d8ebc24fc6 100644 --- a/packages/core/src/step-hydration-memoization.test.ts +++ b/packages/core/src/step-hydration-memoization.test.ts @@ -1,6 +1,10 @@ import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from './correlation-id.js'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { EventsConsumer } from './events-consumer.js'; import type { WorkflowOrchestratorContext } from './private.js'; @@ -42,7 +46,12 @@ function setupWorkflowContext( getPromiseQueue: () => Promise.resolve(), }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateCorrelationId: createCorrelationIdGenerator({ + seed: 'test', + fixedTimestamp: workflowStartedAt, + positional: () => ulid(workflowStartedAt), + perKind: isPerKindCorrelationIdsEnabled(), + }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/step.test.ts b/packages/core/src/step.test.ts index 80c5e5a30a..1c60300fce 100644 --- a/packages/core/src/step.test.ts +++ b/packages/core/src/step.test.ts @@ -7,6 +7,10 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from './correlation-id.js'; import { beforeAll, describe, expect, it, vi } from 'vitest'; import { registerSerializationClass } from './class-serialization.js'; import { EventsConsumer } from './events-consumer.js'; @@ -61,7 +65,12 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { getPromiseQueue: () => Promise.resolve(), }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), // All generated ulids use the workflow's started at time + generateCorrelationId: createCorrelationIdGenerator({ + seed: 'test', + fixedTimestamp: workflowStartedAt, + positional: () => ulid(workflowStartedAt), + perKind: isPerKindCorrelationIdsEnabled(), + }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/step.ts b/packages/core/src/step.ts index c6dbb13494..4b93e21137 100644 --- a/packages/core/src/step.ts +++ b/packages/core/src/step.ts @@ -25,7 +25,7 @@ export function createUseStep(ctx: WorkflowOrchestratorContext) { ): Promise { const { promise, resolve, reject } = withResolvers(); - const correlationId = `step_${ctx.generateUlid()}`; + const correlationId = `step_${ctx.generateCorrelationId('step')}`; const queueItem: StepInvocationQueueItem = { type: 'step', diff --git a/packages/core/src/workflow.test.ts b/packages/core/src/workflow.test.ts index 42653d2b4e..3936e5d447 100644 --- a/packages/core/src/workflow.test.ts +++ b/packages/core/src/workflow.test.ts @@ -225,11 +225,11 @@ describe('runWorkflow', () => { // Turbo's first delivery synthesizes `startedAt` from the local clock, // while later (non-turbo) deliveries load the server-canonical `startedAt`. // Replay matching must NOT depend on `startedAt`: correlation IDs come from - // `generateUlid`, keyed off the run-ID-recovered `fixedTimestamp`, not + // `generateCorrelationId`, keyed off the run-ID-recovered `fixedTimestamp`, not // `startedAt`. Here the recorded `add` event uses the createdAt-derived // correlation ID, but `startedAt` is months away — replay must still // regenerate the same ID and consume the completion rather than throwing - // ReplayDivergenceError. Reverting `generateUlid` to `ulid(+startedAt)` + // ReplayDivergenceError. Keying the generator off `+startedAt` instead // fails this test. const ops: Promise[] = []; const workflowRunId = 'wrun_123'; diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts index 6c09703327..b1a7d3af41 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -10,6 +10,10 @@ import type { Event, WorkflowRun, WorldCapabilities } from '@workflow/world'; import { SPEC_VERSION_SUPPORTS_COMPRESSION } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from './correlation-id.js'; import { EventConsumerResult, EventsConsumer } from './events-consumer.js'; import type { QueueItem } from './global.js'; import { ENOTSUP, WorkflowSuspension } from './global.js'; @@ -174,18 +178,28 @@ export async function runWorkflow( : `http://localhost:${(await getPortLazy()) ?? 3000}` ); + const seed = `${workflowRun.runId}:${workflowRun.workflowName}:${workflowRun.deploymentId}`; + const { context, globalThis: vmGlobalThis, updateTimestamp, } = createContext({ - seed: `${workflowRun.runId}:${workflowRun.workflowName}:${workflowRun.deploymentId}`, + seed, fixedTimestamp, }); const workflowDiscontinuation = withResolvers(); const ulid = monotonicFactory(() => vmGlobalThis.Math.random()); + const generateCorrelationId = createCorrelationIdGenerator({ + seed, + fixedTimestamp, + // Correlation IDs must be replay-stable. `startedAt` differs between a + // turbo delivery and a later server-backed replay, so use fixedTimestamp. + positional: () => ulid(fixedTimestamp), + perKind: isPerKindCorrelationIdsEnabled(), + }); const generateNanoid = nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * vmGlobalThis.Math.random()) ); @@ -217,9 +231,7 @@ export async function runWorkflow( globalThis: vmGlobalThis, onWorkflowError: workflowDiscontinuation.reject, eventsConsumer, - // Correlation IDs must be replay-stable. `startedAt` differs between a - // turbo delivery and a later server-backed replay, so use fixedTimestamp. - generateUlid: () => ulid(fixedTimestamp), + generateCorrelationId, generateNanoid, invocationsQueue: new Map(), // Use getter/setter so the EventsConsumer's getPromiseQueue() always @@ -294,8 +306,14 @@ export async function runWorkflow( // @ts-expect-error - `@types/node` says symbol is not valid, but it does work vmGlobalThis[WORKFLOW_CONTEXT_SYMBOL] = ctx; + // Serialization mints stream ids through this symbol, and calls it with no + // seed time. `monotonicFactory` returns `encodeTime(lastTime)` on its + // increment branch, so one such call latches the *host* wall clock into + // `lastTime` and every id the run mints afterwards carries that timestamp + // instead of `fixedTimestamp` — a value that differs on every replay. + // Binding the seed time here keeps the whole run on one replay-stable clock. // @ts-expect-error - `@types/node` says symbol is not valid, but it does work - vmGlobalThis[STABLE_ULID] = ulid; + vmGlobalThis[STABLE_ULID] = () => generateCorrelationId('stream'); // Workflow code must import the deterministic `fetch` step from `workflow`. vmGlobalThis.fetch = () => { diff --git a/packages/core/src/workflow/abort-controller.ts b/packages/core/src/workflow/abort-controller.ts index 4d5d9a08af..3d94dee210 100644 --- a/packages/core/src/workflow/abort-controller.ts +++ b/packages/core/src/workflow/abort-controller.ts @@ -111,7 +111,7 @@ export function createCreateAbortController(ctx: WorkflowOrchestratorContext) { readonly [ABORT_HOOK_TOKEN]: string; constructor() { - const id = ctx.generateUlid(); + const id = ctx.generateCorrelationId('abort'); const streamName = getAbortStreamId(id); const hookToken = `abrt_${id}`; @@ -120,8 +120,10 @@ export function createCreateAbortController(ctx: WorkflowOrchestratorContext) { this.signal = new WorkflowAbortSignal(streamName, hookToken); // Register an internal system hook in the invocations queue. - // isSystem prevents token namespace conflicts with user hooks. - const correlationId = `hook_${ctx.generateUlid()}`; + // isSystem prevents token namespace conflicts with user hooks. The id + // draws from its own family, not `hook`, so constructing an abort + // controller does not renumber hooks the workflow creates later. + const correlationId = `hook_${ctx.generateCorrelationId('abortHook')}`; ctx.invocationsQueue.set(correlationId, { type: 'hook', correlationId, diff --git a/packages/core/src/workflow/attribute-dispatcher.ts b/packages/core/src/workflow/attribute-dispatcher.ts index 760dee8bbc..95ac76dc08 100644 --- a/packages/core/src/workflow/attribute-dispatcher.ts +++ b/packages/core/src/workflow/attribute-dispatcher.ts @@ -17,7 +17,7 @@ export function createSetAttributes(ctx: WorkflowOrchestratorContext) { options: { allowReservedAttributes?: boolean } = {} ): Promise { const { promise, resolve } = withResolvers(); - const correlationId = `attr_${ctx.generateUlid()}`; + const correlationId = `attr_${ctx.generateCorrelationId('attr')}`; const queueItem: AttributeInvocationQueueItem = { type: 'attribute', correlationId, diff --git a/packages/core/src/workflow/hook.test.ts b/packages/core/src/workflow/hook.test.ts index 0cc4a950e0..84f4dc4035 100644 --- a/packages/core/src/workflow/hook.test.ts +++ b/packages/core/src/workflow/hook.test.ts @@ -7,6 +7,10 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from '../correlation-id.js'; import { describe, expect, it, vi } from 'vitest'; import { aliasSerializationClass, @@ -46,7 +50,12 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { getPromiseQueue: () => Promise.resolve(), }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateCorrelationId: createCorrelationIdGenerator({ + seed: 'test', + fixedTimestamp: workflowStartedAt, + positional: () => ulid(workflowStartedAt), + perKind: isPerKindCorrelationIdsEnabled(), + }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/workflow/hook.ts b/packages/core/src/workflow/hook.ts index d5d5bed8ec..e95998b858 100644 --- a/packages/core/src/workflow/hook.ts +++ b/packages/core/src/workflow/hook.ts @@ -96,7 +96,7 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) { } // Generate hook ID and token - const correlationId = `hook_${ctx.generateUlid()}`; + const correlationId = `hook_${ctx.generateCorrelationId('hook')}`; const token = options.token ?? ctx.generateNanoid(); const tokenRetentionUntil = options.experimental_minRetention === undefined diff --git a/packages/core/src/workflow/sleep.test.ts b/packages/core/src/workflow/sleep.test.ts index a7c4723dd7..4aa012e849 100644 --- a/packages/core/src/workflow/sleep.test.ts +++ b/packages/core/src/workflow/sleep.test.ts @@ -3,6 +3,10 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { + createCorrelationIdGenerator, + isPerKindCorrelationIdsEnabled, +} from '../correlation-id.js'; import { describe, expect, it, vi } from 'vitest'; import { EventsConsumer } from '../events-consumer.js'; import { WorkflowSuspension } from '../global.js'; @@ -37,7 +41,12 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { getPromiseQueue: () => Promise.resolve(), }), invocationsQueue: new Map(), - generateUlid: () => ulid(workflowStartedAt), + generateCorrelationId: createCorrelationIdGenerator({ + seed: 'test', + fixedTimestamp: workflowStartedAt, + positional: () => ulid(workflowStartedAt), + perKind: isPerKindCorrelationIdsEnabled(), + }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) ), diff --git a/packages/core/src/workflow/sleep.ts b/packages/core/src/workflow/sleep.ts index c8848d0c3a..5245745e62 100644 --- a/packages/core/src/workflow/sleep.ts +++ b/packages/core/src/workflow/sleep.ts @@ -15,7 +15,7 @@ export function createSleep(ctx: WorkflowOrchestratorContext) { param: StringValue | Date | number ): Promise { const { promise, resolve } = withResolvers(); - const correlationId = `wait_${ctx.generateUlid()}`; + const correlationId = `wait_${ctx.generateCorrelationId('wait')}`; // Calculate the resume time const resumeAt = parseDurationToDate(param); diff --git a/workbench/nextjs-turbopack/vercel.json b/workbench/nextjs-turbopack/vercel.json index ff2e7944b0..22da2a3327 100644 --- a/workbench/nextjs-turbopack/vercel.json +++ b/workbench/nextjs-turbopack/vercel.json @@ -5,7 +5,8 @@ } }, "env": { - "WORKFLOW_PUBLIC_MANIFEST": "1" + "WORKFLOW_PUBLIC_MANIFEST": "1", + "WORKFLOW_PER_KIND_CORRELATION_IDS": "1" }, "regions": [ "iad1", From b82b39da1d909aa0888dda744d51a5d0a5b155fd Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Mon, 3 Aug 2026 12:22:41 -0700 Subject: [PATCH 2/4] Enable per-kind correlation ids by default Fixture-coupled replay suites pin themselves to the shared sequence, because their event logs record ids that sequence minted. Files whose fixture ids are derived from the generator keep honoring the flag. --- .../docs/v5/configuration/runtime-tuning.mdx | 8 +- .../async-deserialization-ordering.test.ts | 9 +- .../core/src/correlation-id-replay.test.ts | 124 ++++++++++++++++++ packages/core/src/correlation-id.test.ts | 4 +- packages/core/src/correlation-id.ts | 2 +- .../src/delivery-barrier-coverage.test.ts | 9 +- .../core/src/hook-sleep-interaction.test.ts | 9 +- .../runtime/precondition-guard-replay.test.ts | 3 + .../resume-hook.consumer-preload.test.ts | 3 + .../runtime/wait-completion-replay.test.ts | 3 + .../core/src/step-delivery-hop-count.test.ts | 9 +- .../core/src/step-delivery-ordering.test.ts | 9 +- .../src/step-hydration-memoization.test.ts | 9 +- packages/core/src/step.test.ts | 9 +- .../src/test-support/correlation-id-scheme.ts | 26 ++++ packages/core/src/workflow.test.ts | 3 + packages/core/src/workflow/hook.test.ts | 9 +- packages/core/src/workflow/sleep.test.ts | 9 +- packages/core/tsconfig.json | 2 +- workbench/nextjs-turbopack/vercel.json | 3 +- 20 files changed, 207 insertions(+), 55 deletions(-) create mode 100644 packages/core/src/correlation-id-replay.test.ts create mode 100644 packages/core/src/test-support/correlation-id-scheme.ts diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index 0da5273e47..edb27059c7 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -77,12 +77,12 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL ### `WORKFLOW_PER_KIND_CORRELATION_IDS` -- Default: disabled -- Experimental. Gives each kind of entity a workflow creates — steps, waits, hooks, attribute writes, abort controllers, stream IDs — its own sequence of correlation IDs. +- Default: enabled +- Gives each kind of entity a workflow creates — steps, waits, hooks, attribute writes, abort controllers, stream IDs — its own sequence of correlation IDs. - With one sequence shared by every kind, an ID is an ordinal over the whole run, so a single extra draw of any kind shifts every ID after it. Two concurrent replays of the same run that disagree about one `sleep()` then assign different IDs to every step that follows, and each writes events the other can neither match nor consume, which fails the run with `CORRUPTED_EVENT_LOG`. Per-kind sequences confine that to the kind that actually differs. - IDs remain ordered within a kind, so hooks are still listed in creation order. -- Correlation IDs of runs created before the setting changed are not affected: a run keeps replaying on the deployment it started on. On platforms without that guarantee, only enable it while no runs are in flight — a replay that switches schemes mid-run cannot consume its own earlier events. -- Set `1` to enable. +- Correlation IDs of runs created before the setting changed are not affected: a run keeps replaying on the deployment it started on. On platforms without that guarantee, only change it while no runs are in flight — a replay that switches schemes mid-run cannot consume its own earlier events. +- Set `0` to disable. ## Inline execution diff --git a/packages/core/src/async-deserialization-ordering.test.ts b/packages/core/src/async-deserialization-ordering.test.ts index da5416c911..4ad65f662c 100644 --- a/packages/core/src/async-deserialization-ordering.test.ts +++ b/packages/core/src/async-deserialization-ordering.test.ts @@ -2,12 +2,9 @@ import { FatalError } from '@workflow/errors'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; -import { - createCorrelationIdGenerator, - isPerKindCorrelationIdsEnabled, -} from './correlation-id.js'; import { afterEach, beforeAll, describe, expect, it, vi } from 'vitest'; import { registerSerializationClass } from './class-serialization.js'; +import { createCorrelationIdGenerator } from './correlation-id.js'; import { EventsConsumer } from './events-consumer.js'; import type { WorkflowOrchestratorContext } from './private.js'; import { ReplayPayloadCache } from './replay-payload-cache.js'; @@ -65,7 +62,9 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { seed: 'test', fixedTimestamp: workflowStartedAt, positional: () => ulid(workflowStartedAt), - perKind: isPerKindCorrelationIdsEnabled(), + // The event logs in this file hardcode correlation ids the run-wide + // shared sequence minted, so replay only matches under that scheme. + perKind: false, }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) diff --git a/packages/core/src/correlation-id-replay.test.ts b/packages/core/src/correlation-id-replay.test.ts new file mode 100644 index 0000000000..92b1db81b6 --- /dev/null +++ b/packages/core/src/correlation-id-replay.test.ts @@ -0,0 +1,124 @@ +import type { Event } from '@workflow/world'; +import * as nanoid from 'nanoid'; +import { monotonicFactory } from 'ulid'; +import { describe, expect, it, vi } from 'vitest'; +import { createCorrelationIdGenerator } from './correlation-id.js'; +import { EventsConsumer } from './events-consumer.js'; +import type { WorkflowOrchestratorContext } from './private.js'; +import { ReplayPayloadCache } from './replay-payload-cache.js'; +import { dehydrateStepReturnValue } from './serialization.js'; +import { createUseStep } from './step.js'; +import { createContext } from './vm/index.js'; +import { createCreateHook } from './workflow/hook.js'; +import { createSleep } from './workflow/sleep.js'; + +/** + * Correlation-id stability seen through the primitives that actually mint ids, + * rather than through the generator alone: that a step's id survives another + * kind of entity being created alongside it, and that a replay consumes an event + * log carrying the ids a same-seeded replay derives. + * + * The rest of the replay suites author their event logs with literal correlation + * ids from the shared sequence and pin themselves to it. These fixtures derive + * their ids instead, so they hold under either scheme. + */ + +const SEED = 'test'; +const FIXED_TIMESTAMP = 1753481739458; + +function setupWorkflowContext( + events: Event[], + perKind: boolean +): WorkflowOrchestratorContext { + const context = createContext({ + seed: SEED, + fixedTimestamp: FIXED_TIMESTAMP, + }); + const ulid = monotonicFactory(() => context.globalThis.Math.random()); + return { + runId: 'wrun_test', + encryptionKey: undefined, + replayPayloadCache: new ReplayPayloadCache(undefined), + globalThis: context.globalThis, + eventsConsumer: new EventsConsumer(events, { + onUnconsumedEvent: () => {}, + getPromiseQueue: () => Promise.resolve(), + }), + invocationsQueue: new Map(), + generateCorrelationId: createCorrelationIdGenerator({ + seed: SEED, + fixedTimestamp: FIXED_TIMESTAMP, + positional: () => ulid(FIXED_TIMESTAMP), + perKind, + }), + generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => + new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) + ), + onWorkflowError: vi.fn(), + promiseQueue: Promise.resolve(), + pendingDeliveries: 0, + pendingDeliveryBarriers: new Map(), + }; +} + +/** + * The id the next step of a replay would claim. Nothing in the log resolves the + * step, so the returned promise stays pending by design: the queue item is what + * we are after. + */ +function probeStepId( + perKind: boolean, + before?: (ctx: WorkflowOrchestratorContext) => void +): string { + const ctx = setupWorkflowContext([], perKind); + before?.(ctx); + void createUseStep(ctx)('add')(1, 2).catch(() => {}); + const item = [...ctx.invocationsQueue.values()].find( + (entry) => entry.type === 'step' + ); + if (!item) { + throw new Error('expected a step invocation'); + } + return item.correlationId; +} + +function createHookAndSleep(ctx: WorkflowOrchestratorContext): void { + createCreateHook(ctx)(); + void createSleep(ctx)('1h').catch(() => {}); +} + +describe('correlation ids through the replay primitives', () => { + it('keeps a step id when a hook and a sleep are created before it', () => { + expect(probeStepId(true, createHookAndSleep)).toBe(probeStepId(true)); + }); + + it('renumbers that step under one sequence shared by every kind', () => { + // The failure this PR removes, and the reason the assertion above is worth + // making: with a shared sequence the hook and the sleep consume the two + // ordinals the step would otherwise have drawn from. + expect(probeStepId(false, createHookAndSleep)).not.toBe(probeStepId(false)); + }); + + it('consumes a step_completed authored with the derived id', async () => { + const correlationId = probeStepId(true, createHookAndSleep); + const ctx = setupWorkflowContext( + [ + { + eventId: 'evnt_0', + runId: 'wrun_test', + eventType: 'step_completed', + correlationId, + eventData: { + stepName: 'add', + result: await dehydrateStepReturnValue(3, 'wrun_test', undefined), + }, + createdAt: new Date(), + }, + ], + true + ); + createHookAndSleep(ctx); + await expect(createUseStep(ctx)('add')(1, 2)).resolves.toBe(3); + expect(ctx.onWorkflowError).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/core/src/correlation-id.test.ts b/packages/core/src/correlation-id.test.ts index db95c65df4..a63c8bdb94 100644 --- a/packages/core/src/correlation-id.test.ts +++ b/packages/core/src/correlation-id.test.ts @@ -134,11 +134,11 @@ describe('createCorrelationIdGenerator', () => { }); describe('isPerKindCorrelationIdsEnabled', () => { - it('reads WORKFLOW_PER_KIND_CORRELATION_IDS, defaulting to disabled', () => { + it('reads WORKFLOW_PER_KIND_CORRELATION_IDS, defaulting to enabled', () => { const original = process.env.WORKFLOW_PER_KIND_CORRELATION_IDS; try { delete process.env.WORKFLOW_PER_KIND_CORRELATION_IDS; - expect(isPerKindCorrelationIdsEnabled()).toBe(false); + expect(isPerKindCorrelationIdsEnabled()).toBe(true); process.env.WORKFLOW_PER_KIND_CORRELATION_IDS = '1'; expect(isPerKindCorrelationIdsEnabled()).toBe(true); process.env.WORKFLOW_PER_KIND_CORRELATION_IDS = '0'; diff --git a/packages/core/src/correlation-id.ts b/packages/core/src/correlation-id.ts index d3844ff5ed..431dff16e9 100644 --- a/packages/core/src/correlation-id.ts +++ b/packages/core/src/correlation-id.ts @@ -210,5 +210,5 @@ export const CORRELATION_ID_LENGTH = TIME_CHARS + BODY_CHARS; * makes the flag safe to flip there. */ export function isPerKindCorrelationIdsEnabled(): boolean { - return process.env.WORKFLOW_PER_KIND_CORRELATION_IDS === '1'; + return process.env.WORKFLOW_PER_KIND_CORRELATION_IDS !== '0'; } diff --git a/packages/core/src/delivery-barrier-coverage.test.ts b/packages/core/src/delivery-barrier-coverage.test.ts index 5ed905c86e..9766f1240d 100644 --- a/packages/core/src/delivery-barrier-coverage.test.ts +++ b/packages/core/src/delivery-barrier-coverage.test.ts @@ -44,11 +44,8 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; -import { - createCorrelationIdGenerator, - isPerKindCorrelationIdsEnabled, -} from './correlation-id.js'; import { describe, expect, it, vi } from 'vitest'; +import { createCorrelationIdGenerator } from './correlation-id.js'; import { EventsConsumer } from './events-consumer.js'; import { WorkflowSuspension } from './global.js'; import { @@ -93,7 +90,9 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { seed: 'test', fixedTimestamp: workflowStartedAt, positional: () => ulid(workflowStartedAt), - perKind: isPerKindCorrelationIdsEnabled(), + // The event logs in this file hardcode correlation ids the run-wide + // shared sequence minted, so replay only matches under that scheme. + perKind: false, }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) diff --git a/packages/core/src/hook-sleep-interaction.test.ts b/packages/core/src/hook-sleep-interaction.test.ts index 68e0fbca47..c3247c27b8 100644 --- a/packages/core/src/hook-sleep-interaction.test.ts +++ b/packages/core/src/hook-sleep-interaction.test.ts @@ -3,11 +3,8 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; -import { - createCorrelationIdGenerator, - isPerKindCorrelationIdsEnabled, -} from './correlation-id.js'; import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createCorrelationIdGenerator } from './correlation-id.js'; import { EventsConsumer } from './events-consumer.js'; import { WorkflowSuspension } from './global.js'; import type { WorkflowOrchestratorContext } from './private.js'; @@ -65,7 +62,9 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { seed: 'test', fixedTimestamp: workflowStartedAt, positional: () => ulid(workflowStartedAt), - perKind: isPerKindCorrelationIdsEnabled(), + // The event logs in this file hardcode correlation ids the run-wide + // shared sequence minted, so replay only matches under that scheme. + perKind: false, }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) diff --git a/packages/core/src/runtime/precondition-guard-replay.test.ts b/packages/core/src/runtime/precondition-guard-replay.test.ts index fefa5a0bcf..49769f2518 100644 --- a/packages/core/src/runtime/precondition-guard-replay.test.ts +++ b/packages/core/src/runtime/precondition-guard-replay.test.ts @@ -37,6 +37,7 @@ import { dehydrateStepReturnValue, dehydrateWorkflowArguments, } from '../serialization.js'; +import { pinSharedCorrelationIds } from '../test-support/correlation-id-scheme.js'; import { createContext } from '../vm/index.js'; import { getPreconditionMaxInProcessRestarts, @@ -930,6 +931,8 @@ async function inlineClaimRejectionScenario() { }; } +pinSharedCorrelationIds(); + describe('precondition guard through the real replay loop', () => { let originalGuard: string | undefined; let originalRestartBound: string | undefined; diff --git a/packages/core/src/runtime/resume-hook.consumer-preload.test.ts b/packages/core/src/runtime/resume-hook.consumer-preload.test.ts index c7897553ee..3e37a92f2e 100644 --- a/packages/core/src/runtime/resume-hook.consumer-preload.test.ts +++ b/packages/core/src/runtime/resume-hook.consumer-preload.test.ts @@ -27,6 +27,7 @@ import { dehydrateStepReturnValue, dehydrateWorkflowArguments, } from '../serialization.js'; +import { pinSharedCorrelationIds } from '../test-support/correlation-id-scheme.js'; import { createContext } from '../vm/index.js'; import { setWorld } from './world.js'; @@ -252,6 +253,8 @@ async function runResumeConsumerScenario(options: { }; } +pinSharedCorrelationIds(); + describe('lazy hook resume consumer preload (Perf Option A)', () => { afterEach(() => { setWorld(undefined); diff --git a/packages/core/src/runtime/wait-completion-replay.test.ts b/packages/core/src/runtime/wait-completion-replay.test.ts index c4ca46982a..27ca033d37 100644 --- a/packages/core/src/runtime/wait-completion-replay.test.ts +++ b/packages/core/src/runtime/wait-completion-replay.test.ts @@ -14,6 +14,7 @@ import { dehydrateStepReturnValue, dehydrateWorkflowArguments, } from '../serialization.js'; +import { pinSharedCorrelationIds } from '../test-support/correlation-id-scheme.js'; import { createContext } from '../vm/index.js'; import { setWorld } from './world.js'; @@ -434,6 +435,8 @@ function expectHookBranchQueued( ); } +pinSharedCorrelationIds(); + describe('workflow handler wait completion replay', () => { afterEach(() => { setWorld(undefined); diff --git a/packages/core/src/step-delivery-hop-count.test.ts b/packages/core/src/step-delivery-hop-count.test.ts index 9b92fae4e7..4e81080b1d 100644 --- a/packages/core/src/step-delivery-hop-count.test.ts +++ b/packages/core/src/step-delivery-hop-count.test.ts @@ -28,11 +28,8 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; -import { - createCorrelationIdGenerator, - isPerKindCorrelationIdsEnabled, -} from './correlation-id.js'; import { describe, expect, it, vi } from 'vitest'; +import { createCorrelationIdGenerator } from './correlation-id.js'; import { EventsConsumer } from './events-consumer.js'; import { WorkflowSuspension } from './global.js'; import type { WorkflowOrchestratorContext } from './private.js'; @@ -76,7 +73,9 @@ function setupWorkflowContext( seed: 'test', fixedTimestamp: workflowStartedAt, positional: () => ulid(workflowStartedAt), - perKind: isPerKindCorrelationIdsEnabled(), + // The event logs in this file hardcode correlation ids the run-wide + // shared sequence minted, so replay only matches under that scheme. + perKind: false, }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) diff --git a/packages/core/src/step-delivery-ordering.test.ts b/packages/core/src/step-delivery-ordering.test.ts index 823ac7b757..5eb4715267 100644 --- a/packages/core/src/step-delivery-ordering.test.ts +++ b/packages/core/src/step-delivery-ordering.test.ts @@ -3,11 +3,8 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; -import { - createCorrelationIdGenerator, - isPerKindCorrelationIdsEnabled, -} from './correlation-id.js'; import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createCorrelationIdGenerator } from './correlation-id.js'; import { EventsConsumer } from './events-consumer.js'; import { WorkflowSuspension } from './global.js'; import type { WorkflowOrchestratorContext } from './private.js'; @@ -125,7 +122,9 @@ function setupWorkflowContext( seed: 'test', fixedTimestamp: workflowStartedAt, positional: () => ulid(workflowStartedAt), - perKind: isPerKindCorrelationIdsEnabled(), + // The event logs in this file hardcode correlation ids the run-wide + // shared sequence minted, so replay only matches under that scheme. + perKind: false, }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) diff --git a/packages/core/src/step-hydration-memoization.test.ts b/packages/core/src/step-hydration-memoization.test.ts index d8ebc24fc6..8708c02e5d 100644 --- a/packages/core/src/step-hydration-memoization.test.ts +++ b/packages/core/src/step-hydration-memoization.test.ts @@ -1,11 +1,8 @@ import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; -import { - createCorrelationIdGenerator, - isPerKindCorrelationIdsEnabled, -} from './correlation-id.js'; import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createCorrelationIdGenerator } from './correlation-id.js'; import { EventsConsumer } from './events-consumer.js'; import type { WorkflowOrchestratorContext } from './private.js'; import { ReplayPayloadCache } from './replay-payload-cache.js'; @@ -50,7 +47,9 @@ function setupWorkflowContext( seed: 'test', fixedTimestamp: workflowStartedAt, positional: () => ulid(workflowStartedAt), - perKind: isPerKindCorrelationIdsEnabled(), + // The event logs in this file hardcode correlation ids the run-wide + // shared sequence minted, so replay only matches under that scheme. + perKind: false, }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) diff --git a/packages/core/src/step.test.ts b/packages/core/src/step.test.ts index 1c60300fce..faf852a54b 100644 --- a/packages/core/src/step.test.ts +++ b/packages/core/src/step.test.ts @@ -7,12 +7,9 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; -import { - createCorrelationIdGenerator, - isPerKindCorrelationIdsEnabled, -} from './correlation-id.js'; import { beforeAll, describe, expect, it, vi } from 'vitest'; import { registerSerializationClass } from './class-serialization.js'; +import { createCorrelationIdGenerator } from './correlation-id.js'; import { EventsConsumer } from './events-consumer.js'; import { WorkflowSuspension } from './global.js'; import type { WorkflowOrchestratorContext } from './private.js'; @@ -69,7 +66,9 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { seed: 'test', fixedTimestamp: workflowStartedAt, positional: () => ulid(workflowStartedAt), - perKind: isPerKindCorrelationIdsEnabled(), + // The event logs in this file hardcode correlation ids the run-wide + // shared sequence minted, so replay only matches under that scheme. + perKind: false, }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) diff --git a/packages/core/src/test-support/correlation-id-scheme.ts b/packages/core/src/test-support/correlation-id-scheme.ts new file mode 100644 index 0000000000..32a000dfdd --- /dev/null +++ b/packages/core/src/test-support/correlation-id-scheme.ts @@ -0,0 +1,26 @@ +import { afterAll, beforeAll } from 'vitest'; + +/** + * Pins a test file to the run-wide shared correlation-id sequence. + * + * Replay tests that drive the real `workflowEntrypoint` against an event log + * with hardcoded correlation ids can only match under the scheme those ids were + * minted by, and the fixtures in this repo predate per-kind sequences. Files + * whose fixture ids are derived rather than written out run under whichever + * scheme `WORKFLOW_PER_KIND_CORRELATION_IDS` selects; per-kind minting itself is + * covered by `correlation-id.test.ts`. + */ +export function pinSharedCorrelationIds(): void { + let original: string | undefined; + beforeAll(() => { + original = process.env.WORKFLOW_PER_KIND_CORRELATION_IDS; + process.env.WORKFLOW_PER_KIND_CORRELATION_IDS = '0'; + }); + afterAll(() => { + if (original === undefined) { + delete process.env.WORKFLOW_PER_KIND_CORRELATION_IDS; + } else { + process.env.WORKFLOW_PER_KIND_CORRELATION_IDS = original; + } + }); +} diff --git a/packages/core/src/workflow.test.ts b/packages/core/src/workflow.test.ts index 3936e5d447..5b7940af17 100644 --- a/packages/core/src/workflow.test.ts +++ b/packages/core/src/workflow.test.ts @@ -12,12 +12,15 @@ import { dehydrateWorkflowArguments, hydrateWorkflowReturnValue, } from './serialization.js'; +import { pinSharedCorrelationIds } from './test-support/correlation-id-scheme.js'; import { createContext } from './vm/index.js'; import { runWorkflow } from './workflow.js'; // No encryption key = encryption disabled const noEncryptionKey = undefined; +pinSharedCorrelationIds(); + describe('runWorkflow', () => { const getWorkflowTransformCode = (workflowName?: string) => `;globalThis.__private_workflows = new Map(); diff --git a/packages/core/src/workflow/hook.test.ts b/packages/core/src/workflow/hook.test.ts index 84f4dc4035..fbc2c8063e 100644 --- a/packages/core/src/workflow/hook.test.ts +++ b/packages/core/src/workflow/hook.test.ts @@ -7,15 +7,12 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; -import { - createCorrelationIdGenerator, - isPerKindCorrelationIdsEnabled, -} from '../correlation-id.js'; import { describe, expect, it, vi } from 'vitest'; import { aliasSerializationClass, RUN_CLASS_ID, } from '../class-serialization.js'; +import { createCorrelationIdGenerator } from '../correlation-id.js'; import { EventsConsumer } from '../events-consumer.js'; import { WorkflowSuspension } from '../global.js'; import type { WorkflowOrchestratorContext } from '../private.js'; @@ -54,7 +51,9 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { seed: 'test', fixedTimestamp: workflowStartedAt, positional: () => ulid(workflowStartedAt), - perKind: isPerKindCorrelationIdsEnabled(), + // The event logs in this file hardcode correlation ids the run-wide + // shared sequence minted, so replay only matches under that scheme. + perKind: false, }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) diff --git a/packages/core/src/workflow/sleep.test.ts b/packages/core/src/workflow/sleep.test.ts index 4aa012e849..b3ba216e7d 100644 --- a/packages/core/src/workflow/sleep.test.ts +++ b/packages/core/src/workflow/sleep.test.ts @@ -3,11 +3,8 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; -import { - createCorrelationIdGenerator, - isPerKindCorrelationIdsEnabled, -} from '../correlation-id.js'; import { describe, expect, it, vi } from 'vitest'; +import { createCorrelationIdGenerator } from '../correlation-id.js'; import { EventsConsumer } from '../events-consumer.js'; import { WorkflowSuspension } from '../global.js'; import type { WorkflowOrchestratorContext } from '../private.js'; @@ -45,7 +42,9 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext { seed: 'test', fixedTimestamp: workflowStartedAt, positional: () => ulid(workflowStartedAt), - perKind: isPerKindCorrelationIdsEnabled(), + // The event logs in this file hardcode correlation ids the run-wide + // shared sequence minted, so replay only matches under that scheme. + perKind: false, }), generateNanoid: nanoid.customRandom(nanoid.urlAlphabet, 21, (size) => new Uint8Array(size).map(() => 256 * context.globalThis.Math.random()) diff --git a/packages/core/tsconfig.json b/packages/core/tsconfig.json index 98892887a9..7749f99278 100644 --- a/packages/core/tsconfig.json +++ b/packages/core/tsconfig.json @@ -5,5 +5,5 @@ "rootDir": "./src" }, "include": ["src"], - "exclude": ["node_modules", "**/*.test.ts"] + "exclude": ["node_modules", "**/*.test.ts", "src/test-support"] } diff --git a/workbench/nextjs-turbopack/vercel.json b/workbench/nextjs-turbopack/vercel.json index 22da2a3327..ff2e7944b0 100644 --- a/workbench/nextjs-turbopack/vercel.json +++ b/workbench/nextjs-turbopack/vercel.json @@ -5,8 +5,7 @@ } }, "env": { - "WORKFLOW_PUBLIC_MANIFEST": "1", - "WORKFLOW_PER_KIND_CORRELATION_IDS": "1" + "WORKFLOW_PUBLIC_MANIFEST": "1" }, "regions": [ "iad1", From e49c053f65745fe393ed0c0dabae53bb6d2d24e1 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Mon, 3 Aug 2026 15:02:49 -0700 Subject: [PATCH 3/4] Keep per-kind correlation ids opt-in, and pin the STABLE_ULID binding Reverts the default to off, so an SDK upgrade alone never moves a run between id schemes on a platform that does not pin a run to the deployment that started it. The nextjs-turbopack workbench sets the flag so e2e and the race-repro harness exercise the new scheme. Also from review: a runWorkflow-level regression test for the STABLE_ULID binding (reverting it to the raw monotonic factory now fails), a note that serialization mints abort-holder identities through the `stream` family too, a caveat that two kinds mint hook_ ids so listing order is creation order only within a kind, a changeset for the stream-id determinism fix, and organizeImports. --- .changeset/per-kind-correlation-ids.md | 2 +- .changeset/stable-stream-id-timestamps.md | 6 ++ .../docs/v5/configuration/runtime-tuning.mdx | 12 ++- packages/core/src/abort-consistency.test.ts | 2 +- packages/core/src/abort-controller.test.ts | 2 +- .../core/src/abort-replay-ordering.test.ts | 2 +- packages/core/src/correlation-id.test.ts | 4 +- packages/core/src/correlation-id.ts | 38 ++++++-- packages/core/src/workflow.test.ts | 95 ++++++++++++++++++- workbench/nextjs-turbopack/vercel.json | 3 +- 10 files changed, 144 insertions(+), 22 deletions(-) create mode 100644 .changeset/stable-stream-id-timestamps.md diff --git a/.changeset/per-kind-correlation-ids.md b/.changeset/per-kind-correlation-ids.md index 34906fabab..97cba5eb10 100644 --- a/.changeset/per-kind-correlation-ids.md +++ b/.changeset/per-kind-correlation-ids.md @@ -3,4 +3,4 @@ 'workflow': patch --- -Give each kind of entity a workflow creates its own sequence of correlation IDs, so an extra hook or sleep no longer renumbers every step after it +Add experimental `WORKFLOW_PER_KIND_CORRELATION_IDS=1`, which gives each kind of entity a workflow creates its own sequence of correlation IDs so an extra hook or sleep no longer renumbers every step after it. Off by default; a run must replay under the scheme that minted its IDs, so only turn it on while no runs are in flight unless your platform pins a run to the deployment it started on diff --git a/.changeset/stable-stream-id-timestamps.md b/.changeset/stable-stream-id-timestamps.md new file mode 100644 index 0000000000..71e6655ca9 --- /dev/null +++ b/.changeset/stable-stream-id-timestamps.md @@ -0,0 +1,6 @@ +--- +'@workflow/core': patch +'workflow': patch +--- + +Fix stream IDs minted while serializing a step's arguments latching the host wall clock into a run's ID sequence, which gave every entity created afterwards a different correlation ID on each replay diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index edb27059c7..d01161d509 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -77,12 +77,14 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL ### `WORKFLOW_PER_KIND_CORRELATION_IDS` -- Default: enabled -- Gives each kind of entity a workflow creates — steps, waits, hooks, attribute writes, abort controllers, stream IDs — its own sequence of correlation IDs. +- Default: disabled +- Experimental. Gives each kind of entity a workflow creates — steps, waits, hooks, attribute writes, abort controllers, stream IDs — its own sequence of correlation IDs. - With one sequence shared by every kind, an ID is an ordinal over the whole run, so a single extra draw of any kind shifts every ID after it. Two concurrent replays of the same run that disagree about one `sleep()` then assign different IDs to every step that follows, and each writes events the other can neither match nor consume, which fails the run with `CORRUPTED_EVENT_LOG`. Per-kind sequences confine that to the kind that actually differs. -- IDs remain ordered within a kind, so hooks are still listed in creation order. -- Correlation IDs of runs created before the setting changed are not affected: a run keeps replaying on the deployment it started on. On platforms without that guarantee, only change it while no runs are in flight — a replay that switches schemes mid-run cannot consume its own earlier events. -- Set `0` to disable. +- IDs remain ordered within a kind, so hooks created by your workflow are still listed in creation order. A hook the runtime creates for you, such as the one backing an abort controller, draws from its own kind and so is listed at an arbitrary position relative to your hooks rather than at its creation position. +- A run must replay under the scheme that minted its IDs. A replay that switches schemes mid-run assigns IDs its own earlier events do not carry, so it can consume none of them and the run fails. + - On Vercel, a run keeps replaying on the deployment it started on, so it only ever sees the value baked into that deployment. Changing the setting affects new runs only. + - Elsewhere — `@workflow/world-postgres`, `@workflow/world-local`, any self-hosted process — nothing pins a run to the code that started it. Turn the setting on during a quiet window with no runs in flight, and roll the new value out to your whole fleet at once: a rolling deploy that leaves both values live replays one run under two schemes concurrently, which is the failure the setting exists to reduce. +- Set `1` to enable. ## Inline execution diff --git a/packages/core/src/abort-consistency.test.ts b/packages/core/src/abort-consistency.test.ts index 3cf2dd1896..d99b67de1d 100644 --- a/packages/core/src/abort-consistency.test.ts +++ b/packages/core/src/abort-consistency.test.ts @@ -10,11 +10,11 @@ import type { Event, WorkflowRun } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { describe, expect, it, vi } from 'vitest'; import { createCorrelationIdGenerator, isPerKindCorrelationIdsEnabled, } from './correlation-id.js'; -import { describe, expect, it, vi } from 'vitest'; import { EventsConsumer } from './events-consumer.js'; import type { WorkflowSuspension } from './global.js'; import type { WorkflowOrchestratorContext } from './private.js'; diff --git a/packages/core/src/abort-controller.test.ts b/packages/core/src/abort-controller.test.ts index 0fd39fcd66..2ced0b4aca 100644 --- a/packages/core/src/abort-controller.test.ts +++ b/packages/core/src/abort-controller.test.ts @@ -11,11 +11,11 @@ import { withResolvers } from '@workflow/utils'; import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { describe, expect, it, vi } from 'vitest'; import { createCorrelationIdGenerator, isPerKindCorrelationIdsEnabled, } from './correlation-id.js'; -import { describe, expect, it, vi } from 'vitest'; import { DEFERRED_CHECK_DELAY_MS, EventsConsumer } from './events-consumer.js'; import type { WorkflowOrchestratorContext } from './private.js'; import { ReplayPayloadCache } from './replay-payload-cache.js'; diff --git a/packages/core/src/abort-replay-ordering.test.ts b/packages/core/src/abort-replay-ordering.test.ts index e926eb5adb..8e7a1a9e98 100644 --- a/packages/core/src/abort-replay-ordering.test.ts +++ b/packages/core/src/abort-replay-ordering.test.ts @@ -26,11 +26,11 @@ import type { Event } from '@workflow/world'; import * as nanoid from 'nanoid'; import { monotonicFactory } from 'ulid'; +import { describe, expect, it, vi } from 'vitest'; import { createCorrelationIdGenerator, isPerKindCorrelationIdsEnabled, } from './correlation-id.js'; -import { describe, expect, it, vi } from 'vitest'; import { EventsConsumer } from './events-consumer.js'; import { scheduleWhenIdle, diff --git a/packages/core/src/correlation-id.test.ts b/packages/core/src/correlation-id.test.ts index a63c8bdb94..db95c65df4 100644 --- a/packages/core/src/correlation-id.test.ts +++ b/packages/core/src/correlation-id.test.ts @@ -134,11 +134,11 @@ describe('createCorrelationIdGenerator', () => { }); describe('isPerKindCorrelationIdsEnabled', () => { - it('reads WORKFLOW_PER_KIND_CORRELATION_IDS, defaulting to enabled', () => { + it('reads WORKFLOW_PER_KIND_CORRELATION_IDS, defaulting to disabled', () => { const original = process.env.WORKFLOW_PER_KIND_CORRELATION_IDS; try { delete process.env.WORKFLOW_PER_KIND_CORRELATION_IDS; - expect(isPerKindCorrelationIdsEnabled()).toBe(true); + expect(isPerKindCorrelationIdsEnabled()).toBe(false); process.env.WORKFLOW_PER_KIND_CORRELATION_IDS = '1'; expect(isPerKindCorrelationIdsEnabled()).toBe(true); process.env.WORKFLOW_PER_KIND_CORRELATION_IDS = '0'; diff --git a/packages/core/src/correlation-id.ts b/packages/core/src/correlation-id.ts index 431dff16e9..dbed6f8f5c 100644 --- a/packages/core/src/correlation-id.ts +++ b/packages/core/src/correlation-id.ts @@ -27,6 +27,13 @@ import { encodeTime, incrementBase32 } from 'ulid'; * prefixed 26-char ULIDs by the backend, and they stay monotonic *within* a * kind, because `hooks.list` is ordered by hook id. * + * Monotonicity is per kind, and two kinds mint `hook_` ids (`hook` and + * `abortHook`), so listing order is only creation order *within* each of them. + * No world filters system hooks out of a listing, so a run that constructs an + * abort controller and also creates its own hooks lists that system hook at a + * position decided by its kind's hash rather than at its creation position. + * Order among the user's own hooks is unaffected. + * * This does not make ids independent of *ordinal position within their own * kind*: two replays that disagree about how many steps ran still mint * different ids for the next step. That is a narrower failure than the shared @@ -52,9 +59,12 @@ export type CorrelationIdKind = /** `hook_` ids for the internal system hook backing an abort controller. */ | 'abortHook' /** - * Stream ids minted during serialization (`STABLE_ULID`). Not a correlation - * id, but it drew from the same shared sequence, so a workflow that - * serialized a stream renumbered every entity created after it. + * Ids minted during serialization (`STABLE_ULID`): stream names, and an abort + * holder's stream name and `abrt_` hook token when it reaches serialization + * without an identity yet (`reduceAbortWithListener`), which is why `abort` + * above is not the only mint path for an abort identity. Not correlation ids, + * but they drew from the same shared sequence, so a workflow that serialized + * a stream renumbered every entity created after it. */ | 'stream'; @@ -202,13 +212,23 @@ export const CORRELATION_ID_LENGTH = TIME_CHARS + BODY_CHARS; /** * Whether each entity family draws correlation ids from its own sequence rather - * than from one sequence shared by the whole run. + * than from one sequence shared by the whole run. Off unless opted in, so an SDK + * upgrade alone never moves a run between schemes. + * + * The invariant either way: a run must replay under the scheme that minted its + * ids. A replay under the other scheme mints ids its own earlier events do not + * carry, so it can consume none of them and fails the run. Two things can break + * it, and both are about turning the flag on rather than about upgrading: * - * A run cannot change scheme mid-flight: a replay under the other scheme mints - * ids its own earlier events do not carry, so it can consume none of them. On - * Vercel, skew protection keeps a run on the deployment that started it, which - * makes the flag safe to flip there. + * - Enabling it while runs are in flight. On Vercel, skew protection keeps a run + * on the deployment that started it, so a run only ever sees the value baked + * into its own deployment. Elsewhere (world-postgres, world-local, a + * self-hosted process) nothing pins a run to the code that started it, so + * enable it during a quiet window. + * - A rolling deploy that leaves both values live, which puts two schemes on one + * run concurrently — the side-by-side append this whole mechanism exists to + * avoid. Roll the value out to the whole fleet at once. */ export function isPerKindCorrelationIdsEnabled(): boolean { - return process.env.WORKFLOW_PER_KIND_CORRELATION_IDS !== '0'; + return process.env.WORKFLOW_PER_KIND_CORRELATION_IDS === '1'; } diff --git a/packages/core/src/workflow.test.ts b/packages/core/src/workflow.test.ts index 5b7940af17..15aae5e25a 100644 --- a/packages/core/src/workflow.test.ts +++ b/packages/core/src/workflow.test.ts @@ -2,7 +2,7 @@ import { types } from 'node:util'; import { HookConflictError, WorkflowRuntimeError } from '@workflow/errors'; import type { Event, WorkflowRun } from '@workflow/world'; import { SPEC_VERSION_CURRENT } from '@workflow/world'; -import { monotonicFactory } from 'ulid'; +import { decodeTime, monotonicFactory } from 'ulid'; import { afterEach, assert, describe, expect, it, vi } from 'vitest'; import { DEFERRED_CHECK_DELAY_MS } from './events-consumer.js'; import type { WorkflowSuspension } from './global.js'; @@ -303,6 +303,99 @@ describe('runWorkflow', () => { ).toEqual(3); }); + it('keeps IDs minted through STABLE_ULID on the run clock', async () => { + // Serialization mints stream names (and an abort holder's identity) through + // the STABLE_ULID global, calling it with no seed time. `monotonicFactory` + // returns `encodeTime(lastTime)` on its increment branch, so binding the raw + // factory there let one such call latch the *host* wall clock into + // `lastTime`: the stream ID and every correlation ID drawn after it carried a + // timestamp that differs on every replay. This drives the binding through + // `runWorkflow` rather than the generator, so reverting the binding site to + // the raw factory fails here. + const ops: Promise[] = []; + const workflowRunId = 'wrun_stable_ulid'; + const startedAt = new Date('2024-01-01T00:00:00.000Z'); + const workflowRun: WorkflowRun = { + runId: workflowRunId, + workflowName: 'workflow', + status: 'running', + input: await dehydrateWorkflowArguments( + [], + workflowRunId, + noEncryptionKey, + ops + ), + createdAt: startedAt, + updatedAt: startedAt, + startedAt, + deploymentId: 'test-deployment', + }; + + // Derive the IDs the run should mint, in draw order, with the same seeded + // factory runWorkflow uses: the stream draw first, then the step that + // follows it. The step's recorded event only matches if the stream draw left + // the sequence on `fixedTimestamp`. + const seed = `${workflowRunId}:${workflowRun.workflowName}:${workflowRun.deploymentId}`; + const fixedTimestamp = +startedAt; + const vm = createContext({ seed, fixedTimestamp }); + const ulid = monotonicFactory(() => vm.globalThis.Math.random()); + const expectedStreamId = ulid(fixedTimestamp); + const stepCorr = `step_${ulid(fixedTimestamp)}`; + + const events: Event[] = [ + { + eventId: 'event-0', + runId: workflowRunId, + eventType: 'step_started', + correlationId: stepCorr, + eventData: { stepName: 'add' }, + createdAt: startedAt, + }, + { + eventId: 'event-1', + runId: workflowRunId, + eventType: 'step_completed', + correlationId: stepCorr, + eventData: { + stepName: 'add', + result: await dehydrateStepReturnValue( + 3, + workflowRunId, + noEncryptionKey, + ops + ), + }, + createdAt: startedAt, + }, + ]; + + const result = await runWorkflow( + `const add = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("add"); + async function workflow() { + // Stands in for serialization, which is what actually calls this. + const streamId = globalThis[Symbol.for("WORKFLOW_STABLE_ULID")](); + const a = await add(1, 2); + return { streamId, a }; + }${getWorkflowTransformCode('workflow')}`, + workflowRun, + events, + noEncryptionKey + ); + + const returned = (await hydrateWorkflowReturnValue( + result as any, + workflowRunId, + noEncryptionKey, + ops + )) as { streamId: string; a: number }; + + expect(decodeTime(returned.streamId)).toBe(fixedTimestamp); + expect(returned.streamId).toBe(expectedStreamId); + // Consuming the recorded completion is the other half: a stream draw that + // moved the clock would have renamed this step out of its own event log. + expect(returned.a).toBe(3); + }); + // Test that timestamps update correctly as events are consumed it('should update the timestamp in the vm context as events are replayed', async () => { const ops: Promise[] = []; diff --git a/workbench/nextjs-turbopack/vercel.json b/workbench/nextjs-turbopack/vercel.json index ff2e7944b0..22da2a3327 100644 --- a/workbench/nextjs-turbopack/vercel.json +++ b/workbench/nextjs-turbopack/vercel.json @@ -5,7 +5,8 @@ } }, "env": { - "WORKFLOW_PUBLIC_MANIFEST": "1" + "WORKFLOW_PUBLIC_MANIFEST": "1", + "WORKFLOW_PER_KIND_CORRELATION_IDS": "1" }, "regions": [ "iad1", From 8635a428aaa96b3c8a836e84eae5e1f1f0587afb Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Mon, 3 Aug 2026 16:30:34 -0700 Subject: [PATCH 4/4] Apply suggestion from @VaguelySerious Signed-off-by: Peter Wielander --- .changeset/per-kind-correlation-ids.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/per-kind-correlation-ids.md b/.changeset/per-kind-correlation-ids.md index 97cba5eb10..2e13a31a73 100644 --- a/.changeset/per-kind-correlation-ids.md +++ b/.changeset/per-kind-correlation-ids.md @@ -3,4 +3,4 @@ 'workflow': patch --- -Add experimental `WORKFLOW_PER_KIND_CORRELATION_IDS=1`, which gives each kind of entity a workflow creates its own sequence of correlation IDs so an extra hook or sleep no longer renumbers every step after it. Off by default; a run must replay under the scheme that minted its IDs, so only turn it on while no runs are in flight unless your platform pins a run to the deployment it started on +Add env option to split correlation ID derivation into per-entity-type sequential ULIDs, instead of sharing one derivation source