Slot event identity (SDK 3/3): number events by slot in the Local and Postgres Worlds

.changeset/inline-claim-reclaim.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Keep a batch of inline steps together when one of its event writes loses a race, instead of discarding the batch

.changeset/late-hook-delivery-divergence.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'workflow': patch
+---
+
+Stop failing runs with a corrupted-event-log error when a hook delivery arrives after the hook was disposed, or when a step result is still being fetched

.changeset/local-slot-order.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-local': patch
+---
+
+Read a position-numbered event log in position order, so a replay sees the log the order it was written

.changeset/precondition-restart-backoff.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'workflow': patch
+---
+
+Space out in-process replay restarts with a randomized backoff so concurrent replays of one run stop contending in lockstep

.changeset/slimy-weeks-act.md

@@ -0,0 +1,2 @@
+---
+---

.changeset/slot-claims-tail-tight.md

@@ -0,0 +1,7 @@
+---
+'@workflow/core': patch
+'@workflow/world-local': patch
+'@workflow/world-postgres': patch
+---
+
+Take event slot claims one at a time against the event log's tail, so a replay that decided from a log missing an event is rejected instead of committing.

.changeset/slot-duplicate-start-skip.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'workflow': patch
+---
+
+Skip a step whose concurrent start another handler already wrote, instead of restarting the replay to rediscover it

.changeset/slot-event-identity-worlds.md

@@ -0,0 +1,8 @@
+---
+'@workflow/world': minor
+'@workflow/world-local': minor
+'@workflow/world-postgres': minor
+'@workflow/core': minor
+---
+
+Number a run's events by position in the Local and Postgres Worlds when `WORKFLOW_SLOT_IDENTITY` is set, so a reader can prove its copy of an event log is complete.

.changeset/slot-identity-default-on.md

@@ -0,0 +1,8 @@
+---
+'@workflow/world': minor
+'@workflow/world-vercel': minor
+'@workflow/world-local': minor
+'@workflow/world-postgres': minor
+---
+
+Number new runs' events by position by default, in every World including the Vercel one. Set `WORKFLOW_SLOT_IDENTITY=0` to keep minting ULID event ids.

.changeset/slot-restart-budget.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'workflow': patch
+---
+
+Let a run that numbers its events by position absorb more concurrent-write rejections in one invocation, instead of falling back to a delayed re-invocation

.changeset/slot-restart-cursor-top-up.md

@@ -0,0 +1,6 @@
+---
+'@workflow/core': patch
+'workflow': patch
+---
+
+Recover faster from a concurrent write on runs that number their events by slot, by topping the event log up from its cursor instead of reloading it in full

.changeset/tidy-moons-observe.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Name the divergent event's pending invocations and the fenced member of an inline step batch in replay-divergence logs

.changeset/turbo-run-started-occurred-at.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Report the same workflow start time on an optimistically started run's first pass and its replays

.github/scripts/render-event-log-race-repro-results.js

@@ -407,6 +407,11 @@ function renderLatestFailures(entry) {
       `\nShowing 20 of ${entry.failing.length + entry.truncatedFailingCount} non-completed runs.`
     );
   }
+  // Deliberately not inlined here: the slices are large and this comment has a
+  // size limit, while the artifact has none.
+  console.log(
+    '\nThe `event-log-race-repro-results` artifact carries a window of the committed log around the divergent event, for a sample of the corruptions.'
+  );
   console.log('');
 }
 

.gitignore

@@ -43,6 +43,9 @@ workbench/nextjs-*/public/.well-known/workflow
 workbench/sveltekit/static/.well-known/workflow
 
 
+# Local e2e diagnostics dumps
+e2e-diagnostics-*.json
+
 # Event log race repro output (written to the repo root by the harness and by
 # scripts/event-log-race-repro-local.sh)
 event-log-race-repro-results.json

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

@@ -52,9 +52,11 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL
 
 ### `WORKFLOW_PRECONDITION_MAX_INPROCESS_RESTARTS`
 
-- Default: `3`
+- Default: `3`, or `12` on a run numbering its events by position (see [`WORKFLOW_SLOT_IDENTITY`](#workflow_slot_identity))
 - How many times a single invocation restarts its replay in-process after a rejected event creation before it falls back to a re-invocation.
 - A restart reloads the event log and rebuilds the workflow from scratch, so it costs a replay but no queue round trip. A World may attach the missing events to its rejection, in which case the first restart needs no event-log request at all.
+- A run numbering its events by position reads only the page after its cursor instead of reloading the log, since positions are allocated in order and every event it was missing sorts above what it already has. Restarts are cheap enough there that the higher default is worth taking before a re-invocation and its delay.
+- Setting this overrides both defaults.
 
 ### `WORKFLOW_PRECONDITION_MAX_REINVOCATIONS`
 
@@ -68,6 +70,27 @@ 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_PRECONDITION_RESTART_BACKOFF_MS`
+
+- Default: `15`
+- Base for the randomized wait before an in-process replay restart re-derives, in milliseconds. The wait doubles with each restart the invocation has spent, up to [`WORKFLOW_PRECONDITION_RESTART_BACKOFF_MAX_MS`](#workflow_precondition_restart_backoff_max_ms), and is drawn uniformly from zero to that bound.
+- The wait exists for runs with several replays in flight at once — a fan-out of steps completing together, or a burst of hooks. A rejected event creation means another replay got there first; if every loser re-derives immediately they all contend again, and none of them pulls far enough ahead to finish. Drawing each wait from the full range spreads the retries apart.
+- Set to `0` to restart without waiting.
+
+### `WORKFLOW_PRECONDITION_RESTART_BACKOFF_MAX_MS`
+
+- Default: `400`
+- Ceiling on the wait described above.
+
+### `WORKFLOW_SLOT_IDENTITY`
+
+- Default: enabled
+- Numbers a new run's events by position instead of by ULID: `evnt_…001` is the run's first event, `evnt_…002` its second. Positions are allocated in order, so the log reads in the order it was written regardless of clock skew between writers.
+- Contention becomes explicit rather than silent. Two writers proposing the same position cannot both win: the loser gets a 409 ([`SlotConflictError`](/docs/api-reference/workflow-errors/slot-conflict-error)) carrying the events it was missing, and the run replays from the top against a log that now includes them.
+- Applies only to runs created while it is enabled. A run keeps the identity scheme it was created with for life, so turning the flag on or off never affects runs already in flight.
+- Requires a World that supports it. A World that does not rejects the run outright rather than mis-numbering its events.
+- Set `0` or `false` to disable, which numbers new runs by ULID as before.
+
 ## Inline execution
 
 ### `WORKFLOW_V2_TIMEOUT_MS`
@@ -209,6 +232,12 @@ These variables are primarily for tests, debugging, or unusual deployments.
 - Delay before the unconsumed-event check fires.
 - Minimum: `10`.
 
+### `WORKFLOW_DEFERRED_CHECK_MAX_GRACE_MS`
+
+- Default: `15000`
+- How long the unconsumed-event check keeps waiting while a step result or hook payload is still on its way to the workflow. An event whose consumer has not been registered yet looks exactly like an orphaned one, so the check waits rather than failing the run.
+- Once this budget is spent the check reports regardless, so a delivery that never lands cannot keep a genuinely orphaned event from being detected.
+
 ### `WORKFLOW_LOCK_POLL_INTERVAL_MS`
 
 - Default: `10`

packages/core/e2e/event-log-race-repro.test.ts

@@ -137,6 +137,31 @@ interface ReproRunResult {
     resumesFailed: number;
     stragglers?: number;
   };
+  /**
+   * The committed log around the divergent event, for corruptions only. A
+   * divergence is a disagreement between the order the log records and the
+   * order a replay reconstructs, so the log's own ordering is the only
+   * evidence that distinguishes the candidate causes — and the run is on an
+   * ephemeral preview deployment, so it has to be captured while the job is
+   * still running rather than read back afterwards.
+   */
+  logSlice?: LogSliceEntry[];
+}
+
+/**
+ * One committed event, projected to the fields that decide replay order:
+ * its position (`slot`), what it resolves (`eventType`/`correlationId`), and
+ * both clock domains. `occurredAt` is the client/VM moment and `createdAt`
+ * the persisted event time the sandbox clock is driven from; a race between a
+ * `sleep` and a step is decided by that clock, so the two have to be
+ * comparable side by side.
+ */
+interface LogSliceEntry {
+  slot: number | string;
+  eventType: string;
+  correlationId?: string;
+  occurredAt?: string;
+  createdAt?: string;
 }
 
 function envNumber(name: string, fallback: number) {
@@ -392,6 +417,98 @@ function validateStormReturn(value: unknown): {
   return { stragglers };
 }
 
+/**
+ * Reads a terminal-failed run's error through `returnValue()`, which hydrates
+ * the stored payload into an Error. Returns undefined when the read itself
+ * fails — the outcome is already known from `errorCode`, so a missing message
+ * degrades the report rather than the classification.
+ */
+async function readFailureMessage(
+  run: Run<unknown>
+): Promise<{ name?: string; message?: string } | undefined> {
+  try {
+    await run.returnValue();
+    return undefined;
+  } catch (err) {
+    if (WorkflowRunFailedError.is(err)) {
+      const cause = err.cause;
+      return {
+        name: cause instanceof Error ? cause.name : err.name,
+        message: cause instanceof Error ? cause.message : err.message,
+      };
+    }
+    return undefined;
+  }
+}
+
+/**
+ * How many events either side of the divergent one to keep. The window has to
+ * span a whole round of the storm — width branches, each with a step create,
+ * start and completion, plus the round's waits — or it can miss the very
+ * event whose position explains the divergence.
+ */
+const LOG_SLICE_RADIUS = envNumber('EVENT_LOG_RACE_REPRO_LOG_SLICE_RADIUS', 45);
+
+/**
+ * Cap on how many corruptions carry a slice. The results JSON is rendered into
+ * a PR comment, and a body over GitHub's limit is rejected outright, so the
+ * slices are a sample rather than a complete record.
+ */
+const LOG_SLICE_MAX_RUNS = envNumber('EVENT_LOG_RACE_REPRO_LOG_SLICE_RUNS', 6);
+
+let logSlicesCaptured = 0;
+
+/** Worlds hand timestamps back as a Date or as the stored ISO string. */
+function isoOrUndefined(value: unknown): string | undefined {
+  if (value instanceof Date) return value.toISOString();
+  return typeof value === 'string' ? value : undefined;
+}
+
+/** Ordinal of a slot-numbered id, or the raw id when it is a ULID. */
+function idOrdinal(id: string): number | string {
+  const body = id.slice(id.indexOf('_') + 1);
+  return /^\d+$/.test(body) ? Number(body) : id;
+}
+
+/**
+ * Reads the committed log and returns the window around the divergent event
+ * named in `message`. Best-effort: the report is a measurement, so a failed
+ * read costs a slice rather than the attempt's classification.
+ */
+async function readLogSlice(
+  runId: string,
+  message: string | undefined
+): Promise<LogSliceEntry[] | undefined> {
+  if (logSlicesCaptured >= LOG_SLICE_MAX_RUNS) return undefined;
+  try {
+    const world = await getWorld();
+    const { data: events } = await world.events.list({ runId });
+    const projected: LogSliceEntry[] = events.map((event) => ({
+      slot: idOrdinal(event.eventId),
+      eventType: event.eventType,
+      correlationId: event.correlationId,
+      occurredAt: isoOrUndefined(event.occurredAt),
+      createdAt: isoOrUndefined(event.createdAt),
+    }));
+
+    // The corruption message names the last divergent event; centre on it when
+    // it is there, and otherwise keep the tail, where a divergence that ran out
+    // of recovery replays ends up.
+    const divergent = message?.match(/evnt_[0-9A-Z]+/)?.[0];
+    const at = divergent
+      ? projected.findIndex((entry) => entry.slot === idOrdinal(divergent))
+      : -1;
+    const centre = at >= 0 ? at : projected.length - 1;
+    logSlicesCaptured += 1;
+    return projected.slice(
+      Math.max(0, centre - LOG_SLICE_RADIUS),
+      centre + LOG_SLICE_RADIUS + 1
+    );
+  } catch {
+    return undefined;
+  }
+}
+
 async function pollTerminalRun(
   run: Run<unknown>,
   startedAt: number,
@@ -431,14 +548,27 @@ async function pollTerminalRun(
         errorCode?: string;
         error?: { name?: string; message?: string };
       };
+      // `runs.get` hands back the raw serialized error payload, not an Error, so
+      // reading `.message` off it yields undefined and the report records the
+      // code with no diagnosis. Read the failure through the public
+      // return-value path, which hydrates it. For a corruption that message
+      // carries the divergent event and what the replay was waiting for, which
+      // is the whole reason to keep the report.
+      const hydrated = await readFailureMessage(run);
+      const outcome = classifyFailure(failure.errorCode);
+      const errorMessage = hydrated?.message ?? failure.error?.message;
       return {
         ...base,
-        outcome: classifyFailure(failure.errorCode),
+        outcome,
         status: runData.status,
         errorCode: failure.errorCode,
-        errorMessage: failure.error?.message,
-        errorName: failure.error?.name,
+        errorMessage,
+        errorName: hydrated?.name ?? failure.error?.name,
         durationMs: Date.now() - startedAt,
+        logSlice:
+          outcome === 'CORRUPTED_EVENT_LOG'
+            ? await readLogSlice(run.runId, errorMessage)
+            : undefined,
       };
     }
 

packages/core/e2e/utils.ts

@@ -659,7 +659,11 @@ async function getRunDiagnostics(tracked: TrackedRun): Promise<string> {
           const elapsed = baseTime
             ? ((event.createdAt?.getTime?.() ?? 0) - baseTime) / 1000
             : 0;
-          const prefix = `  +${elapsed.toFixed(1)}s`;
+          // The event's own id leads the line: on a run numbering its events
+          // by position it says where in the log the event sits, which is what
+          // a diagnosis of an out-of-order or gapped log needs and what the
+          // correlation id below cannot report.
+          const prefix = `  ${event.eventId} +${elapsed.toFixed(1)}s`;
           let detail = event.eventType;
           if ('eventData' in event) {
             const data = (event as any).eventData;