[core] Derive correlation ids from per-kind sequences (opt-in)

.changeset/per-kind-correlation-ids.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'workflow': patch
+---
+
+Add env option to split correlation ID derivation into per-entity-type sequential ULIDs, instead of sharing one derivation source

.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

docs/content/docs/v5/configuration/runtime-tuning.mdx

@@ -75,6 +75,17 @@ 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 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
 
 ### `WORKFLOW_V2_TIMEOUT_MS`

packages/core/src/abort-consistency.test.ts

@@ -11,6 +11,10 @@ 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 { EventsConsumer } from './events-consumer.js';
 import type { WorkflowSuspension } from './global.js';
 import type { WorkflowOrchestratorContext } from './private.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())
     ),

packages/core/src/abort-controller.test.ts

@@ -12,6 +12,10 @@ 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 { DEFERRED_CHECK_DELAY_MS, EventsConsumer } from './events-consumer.js';
 import type { WorkflowOrchestratorContext } from './private.js';
 import { ReplayPayloadCache } from './replay-payload-cache.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())
     ),

packages/core/src/abort-replay-ordering.test.ts

@@ -27,6 +27,10 @@ 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 { EventsConsumer } from './events-consumer.js';
 import {
   scheduleWhenIdle,
@@ -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())
     ),

packages/core/src/async-deserialization-ordering.test.ts

@@ -4,6 +4,7 @@ import * as nanoid from 'nanoid';
 import { monotonicFactory } from 'ulid';
 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';
@@ -57,7 +58,14 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext {
       getPromiseQueue: () => Promise.resolve(),
     }),
     invocationsQueue: new Map(),
-    generateUlid: () => ulid(workflowStartedAt),
+    generateCorrelationId: createCorrelationIdGenerator({
+      seed: 'test',
+      fixedTimestamp: workflowStartedAt,
+      positional: () => ulid(workflowStartedAt),
+      // 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())
     ),

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();
+  });
+});