diff --git a/.changeset/slot-only-precondition.md b/.changeset/slot-only-precondition.md new file mode 100644 index 0000000000..a5f6ab3f2d --- /dev/null +++ b/.changeset/slot-only-precondition.md @@ -0,0 +1,8 @@ +--- +'@workflow/core': patch +'@workflow/errors': patch +'@workflow/world': patch +'@workflow/world-vercel': patch +--- + +Replay-context event writes now always report the log position they replayed from, and the `WORKFLOW_PRECONDITION_GUARD` flag is removed. diff --git a/docs/content/docs/v5/api-reference/workflow-errors/precondition-failed-error.mdx b/docs/content/docs/v5/api-reference/workflow-errors/precondition-failed-error.mdx index e9bf2772c3..fc4a994e4f 100644 --- a/docs/content/docs/v5/api-reference/workflow-errors/precondition-failed-error.mdx +++ b/docs/content/docs/v5/api-reference/workflow-errors/precondition-failed-error.mdx @@ -8,9 +8,9 @@ related: - /docs/api-reference/workflow-errors/entity-conflict-error --- -`PreconditionFailedError` is thrown by world implementations when an event creation is rejected because the client's event-log snapshot is stale — either a newer out-of-band event (such as a received hook or a completed step) was recorded after the snapshot the client replayed from, or the snapshot is missing an event recorded at or before it. It corresponds to HTTP 412 Precondition Failed semantics. +`PreconditionFailedError` is thrown by world implementations when an event creation is rejected because the client's event-log snapshot is stale: the log already held more events than the position the creation named. It corresponds to HTTP 412 Precondition Failed semantics. -This only occurs while the optimistic-concurrency guard is enabled (`WORKFLOW_PRECONDITION_GUARD`, on by default — see [Runtime Tuning](/docs/configuration/runtime-tuning)); event creations that carry no snapshot are never rejected with this error. +This only occurs against a world that fences on that position (`capabilities.preconditionGuard` — see [Stale-write rejection](/docs/configuration/runtime-tuning#stale-write-rejection)); event creations that carry no position are never rejected with this error. A world rejects only on evidence and accepts the creation whenever it cannot decide, so this error always means the snapshot really was stale — but not receiving it does not prove the snapshot was current. diff --git a/docs/content/docs/v5/configuration/runtime-tuning.mdx b/docs/content/docs/v5/configuration/runtime-tuning.mdx index b3e1c85824..352a53337f 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -75,21 +75,20 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL - Default: enabled - When a suspension hands newly created steps to the queue, the runtime publishes each step's execution message in parallel with its `step_created` event write instead of sequencing them, cutting a round trip per dispatched step. The message also carries the serialized step input (`stepInput`), so a transient `step_created` write failure (429 / 5xx / transport) still executes the step — the queue consumer idempotently re-ensures the event before running it, converging with the producer's write on the step's correlation ID. This mirrors resilient start (`runInput`) and the lazy hook resume (`hookInput`). -- The runtime falls back to the sequential create-then-publish dispatch automatically when the step input is too large to inline on the queue message, when the run's queue transport cannot carry binary payloads (pre-CBOR spec versions), or — on the `node` VM engine, whose suspension writes carry the [precondition guard](#workflow_precondition_guard) snapshot — when the World enforces that guard (`capabilities.preconditionGuard`; the Vercel World does): a guard-rejected `step_created` must not be materializable through the queue side-channel, and only sequencing the publish after the create gives the message a happens-after edge over the create's guard verdict. The `quickjs` engine's suspension writes are unguarded, so it uses resilient dispatch against every World. +- The runtime falls back to the sequential create-then-publish dispatch automatically when the step input is too large to inline on the queue message, when the run's queue transport cannot carry binary payloads (pre-CBOR spec versions), or — on the `node` VM engine, whose suspension writes are replay-context writes — when the World can [reject a write as stale](#stale-write-rejection) (`capabilities.preconditionGuard`; the Vercel World declares it): a rejected `step_created` must not be materializable through the queue side-channel, and only sequencing the publish after the create gives the message a happens-after edge over the create's verdict. The `quickjs` engine's suspension writes are not replay-context writes, so it uses resilient dispatch against every World. - Producer-side recoveries are reported on the suspension span as `workflow.step.resilient_dispatch_recovered`; a consumer that materialized the event reports `workflow.step.resilient_dispatch_materialized`. - Set `0` to force the sequential dispatch as a kill switch. -### `WORKFLOW_PRECONDITION_GUARD` +### Stale-write rejection -- Default: enabled -- An optimistic-concurrency guard for event creation: replay-context event creations describe the snapshot they replayed from — its latest event timestamp (`stateUpdatedAt`), the number of events it contains (`stateEventCount`), and its event-log cursor (`stateCursor`) — and a backend that supports the guard rejects a creation with 412 ([`PreconditionFailedError`](/docs/api-reference/workflow-errors/precondition-failed-error)) when a newer out-of-band event (a received hook or a completed step) was recorded after that snapshot, or when the snapshot is missing an event recorded at or before it. +- Not a variable: this is what a World declaring `capabilities.preconditionGuard` does, and what the runtime does about it. +- A replay-context event creation names the position it replayed from (`eventCount`, the number of events the replay had loaded), and a World that fences on it rejects the creation with 412 ([`PreconditionFailedError`](/docs/api-reference/workflow-errors/precondition-failed-error)) when the log already held more than that. The position is derived from the run's event IDs, so it is sent only for a run whose World numbers events by position; a run on the older ID scheme, and any caller with no loaded log to be stale against, sends none and is never rejected. - On rejection the runtime restarts the replay in the same invocation from a corrected event log, and falls back to a re-invocation with a fresh replay once the restart budget is spent. The rejected write is never retried as-is: a replay working from a corrected log derives different events, so only a fresh replay may write again. -- When enabled — and the World declares that it enforces the guard (`capabilities.preconditionGuard`; the Vercel World does) — the runtime also keeps the per-step event-log delta optimization (consuming the delta returned by a step's terminal write instead of issuing an extra `events.list` per step) active while the run has an open hook. Without an enforced guard, an open hook disables it. -- While a hook is open on a guard-enforcing deployment, inline steps take the await-then-run path even when optimistic inline start is enabled: the step's `step_started` claim carries the snapshot and is awaited before the body runs, so a claim the backend rejects as stale never executes user code. -- Backends that do not support the guard ignore the snapshot; they must not declare the capability, so guard-dependent optimizations stay off against them even when the flag is set. -- The guard only ever rejects on evidence, and it fails open in every other case: a backend that cannot decide — because its record of recent events is incomplete, has expired, or covers only part of the run's history — must accept the write. A rejection therefore always means the snapshot really was incomplete, but the absence of one does not prove it was complete. Busy runs (wide step fan-outs, high hook volume) are the most likely to skip the check. -- As a result, 412 volume describes a workload rather than the health of a deployment, and a run that never sees one is not evidence the guard is inactive. -- Set `0` to disable. +- Against a fencing World the runtime also keeps the per-step event-log delta optimization (consuming the delta returned by a step's terminal write instead of issuing an extra `events.list` per step) active while the run has an open hook. Without a fence, an open hook disables it. +- While a hook is open on a fencing World, inline steps take the await-then-run path even when optimistic inline start is enabled: the step's `step_started` claim is awaited before the body runs, so a claim rejected as stale never executes user code. +- Worlds that do not fence ignore the position and must not declare the capability, so the dependent optimizations stay off against them. +- A fence only ever rejects on evidence, and fails open in every other case: a World that cannot decide must accept the write. A rejection therefore always means the position really was stale, but the absence of one does not prove it was current. +- The Vercel World declares the capability. It does not fence a run that uses [slot-numbered event IDs](/docs/how-it-works/event-sourcing#event-ids), because such a run has no position to reject: the World assigns each event its slot at commit time and reports back the slots the write skipped over. ### `WORKFLOW_SLOT_GAP_CHECK` @@ -102,7 +101,7 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL ### `WORKFLOW_PRECONDITION_MAX_INPROCESS_RESTARTS` - Default: `3` -- How many times a single invocation restarts its replay in-process after a rejected event creation before it falls back to a re-invocation. +- How many times a single invocation restarts its replay in-process after an event creation is [rejected as stale](#stale-write-rejection) 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. ### `WORKFLOW_PRECONDITION_MAX_REINVOCATIONS` diff --git a/docs/content/worlds/v5/building-a-world.mdx b/docs/content/worlds/v5/building-a-world.mdx index 33062c2044..757ac23a7a 100644 --- a/docs/content/worlds/v5/building-a-world.mdx +++ b/docs/content/worlds/v5/building-a-world.mdx @@ -49,7 +49,7 @@ interface World extends Storage, Queue, Streamer { } ``` -The optional `capabilities` object advertises additional behavior. Set `hookRetention.active` to `true` only when the World implements Hook token retention, `slotEventIds` when it allocates [slot-numbered event IDs](#event-id-allocation), and `preconditionGuard` when it enforces the [precondition guard](#optional-the-event-creation-precondition-guard). The optional `start()` method initializes background tasks (for example, queue polling). The optional `close()` method releases resources like connection pools and listeners. The optional `getEncryptionKeyForRun()` method returns the AES-256 key used to encrypt data for a run; if it is not implemented, encryption is disabled. +The optional `capabilities` object advertises additional behavior. Set `hookRetention.active` to `true` only when the World implements Hook token retention, `slotEventIds` when it allocates [slot-numbered event IDs](#event-id-allocation), and `preconditionGuard` when it can [reject a stale write](#optional-rejecting-a-stale-write). The optional `start()` method initializes background tasks (for example, queue polling). The optional `close()` method releases resources like connection pools and listeners. The optional `getEncryptionKeyForRun()` method returns the AES-256 key used to encrypt data for a run; if it is not implemented, encryption is disabled. ## The Event Log Model @@ -119,29 +119,20 @@ Two properties have to hold, and both are about what a reader can conclude from `events.create()` params carry `eventCount`: how many events the writer held in the log it replayed from, which is the position it expects to land on minus one. Attempt `eventCount + 1`. When that position is taken, **do not reject the write**. Advance to the next free position, commit there, and return the events occupying the positions you skipped over on the success response, in `events` with a matching `cursor` and `hasMore`. The writer merges them into its own log and replays once, rather than paying a second round trip to discover it was behind. A caller with a stale count is the normal case for a fan-out, and rejecting it would serialize writes the runtime issues in parallel. -`eventCount` supersedes the `stateUpdatedAt` / `stateEventCount` / `stateCursor` triple below for a World that allocates positions. The triple approximates a position with a timestamp watermark plus a count of events at or below it, which a complete-but-stale snapshot passes: every event the writer holds is at or below its own watermark, so the count matches and no fence fires. A dense position has no such blind spot. +### Optional: Rejecting a Stale Write -### Optional: The Event Creation Precondition Guard +A replay writes events derived from the event log it loaded, so a write made from an event log that no longer matches the store can commit events no correct replay would produce. Rejecting one means answering `events.create()` with a `PreconditionFailedError` when the run's log already holds more events than the caller's `eventCount` says it had loaded. -A replay writes events derived from the event log it loaded, so a write made from an event log that no longer matches the store can commit events no correct replay would produce. To let a World fence those writes, `events.create()` params may carry a description of the snapshot the caller replayed from: - -- `stateUpdatedAt` — the timestamp encoded in the latest loaded event's ID. -- `stateEventCount` — how many events the caller loaded. Only sent together with `stateUpdatedAt`; ignore a count that arrives without one. -- `stateCursor` — the caller's event-log cursor. Advisory, and only meaningful on the reject path (see below). - -Enforcing the guard means rejecting the creation with a `PreconditionFailedError` when either holds: - -1. An event was recorded after `stateUpdatedAt` that the caller could not have loaded. -2. More events were recorded at or before `stateUpdatedAt` than `stateEventCount` — the caller's snapshot is missing one. +Check first whether the field can reach you at all. The runtime derives `eventCount` from the highest slot in the log it loaded, and sends nothing when any loaded event ID is not a slot, so a World whose IDs are not positions never receives it and has nothing to fence on. A World that does allocate positions has the better mechanism already: commit above the contention and report the skipped events back, which costs the writer no replay. That leaves this worth implementing in one case — your store allocates positions, but not atomically with the commit, so refusing a stale write is safer than accepting one out of order. Two rules make this safe: -- **Compare at or below `stateUpdatedAt`, not strictly below.** Events routinely share a millisecond with the caller's latest event, and a strict comparison misses exactly those. -- **Every uncertainty must allow the write.** If your record of a run's events is incomplete, expired, or cannot answer the question, accept the creation. A rejection must always mean a real discrepancy, because the runtime responds to one by discarding a replay. +- **Only reject on evidence.** If your record of a run's events is incomplete, expired, or cannot answer the question, accept the creation. A rejection must always mean a real discrepancy, because the runtime responds to one by discarding a replay. +- **Accept a creation that carries no `eventCount`.** It came from a caller with no loaded log to be stale (a queued step body, an out-of-band writer), not from a caller claiming the log is empty. A rejection may optionally carry the events the caller was missing, as `{ events, cursor }` on the error's `details`. Only include them when you can prove the set is complete — that those events fully account for the discrepancy and are not truncated — and that every one of them belongs to the run being written. The runtime merges them straight into the replay's event log, so anything else there is worse than no delta at all. Otherwise omit them, and the runtime performs a full reload instead. -A World that enforces the guard should declare `capabilities.preconditionGuard`, which the runtime also reads to keep event-log delta optimizations enabled. A World that ignores these params must not declare it. +Declare `capabilities.preconditionGuard` if your World can refuse a write this way. The runtime reads it as "a write can come back refused", not as a promise that any particular one will be, and three behaviors key on it: the per-step event-log delta optimization stays enabled while the run has an open hook, an inline step's `step_started` claim is awaited before the body runs, and a `step_created` publish is sequenced after the create on the `node` VM engine. A World that accepts `eventCount` and ignores it must leave the capability unset — sending a position is not the same as one being enforced. ## Queue Interface diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index 8a2e153e95..108b07d870 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -8,6 +8,7 @@ import { import { type Event, SPEC_VERSION_CURRENT, + slotToEventId, type WorkflowRun, } from '@workflow/world'; import { ulid } from 'ulid'; @@ -2450,20 +2451,13 @@ describe('workflowEntrypoint turbo mode', () => { }); describe('workflowEntrypoint inline-delta gate with open hooks', () => { - const ORIG_GUARD = process.env.WORKFLOW_PRECONDITION_GUARD; const ORIG_OPT = process.env.WORKFLOW_OPTIMISTIC_INLINE_START; beforeEach(() => { - delete process.env.WORKFLOW_PRECONDITION_GUARD; delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START; deltaGateBodyRuns = []; }); afterEach(() => { - if (ORIG_GUARD === undefined) { - delete process.env.WORKFLOW_PRECONDITION_GUARD; - } else { - process.env.WORKFLOW_PRECONDITION_GUARD = ORIG_GUARD; - } if (ORIG_OPT === undefined) { delete process.env.WORKFLOW_OPTIMISTIC_INLINE_START; } else { @@ -2509,7 +2503,7 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => { * off and the initial events.list — which supplies the cursor the delta * diffs against — runs). Returns the events.create mock so tests can * inspect the step-terminal write's params for `sinceCursor` and the - * step_started claims' params for `stateUpdatedAt`. + * step_started claims' params for `eventCount`. */ async function driveDeltaGate( runId: string, @@ -2542,10 +2536,11 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => { const durableEvents: Event[] = []; const recordEvent = (data: any): Event => { - // ULID-shaped event IDs so the runtime's stateUpdatedAt snapshot - // (derived from the latest event id's ULID timestamp) is computable. + // Slot-numbered event ids, so the runtime's snapshot (the highest slot + // its loaded log occupies) is computable. This is the only kind of run + // that reaches a fencing backend. const created = { - eventId: `evnt_${ulid()}`, + eventId: slotToEventId(durableEvents.length + 1), runId, createdAt: new Date(), ...data, @@ -2658,8 +2653,7 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => { return call?.[2] as { sinceCursor?: string } | undefined; } - it('requests the inline delta despite the open hook when the precondition guard is enabled and the World enforces it', async () => { - process.env.WORKFLOW_PRECONDITION_GUARD = '1'; + it('requests the inline delta despite the open hook when the World enforces the precondition guard', async () => { const { res, eventsCreate } = await driveDeltaGate( 'wrun_delta_gate_guard_on', { capabilities: { preconditionGuard: true } } @@ -2678,7 +2672,7 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => { ); }); - it('does not request the inline delta with an open hook when the guard is disabled', async () => { + it('does not request the inline delta with an open hook when no World fence exists', async () => { const { res, eventsCreate } = await driveDeltaGate( 'wrun_delta_gate_guard_off' ); @@ -2688,20 +2682,7 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => { expect(stepCompletedParams(eventsCreate)?.sinceCursor).toBeUndefined(); }); - it('does not request the inline delta when the env flag is set but the World does not enforce the guard', async () => { - process.env.WORKFLOW_PRECONDITION_GUARD = '1'; - // No capabilities declared: the env flag only makes the runtime SEND - // snapshots — a World that ignores stateUpdatedAt provides no 412 fence, - // so the relaxation must fail closed to the conservative gate. - const { res, eventsCreate } = await driveDeltaGate( - 'wrun_delta_gate_guard_no_capability' - ); - expect(res.status).toBe(204); - expect(stepCompletedParams(eventsCreate)?.sinceCursor).toBeUndefined(); - }); - it('restarts the replay in-process and still completes the run when a stale lazy claim is rejected by the guard (interleaved hook_received)', async () => { - process.env.WORKFLOW_PRECONDITION_GUARD = '1'; // Simulates the interleaving the fence exists for: after step A's // terminal write, an out-of-band hook_received bumps the run's marker; // the next replay (working from a view that misses it) schedules step B, @@ -2714,7 +2695,7 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => { rejectClaimOnce: { stepName: 'deltaGateStepB', error: new PreconditionFailedError( - 'stale stateUpdatedAt: a newer outside event exists' + 'stale snapshot: a newer outside event exists' ), }, } @@ -2722,17 +2703,17 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => { // The handler responds normally: the rejection restarts the replay inside // this delivery, never a run_failed. expect(res.status).toBe(204); - // Step B's claim was issued from a loaded (non-empty) log, so it carried - // the guard snapshot — that is what lets the backend fence it. (The very - // first batch of a run loads an empty log and has no snapshot to send; - // the guard is best-effort there, matching the suspension creates.) + // Step B's claim was issued from a loaded (non-empty) log, so it named the + // position it was decided against. (The very first batch of a run loads an + // empty log and has no position to name; reporting is best-effort there, + // matching the suspension creates.) const rejectedClaim = eventsCreate.mock.calls.find( (c) => (c[1] as any).eventType === 'step_started' && ((c[1] as any).eventData as { stepName?: string })?.stepName === 'deltaGateStepB' ); - expect(typeof (rejectedClaim?.[2] as any)?.stateUpdatedAt).toBe('number'); + expect(typeof (rejectedClaim?.[2] as any)?.eventCount).toBe('number'); // The fenced claim's body never ran: step B executes exactly once, on the // restarted replay whose claim the backend accepted. expect(deltaGateBodyRuns).toEqual(['B']); @@ -2760,7 +2741,6 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => { }); it('suppresses optimistic start on guarded stale-sensitive batches: a 412-fenced step never runs its body even with WORKFLOW_OPTIMISTIC_INLINE_START=1', async () => { - process.env.WORKFLOW_PRECONDITION_GUARD = '1'; process.env.WORKFLOW_OPTIMISTIC_INLINE_START = '1'; // Same interleaving as above, but with optimistic start enabled globally. // Without suppression, executeStep would begin step B's body immediately @@ -2777,7 +2757,7 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => { rejectClaimOnce: { stepName: 'deltaGateStepB', error: new PreconditionFailedError( - 'stale stateUpdatedAt: a newer outside event exists' + 'stale snapshot: a newer outside event exists' ), }, } diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index df913d79d4..0f2db2550d 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -68,22 +68,22 @@ import { type ReenqueueArgs, } from './runtime/deployment-guard.js'; import { + absorbSkippedSlotReport, appendUniqueEvents, getQueueOverhead, getWorkflowQueueName, handleHealthCheckMessage, insertEventByEventId, - isPreconditionGuardEnabled, isSlotGapCheckEnabled, type LoadedEventLog, loadWorkflowRunEvents, memoizeEncryptionKey, - mergeReportedEvents, parseHealthCheckPayload, preconditionEventDelta, - preconditionSnapshotParams, queueMessage, + type SlotSnapshotParams, settleEventSlotGap, + slotSnapshotParams, stepDispatchIdempotencyKey, withHealthCheck, } from './runtime/helpers.js'; @@ -890,10 +890,11 @@ export function workflowEntrypoint( params?: CreateEventParams ) => { const sinceCursor = deltaRequestCursor(data, params); + const withSnapshot = { ...slotSnapshot(), ...params }; const result = await replayRecoveryReporter.withEventCreate( sinceCursor === undefined - ? params - : { ...params, sinceCursor }, + ? withSnapshot + : { ...withSnapshot, sinceCursor }, (p) => world.events.create(runId, data, p) ); if (sinceCursor !== undefined) { @@ -902,6 +903,30 @@ export function workflowEntrypoint( return result; }; + /** + * The slot snapshot for a write issued from this loop: how + * much of the run's log the decision behind it was made + * against. + * + * Every write goes through here, including the ones that ask + * for an inline delta. The two answer different questions and + * a World is free to serve both: the cursor says where to + * start listing from, the slot says which events this writer + * had already decided without. Worlds that treat the delta as + * the better answer simply ignore the slot. + * + * Taken from whatever is loaded, including the stale + * `loadAfter` state. Understating is safe: the World reports + * back a wider span than strictly needed, and + * `mergeReportedEvents` drops what the log already holds. + * Overstating is not, so there is no guessing when nothing is + * loaded at all. + */ + const slotSnapshot = (): SlotSnapshotParams => + eventLog.type === 'loadAll' + ? {} + : slotSnapshotParams(eventLog.events); + /** * The cursor to ask for an inline delta against, or * undefined to not ask. @@ -942,6 +967,15 @@ export function workflowEntrypoint( * already has the log and the loop's incremental * `events.list` finds nothing left to fetch. * + * Deliberately not `absorbSkippedSlotReport`, even though the + * `hasMore` half of the two policies coincides. A delta + * extends the tail and carries a cursor that has to move with + * it, so it appends without re-sorting and needs the cursor + * guard below; a skipped-slot report is a window strictly + * below the write, carries no cursor, and has to be sorted + * back into place. Sharing an implementation would mean one + * of them doing the other's work. + * * Declining is always safe — an unabsorbed delta is a delta * the next `events.list` returns — so the guards are free to * be strict: @@ -2556,6 +2590,7 @@ export function workflowEntrypoint( requestId, attempt: metadata.attempt, limitMs: replayBudget.configuredLimitMs, + slotSnapshot: slotSnapshot(), }); // Only the terminal attempt returns, after run_failed is // durable. Earlier attempts reject for queue redelivery. @@ -2738,32 +2773,20 @@ export function workflowEntrypoint( try { const created = await createEvent(waitEvent, { requestId, - ...preconditionSnapshotParams( - eventLog.events, - eventLog.cursor - ), }); // Bump-and-report: fold what this write skipped over - // into the snapshot the remaining waits are guarded - // against, so each asks for a slot above it. + // into the log, which is what `createEvent` reads the + // slot snapshot off, so each remaining wait asks for + // a slot above it. // - // Only a complete answer. `hasMore` means the World - // returned part of what it was asked for, and the - // missing-completion check below is what decides - // whether this handler still has to fetch. Folding in - // a partial page would make the log look like it - // holds the completion when the rest of the page is - // still unread, so the fetch would be skipped on a - // snapshot that is short of the World's. - if ( - created.events?.length && - created.hasMore !== true - ) { - mergeReportedEvents( - eventLog.events, - created.events - ); - } + // The truncated case matters twice over here. Beyond + // the position accounting `absorbSkippedSlotReport` + // rejects it for, the missing-completion check below + // decides from this log whether the handler still has + // to fetch, so a partial page would make the log look + // like it holds a completion whose page is unread and + // skip the fetch on a snapshot short of the World's. + absorbSkippedSlotReport(eventLog.events, created); } catch (err) { if (EntityConflictError.is(err)) { runtimeLogger.info( @@ -2968,12 +2991,12 @@ export function workflowEntrypoint( }); replayRecoveryReporter.activate(); - // Workflow completed. Send the snapshot but do NOT - // reload-and-retry the create in place: `result` was - // computed by this replay, so a stale (412) rejection must - // force a *fresh replay* (which may observe the new event - // and produce a different result), not re-commit the stale - // result. The catch below restarts the replay in-process. + // Workflow completed. Do NOT reload-and-retry the create + // in place: `result` was computed by this replay, so a + // stale (412) rejection must force a *fresh replay* (which + // may observe the new event and produce a different + // result), not re-commit the stale result. The catch below + // restarts the replay in-process. try { // Turbo: a workflow that finishes with no steps reaches // here before the backgrounded run_started; order the @@ -2985,13 +3008,7 @@ export function workflowEntrypoint( specVersion: SPEC_VERSION_CURRENT, eventData: { output: result }, }, - { - requestId, - ...preconditionSnapshotParams( - eventLog.events, - eventLog.cursor - ), - } + { requestId } ); } catch (err) { if ( @@ -3672,11 +3689,8 @@ export function workflowEntrypoint( // does not bump the outside-event marker, so nothing // fences a replay from the stale delta. // - No open (or this-suspension-created) hook — UNLESS - // the precondition guard is enabled AND the World - // declares it actually enforces the guard - // (`capabilities.preconditionGuard`; the env flag - // alone only makes the runtime SEND snapshots, which - // an unsupporting backend ignores — no fence). The + // the World declares it fences stale writes + // (`capabilities.preconditionGuard`). The // delta snapshots the log at the step_completed // write but is consumed on the next replay, so an // out-of-band `hook_received` landing in that window @@ -3685,24 +3699,24 @@ export function workflowEntrypoint( // it. That staleness is qualitatively the same // read-to-write race the fetch path already has (an // event can land right after `events.list` returns - // and before the suspension's writes); with an - // enforced guard it is also fenced: `hook_received` - // bumps the per-run outside-event marker, so every - // durable write the stale replay attempts is - // rejected with 412 — its guarded suspension creates - // (retried over the reloaded log, or exhausted into - // a queue re-invocation), AND the lazy step_started - // claim of its next inline step, which carries the - // snapshot too (threaded below via - // `stateUpdatedAt`; on rejection the batch is - // abandoned and re-invoked for a fresh replay, so a - // stale view can never commit a step). Hooks created + // and before the suspension's writes); on a fencing + // World it is also fenced: `hook_received` bumps the + // per-run outside-event marker, so every durable + // write the stale replay attempts is rejected with + // 412 — its guarded suspension creates (retried over + // the reloaded log, or exhausted into a queue + // re-invocation), AND the lazy step_started claim of + // its next inline step, which carries the snapshot + // too (threaded below via `slotSnapshot`; on + // rejection the batch is abandoned and re-invoked for + // a fresh replay, so a stale view can never commit a + // step). Hooks created // by THIS suspension are inside the delta (their // `hook_created` lands before the step-terminal // write), so only their `hook_received` responses - // are subject to the same fenced window. Without an - // enforced guard there is no fence, so keep the - // conservative gate. + // are subject to the same fenced window. Without a + // fencing World there is nothing to reject a stale + // write, so keep the conservative gate. // - With no hook or wait open at all, the only // out-of-band writer is cancellation, which is safe // to observe one iteration late. See @@ -3712,12 +3726,11 @@ export function workflowEntrypoint( // own events and the per-write delta would be partial, so // the delta is not requested (the gate below is false for // multi-step) and the next iteration does a normal fetch. - // Whether the precondition guard is actually in force: - // enabled by env AND enforced by the World. The env - // flag alone only makes the runtime send snapshots, - // which an unsupporting backend ignores (no fence). + // Whether the World fences a stale write. Sending a + // snapshot is not the same as it being enforced: a + // World that does not declare the capability ignores + // what it is sent, so nothing rejects. const guardEnforced = - isPreconditionGuardEnabled() && world.capabilities?.preconditionGuard === true; const requestInlineDelta = @@ -3836,20 +3849,19 @@ export function workflowEntrypoint( turbo, }); - // Precondition-guard snapshot for the inline - // step_started claims: the lazy claim is the first - // durable write of a hot-path step (its step_created - // is deferred), so without a snapshot it would bypass - // the guard entirely and a stale replay could claim — - // and commit — a step scheduled off a view that misses - // an event it never loaded. - // `preconditionSnapshotParams` returns an empty object - // when the guard env flag is off, so this is a no-op - // outside guarded deployments; Worlds that don't - // enforce the guard ignore it. - const inlineClaimSnapshot = preconditionSnapshotParams( - eventLog.events, - eventLog.cursor + // Slot snapshot for the inline step_started claims: the + // lazy claim is the first durable write of a hot-path + // step (its step_created is deferred), so without a + // snapshot it would name no position at all and a stale + // replay could claim — and commit — a step scheduled off + // a view that misses an event it never loaded. + // + // Taken here rather than inside the executor because + // this is the view the scheduling decision was made + // against. The executor advances from it as its own + // writes land; see `slotSnapshot` in step-executor. + const inlineClaimSnapshot = slotSnapshotParams( + eventLog.events ); // TTR: consumed by this batch. Every step is handed @@ -3943,7 +3955,7 @@ export function workflowEntrypoint( // see suppressOptimisticStart above. suppressOptimisticStart, runReadyBarrier, - preconditionSnapshot: inlineClaimSnapshot, + slotSnapshot: inlineClaimSnapshot, ...(stepIndex === 0 && s.lazyStepInput !== undefined && latencyTracking @@ -4376,16 +4388,14 @@ export function workflowEntrypoint( // type identity and custom properties round-trip // through the event log. // - // Precondition-guard asymmetry: unlike `run_completed`, - // this terminal `run_failed` sends no `stateUpdatedAt` - // snapshot, so it is never 412-rejected even if a hook - // landed mid-replay and could have changed the path that - // threw. This is intentional and fail-open: a spurious - // failure is recoverable (the run can be re-run from the + // Like every other write from this loop it carries the + // slot snapshot, so a World that fences can reject it + // and a slot-allocating one reports back what the + // failing replay had not seen. Fail-open on purpose + // where neither applies: a spurious failure is + // recoverable (the run can be re-run from the // dashboard), whereas a spurious *completion* commits a - // wrong result. Guarding this write symmetrically would - // also need the loaded event log, which is scoped to the - // replay `try` above and not available in this catch. + // wrong result. try { // Turbo: order the terminal write after the // backgrounded run_started so the run exists. diff --git a/packages/core/src/runtime/helpers.test.ts b/packages/core/src/runtime/helpers.test.ts index 4623f006a4..15b84ba5e0 100644 --- a/packages/core/src/runtime/helpers.test.ts +++ b/packages/core/src/runtime/helpers.test.ts @@ -18,15 +18,14 @@ import { handleHealthCheckMessage, healthCheck, insertEventByEventId, - latestEventStateUpdatedAt, loadWorkflowRunEvents, maxEventSlot, memoizeEncryptionKey, mergeReportedEvents, preconditionEventDelta, - preconditionSnapshotParams, SLOT_GAP_RECHECK_ATTEMPTS, settleEventSlotGap, + slotSnapshotParams, } from './helpers.js'; // Mock the logger to suppress output during tests @@ -599,135 +598,11 @@ const makeUlidEvent = (time: number): Event => createdAt: new Date(time), }) as unknown as Event; -describe('latestEventStateUpdatedAt', () => { - it('returns undefined for an empty event list', () => { - expect(latestEventStateUpdatedAt([])).toBeUndefined(); - }); - - it('decodes the ULID time of the newest event, stripping the prefix', () => { - const time = 1_700_000_000_000; - // ULID time resolution is whole milliseconds. - expect( - latestEventStateUpdatedAt([ - makeUlidEvent(time - 1000), - makeUlidEvent(time), - ]) - ).toBe(time); - }); - - it('reports the maximum, not the tail, when the log is not id-ordered', () => { - // A World's canonical order need not be event-id order (world-local orders - // by `(createdAt, eventId)`), so the watermark cannot be read off the tail. - const time = 1_700_000_002_000; - - expect( - latestEventStateUpdatedAt([ - makeUlidEvent(time), - makeUlidEvent(1_700_000_000_000), - makeUlidEvent(1_700_000_001_000), - ]) - ).toBe(time); - }); - - it('returns undefined when the newest event id is not a decodable ULID', () => { - expect( - latestEventStateUpdatedAt([makeEvent('evnt_not-a-ulid')]) - ).toBeUndefined(); - }); -}); - -describe('preconditionSnapshotParams', () => { - let originalGuard: string | undefined; - - beforeEach(() => { - originalGuard = process.env.WORKFLOW_PRECONDITION_GUARD; - process.env.WORKFLOW_PRECONDITION_GUARD = '1'; - }); - - afterEach(() => { - if (originalGuard !== undefined) { - process.env.WORKFLOW_PRECONDITION_GUARD = originalGuard; - } else { - delete process.env.WORKFLOW_PRECONDITION_GUARD; - } - }); - - it('sends the watermark, the count and the cursor together', () => { - const time = 1_700_000_000_000; - const events = [makeUlidEvent(time - 1000), makeUlidEvent(time)]; - - expect(preconditionSnapshotParams(events, 'eid:abc')).toEqual({ - stateUpdatedAt: time, - stateEventCount: events.length, - stateCursor: 'eid:abc', - }); - }); - - it('sends the count without a cursor when the caller has none', () => { - const time = 1_700_000_000_000; - - expect(preconditionSnapshotParams([makeUlidEvent(time)], null)).toEqual({ - stateUpdatedAt: time, - stateEventCount: 1, - }); - }); - - it('sends the snapshot by default (guard is on unless disabled)', () => { - delete process.env.WORKFLOW_PRECONDITION_GUARD; - const time = 1_700_000_000_000; - - expect( - preconditionSnapshotParams([makeUlidEvent(time)], 'eid:abc') - ).toEqual({ - stateUpdatedAt: time, - stateEventCount: 1, - stateCursor: 'eid:abc', - }); - }); - - it('omits every field when the guard is disabled', () => { - process.env.WORKFLOW_PRECONDITION_GUARD = '0'; - - expect( - preconditionSnapshotParams([makeUlidEvent(1_700_000_000_000)], 'eid:abc') - ).toEqual({}); - }); - - it('omits every field on an empty log', () => { - expect(preconditionSnapshotParams([], 'eid:abc')).toEqual({}); - }); - - it('omits every field when the latest event id is not a decodable ULID', () => { - // A count without a watermark would be meaningless to the backend, so the - // three fields have to fail open together. - expect( - preconditionSnapshotParams([makeEvent('evnt_not-a-ulid')], 'eid:abc') - ).toEqual({}); - }); -}); - -describe('preconditionSnapshotParams on a slot-numbered run', () => { - let originalGuard: string | undefined; - - beforeEach(() => { - originalGuard = process.env.WORKFLOW_PRECONDITION_GUARD; - process.env.WORKFLOW_PRECONDITION_GUARD = '1'; - }); - - afterEach(() => { - if (originalGuard !== undefined) { - process.env.WORKFLOW_PRECONDITION_GUARD = originalGuard; - } else { - delete process.env.WORKFLOW_PRECONDITION_GUARD; - } - }); - - it('sends eventCount instead of the ULID triple', () => { +describe('slotSnapshotParams', () => { + it('sends the highest slot the loaded log occupies', () => { const events = [1, 2, 3].map((slot) => makeEvent(slotToEventId(slot))); - expect(preconditionSnapshotParams(events, 'eid:abc')).toEqual({ - eventCount: 3, - }); + expect(slotSnapshotParams(events)).toEqual({ eventCount: 3 }); }); it('reports the highest slot, not the number of events', () => { @@ -737,37 +612,35 @@ describe('preconditionSnapshotParams on a slot-numbered run', () => { // on every single create. const events = [1, 2, 5].map((slot) => makeEvent(slotToEventId(slot))); - expect(preconditionSnapshotParams(events, 'eid:abc')).toEqual({ - eventCount: 5, - }); + expect(slotSnapshotParams(events)).toEqual({ eventCount: 5 }); }); it('is invariant under the order the World returned the log in', () => { const forward = [1, 2, 3].map((slot) => makeEvent(slotToEventId(slot))); - expect(preconditionSnapshotParams([...forward].reverse(), null)).toEqual( - preconditionSnapshotParams(forward, null) + expect(slotSnapshotParams([...forward].reverse())).toEqual( + slotSnapshotParams(forward) ); }); - it('omits eventCount when the guard is disabled', () => { - process.env.WORKFLOW_PRECONDITION_GUARD = '0'; + it('sends nothing on an empty log', () => { + expect(slotSnapshotParams([])).toEqual({}); + }); - expect( - preconditionSnapshotParams([makeEvent(slotToEventId(1))], null) - ).toEqual({}); + it('sends nothing for a run whose events are not slot-numbered', () => { + expect(slotSnapshotParams([makeUlidEvent(1_700_000_000_000)])).toEqual({}); }); - it('falls back to the ULID triple when one event is not a slot', () => { + it('sends nothing when one event of the log is not a slot', () => { // A log may not mix the two schemes. If it somehow does, the slot reading - // is meaningless, so the run is treated as ULID-numbered. - const time = 1_700_000_000_000; - const events = [makeEvent(slotToEventId(1)), makeUlidEvent(time)]; + // is meaningless, and a count derived from part of the log would understate + // the writer's position in a way the World cannot detect. + const events = [ + makeEvent(slotToEventId(1)), + makeUlidEvent(1_700_000_000_000), + ]; - expect(preconditionSnapshotParams(events, null)).toEqual({ - stateUpdatedAt: time, - stateEventCount: 2, - }); + expect(slotSnapshotParams(events)).toEqual({}); }); }); @@ -1014,21 +887,16 @@ describe('appendUniqueEvents', () => { expect(target.map((e) => e.eventId)).toEqual([b.eventId, a.eventId]); }); - it('leaves the watermark correct even when the merge is not id-ordered', () => { - // Why the merge needs no sort: the snapshot reads the maximum ULID time - // across the log rather than the tail, so an out-of-order tail costs nothing - // and every loaded event stays at or below the watermark. - const time = 1_700_000_002_000; - const target = [makeUlidEvent(1_700_000_000_000), makeUlidEvent(time)]; + it('leaves the snapshot correct even when the merge is not id-ordered', () => { + // Why the merge needs no sort of its own: the snapshot reads the maximum + // slot across the log rather than the tail, so an out-of-order tail costs + // nothing. + const target = [makeEvent(slotToEventId(1)), makeEvent(slotToEventId(3))]; - appendUniqueEvents(target, [makeUlidEvent(1_700_000_001_000)]); + appendUniqueEvents(target, [makeEvent(slotToEventId(2))]); - expect(latestEventStateUpdatedAt(target)).toBe(time); - expect(preconditionSnapshotParams(target, 'eid:abc')).toEqual({ - stateUpdatedAt: time, - stateEventCount: 3, - stateCursor: 'eid:abc', - }); + expect(target.at(-1)?.eventId).toBe(slotToEventId(2)); + expect(slotSnapshotParams(target)).toEqual({ eventCount: 3 }); }); }); diff --git a/packages/core/src/runtime/helpers.ts b/packages/core/src/runtime/helpers.ts index 7c8d1cfcee..be1d867655 100644 --- a/packages/core/src/runtime/helpers.ts +++ b/packages/core/src/runtime/helpers.ts @@ -22,7 +22,6 @@ import { resolveQueueNamespace, SPEC_VERSION_CURRENT, SPEC_VERSION_LEGACY, - ulidToDate, } from '@workflow/world'; import { monotonicFactory } from 'ulid'; import { runtimeLogger } from '../logger.js'; @@ -491,12 +490,14 @@ function recordRequestedEventCursor( * * Events are appended in the order the World returned them, and are not * re-sorted: a World's canonical order is its own, and the runtime cannot - * reproduce it from event ids alone. `world-vercel` orders by event id, while - * `world-local` orders by `(createdAt, eventId)` and deliberately re-mints keys - * so that the two diverge. Every append source is already in canonical order + * reproduce it from event ids alone. That is true even though the id schemes in + * this repo happen to sort correctly today — a slot-numbered run orders by id + * everywhere, and `world-local` falls back to `(createdAt, eventId)` only for a + * run minted before slots, where it re-mints keys and the two orders diverge. + * Every append source is already in canonical order * relative to the tail (a cursor-delimited page, or a write-response delta), so * receipt order is the order to keep. Nothing downstream may assume the tail is - * the newest event — see {@link latestEventStateUpdatedAt}. + * the newest event — see {@link maxEventSlot}. */ export function appendUniqueEvents( target: Event[], @@ -525,8 +526,15 @@ export function appendUniqueEvents( * `preloadedEvents` is loaded `sortOrder: 'asc'` and is never re-sorted * client-side, so a `hook_received` spliced in by the lazy-resume consumer must * land in `eventId` order — a plain `push` would place a late-committing - * earlier event after events that sort before it, corrupting replay. Event IDs - * are ULIDs, so lexicographic string order matches commit order. + * earlier event after events that sort before it, corrupting replay. + * + * Lexicographic string order matches commit order under both id schemes, which + * is why this needs no slot gate the way {@link mergeReportedEvents} does: a + * slot id is a fixed-width zero-padded position, and a ULID is monotonic by + * construction. What differs is the guarantee. On a slot run the id order *is* + * the World's canonical order, while on a ULID run it is the runtime's best + * reconstruction of it, which is enough for a single splice into a page that + * was already loaded in that order. */ export function insertEventByEventId(target: Event[], event: Event): void { // Linear scan from the end: the spliced event is almost always the newest @@ -710,86 +718,55 @@ export interface LoadedEventLog { cursor: string | null; } -/** - * Whether the optimistic-concurrency guard for event creation is enabled. - * **On by default** where the runtime executes: replay-context creates send a - * `stateUpdatedAt` snapshot (and can be rejected with 412 by a supporting - * backend) unless `WORKFLOW_PRECONDITION_GUARD` is set to `0`. Backends without - * guard support ignore the snapshot, so enabling by default is - * backward-compatible. - */ -export function isPreconditionGuardEnabled(): boolean { - return process.env.WORKFLOW_PRECONDITION_GUARD !== '0'; -} - /** * Whether a replay refuses to run over a log with a hole in it (see * {@link findEventSlotGap}). **On by default**; set * `WORKFLOW_SLOT_GAP_CHECK=0` to replay across holes instead. * - * The switch exists because the check trades one failure for another. A hole is - * a position claimed by a write that then failed, so most of them stand for an - * event that never happened and replaying past one is correct. But a hole - * standing for an event that *did* happen is indistinguishable from that, and - * replaying past that one produces a run whose result is wrong with nothing to - * show for it. Failing loudly is the recoverable side of the trade, and this is - * the way back out if a fleet turns out to carry benign holes. + * A World that allocates a position at the moment it commits leaves no hole + * behind when a write fails, so density is a property the log has by + * construction rather than one this check maintains. What the check is for is + * the reads and the Worlds where that does not hold, and by the time it runs + * the benign explanations are spent: a position missing because a concurrent + * commit is not visible yet clears on a re-read, which is what + * {@link settleEventSlotGap} does first. + * + * What is left is a hole that persists, and its two causes are + * indistinguishable from the log. Either a World allocated the position outside + * the commit and lost the write, in which case nothing happened there and + * replaying past it is correct, or an event that did happen is missing, in + * which case replaying past it decides a branch on absence and produces a wrong + * result with nothing to show for it. Failing is the recoverable side of that + * trade, and this is the way back out if a fleet turns out to carry holes of + * the first kind. */ export function isSlotGapCheckEnabled(): boolean { return process.env.WORKFLOW_SLOT_GAP_CHECK !== '0'; } -/** - * The `stateUpdatedAt` value to send with a replay-context event creation: the - * *maximum* ULID time (epoch ms) over the events the runtime has loaded. Returns - * `undefined` when there are no events or that id is not a decodable ULID. +/* + * Merging into a log a replay is midway through reading. * - * It is the maximum rather than the tail's because the loaded log is in the - * World's canonical order, which is not necessarily event-id order (see - * {@link appendUniqueEvents}). The maximum is what lets the count sent alongside - * it be read as "events at or below this watermark": every loaded event is at or - * below it, so the count is exactly `events.length`. Reading the tail instead - * would understate the watermark on a World whose order is not id-ordered, which - * is safe (it can only weaken detection) but needlessly imprecise. + * Three paths add events to a loaded log after the replay has started: a + * bump-and-report write hands back the slots it skipped ({@link + * mergeReportedEvents}), an inline delta extends the tail (`absorbCreateDelta` + * in `runtime.ts`), and a listed page appends ({@link appendUniqueEvents}). + * They look alarming — the replay is reading an array while something else + * writes to it — and they are safe for one reason worth stating plainly, since + * every correctness argument in this file leans on it. * - * The maximum is found by lexicographic id comparison, decoding only once: the - * 26-character Crockford ULID encodes its timestamp in the leading 10 - * characters, so the greatest id also carries the greatest time. + * **An event in the log is a fact, and a longer log cannot retract one.** The + * log is append-only and positions are never reused, so a replay that produced + * event E from the prefix it had loaded has made E true for every replay that + * follows. A later replay reading a fuller log does not get to decide E should + * not be there; it consumes E and reconciles to it, which is what the events + * consumer does when it walks a log holding writes from a replay that raced it. * - * Granularity: snapshots are epoch-milliseconds, and the backend allows an - * equal-timestamp snapshot (an up-to-date client must not be rejected). Two - * out-of-band events landing in the same millisecond where only the first was - * loaded therefore pass this half of the guard undetected — that is exactly - * the hole `stateEventCount` closes, since the count of events at or below the - * watermark differs even when the watermarks are equal. + * Merging is therefore monotone: it can only add facts this replay has yet to + * reconcile to, never remove one it already has. That is why absorbing is + * always optional and never wrong to decline — an unabsorbed event is one the + * next read returns — and why the guards below can afford to be strict. */ -export function latestEventStateUpdatedAt(events: Event[]): number | undefined { - let latest: string | undefined; - for (const event of events) { - if (latest === undefined || event.eventId > latest) { - latest = event.eventId; - } - } - if (latest === undefined) { - return undefined; - } - // Event IDs are prefixed ULIDs (e.g. `evnt_01ARYZ...`); ulidToDate only - // decodes the bare 26-char ULID, so strip the prefix first. - const eventId = latest; - const underscore = eventId.lastIndexOf('_'); - const rawUlid = underscore === -1 ? eventId : eventId.slice(underscore + 1); - const time = ulidToDate(rawUlid)?.getTime(); - if (time === undefined) { - // Fail open: a non-decodable id disarms the guard for this create (no - // snapshot sent). Log so a fleet-wide silent disarm is diagnosable. - runtimeLogger.debug( - 'Precondition guard: latest event id is not a decodable ULID; sending no snapshot', - { eventId } - ); - return undefined; - } - return time; -} /** * Merge the events a bump-and-report write handed back into the log it was @@ -817,6 +794,56 @@ export function mergeReportedEvents( return added; } +/** What {@link absorbSkippedSlotReport} did with a write's report. */ +export interface SkippedSlotReport { + /** How many of the reported events were new to the log. Zero if dropped. */ + added: number; + /** How many events the report offered, whether or not they were taken. */ + offered: number; + /** The report was truncated, so it was dropped whole instead of merged. */ + truncated: boolean; +} + +/** + * Apply a write's skipped-slot report to the log it was derived from, deciding + * whether the report may be taken at all. + * + * Every replay-context write can come back carrying the events on the slots it + * skipped over, and every caller wants the same thing from them: fold them in + * so the writes that follow name a position above them, and so the replay + * resuming from this log sees them without a reload. + * + * The one policy is that a **truncated report is dropped whole**. It covers a + * span of positions but carries only some of the events on them, so merging it + * would raise the log's highest position past a position whose event is + * missing. Later writes read that maximum to say what they have seen, so each + * would claim a position it never saw, and the World only reports the span a + * write skips — it would never send the missing one. Dropping costs one more + * round of the same events on the next write and keeps the log a prefix of the + * truth, which the note above {@link mergeReportedEvents} explains is always an + * available answer. + * + * Callers that log do so from the returned counts; the decision is not theirs + * to re-derive. + */ +export function absorbSkippedSlotReport( + target: Event[], + result: { events?: readonly Event[]; hasMore?: boolean } +): SkippedSlotReport { + const offered = result.events?.length ?? 0; + if (offered === 0) { + return { added: 0, offered: 0, truncated: false }; + } + if (result.hasMore === true) { + return { added: 0, offered, truncated: true }; + } + return { + added: mergeReportedEvents(target, result.events ?? []), + offered, + truncated: false, + }; +} + /** * The highest slot the loaded log occupies, or `undefined` when the run is not * slot-numbered. A run keeps the id scheme it was created under, so one event @@ -971,66 +998,42 @@ export async function settleEventSlotGap( } /** - * The precondition snapshot a replay-context event creation sends, describing - * the event log the replay derived the event from. + * How much of its run's log a replay-context event creation had loaded when it + * decided to write, as the highest slot that log occupies. + * + * One integer says it because the World keeps its positions dense: a writer + * that names slot N is claiming to hold every event from 1 to N and nothing + * above. The World answers by numbering the write above whatever the log has + * actually reached and handing back the events on the slots in between — the + * ones this writer decided without. + * + * Density is the World's invariant, not a claim about this particular read. A + * reader that is short of a position it holds no event for names a lower N, + * which understates what it has seen and only costs it a wider report. Naming + * a position it cannot account for is the direction that is unsafe, which is + * why {@link slotSnapshotParams} takes the maximum rather than the count. * - * On a slot-numbered run this is `eventCount`, optionally with the set of - * correlation ids the writer is blocked on. On a ULID-numbered run it is the - * `stateUpdatedAt` / `stateEventCount` / `stateCursor` triple, whose three - * fields are one indivisible unit: the backend reads the count only relative to - * the watermark, and returns its inline delta only relative to the cursor. - * Passing them as a single object is what keeps them from drifting apart at a - * call site. + * Its own object rather than a bare number so a call site cannot half-send it, + * and so the empty case (a run that is not slot-numbered) spreads to nothing. */ -export interface PreconditionSnapshotParams { - stateUpdatedAt?: number; - stateEventCount?: number; - stateCursor?: string; +export interface SlotSnapshotParams { eventCount?: number; } /** - * Build the precondition snapshot to attach to a replay-context event creation. + * Build the slot snapshot to attach to a replay-context event creation. * - * Returns an empty object — no guard, backend behaves as before — when the - * guard is disabled or the watermark is not derivable. All three fields fail - * open together: a count without a watermark is meaningless to the backend, and - * a cursor without either would invite a delta nobody asked for. + * Empty for a run that is not slot-numbered: there is no position to name, and + * a World that numbers by ULID has nothing to compare against. * - * `stateEventCount` is `events.length` because the watermark is the log's - * *maximum* ULID time, so every loaded event is at or below it regardless of the - * order the World returned them in. - * - * Both fields are therefore invariant under permutation of the log: a maximum is - * order-independent, and the length is set cardinality once `appendUniqueEvents` - * has deduped by event id. Two replays that consume the same events in different - * orders send an identical snapshot, so this guard detects that a log is missing - * an event and can never detect that a replay consumed one in a different order. + * The maximum rather than the length, for the reason {@link maxEventSlot} + * gives: a partially-read log holds fewer events than its highest position, and + * counting those would make the write claim to have seen less than it has, so + * the World would report the same events back on every attempt. */ -export function preconditionSnapshotParams( - events: Event[], - cursor?: string | null -): PreconditionSnapshotParams { - if (!isPreconditionGuardEnabled()) { - return {}; - } - // A slot-numbered run says with one integer everything the triple was - // approximating, so the two are alternatives rather than a pair. Sending the - // triple here would also be futile: a slot id carries no time, so - // `latestEventStateUpdatedAt` would fail open on every single write. +export function slotSnapshotParams(events: Event[]): SlotSnapshotParams { const eventCount = maxEventSlot(events); - if (eventCount !== undefined) { - return { eventCount }; - } - const stateUpdatedAt = latestEventStateUpdatedAt(events); - if (stateUpdatedAt === undefined) { - return {}; - } - return { - stateUpdatedAt, - stateEventCount: events.length, - ...(cursor ? { stateCursor: cursor } : {}), - }; + return eventCount === undefined ? {} : { eventCount }; } /** diff --git a/packages/core/src/runtime/precondition-guard-replay.test.ts b/packages/core/src/runtime/precondition-guard-replay.test.ts index fefa5a0bcf..0f9fe96f89 100644 --- a/packages/core/src/runtime/precondition-guard-replay.test.ts +++ b/packages/core/src/runtime/precondition-guard-replay.test.ts @@ -7,8 +7,8 @@ * re-post the rejected payload. The payload's correlation ids were minted * by the rejected replay's seeded ULID sequence, so a corrected event log * generally implies different ids; only a fresh replay may write again. - * 2. The restart reloads the whole event log with no cursor, because a hole is - * defined by ULID time while a cursor filters lexicographically — unless + * 2. The restart reloads the whole event log with no cursor, because a hole + * can sit below the held cursor and survive an incremental load — unless * the World attached the missing events to the 412, which the runtime * consumes with no events.list round trip at all (first restart only). * 3. Restarts are bounded; once the bound is spent the runtime schedules a @@ -16,14 +16,17 @@ * itself counted on the queue message, so a run that can never observe its * own log completely fails loudly rather than cycling restart chains. * - * Modeled on wait-completion-replay.test.ts, but with real ULID event IDs so - * latestEventStateUpdatedAt() actually derives snapshot times. + * Modeled on wait-completion-replay.test.ts, but with slot-numbered event ids, + * which is what lets the runtime name a position at all: the snapshot it sends + * is the highest slot it holds, and a log of ids it cannot read as slots + * produces no snapshot. */ import { PreconditionFailedError, RUN_ERROR_CODES } from '@workflow/errors'; import { type CreateEventRequest, type Event, SPEC_VERSION_CURRENT, + slotToEventId, type WorkflowRun, type World, } from '@workflow/world'; @@ -85,9 +88,7 @@ function buildStepEntity( interface SnapshotParams { eventType: string; - stateUpdatedAt: number | undefined; - stateEventCount: number | undefined; - stateCursor: string | undefined; + eventCount: number | undefined; } async function runPreconditionScenario(options: { @@ -133,18 +134,18 @@ async function runPreconditionScenario(options: { updatedAt: startedAt, }; - // Real ULID event IDs at controlled times so latestEventStateUpdatedAt() - // resolves an actual epoch-ms snapshot from the loaded log. - const hostUlid = monotonicFactory(); - let eventIndex = 0; + // Dense slot-numbered event ids in commit order, so the runtime can derive a + // snapshot from the loaded log at all. `atMs` only moves `createdAt`: an + // event's position is its slot, which is always the next one. + let eventSlot = 0; const event = (data: CreateEventRequest, atMs?: number): Event => { - const t = atMs ?? +startedAt + ++eventIndex * 100; + const slot = ++eventSlot; return { ...data, specVersion: data.specVersion ?? SPEC_VERSION_CURRENT, runId, - eventId: `evnt_${hostUlid(t)}`, - createdAt: new Date(t), + eventId: slotToEventId(slot), + createdAt: new Date(atMs ?? +startedAt + slot * 100), } as Event; }; @@ -194,10 +195,9 @@ async function runPreconditionScenario(options: { eventData: { resumeAt: new Date(+startedAt - 1_000) }, }), ]; - const staleSnapshotMs = +startedAt + staleEvents.length * 100; - expect(staleSnapshotMs).toBe(+startedAt + 700); - const staleEventsCursor = 'cursor-after-stale-events'; + // The hook lands far enough in the future that the replay's own `sleep("5s")` + // has already elapsed against it. const OUTSIDE_EVENT_MS = +startedAt + 5_000; const hookReceivedEvent = event( { @@ -258,17 +258,11 @@ async function runPreconditionScenario(options: { async ( _runId: string, request: CreateEventRequest, - params?: { - stateUpdatedAt?: number; - stateEventCount?: number; - stateCursor?: string; - } + params?: { eventCount?: number } ) => { createParams.push({ eventType: request.eventType, - stateUpdatedAt: params?.stateUpdatedAt, - stateEventCount: params?.stateEventCount, - stateCursor: params?.stateCursor, + eventCount: params?.eventCount, }); if (request.eventType === 'run_started') { @@ -418,8 +412,7 @@ async function runPreconditionScenario(options: { queue, runId, staleEventsCursor, - staleSnapshotMs, - OUTSIDE_EVENT_MS, + staleEventCount: staleEvents.length, waitCorrelationId, waitCompletedRejectionCount: () => waitCompletedRejections, }; @@ -464,16 +457,15 @@ async function runCompletedRejectionScenario({ updatedAt: startedAt, }; - const hostUlid = monotonicFactory(); - let eventIndex = 0; + let eventSlot = 0; const event = (data: CreateEventRequest, atMs?: number): Event => { - const t = atMs ?? +startedAt + ++eventIndex * 100; + const slot = ++eventSlot; return { ...data, specVersion: data.specVersion ?? SPEC_VERSION_CURRENT, runId, - eventId: `evnt_${hostUlid(t)}`, - createdAt: new Date(t), + eventId: slotToEventId(slot), + createdAt: new Date(atMs ?? +startedAt + slot * 100), } as Event; }; @@ -506,17 +498,11 @@ async function runCompletedRejectionScenario({ async ( _runId: string, request: CreateEventRequest, - params?: { - stateUpdatedAt?: number; - stateEventCount?: number; - stateCursor?: string; - } + params?: { eventCount?: number } ) => { createParams.push({ eventType: request.eventType, - stateUpdatedAt: params?.stateUpdatedAt, - stateEventCount: params?.stateEventCount, - stateCursor: params?.stateCursor, + eventCount: params?.eventCount, }); createRequests.push(request); if (request.eventType === 'run_started') { @@ -597,7 +583,6 @@ async function runCompletedRejectionScenario({ queue, staleEventCount: staleEvents.length, staleEventsCursor, - runStartedSnapshotMs: +startedAt + 200, }; } @@ -627,17 +612,18 @@ async function attributeSnapshotScenario() { updatedAt: startedAt, }; - const hostUlid = monotonicFactory(); - let eventIndex = 0; + let eventSlot = 0; const durableEvents: Event[] = []; - const event = (data: CreateEventRequest): Event => - ({ + const event = (data: CreateEventRequest): Event => { + const slot = ++eventSlot; + return { ...data, specVersion: data.specVersion ?? SPEC_VERSION_CURRENT, runId, - eventId: `evnt_${hostUlid(+startedAt + ++eventIndex * 100)}`, - createdAt: new Date(+startedAt + eventIndex * 100), - }) as Event; + eventId: slotToEventId(slot), + createdAt: new Date(+startedAt + slot * 100), + } as Event; + }; durableEvents.push( event({ @@ -648,6 +634,7 @@ async function attributeSnapshotScenario() { event({ eventType: 'run_started', specVersion: SPEC_VERSION_CURRENT }) ); const preloadedCursor = 'cursor-after-run-started'; + const preloadedEventCount = durableEvents.length; const createParams: SnapshotParams[] = []; let capturedHandler: @@ -661,17 +648,11 @@ async function attributeSnapshotScenario() { async ( _runId: string, request: CreateEventRequest, - params?: { - stateUpdatedAt?: number; - stateEventCount?: number; - stateCursor?: string; - } + params?: { eventCount?: number } ) => { createParams.push({ eventType: request.eventType, - stateUpdatedAt: params?.stateUpdatedAt, - stateEventCount: params?.stateEventCount, - stateCursor: params?.stateCursor, + eventCount: params?.eventCount, }); if (request.eventType === 'run_started') { return { @@ -731,7 +712,7 @@ async function attributeSnapshotScenario() { } ); - return { createParams, preloadedCursor }; + return { createParams, preloadedEventCount }; } /** @@ -762,16 +743,17 @@ async function inlineClaimRejectionScenario() { updatedAt: startedAt, }; - const hostUlid = monotonicFactory(); - let eventIndex = 0; - const event = (data: CreateEventRequest): Event => - ({ + let eventSlot = 0; + const event = (data: CreateEventRequest): Event => { + const slot = ++eventSlot; + return { ...data, specVersion: data.specVersion ?? SPEC_VERSION_CURRENT, runId, - eventId: `evnt_${hostUlid(+startedAt + ++eventIndex * 100)}`, - createdAt: new Date(+startedAt + eventIndex * 100), - }) as Event; + eventId: slotToEventId(slot), + createdAt: new Date(+startedAt + slot * 100), + } as Event; + }; const staleEvents: Event[] = [ event({ @@ -808,17 +790,11 @@ async function inlineClaimRejectionScenario() { async ( _runId: string, request: CreateEventRequest, - params?: { - stateUpdatedAt?: number; - stateEventCount?: number; - stateCursor?: string; - } + params?: { eventCount?: number } ) => { createParams.push({ eventType: request.eventType, - stateUpdatedAt: params?.stateUpdatedAt, - stateEventCount: params?.stateEventCount, - stateCursor: params?.stateCursor, + eventCount: params?.eventCount, }); if (request.eventType === 'run_started') { @@ -931,16 +907,13 @@ async function inlineClaimRejectionScenario() { } describe('precondition guard through the real replay loop', () => { - let originalGuard: string | undefined; let originalRestartBound: string | undefined; let originalInlineCap: string | undefined; beforeEach(() => { - originalGuard = process.env.WORKFLOW_PRECONDITION_GUARD; originalRestartBound = process.env.WORKFLOW_PRECONDITION_MAX_INPROCESS_RESTARTS; originalInlineCap = process.env.WORKFLOW_MAX_INLINE_STEPS; - process.env.WORKFLOW_PRECONDITION_GUARD = '1'; // The inline-claim scenario needs a whole two-step batch to run inline, // which the default cap allows. delete process.env.WORKFLOW_MAX_INLINE_STEPS; @@ -948,7 +921,6 @@ describe('precondition guard through the real replay loop', () => { afterEach(() => { for (const [name, value] of [ - ['WORKFLOW_PRECONDITION_GUARD', originalGuard], ['WORKFLOW_PRECONDITION_MAX_INPROCESS_RESTARTS', originalRestartBound], ['WORKFLOW_MAX_INLINE_STEPS', originalInlineCap], ] as const) { @@ -980,18 +952,17 @@ describe('precondition guard through the real replay loop', () => { (c) => c.eventType === 'wait_completed' ); expect(waitCreates).toHaveLength(2); - // First attempt carried the stale snapshot (ULID time of wait_created)... - expect(waitCreates[0]?.stateUpdatedAt).toBe(result.staleSnapshotMs); - // ...the restarted replay carried the corrected one (ULID time of - // hook_received), with a count covering the event it had been missing. - expect(waitCreates[1]?.stateUpdatedAt).toBe(result.OUTSIDE_EVENT_MS); - expect(waitCreates[1]?.stateEventCount).toBe( - (waitCreates[0]?.stateEventCount ?? 0) + 1 + // First attempt named the stale position (the last slot it had loaded)... + expect(waitCreates[0]?.eventCount).toBe(result.staleEventCount); + // ...the restarted replay named one slot higher, covering the event it had + // been missing. + expect(waitCreates[1]?.eventCount).toBe( + (waitCreates[0]?.eventCount ?? 0) + 1 ); // The restart reloaded the full log rather than reading from the held - // cursor: an `eid:` cursor filters lexicographically, so a hole defined by - // ULID time can sort below it and survive an incremental load. + // cursor: a hole can sit below that cursor and survive an incremental + // load. expect(cursorlessLoads(result.listEvents)).toBe(1); // Replay after the restart observed the hook and took the hook branch. @@ -1040,7 +1011,7 @@ describe('precondition guard through the real replay loop', () => { (c) => c.eventType === 'wait_completed' ); expect(waitCreates).toHaveLength(2); - expect(waitCreates[1]?.stateUpdatedAt).toBe(result.OUTSIDE_EVENT_MS); + expect(waitCreates[1]?.eventCount).toBe(result.staleEventCount + 1); expect(result.createdEvents).toEqual( expect.arrayContaining([ expect.objectContaining({ @@ -1160,8 +1131,7 @@ describe('precondition guard through the real replay loop', () => { 1 + getPreconditionMaxInProcessRestarts() ); for (const create of runCompletedCreates) { - expect(create.stateUpdatedAt).toBe(result.runStartedSnapshotMs); - expect(create.stateEventCount).toBe(result.staleEventCount); + expect(create.eventCount).toBe(result.staleEventCount); } // And the runtime must not convert the rejection into a run failure. expect( @@ -1227,8 +1197,6 @@ describe('precondition guard through the real replay loop', () => { (c) => c.eventType === 'attr_set' ); expect(attrCreate).toBeDefined(); - expect(attrCreate?.stateUpdatedAt).toBeTypeOf('number'); - expect(attrCreate?.stateEventCount).toBe(2); - expect(attrCreate?.stateCursor).toBe(result.preloadedCursor); + expect(attrCreate?.eventCount).toBe(result.preloadedEventCount); }); }); diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts index ce88f86906..bd81fb6981 100644 --- a/packages/core/src/runtime/quickjs-entrypoint.ts +++ b/packages/core/src/runtime/quickjs-entrypoint.ts @@ -676,15 +676,14 @@ async function dispatchPendingOps(params: { * This replaces the `node:vm` replay path (runWorkflow + EventsConsumer) * with a QuickJS VM invocation that performs the same full event replay. * - * KNOWN GAP — precondition guard: unlike the node:vm path, no event write - * in this file participates in the optimistic-concurrency precondition - * guard (`withPreconditionRetry` + `stateUpdatedAtForCreate`), which - * protects a writer holding a stale event-log snapshot from clobbering a - * concurrent one. The engine currently relies on per-(runId, - * correlationId) event uniqueness (EntityConflictError dedup) alone. This - * is a deliberate simplification while the engine is experimental — wiring - * the guard is tracked follow-up work; anyone adding new write paths here - * should not assume parity with the node engine on this axis. + * KNOWN GAP — slot snapshot: unlike the node:vm path, no event write in + * this file carries {@link CreateEventParams.eventCount}, so a World never + * learns which events the writer had not seen and never reports them back. + * The engine currently relies on per-(runId, correlationId) event + * uniqueness (EntityConflictError dedup) alone. This is a deliberate + * simplification while the engine is experimental — wiring the snapshot is + * tracked follow-up work; anyone adding new write paths here should not + * assume parity with the node engine on this axis. */ export async function runWorkflowWithQuickJS(params: { workflowCode: string; diff --git a/packages/core/src/runtime/replay-budget.ts b/packages/core/src/runtime/replay-budget.ts index 9584ca672c..53f70f3b1f 100644 --- a/packages/core/src/runtime/replay-budget.ts +++ b/packages/core/src/runtime/replay-budget.ts @@ -4,7 +4,7 @@ import { describeError } from '../describe-error.js'; import { runtimeLogger } from '../logger.js'; import { dehydrateRunError } from '../serialization.js'; import { getReplayTimeoutMaxRetries, getReplayTimeoutMs } from './constants.js'; -import { memoizeEncryptionKey } from './helpers.js'; +import { memoizeEncryptionKey, type SlotSnapshotParams } from './helpers.js'; import { getWorld } from './world.js'; /** @@ -119,8 +119,16 @@ export async function handleReplayBudgetExhausted(args: { requestId: string | undefined; attempt: number; limitMs: number; + /** + * How much of the log the replay that ran out of budget had loaded. Carried + * onto the terminal write for the same reason every other write from that + * replay carries it: the run is being failed on a view of the log, and the + * World should be told which view. + */ + slotSnapshot?: SlotSnapshotParams; }): Promise { - const { runId, workflowName, requestId, attempt, limitMs } = args; + const { runId, workflowName, requestId, attempt, limitMs, slotSnapshot } = + args; const runLogger = runtimeLogger.forRun(runId, workflowName); const maxRetries = getReplayTimeoutMaxRetries(); @@ -170,6 +178,6 @@ export async function handleReplayBudgetExhausted(args: { errorCode: RUN_ERROR_CODES.REPLAY_TIMEOUT, }, }, - { requestId } + { requestId, ...slotSnapshot } ); } diff --git a/packages/core/src/runtime/step-executor.test.ts b/packages/core/src/runtime/step-executor.test.ts index 4bbe37fa56..ad80d8e1b7 100644 --- a/packages/core/src/runtime/step-executor.test.ts +++ b/packages/core/src/runtime/step-executor.test.ts @@ -171,7 +171,7 @@ describe('executeStep — compute instance stamping', () => { counter += 1; }); - it('stamps computeInstanceId on step_started without displacing the precondition snapshot', async () => { + it('stamps computeInstanceId on step_started without displacing the slot snapshot', async () => { const world = makeWorld(); const stepName = uniqueStepName(); const { runId, stepId } = await setupRunningStep({ @@ -184,11 +184,7 @@ describe('executeStep — compute instance stamping', () => { // persist — so observe the call itself rather than the stored event. const createSpy = vi.spyOn(world.events, 'create'); - const preconditionSnapshot = { - stateUpdatedAt: 1_700_000_000_000, - stateEventCount: 7, - stateCursor: 'eid:evnt_01H0000000000000000000000', - }; + const slotSnapshot = { eventCount: 7 }; await executeStep({ world, @@ -197,7 +193,7 @@ describe('executeStep — compute instance stamping', () => { workflowStartedAt: Date.now(), stepId, stepName, - preconditionSnapshot, + slotSnapshot, }); const started = createSpy.mock.calls.filter( @@ -205,8 +201,49 @@ describe('executeStep — compute instance stamping', () => { ); expect(started).toHaveLength(1); expect(started[0]?.[2]?.computeInstanceId).toBe(COMPUTE_INSTANCE_ID); - // Both ride the same params object — neither may clobber the other, and the - // three snapshot fields must arrive as one unit. - expect(started[0]?.[2]).toMatchObject(preconditionSnapshot); + // Both ride the same params object and neither may clobber the other. + expect(started[0]?.[2]?.eventCount).toBe(slotSnapshot.eventCount); + }); + + it('advances the snapshot it sends as its own writes land', async () => { + // The executor writes twice for one step. If the second write still named + // the position its caller scheduled against, the World would report the + // first one back to it on every step, forever. + const world = makeWorld(); + const stepName = uniqueStepName(); + const { runId, stepId } = await setupRunningStep({ + world, + stepName, + onBody: () => {}, + }); + + // The position the caller would have scheduled against, taken from the log + // rather than written down, so the seed stays below the slots the executor + // is about to commit at. Seeding it above them would leave `observeSlot` + // with nothing to raise and the test would pass without exercising it. + const { data: seeded } = await world.events.list({ runId }); + const scheduledAt = seeded.length; + + const createSpy = vi.spyOn(world.events, 'create'); + + await executeStep({ + world, + workflowRunId: runId, + workflowName: 'wf', + workflowStartedAt: Date.now(), + stepId, + stepName, + slotSnapshot: { eventCount: scheduledAt }, + }); + + // world-local mints slots for a run created on this scheme, so every write + // reads back as a position and each one has to name the position its + // predecessor landed on. + const counts = createSpy.mock.calls.map((call) => call[2]?.eventCount); + expect(counts.length).toBeGreaterThan(1); + expect(counts[0]).toBe(scheduledAt); + for (let i = 1; i < counts.length; i++) { + expect(counts[i]).toBeGreaterThan(counts[i - 1] as number); + } }); }); diff --git a/packages/core/src/runtime/step-executor.ts b/packages/core/src/runtime/step-executor.ts index 70f5a6831c..390e315889 100644 --- a/packages/core/src/runtime/step-executor.ts +++ b/packages/core/src/runtime/step-executor.ts @@ -23,6 +23,7 @@ import type { World, } from '@workflow/world'; import { + eventIdToSlot, SPEC_VERSION_CURRENT, SPEC_VERSION_SUPPORTS_COMPRESSION, } from '@workflow/world'; @@ -52,8 +53,9 @@ import { } from './constants.js'; import { getPortLazy } from './get-port-lazy.js'; import { + maxEventSlot, memoizeEncryptionKey, - type PreconditionSnapshotParams, + type SlotSnapshotParams, } from './helpers.js'; import { ReplayRecoveryReporter } from './replay-recovery-reporter.js'; import { @@ -145,20 +147,25 @@ export interface StepExecutorParams { */ inlineDeltaSinceCursor?: string; /** - * Precondition-guard snapshot of the event log the caller's replay loaded, to - * attach to this step's `step_started` claim. On the lazy inline path the - * claim is the step's FIRST durable write (its `step_created` is deferred), - * so without this the claim would bypass the optimistic-concurrency guard - * entirely: a replay working from a stale view could claim — and then commit - * — a step scheduled without observing an event it never loaded. A - * guard-enforcing World rejects a stale claim with `PreconditionFailedError` + * How much of the run's log the caller's replay had loaded when it scheduled + * this step, as the highest slot that log occupies. Seeds the snapshot every + * write this executor makes carries; each committed event advances it to its + * own slot, so a later write never names a position that predates an earlier + * one from the same step. + * + * It matters most on the lazy inline path, where the `step_started` claim is + * the step's FIRST durable write (its `step_created` is deferred): without a + * seed the claim would name no position at all, and a replay working from a + * stale view could claim — and then commit — a step scheduled without + * observing an event it never loaded. + * + * A World that fences rejects a stale claim with `PreconditionFailedError` * (412); executeStep does NOT translate that rejection (re-claiming in place * would still commit the stale schedule), so it propagates for the caller to - * abandon the batch and restart its replay. Undefined when the guard is - * disabled or the caller has no snapshot; Worlds that don't enforce the guard - * ignore it. + * abandon the batch and restart its replay. Undefined for a run that is not + * slot-numbered, or a caller with nothing loaded. */ - preconditionSnapshot?: PreconditionSnapshotParams; + slotSnapshot?: SlotSnapshotParams; /** * Suppress optimistic inline start for this step regardless of * `WORKFLOW_OPTIMISTIC_INLINE_START` / `forceOptimisticStart`: take the @@ -286,13 +293,48 @@ export async function executeStep( (params.runSpecVersion ?? 0) >= SPEC_VERSION_SUPPORTS_COMPRESSION; const replayRecoveryReporter = params.replayRecoveryReporter ?? ReplayRecoveryReporter.inert(); - const createEvent = ( + /** + * The highest log slot this executor knows about, seeded from the view its + * caller scheduled the step against and advanced by every event it commits. + * + * Advancing is what keeps the snapshot honest across a step's own writes. A + * step commits `step_started` and then `step_completed`; if the second still + * named the caller's original position, the World would report the first one + * back as an event this writer had not seen, on every step, forever. + * + * This reads a report for its highest position and then discards it, where + * the replay loop and the suspension handler merge theirs with + * `absorbSkippedSlotReport`. That is the difference between the callers, not + * an oversight: an executor holds no loaded log to merge into. It runs from a + * queued delivery whose only view of the log is the integer its caller passed + * in, so the position is the entire value the report has to it. Whoever + * replays next loads the log and gets the events themselves. + */ + let knownSlot = params.slotSnapshot?.eventCount; + const observeSlot = (result: { event?: Event; events?: Event[] }): void => { + if (knownSlot === undefined) { + return; + } + const committed = result.event ? eventIdToSlot(result.event.eventId) : null; + for (const slot of [committed, maxEventSlot(result.events ?? []) ?? null]) { + if (slot !== null && slot > knownSlot) { + knownSlot = slot; + } + } + }; + const createEvent = async ( data: T, eventParams?: CreateEventParams - ) => - replayRecoveryReporter.withEventCreate(eventParams, (p) => - world.events.create(workflowRunId, data, p) + ) => { + const result = await replayRecoveryReporter.withEventCreate( + knownSlot === undefined + ? eventParams + : { eventCount: knownSlot, ...eventParams }, + (p) => world.events.create(workflowRunId, data, p) ); + observeSlot(result); + return result; + }; const spanName = `step.execute ${stepDisplayName(stepName)}`; return trace(spanName, {}, async (span) => { @@ -544,13 +586,11 @@ export async function executeStep( !isOptimisticInlineStartExplicitlyDisabled())); let step: StartedStep; - // Params for the `step_started` create on either path below: the ambient - // compute-instance stamp plus the optimistic-concurrency claim guard. + // Params for the `step_started` create on either path below. The slot + // snapshot is not spread here: `createEvent` attaches it to every write, + // this one included. const startEventParams: CreateEventParams = { computeInstanceId: COMPUTE_INSTANCE_ID, - // Spread as a unit: the three snapshot fields describe one snapshot and - // must travel together — see StepExecutorParams.preconditionSnapshot. - ...params.preconditionSnapshot, }; // `Date.now()` taken immediately before the `step_started` create is // issued (either path below) — anchors RSFS's end point. See @@ -613,7 +653,7 @@ export async function executeStep( : {}), }, }, - // Guard the claim — see StepExecutorParams.preconditionSnapshot. A + // Guard the claim — see StepExecutorParams.slotSnapshot. A // stale (412) rejection surfaces via reconcileOptimisticStart as a // non-translatable error: the body result is discarded and the // rejection propagates to the caller. @@ -673,7 +713,7 @@ export async function executeStep( } : { stepName, ...ownershipStamp }, }, - // Guard the claim — see StepExecutorParams.preconditionSnapshot. A + // Guard the claim — see StepExecutorParams.slotSnapshot. A // stale (412) rejection is intentionally NOT translated by // startErrorToResult below, so it propagates to the caller for a // fresh replay. diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index 13363a196c..713b0a1157 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -41,12 +41,11 @@ import { MAX_RESILIENT_STEP_INPUT_BYTES, } from './constants.js'; import { + absorbSkippedSlotReport, type EventCreator, - isPreconditionGuardEnabled, type LoadedEventLog, - mergeReportedEvents, - preconditionSnapshotParams, queueMessage, + slotSnapshotParams, stepDispatchIdempotencyKey, } from './helpers.js'; import { ReplayRecoveryReporter } from './replay-recovery-reporter.js'; @@ -356,40 +355,28 @@ export async function handleSuspension({ const log = eventLog; const result = await createEvent(data, { ...params, - ...preconditionSnapshotParams(log.events, log.cursor), + ...slotSnapshotParams(log.events), }); - // Bump-and-report: the write landed above the slot it asked for, so these - // are the events it was decided without. Merging them here rather than at - // each call site means the rest of this phase's writes — which read the - // same array to build their own snapshot — ask for a slot above them, and - // the replay that resumes from this log sees them without a reload. - // - // A truncated report (`hasMore`) is dropped whole rather than merged, the - // same way the wait loop treats one. It covers a span of positions but - // carries only some of the events on them, so merging it would raise the - // log's highest position past a position whose event is missing. Every - // later write of this phase reads that maximum to say what it has seen, so - // each would claim a position it never saw and the World, which only - // reports the span a write skips, would never send it. Dropping the report - // costs one more round of the same events on the next write and keeps the - // log a prefix of the truth. - if (result.events?.length && result.hasMore !== true) { - const added = mergeReportedEvents(log.events, result.events); - reportedEvents += added; - if (added > 0) { - runtimeLogger.debug('Suspension write skipped occupied slots', { - workflowRunId: runId, - eventType: data.eventType, - eventId: result.event?.eventId, - reported: added, - }); - } - } else if (result.events?.length) { + // Bump-and-report: the write landed above the slot it asked for, so the + // report holds the events it was decided without. Absorbing here rather + // than at each call site means the rest of this phase's writes — which read + // the same array to build their own snapshot — ask for a slot above them, + // and the replay that resumes from this log sees them without a reload. + const report = absorbSkippedSlotReport(log.events, result); + reportedEvents += report.added; + if (report.truncated) { runtimeLogger.debug('Dropped a truncated skipped-slot report', { workflowRunId: runId, eventType: data.eventType, eventId: result.event?.eventId, - offered: result.events.length, + offered: report.offered, + }); + } else if (report.added > 0) { + runtimeLogger.debug('Suspension write skipped occupied slots', { + workflowRunId: runId, + eventType: data.eventType, + eventId: result.event?.eventId, + reported: report.added, }); } return result; @@ -691,7 +678,8 @@ export async function handleSuspension({ // - The caller provided a dispatch target (`stepDispatch`) — terminal // drains and other create-only callers never queue. // - The feature is enabled (`WORKFLOW_RESILIENT_STEP_DISPATCH` opt-out). - // - The optimistic-concurrency guard is not in effect. A guard-enforcing + // - The World does not fence stale writes + // (`capabilities.preconditionGuard`). A guard-enforcing // backend can reject the step_created as stale (412) and the caller then // restarts the replay — but a queue message carrying the payload would // already be out, letting the consumer materialize a step the guard @@ -708,10 +696,7 @@ export async function handleSuspension({ const resilientDispatchEligible = stepDispatch !== undefined && isResilientStepDispatchEnabled() && - !( - isPreconditionGuardEnabled() && - world.capabilities?.preconditionGuard === true - ) && + world.capabilities?.preconditionGuard !== true && (run.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT; // The trace carrier for resilient step dispatches, resolved at most once per diff --git a/packages/errors/src/index.ts b/packages/errors/src/index.ts index fb53982131..bd841df150 100644 --- a/packages/errors/src/index.ts +++ b/packages/errors/src/index.ts @@ -869,8 +869,10 @@ export class ThrottleError extends WorkflowWorldError { * does not read this field. * @property details - Optional rejection detail supplied by the World. A World * MAY attach the events the client's snapshot was missing so the client can - * correct its log without a follow-up fetch; see the `stateCursor` contract - * on `CreateEventParams`. Typed `unknown` because this package cannot depend + * correct its log without a follow-up fetch. The attached set must account + * for the whole discrepancy the rejection reported or be omitted entirely; a + * client that receives nothing does the authoritative full reload, which is + * always correct. Typed `unknown` because this package cannot depend * on the event type — consumers narrow it themselves and must treat a * missing or malformed value as "no detail" (a full reload is always * correct). diff --git a/packages/world-sim/DESIGN.md b/packages/world-sim/DESIGN.md index a973bb0b1f..0da704e759 100644 --- a/packages/world-sim/DESIGN.md +++ b/packages/world-sim/DESIGN.md @@ -294,19 +294,23 @@ about nothing interesting. The fence is off by default and set per scenario (`ScenarioSpec.preconditionGuard`, which flows into `SimWorldOptions`), which is how a scenario can be run one flag apart from its neighbour. `countGuard` -**follows the fence** unless a spec says otherwise, because that is what -production does — see below. +**follows the fence** unless a spec says otherwise, because a World that fences +arms both halves — see below. **`preconditionGuard`** models `WorldCapabilities.preconditionGuard`: reject a -replay-context write whose `stateUpdatedAt` snapshot predates the newest -externally-originated event. In the SDK this is declared by **world-vercel -only** (`packages/world-vercel/src/index.ts:40`); `world-local` and -`world-postgres` declare neither it nor `maxConcurrency`. +replay-context write whose snapshot predates the newest externally-originated +event. In the SDK the capability is declared by **world-vercel only** +(`packages/world-vercel/src/index.ts`); `world-local` and `world-postgres` +declare neither it nor `maxConcurrency`. It no longer describes what world-vercel +does to a run, though: a slot-identity run has no snapshot to reject, because the +World allocates the event's position at commit time and reports the positions the +write skipped over. What the sim's fence still covers is the 412 *reception* path +the runtime keeps for Worlds that do fence, and the predicate itself. Its predicate is narrower than the bug class, and the reason is its *shape*, not the event type it watches. The marker advances on `hook_received` **or** `step_completed`, but it is a **high-water mark** — the newest such write — and -the test is `stateUpdatedAt < marker`, strictly. So it detects a log truncated +the test is `snapshot.updatedAt < marker`, strictly. So it detects a log truncated at the end and is blind to a hole in the middle: when the withheld event is *older* than one the reader can see, the reader's snapshot is never strictly older than the mark. The hook direction is caught for the mirror-image reason — @@ -315,25 +319,44 @@ orchestrator's snapshot predates the sleep, so the fence fires and the run reconciles. **`countGuard`** adds the count half: how many events the log holds at or below -`stateUpdatedAt`, compared against how many the caller loaded. It closes the -hole the watermark cannot see. It requires the caller to send -`stateEventCount` — and since #3145 (`1471f252f`) `@workflow/core` sends it on -every replay-context create, gated only by the `WORKFLOW_PRECONDITION_GUARD` -kill-switch, with workflow-server's own count guard defaulting on. So both -halves are armed together in production, and `countGuard` defaults here to -whatever the fence is set to. A run with the fence on and the count off is a -world that exists nowhere; the two scenarios that ask for it -(`step-vs-step-fork-fenced`, `in-flight-before-decision`) do so explicitly, -because isolating the watermark half is their whole subject. - -Where the runtime sends a count, the sim uses **that** value rather than its own -reconstruction (`loadedCount()`), so the guard is tested against the number the -real client computes; the reconstruction is the fallback for writes core does -not count. - -**Two ways the sim's guards are stronger than production's.** Both are -deliberate, and both mean a fenced green here is a claim about the *predicate*, -not about production's *deployment* of it: +the caller's watermark, compared against how many the caller loaded. It closes +the hole the watermark cannot see. A World that fences arms both halves +together, so `countGuard` defaults here to whatever the fence is set to; a run +with the fence on and the count off is a world that exists nowhere, and the two +scenarios that ask for it (`step-vs-step-fork-fenced`, +`in-flight-before-decision`) do so explicitly, because isolating the watermark +half is their whole subject. + +**Where the snapshot comes from.** Both halves read one +`SimCreateParams.snapshot`, and the sim **reconstructs** it rather than reading +it off the wire. `@workflow/core` states its position as a slot count +(`eventCount`), derived from slot-numbered event IDs; the sim mints ULIDs, so a +create arrives with nothing the fence can compare against a ULID watermark. The +world facade therefore derives `{ updatedAt, count }` from the pages the writer +read: newest loaded position, and how many loaded events sit at or below it. +That is the same derivation the client used to make, and it is what lets the +count half see a hole *behind* the watermark at all. + +Two rules keep that derivation honest, and both were learned by getting them +wrong. **A read is what starts the set**: a runtime that writes before loading +anything (a fresh delivery's `run_started`) sends no snapshot, so crediting it +with its own write would hand the fence a position lower than the log holds and +reject the next concurrent write on a claim nobody made. **The set lives for one +delivery**: it is scoped to the queue handler invocation, because a cold- +starting replay holds nothing until it reads, and inheriting the previous +delivery's view fences it for a log it never loaded. + +One consequence worth holding: the discriminator for "this write did not come +from a replay context" is now **the facade attached no snapshot**, where it used +to be "the client sent no watermark". Those differ for a write issued from a +step body that did load a log. Such a write is fenced here where it previously +advanced the out-of-band marker instead. + +**Two ways the sim's guards are stronger than a deployed fence.** The +comparison is against the fence as world-vercel ran it for ULID runs, which is +the only shape it ever ran against. Both differences are deliberate, and both +mean a fenced green here is a claim about the *predicate*, not about any +deployment of it: - The server's retained-id window is a FIFO in **insertion (commit) order** — its Lua script prunes with `table.remove(ids, 1)`, oldest-inserted — while @@ -706,14 +729,13 @@ So: **two** of the six have their fix demonstrated by a paired green scenario, Writing the three missing pairs is the obvious next increment. Note what the `fix` column does *not* mean. `countGuard` closing doc-29 is a -statement about the World implementation *and* about production's predicate: -core has sent `stateEventCount` on every replay-context create since #3145 and -the server's count guard defaults on (§5). What it is not is a statement about -production's *deployment* of that predicate, which is region-local, fails open, -and prunes its window in a different order than this store does — all three -noted in §5. So the honest reading of the column is: four of the six have a fix -whose predicate is armed in production today, and whether it fires there depends -on conditions the sim does not model. +statement about a fencing World's *predicate*, and nothing more. It is not a +statement about world-vercel, which does not fence a slot-identity run at all: +positions there are assigned at commit, so a write that named a stale one still +commits and comes back carrying the events it skipped over. That is the +append-only column below, reached by a different mechanism. Read the `fix` +column as "which predicate would have caught this fault", and read the +append-only column for what production actually does about it. **The append-only log closes all six, in two different senses — and the split is four and two, not three and three.** Four (doc-23, doc-25, doc-26, doc-27) @@ -812,7 +834,7 @@ package about hook races. **Also untested:** turbo / optimistic-inline-start, which skip replays and so give a stale branch somewhere to hide; and the fence's same-millisecond -behaviour, where an equal `stateUpdatedAt` passes by design as anti-livelock. +behaviour, where an equal snapshot watermark passes by design as anti-livelock. **Not modelled at all:** the concurrency machinery `world-local` needs and this store omits — claim files, per-entity locks, staged/promoted hook events, diff --git a/packages/world-sim/README.md b/packages/world-sim/README.md index 6646c36578..c78ee5ab5b 100644 --- a/packages/world-sim/README.md +++ b/packages/world-sim/README.md @@ -149,15 +149,22 @@ which is why `StaleRead` reports `{ eventId, hidden, truncated }` and the trace distinguishes a lagging read from a stale one. **Precondition fence** (`preconditionGuard: true`) — rejects a write whose -`stateUpdatedAt` snapshot is strictly older than the newest externally -originated event. It is a high-water mark, so it sees a log truncated at the end -and is blind to a hole in the middle. +snapshot is strictly older than the newest externally originated event. It is a +high-water mark, so it sees a log truncated at the end and is blind to a hole in +the middle. **Count guard** (`countGuard: true`) — adds the other half: how many events the -log holds at or below `stateUpdatedAt`, against how many the caller loaded. It -closes the hole a watermark cannot see, and requires the caller to send -`stateEventCount`. It is evaluated inside the fence's predicate, so it is only -live when the fence is. +log holds at or below that watermark, against how many the caller loaded. It +closes the hole a watermark cannot see. It is evaluated inside the fence's +predicate, so it is only live when the fence is. + +Both halves read one snapshot, and the sim reconstructs it rather than reading +it off the wire: a client on slot-numbered event IDs sends a slot count, the sim +mints ULIDs, so the facade derives `{ updatedAt, count }` from the pages the +writer actually read, within the delivery that read them. The derivation is the +client's own — newest loaded position, and how many loaded events sit at or +below it. A write the facade attached no snapshot to did not come from a replay +context and is never fenced. Each is a spec field, and `RunScenarioOptions` carries a run-wide override — `pnpm sim --append-only`, `--fence` / `--no-fence` — where `undefined` leaves diff --git a/packages/world-sim/src/scenario.ts b/packages/world-sim/src/scenario.ts index 10a841ccf2..b1566027ac 100644 --- a/packages/world-sim/src/scenario.ts +++ b/packages/world-sim/src/scenario.ts @@ -133,11 +133,9 @@ export interface ScenarioSpec { * Also enforce the count half of the fence: reject a write whose caller loaded * fewer events at or below its watermark than the log now holds. * - * Defaults to `preconditionGuard`, because that is production: since #3145 - * `@workflow/core` sends `stateEventCount` on every replay-context create and - * workflow-server's count guard is on by default, so a fence is a fence with - * both halves. Set it to `false` alongside `preconditionGuard: true` to model - * the watermark alone. + * Defaults to `preconditionGuard`, because that is production: a world that + * fences at all fences with both halves. Set it to `false` alongside + * `preconditionGuard: true` to model the watermark alone. */ countGuard?: boolean; /** diff --git a/packages/world-sim/src/store.test.ts b/packages/world-sim/src/store.test.ts index f72a987b04..9041627967 100644 --- a/packages/world-sim/src/store.test.ts +++ b/packages/world-sim/src/store.test.ts @@ -424,9 +424,12 @@ describe('sim store', () => { }); guarded.tick(10); - const snapshot = guarded.nowMs(); + const snapshot = { + updatedAt: guarded.nowMs(), + count: guarded.store.allEvents(RUN).length, + }; guarded.tick(10); - // An out-of-band resume: no stateUpdatedAt, so it advances the marker. + // An out-of-band resume: no snapshot, so it advances the marker. await guarded.store.events.create(RUN, { eventType: 'hook_received', specVersion: SPEC, @@ -443,7 +446,7 @@ describe('sim store', () => { correlationId: 'step_1', eventData: { stepName: 'step//./w//s', input: new Uint8Array() }, }, - { stateUpdatedAt: snapshot } + { snapshot } ) ).rejects.toThrow(/out of band/); @@ -457,7 +460,12 @@ describe('sim store', () => { correlationId: 'step_1', eventData: { stepName: 'step//./w//s', input: new Uint8Array() }, }, - { stateUpdatedAt: guarded.nowMs() } + { + snapshot: { + updatedAt: guarded.nowMs(), + count: guarded.store.allEvents(RUN).length, + }, + } ) ).resolves.toBeTruthy(); }); diff --git a/packages/world-sim/src/store.ts b/packages/world-sim/src/store.ts index c2d007279d..abde8a4193 100644 --- a/packages/world-sim/src/store.ts +++ b/packages/world-sim/src/store.ts @@ -99,11 +99,29 @@ interface SimCreateParams { */ minted?: MintedEvent; /** - * How many events the caller had loaded when it decided to make this write. - * Mirrors workflow-server's `stateEventCount`, and since #3145 the runtime - * sends it; see `SimStoreOptions.countGuard`. + * The log this write was decided against, or absent when the write did not + * come from a replay context at all (an out-of-band writer, or a store driven + * directly by a unit test). + * + * Reconstructed by the world facade from the pages the writer read, because + * the runtime no longer states it: `@workflow/core` describes its snapshot as + * a slot count, and the sim mints ULIDs, so there is nothing on the wire for + * the fence to read. The reconstruction is the same derivation the client + * used to make — the newest loaded position, and how many events sit at or + * below it — which is what lets the fence spot a hole *behind* the watermark + * that no comparison against the watermark alone can see. + * + * See `SimStoreOptions.preconditionGuard` and `SimWorldOptions.countGuard`. */ - stateEventCount?: number; + snapshot?: LoadedSnapshot; +} + +/** What a replay-context writer had loaded when it decided to write. */ +export interface LoadedSnapshot { + /** ULID time of the newest loaded event. */ + updatedAt: number; + /** How many loaded events sit at or below {@link updatedAt}. */ + count: number; } /** Per run: the tail of the log, for the count guard. See `countRecordedAtOrBelow`. */ @@ -113,8 +131,8 @@ interface RunEventIndex { } /** - * How many events the log holds at or below `stateUpdatedAt`, or `null` when - * the retained window cannot prove it. + * How many events the log holds at or below the caller's watermark, or `null` + * when the retained window cannot prove it. * * Ported from workflow-server's `countRecordedAtOrBelow`, including its * exactness argument: pruning always drops the oldest id, so `total - above` is @@ -124,10 +142,10 @@ interface RunEventIndex { */ function countRecordedAtOrBelow( index: RunEventIndex, - stateUpdatedAt: number + updatedAt: number ): number | null { const above = index.recentEventIds.filter( - (id) => ulidTimeOf(id) > stateUpdatedAt + (id) => ulidTimeOf(id) > updatedAt ).length; const pruned = index.total > index.recentEventIds.length; if (pruned && above === index.recentEventIds.length) return null; @@ -140,7 +158,7 @@ export interface SimStoreOptions { /** * Enforce the optimistic-concurrency precondition guard described in * `WorldCapabilities.preconditionGuard`: reject a replay-context write whose - * `stateUpdatedAt` snapshot predates the newest externally-originated event. + * snapshot predates the newest externally-originated event. * * Off by default. Turning it on is the point of a simulation — it lets a * scenario check that the runtime recovers from a 412 fence — but it also @@ -157,11 +175,9 @@ export interface SimStoreOptions { * truncated at the end; the count answers "is anything missing *behind* my * snapshot?", which is the hole two concurrent writers actually produce. * - * Requires `preconditionGuard` (it reuses `stateUpdatedAt` as the watermark to - * count against) and a client that sends `stateEventCount` — which - * `@workflow/core` has done on every replay-context create since #3145. The - * sim uses that value when it is there and reconstructs one for the writes - * core does not count; see `SimWorldOptions.countGuard`. + * Requires `preconditionGuard`: it counts against the same watermark. Both + * halves read `SimCreateParams.snapshot`, which the world facade + * reconstructs; see `SimWorldOptions.countGuard`. */ countGuard?: boolean; /** @@ -399,7 +415,7 @@ export function createSimStore(options: SimStoreOptions): SimStore { const disposedHooks = new Set(); /** * Per run: ULID time of the newest externally-originated event. Only read - * when `preconditionGuard` is on. See `CreateEventParams.stateUpdatedAt`. + * when `preconditionGuard` is on. See `SimCreateParams.snapshot`. */ const externalWriteMarker = new Map(); /** @@ -855,31 +871,31 @@ export function createSimStore(options: SimStoreOptions): SimStore { // workflow-server's handler. The first is a high-water mark; the second is // a count. They fail on different shapes, and only together do they cover // both halves of a two-writer race. - if (options.preconditionGuard && params?.stateUpdatedAt !== undefined) { + const snapshot = internal?.snapshot; + if (options.preconditionGuard && snapshot) { const marker = externalWriteMarker.get(runId); - if (marker !== undefined && params.stateUpdatedAt < marker) { + if (marker !== undefined && snapshot.updatedAt < marker) { throw new PreconditionFailedError( `Run "${runId}" changed out of band since the caller's snapshot` ); } - // The count guard. `recorded > stateEventCount` means the log holds an + // The count guard. `recorded > snapshot.count` means the log holds an // event at or below the caller's own watermark that the caller never // loaded: a hole, which the marker comparison above passes by // construction because the missing event is *older* than the newest one // the caller did see. A `null` count is indeterminate (the window pruned // past the snapshot) and is never treated as stale — the guard is // deliberately one-sided. - const stateEventCount = internal?.stateEventCount; - if (options.countGuard && stateEventCount !== undefined) { + if (options.countGuard) { const index = runEventIndex.get(runId); const recorded = index - ? countRecordedAtOrBelow(index, params.stateUpdatedAt) + ? countRecordedAtOrBelow(index, snapshot.updatedAt) : null; - if (recorded !== null && recorded > stateEventCount) { + if (recorded !== null && recorded > snapshot.count) { throw new PreconditionFailedError( `Run "${runId}" holds ${recorded} events at or below the caller's ` + - `watermark, but the caller loaded ${stateEventCount}` + `watermark, but the caller loaded ${snapshot.count}` ); } } @@ -1157,7 +1173,7 @@ export function createSimStore(options: SimStoreOptions): SimStore { event = append(event); // Track externally-originated writes for the precondition fence. A write - // that carries no `stateUpdatedAt` did not come from a replay context, so + // the facade attached no snapshot to did not come from a replay context, so // it is exactly the kind of out-of-band change a replaying caller needs to // be fenced against. // @@ -1165,16 +1181,16 @@ export function createSimStore(options: SimStoreOptions): SimStore { // `recordOutsideEvent`: // // - The mark is the event's *own* position time, not the commit instant. It - // has to be the same derivation as the client's `stateUpdatedAt` (the - // position time of its newest loaded event) or a client holding exactly - // this event would compare as older and 412 forever. + // has to be the same derivation as a caller's watermark (the position + // time of its newest loaded event) or a caller holding exactly this event + // would compare as older and 412 forever. // - The write is forward-only. Concurrent out-of-band events can commit out // of position order — the whole subject of these scenarios — and letting a // late-committing older event drag the mark backwards would silently // disarm the guard for the newer one. if ( options.preconditionGuard && - params?.stateUpdatedAt === undefined && + snapshot === undefined && (data.eventType === 'hook_received' || data.eventType === 'step_completed' || data.eventType === 'step_failed') diff --git a/packages/world-sim/src/world.ts b/packages/world-sim/src/world.ts index 2579319a9c..0938126a29 100644 --- a/packages/world-sim/src/world.ts +++ b/packages/world-sim/src/world.ts @@ -36,7 +36,12 @@ import { import { createVirtualClock, type VirtualClock } from './clock.js'; import { createIdFactory, type IdFactory, ulidTimeOf } from './ids.js'; import { createSimQueue, type DirectHandler, type SimQueue } from './queue.js'; -import { createSimStore, type MintedEvent, type SimStore } from './store.js'; +import { + createSimStore, + type LoadedSnapshot, + type MintedEvent, + type SimStore, +} from './store.js'; import { createSimStreamer, type SimStreamer } from './streams.js'; import type { CallContext, @@ -82,16 +87,11 @@ export interface SimWorldOptions { /** See `SimStoreOptions.preconditionGuard`. */ preconditionGuard?: boolean; /** - * Also enforce the count half of the fence (see `SimStoreOptions.countGuard`), - * and supply `stateEventCount` for the writes the runtime does not count. + * Also enforce the count half of the fence (see `SimStoreOptions.countGuard`). * - * Production arms both halves. Since #3145 `@workflow/core` sends - * `stateEventCount` on every replay-context create - * (`preconditionSnapshotParams`), gated only by the - * `WORKFLOW_PRECONDITION_GUARD` kill-switch, and workflow-server's count guard - * is on by default — so this tracks `preconditionGuard` rather than being - * opted into per scenario. A run with the fence on and the count off would be - * a world that exists nowhere. + * Production arms both halves, so this tracks `preconditionGuard` rather than + * being opted into per scenario. A run with the fence on and the count off + * would be a world that exists nowhere. */ countGuard?: boolean; /** @@ -535,69 +535,85 @@ export function createSimWorld(options: SimWorldOptions = {}): SimWorld { /** * Which events the log the *runtime* is holding contains, keyed by run. * - * This reconstructs the array the client has in memory, because the count - * guard compares against that array and nothing else: `stateEventCount` is - * defined as the number of loaded events whose ULID time is at or below - * `stateUpdatedAt` — and since `stateUpdatedAt` *is* the maximum of those - * times, that is the whole array. The pair "I loaded N events, the newest at - * T" is what lets the world spot a hole *behind* T, which no comparison - * against T alone can see. + * This reconstructs the array the client has in memory, which is the whole + * input to the fence. The runtime does not state it: `@workflow/core` + * describes its snapshot as a slot count and the sim mints ULIDs, so the + * facade derives the pair the fence needs — the newest loaded position, and + * how many events sit at or below it. Since the watermark *is* the maximum of + * those times, the count is the size of the array. "I loaded N events, the + * newest at T" is what lets the world spot a hole *behind* T, which no + * comparison against T alone can see. + * + * Keyed by run *within one delivery*, not by writer: the orchestrator and the + * inline step bodies of one delivery are sim-level writers, but they are one + * process sharing one loaded log, and that log is what the count describes. + * Two writers that are genuinely separate processes get separate sets: the + * out-of-band writer, excluded by the same `isExternal()` rule that keeps it + * from being a call point, and a concurrent delivery of the same run, which + * is the shape every replay-race scenario is built on. A map that outlived + * the delivery would hand a cold-starting replay the previous delivery's + * view, and the fence would reject it for a log it never claimed to hold. * - * Keyed by run, not by writer: the orchestrator and the inline step bodies of - * one delivery are sim-level writers, but they are one process sharing one - * loaded log, and that log is what the count describes. The out-of-band - * writer is the exception — a different process with its own log — so its - * calls are excluded, by the same `isExternal()` rule that keeps them from - * being call points. + * **A read is what starts a set.** Until a writer lists the log it holds + * nothing, and no write of its own changes that: the client's own snapshot + * comes from `eventLog`, which is empty until the runtime loads it, and a + * runtime writing `run_started` before loading anything sends no snapshot at + * all. Crediting it with that write would hand the fence a position the + * client never claimed, and a *lower* one than the log holds, so a later + * concurrent write would be rejected as stale on the strength of a claim + * nobody made. * - * Everything a caller's own write appends counts as loaded, including events - * the write produces as a side effect (a `step_started` claim also appends - * the `step_created` ahead of it). They have to count: the client takes - * `stateUpdatedAt` over a log that includes what it just appended, so - * counting less would leave the count below the watermark it is paired with - * and reject perfectly current writes. + * Once a set exists, everything the caller's own writes append counts, side + * effects included (a `step_started` claim also appends the `step_created` + * ahead of it). They have to count: the caller's watermark is taken over a + * log that includes what it just appended, so counting less would leave the + * count below the watermark it is paired with and reject current writes. * * A scan that starts without a cursor replaces the set rather than adding to - * it — that is a fresh delivery re-reading the log from the beginning, and its + * it — that is the runtime re-reading the log from the beginning, and its * earlier view should not linger. */ - const loadedEvents = new Map>(); - - function loadedSet(runId: string): Set { - let set = loadedEvents.get(runId); - if (!set) { - set = new Set(); - loadedEvents.set(runId, set); - } - return set; - } + const deliveryCtx = new AsyncLocalStorage>>(); + const loadedEvents = (): Map> | undefined => + deliveryCtx.getStore(); function noteLoadedEvents( runId: string, args: readonly unknown[], result: unknown ): void { + const perRun = loadedEvents(); + if (!perRun) return; const page = (result as { data?: { eventId?: string }[] } | undefined) ?.data; if (!page) return; const cursor = (args[0] as { pagination?: { cursor?: string } } | undefined) ?.pagination?.cursor; - if (!cursor) loadedEvents.set(runId, new Set()); - const set = loadedSet(runId); + const set: Set = cursor + ? (perRun.get(runId) ?? new Set()) + : new Set(); for (const event of page) if (event.eventId) set.add(event.eventId); + perRun.set(runId, set); } - /** The `stateEventCount` the loaded log implies at `stateUpdatedAt`. */ - function loadedCount( - runId: string | undefined, - stateUpdatedAt: number - ): number { - if (!runId) return 0; - let count = 0; - for (const eventId of loadedSet(runId)) { - if (ulidTimeOf(eventId) <= stateUpdatedAt) count++; + /** + * The snapshot a replay-context write was decided against, or `undefined` + * when this writer has read nothing for the run and so has no position to + * name. A writer with an empty log is indistinguishable from an out-of-band + * one, and the fence treats it that way: there is nothing to compare. + */ + function loadedSnapshot( + runId: string | undefined + ): LoadedSnapshot | undefined { + if (!runId) return undefined; + const set = loadedEvents()?.get(runId); + if (!set || set.size === 0) return undefined; + let updatedAt = 0; + for (const eventId of set) { + const at = ulidTimeOf(eventId); + if (at > updatedAt) updatedAt = at; } - return count; + return { updatedAt, count: set.size }; } /** Wrap one world method so it becomes a call point. */ @@ -645,39 +661,37 @@ export function createSimWorld(options: SimWorldOptions = {}): SimWorld { { ...params, minted, - // The runtime's own count wins when it sent one. Since #3145 - // `preconditionSnapshotParams` puts `stateEventCount` on every - // replay-context create, so the value under test is normally the - // real client's, not ours. + // The snapshot the fence reads. Reconstructed rather than taken off + // the wire: the runtime states its position as a slot count, and + // this store mints ULIDs, so there is nothing on the wire a ULID + // fence can compare. What the facade tracks is the same array the + // client is holding (see `loadedEvents`), so the pair it derives is + // the pair the client would have sent. // - // The reconstruction is the fallback, for the writes core does not - // count: a step body committing outside a replay context, and any - // create made while the precondition guard's env kill-switch is off - // (`preconditionSnapshotParams` returns `{}` wholesale then). It is - // the size of the last page this writer read, which is what the - // client would have counted had it counted. - ...(options.countGuard && - !isExternal() && - typeof params.stateUpdatedAt === 'number' - ? { - stateEventCount: - typeof params.stateEventCount === 'number' - ? params.stateEventCount - : loadedCount(runId, params.stateUpdatedAt), - } + // Attached whenever the fence is armed, not just for the count + // half: `SimCreateParams.snapshot` is also what marks a write as + // replay-origin, and an out-of-band write must stay unmarked so it + // can advance the store's external-write marker. + ...(options.preconditionGuard && !isExternal() + ? { snapshot: loadedSnapshot(runId) } : {}), }, ] as unknown as Parameters; entered = { ...base, args: callArgs }; } - // For a create, what the log held going in — so the caller can be - // credited with everything its own write appended, not just the event the - // call handed back. A `step_started` claim also appends the `step_created` - // ahead of it, and a client that did not count both would look like it was - // holding a hole it had itself just made. + // For a create by a writer that has already loaded the log, what that log + // held going in — so the caller can be credited with everything its own + // write appended, not just the event the call handed back. A + // `step_started` claim also appends the `step_created` ahead of it, and a + // client that did not count both would look like it was holding a hole it + // had itself just made. A writer that has loaded nothing is left alone; + // see `loadedEvents`. const before = - call === 'events.create' && runId && !isExternal() + call === 'events.create' && + runId && + !isExternal() && + loadedEvents()?.has(runId) ? new Set(store.allEvents(runId).map((e) => e.eventId)) : undefined; @@ -695,9 +709,11 @@ export function createSimWorld(options: SimWorldOptions = {}): SimWorld { if (call === 'events.list') { noteLoadedEvents(runId, args, result); } else if (before) { - const set = loadedSet(runId); - for (const event of store.allEvents(runId)) { - if (!before.has(event.eventId)) set.add(event.eventId); + const set = loadedEvents()?.get(runId); + if (set) { + for (const event of store.allEvents(runId)) { + if (!before.has(event.eventId)) set.add(event.eventId); + } } } } @@ -810,8 +826,13 @@ export function createSimWorld(options: SimWorldOptions = {}): SimWorld { snapshot, trace, + // Each delivery is a separate process, and the log it holds is its own. + // Opening the scope around the handler is what makes `loadedEvents` + // describe one replay rather than every replay this run has ever had. registerHandler: (prefix, handler) => - simQueue.registerHandler(prefix as never, handler), + simQueue.registerHandler(prefix as never, (req) => + deliveryCtx.run(new Map(), () => handler(req)) + ), addWatch(watch) { const entry = { watch, matches: 0 }; watches.push(entry); diff --git a/packages/world-sim/src/writers.ts b/packages/world-sim/src/writers.ts index 971f7e5a75..61814ce491 100644 --- a/packages/world-sim/src/writers.ts +++ b/packages/world-sim/src/writers.ts @@ -9,13 +9,12 @@ * against each other, and the World API exposes no primitive that could — at * most an optimistic fence, which is two checks and not isolation: * - * - `stateUpdatedAt` is a high-water mark on one class of write ("is there an + * - a watermark is a high-water mark on one class of write ("is there an * out-of-band event newer than my snapshot?"). It sees a log truncated at the * end; it cannot see a hole in the middle. * - a count of the events the caller loaded at or below that mark closes the * hole, but only for events already committed when the write is checked, and - * only within a bounded window of the log's tail. It is also dark in - * production, since no client sends the count (`countGuard` arms it). + * only within a bounded window of the log's tail. * * So a hole in the middle is what the step-vs-step scenarios exploit, and a hole * that opens *after* the write it should have fenced is beyond either check. diff --git a/packages/world-vercel/src/events-v4.test.ts b/packages/world-vercel/src/events-v4.test.ts index 161195fe26..120ab1a384 100644 --- a/packages/world-vercel/src/events-v4.test.ts +++ b/packages/world-vercel/src/events-v4.test.ts @@ -1027,9 +1027,7 @@ describe('createWorkflowRunEventV4 over HTTP', () => { eventType: 'wait_created', specVersion: 5, correlationId: 'wait_1', - stateUpdatedAt: 1747742400000, - stateEventCount: 3, - stateCursor: 'eid:evnt_3', + maxSlot: 3, }, { token: 'test-token', dispatcher: agent } ).catch((err: unknown) => err); @@ -1152,9 +1150,7 @@ describe('createWorkflowRunEventV4 over HTTP', () => { eventType: 'wait_created', specVersion: 5, correlationId: 'wait_1', - stateUpdatedAt: 1747742400000, - stateEventCount: 3, - stateCursor: 'eid:evnt_3', + maxSlot: 3, }, { token: 'test-token', dispatcher: agent } ).catch((err: unknown) => err); @@ -1205,9 +1201,7 @@ describe('createWorkflowRunEventV4 over HTTP', () => { eventType: 'wait_created', specVersion: 5, correlationId: 'wait_1', - stateUpdatedAt: 1747742400000, - stateEventCount: 3, - stateCursor: 'eid:evnt_3', + maxSlot: 3, }, { token: 'test-token', dispatcher: agent } ).catch((err: unknown) => err); @@ -1222,124 +1216,6 @@ describe('createWorkflowRunEventV4 over HTTP', () => { agent.assertNoPendingInterceptors(); }); - it('forwards stateUpdatedAt in the frame meta (precondition guard)', async () => { - const origin = - WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; - const agent = new MockAgent(); - agent.disableNetConnect(); - - let capturedMeta: Record | undefined; - agent - .get(origin) - .intercept({ - path: '/api/v4/runs/wrun_1/events/wait_created', - method: 'POST', - }) - .reply( - 200, - (opts: { body?: unknown }) => { - const bytes = new Uint8Array(opts.body as ArrayBufferLike); - const metaLen = new DataView( - bytes.buffer, - bytes.byteOffset, - bytes.byteLength - ).getUint32(0, false); - capturedMeta = decode(bytes.subarray(4, 4 + metaLen)) as Record< - string, - unknown - >; - return createEventBody({ - eventType: 'wait_created', - specVersion: 5, - correlationId: 'wait_1', - eventData: { resumeAt: CREATED_AT }, - }); - }, - { - headers: { - 'x-wf-event-id': 'evnt_1', - 'x-wf-run-id': 'wrun_1', - 'x-wf-created-at': '2026-06-10T00:00:00.000Z', - }, - } - ); - - await createWorkflowRunEventV4( - { - runId: 'wrun_1', - eventType: 'wait_created', - specVersion: 5, - correlationId: 'wait_1', - stateUpdatedAt: 1747742400000, - }, - { token: 'test-token', dispatcher: agent } - ); - - expect(capturedMeta?.eventType).toBe('wait_created'); - expect(capturedMeta?.stateUpdatedAt).toBe(1747742400000); - agent.assertNoPendingInterceptors(); - }); - - it('forwards stateEventCount and stateCursor in the frame meta', async () => { - const origin = - WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; - const agent = new MockAgent(); - agent.disableNetConnect(); - - let capturedMeta: Record | undefined; - agent - .get(origin) - .intercept({ - path: '/api/v4/runs/wrun_1/events/wait_created', - method: 'POST', - }) - .reply( - 200, - (opts: { body?: unknown }) => { - const bytes = new Uint8Array(opts.body as ArrayBufferLike); - const metaLen = new DataView( - bytes.buffer, - bytes.byteOffset, - bytes.byteLength - ).getUint32(0, false); - capturedMeta = decode(bytes.subarray(4, 4 + metaLen)) as Record< - string, - unknown - >; - return createEventBody({ - eventType: 'wait_created', - specVersion: 5, - correlationId: 'wait_1', - eventData: { resumeAt: CREATED_AT }, - }); - }, - { - headers: { - 'x-wf-event-id': 'evnt_1', - 'x-wf-run-id': 'wrun_1', - 'x-wf-created-at': '2026-06-10T00:00:00.000Z', - }, - } - ); - - await createWorkflowRunEventV4( - { - runId: 'wrun_1', - eventType: 'wait_created', - specVersion: 5, - correlationId: 'wait_1', - stateUpdatedAt: 1747742400000, - stateEventCount: 12, - stateCursor: 'eid:evnt_1', - }, - { token: 'test-token', dispatcher: agent } - ); - - expect(capturedMeta?.stateEventCount).toBe(12); - expect(capturedMeta?.stateCursor).toBe('eid:evnt_1'); - agent.assertNoPendingInterceptors(); - }); - it('forwards maxSlot in the frame meta', async () => { const origin = WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; @@ -1452,121 +1328,6 @@ describe('createWorkflowRunEventV4 over HTTP', () => { expect('maxSlot' in (capturedMeta ?? {})).toBe(false); agent.assertNoPendingInterceptors(); }); - - it('omits stateEventCount and stateCursor from the frame meta when not set', async () => { - const origin = - WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; - const agent = new MockAgent(); - agent.disableNetConnect(); - - let capturedMeta: Record | undefined; - agent - .get(origin) - .intercept({ - path: '/api/v4/runs/wrun_1/events/wait_created', - method: 'POST', - }) - .reply( - 200, - (opts: { body?: unknown }) => { - const bytes = new Uint8Array(opts.body as ArrayBufferLike); - const metaLen = new DataView( - bytes.buffer, - bytes.byteOffset, - bytes.byteLength - ).getUint32(0, false); - capturedMeta = decode(bytes.subarray(4, 4 + metaLen)) as Record< - string, - unknown - >; - return createEventBody({ - eventType: 'wait_created', - specVersion: 5, - correlationId: 'wait_1', - eventData: { resumeAt: CREATED_AT }, - }); - }, - { - headers: { - 'x-wf-event-id': 'evnt_1', - 'x-wf-run-id': 'wrun_1', - 'x-wf-created-at': '2026-06-10T00:00:00.000Z', - }, - } - ); - - await createWorkflowRunEventV4( - { - runId: 'wrun_1', - eventType: 'wait_created', - specVersion: 5, - correlationId: 'wait_1', - stateUpdatedAt: 1747742400000, - }, - { token: 'test-token', dispatcher: agent } - ); - - expect('stateEventCount' in (capturedMeta ?? {})).toBe(false); - expect('stateCursor' in (capturedMeta ?? {})).toBe(false); - agent.assertNoPendingInterceptors(); - }); - - it('omits stateUpdatedAt from the frame meta when not set', async () => { - const origin = - WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; - const agent = new MockAgent(); - agent.disableNetConnect(); - - let capturedMeta: Record | undefined; - agent - .get(origin) - .intercept({ - path: '/api/v4/runs/wrun_1/events/wait_created', - method: 'POST', - }) - .reply( - 200, - (opts: { body?: unknown }) => { - const bytes = new Uint8Array(opts.body as ArrayBufferLike); - const metaLen = new DataView( - bytes.buffer, - bytes.byteOffset, - bytes.byteLength - ).getUint32(0, false); - capturedMeta = decode(bytes.subarray(4, 4 + metaLen)) as Record< - string, - unknown - >; - return createEventBody({ - eventType: 'wait_created', - specVersion: 5, - correlationId: 'wait_1', - eventData: { resumeAt: CREATED_AT }, - }); - }, - { - headers: { - 'x-wf-event-id': 'evnt_1', - 'x-wf-run-id': 'wrun_1', - 'x-wf-created-at': '2026-06-10T00:00:00.000Z', - }, - } - ); - - await createWorkflowRunEventV4( - { - runId: 'wrun_1', - eventType: 'wait_created', - specVersion: 5, - correlationId: 'wait_1', - }, - { token: 'test-token', dispatcher: agent } - ); - - expect(capturedMeta?.eventType).toBe('wait_created'); - expect('stateUpdatedAt' in (capturedMeta ?? {})).toBe(false); - agent.assertNoPendingInterceptors(); - }); }); /** diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index 0b6eb2b867..e8de126e6b 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -222,39 +222,15 @@ interface CreateEventV4InputBase { * other event types; older servers ignore it entirely (the runtime then * falls back to events.list). */ sinceCursor?: string; - /** - * Epoch ms (the ULID time of the latest event the runtime has loaded - * during replay). Sent by replay-context creates so the backend can - * reject the event when a newer out-of-band event was recorded after this - * snapshot, enabling an optimistic-concurrency guard. Omitted by callers - * without a loaded event log; older servers ignore it entirely. - */ - stateUpdatedAt?: number; - /** - * Number of loaded events at or below `stateUpdatedAt` (i.e. the loaded - * log's length). Sent with `stateUpdatedAt` so the backend can also reject - * a snapshot that is *missing* an event at or below its watermark — the - * corruption case a watermark alone cannot detect. Older servers ignore it. - */ - stateEventCount?: number; - /** - * The runtime's event-log cursor at snapshot time. Advisory: sent so a - * rejecting backend MAY return the missing events on the 412 body, saving a - * follow-up events.list. Distinct from `sinceCursor`, which the server acts - * on for the *accepted* path. - */ - stateCursor?: string; /** * Highest event slot the writer had loaded, i.e. the length of its loaded * log under slot identity. Named `maxSlot` on the wire because the meta * already carries an unrelated telemetry `eventCount`. * - * Supersedes the `stateUpdatedAt`/`stateEventCount`/`stateCursor` triple for - * slot-identity runs: with dense positions one integer says everything the - * watermark approximated. The server allocates from the tail regardless, and - * uses this only to report which slots the write skipped over (returned on - * the success response as `events`/`cursor`/`hasMore`). Older servers ignore - * it. + * Sent by every replay-context create on a slot-identity run. The server + * allocates from the tail regardless, and uses this only to report which + * slots the write skipped over (returned on the success response as + * `events`/`cursor`/`hasMore`). Older servers ignore it. */ maxSlot?: number; /** Number of consecutive replay divergences resolved by this write. */ @@ -463,13 +439,6 @@ function buildPostFrameMeta( } if (input.sinceCursor !== undefined) meta.sinceCursor = input.sinceCursor; if (input.skipPreload) meta.skipPreload = true; - if (input.stateUpdatedAt !== undefined) { - meta.stateUpdatedAt = input.stateUpdatedAt; - } - if (input.stateEventCount !== undefined) { - meta.stateEventCount = input.stateEventCount; - } - if (input.stateCursor !== undefined) meta.stateCursor = input.stateCursor; if (input.maxSlot !== undefined) meta.maxSlot = input.maxSlot; if (input.replayDivergenceCount !== undefined) { meta.replayDivergenceCount = input.replayDivergenceCount; diff --git a/packages/world-vercel/src/events.test.ts b/packages/world-vercel/src/events.test.ts index ab1eb1632a..5fea9675a4 100644 --- a/packages/world-vercel/src/events.test.ts +++ b/packages/world-vercel/src/events.test.ts @@ -179,54 +179,14 @@ describe('createWorkflowRunEvent with v1Compat', () => { }); /** - * The optimistic-concurrency precondition guard: a replay-context create - * describes the runtime's loaded snapshot with three params — `stateUpdatedAt` - * (the ULID time of the latest loaded event), `stateEventCount` (how many - * events that snapshot holds at or below it) and `stateCursor` (so a rejecting - * backend may return the missing events inline). Locks in that each reaches - * the v4 frame meta, and that all are omitted when the caller has no loaded - * snapshot — an unsent field disables the corresponding backend check. + * A replay-context create names the position its decisions were made at: + * `eventCount`, the highest event slot the runtime had loaded. Locks in that it + * reaches the v4 frame meta under the wire name the backend reads, and that it + * is omitted when the caller has no loaded snapshot — an unsent field leaves + * the backend with no position to report a skipped span against. */ -describe('createWorkflowRunEvent precondition snapshot wire fields', () => { - it('includes stateUpdatedAt in the v4 frame meta when provided', async () => { - const agent = mockAgent(); - let capturedMeta: Record | undefined; - - agent - .get(ORIGIN) - .intercept({ - path: '/api/v4/runs/wrun_1/events/run_started', - method: 'POST', - }) - .reply( - 200, - (opts: { body?: unknown }) => { - capturedMeta = decodePostedMeta(opts.body); - return runStartedResponse(); - }, - { - headers: { - 'content-type': V4_FRAME_CONTENT_TYPE, - 'x-wf-event-id': 'evnt_1', - 'x-wf-run-id': 'wrun_1', - 'x-wf-created-at': '2026-06-10T00:00:00.000Z', - 'x-wf-max-events': '10000', - }, - } - ); - - await createWorkflowRunEvent( - 'wrun_1', - { eventType: 'run_started', specVersion: 2 } as AnyEventRequest, - { stateUpdatedAt: 1_700_000_000_000 }, - { token: 'test-token', dispatcher: agent } - ); - - expect(capturedMeta?.stateUpdatedAt).toBe(1_700_000_000_000); - agent.assertNoPendingInterceptors(); - }); - - it('omits stateUpdatedAt from the v4 frame meta when not provided', async () => { +describe('createWorkflowRunEvent slot snapshot wire fields', () => { + it('omits maxSlot from the v4 frame meta when no snapshot is provided', async () => { const agent = mockAgent(); let capturedMeta: Record | undefined; @@ -260,89 +220,7 @@ describe('createWorkflowRunEvent precondition snapshot wire fields', () => { { token: 'test-token', dispatcher: agent } ); - expect('stateUpdatedAt' in (capturedMeta ?? {})).toBe(false); - agent.assertNoPendingInterceptors(); - }); - - it('includes stateEventCount and stateCursor in the v4 frame meta when provided', async () => { - const agent = mockAgent(); - let capturedMeta: Record | undefined; - - agent - .get(ORIGIN) - .intercept({ - path: '/api/v4/runs/wrun_1/events/run_started', - method: 'POST', - }) - .reply( - 200, - (opts: { body?: unknown }) => { - capturedMeta = decodePostedMeta(opts.body); - return runStartedResponse(); - }, - { - headers: { - 'content-type': V4_FRAME_CONTENT_TYPE, - 'x-wf-event-id': 'evnt_1', - 'x-wf-run-id': 'wrun_1', - 'x-wf-created-at': '2026-06-10T00:00:00.000Z', - 'x-wf-max-events': '10000', - }, - } - ); - - await createWorkflowRunEvent( - 'wrun_1', - { eventType: 'run_started', specVersion: 2 } as AnyEventRequest, - { - stateUpdatedAt: 1_700_000_000_000, - stateEventCount: 7, - stateCursor: 'eid:evnt_1', - }, - { token: 'test-token', dispatcher: agent } - ); - - expect(capturedMeta?.stateEventCount).toBe(7); - expect(capturedMeta?.stateCursor).toBe('eid:evnt_1'); - agent.assertNoPendingInterceptors(); - }); - - it('omits stateEventCount and stateCursor from the v4 frame meta when not provided', async () => { - const agent = mockAgent(); - let capturedMeta: Record | undefined; - - agent - .get(ORIGIN) - .intercept({ - path: '/api/v4/runs/wrun_1/events/run_started', - method: 'POST', - }) - .reply( - 200, - (opts: { body?: unknown }) => { - capturedMeta = decodePostedMeta(opts.body); - return runStartedResponse(); - }, - { - headers: { - 'content-type': V4_FRAME_CONTENT_TYPE, - 'x-wf-event-id': 'evnt_1', - 'x-wf-run-id': 'wrun_1', - 'x-wf-created-at': '2026-06-10T00:00:00.000Z', - 'x-wf-max-events': '10000', - }, - } - ); - - await createWorkflowRunEvent( - 'wrun_1', - { eventType: 'run_started', specVersion: 2 } as AnyEventRequest, - { stateUpdatedAt: 1_700_000_000_000 }, - { token: 'test-token', dispatcher: agent } - ); - - expect('stateEventCount' in (capturedMeta ?? {})).toBe(false); - expect('stateCursor' in (capturedMeta ?? {})).toBe(false); + expect('maxSlot' in (capturedMeta ?? {})).toBe(false); agent.assertNoPendingInterceptors(); }); @@ -389,8 +267,9 @@ describe('createWorkflowRunEvent precondition snapshot wire fields', () => { }); it('never sends the snapshot on the legacy v1Compat path', async () => { - // Pre-event-sourcing runs have no event log to fence, and the legacy - // endpoint has no field for the snapshot: the params are dropped whole. + // Pre-event-sourcing runs have no slot-numbered log to name a position + // in, and the legacy endpoint has no field for one: the params are dropped + // whole. const agent = mockAgent(); let capturedBody = ''; @@ -425,19 +304,13 @@ describe('createWorkflowRunEvent precondition snapshot wire fields', () => { specVersion: 1, eventData: { resumeAt: '2026-06-10T00:00:00.000Z' }, } as AnyEventRequest, - { - v1Compat: true, - stateUpdatedAt: 1_700_000_000_000, - stateEventCount: 7, - stateCursor: 'eid:evnt_1', - }, + { v1Compat: true, eventCount: 7 }, { token: 'test-token', dispatcher: agent } ); expect(capturedBody).toContain('wait_completed'); - expect(capturedBody).not.toContain('stateUpdatedAt'); - expect(capturedBody).not.toContain('stateEventCount'); - expect(capturedBody).not.toContain('stateCursor'); + expect(capturedBody).not.toContain('maxSlot'); + expect(capturedBody).not.toContain('eventCount'); agent.assertNoPendingInterceptors(); }); }); diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index 1e8c482844..8ecc213af6 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -603,11 +603,7 @@ async function createWorkflowRunEventInner( ...(params?.computeInstanceId ? { computeInstanceId: params.computeInstanceId } : {}), - stateUpdatedAt: params?.stateUpdatedAt, - stateEventCount: params?.stateEventCount, - ...(params?.stateCursor ? { stateCursor: params.stateCursor } : {}), - // Slot-identity snapshot. The runtime sends `eventCount` instead of the - // watermark triple once the run's own ids are slot-shaped; it rides as + // Slot-identity snapshot: how much of the log the writer held. Rides as // `maxSlot` because the v4 meta already has an unrelated telemetry // `eventCount`. ...(params?.eventCount !== undefined ? { maxSlot: params.eventCount } : {}), diff --git a/packages/world-vercel/src/index.ts b/packages/world-vercel/src/index.ts index e94e98c07e..f1fb2cbe53 100644 --- a/packages/world-vercel/src/index.ts +++ b/packages/world-vercel/src/index.ts @@ -38,10 +38,16 @@ export function createWorld(config?: APIConfig): World { specVersion: SPEC_VERSION_SUPPORTS_SLOT_IDENTITY, capabilities: { hookRetention: { active: true }, - // workflow-server enforces the `stateUpdatedAt` optimistic-concurrency - // guard: creations carrying a stale snapshot are rejected with 412 - // (PreconditionFailedError) when the run's outside-event marker is - // newer. See vercel/workflow-server#484. + // The backend rejects a stale create with 412 (PreconditionFailedError) + // rather than committing it. + // + // No write this adapter now sends can be rejected that way: the backend + // evaluates the fence only for a create carrying a ULID-era snapshot, + // and this SDK sends none. A run created here is v6, where staleness is + // handled by allocating the write above the contention and reporting + // back the slots it skipped. The capability is kept because the runtime + // reads it as "a write can be refused", and the choices keyed on it stay + // the conservative ones while the slot path carries the load. preconditionGuard: true, // Vercel Queues supports maxConcurrency-limited consumers, which // WORKFLOW_SEQUENTIAL_REPLAYS=1 uses for per-run `maxConcurrency: 1` diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index 545cd59376..6373bc3678 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -813,82 +813,11 @@ export interface CreateEventParams { * `AnalyticsEventSchema` / `AnalyticsStepSchema`. */ computeInstanceId?: string; - /** - * Epoch ms (the ULID time of the latest event the runtime has loaded during - * replay). Sent by replay-context creates so the backend can reject the event - * when a newer out-of-band event was recorded after this snapshot, enabling - * an optimistic-concurrency guard. Omitted by callers without a loaded event - * log. - * - * Backend contract (for World implementers who want to support the guard): - * maintain a per-run marker holding the ULID time of the most recent - * *externally-originated* event — a `hook_received` or `step_completed` - * created **without** a `stateUpdatedAt` (replay-origin events carry one and - * must not advance the marker). On a create that carries `stateUpdatedAt`, - * reject with 412 when `stateUpdatedAt < marker` (strictly older); an equal - * timestamp must pass (anti-livelock, so an up-to-date client is never - * rejected). A backend that ignores this field simply disables the guard — - * the client falls open and behaves as before. - * - * A watermark alone cannot see an event *missing at or below* it, which is - * the failure that actually corrupts a replay — see {@link stateEventCount} - * for the second half of the guard. - */ - stateUpdatedAt?: number; - /** - * How many loaded events have a ULID time at or below {@link stateUpdatedAt}. - * Since `stateUpdatedAt` is the *maximum* ULID time in the loaded log, this - * equals the loaded array's length. Sent **only** together with - * `stateUpdatedAt`; a World must ignore a count that arrives without one. - * - * This closes the hole a watermark cannot: the watermark proves only "no - * newer event exists", while a replay corrupts its log by missing an event - * at or *below* its own frontier — a concurrent writer commits in the same - * ULID millisecond as the client's last loaded event, so the two watermarks - * compare equal and the write is accepted against a log that is one event - * short. Because correlation IDs are positional ordinals of a single seeded - * sequence, that one-event difference renames every entity after it. - * - * Backend contract (for World implementers who want to support this half): - * - * - Count **every** created event for the run, including replay-origin ones. - * Unlike the watermark, this is not restricted to out-of-band writes: the - * race being fenced is one replay against another. - * - Reject with 412 when the count of recorded events at ULID time - * `<= stateUpdatedAt` is strictly **greater** than `stateEventCount`. - * - Compare **at or below** `stateUpdatedAt`, never strictly below (the - * missing event routinely shares the client's frontier millisecond) and - * never against a total (all the creates of one suspension share one - * snapshot, so a total would reject every sibling after the first). - * - **One-sided safety is mandatory.** Anything that makes the backend's - * count incomplete, uncomputable, or expired must *allow* the write. A - * rejection has to imply a real hole, because the client responds to it by - * discarding and re-deriving its whole replay. - * - * See also the millisecond-granularity caveat on `stateUpdatedAt`: the count - * is what makes an equal-timestamp snapshot safe to accept. - */ - stateEventCount?: number; - /** - * The client's current event-log cursor (advisory). Sent alongside the other - * two snapshot fields so a World that rejects the write MAY return the - * events the client is missing on the 412 itself, saving the client a - * follow-up `events.list`. - * - * Distinct from {@link sinceCursor}: a World must **not** compute a delta for - * this on the accepted path — it exists purely to make a rejection cheaper. - * Returning events on a 412 is OPTIONAL, and the returned set must be - * provably complete (it must account for the entire discrepancy the - * rejection reported) or omitted entirely: a cursor filters by lexicographic - * event id while a hole is defined by ULID time, so a naive - * "everything after the cursor" delta can silently exclude the very event - * the client is missing. A client that receives nothing does the - * authoritative full reload, which is always correct. - */ - stateCursor?: string; /** * How many events the writer held in its loaded log when it decided to write - * this one — equivalently, the slot it expects to land on minus one. + * this one — equivalently, the slot it expects to land on minus one. Sent by + * every replay-context create; omitted by callers with no loaded log, and by + * a run whose events are not slot-numbered (there is no position to name). * * Only meaningful against a World that declares * `WorldCapabilities.slotEventIds`, where slots are dense and 1-based so a @@ -900,13 +829,10 @@ export interface CreateEventParams { * response in {@link EventResult.events} / `cursor` / `hasMore`, so the * writer learns exactly what it had not seen. * - * This supersedes the {@link stateUpdatedAt} / {@link stateEventCount} / - * {@link stateCursor} triple for slot Worlds. That triple approximates a - * position with a ULID-time watermark plus a count of events at or below it, - * which is why a *complete but stale* prefix passes it: every event the - * writer holds is at or below its own watermark, so the count matches and no - * fence fires. A dense position has no such blind spot. Worlds without slots - * ignore this field and keep using the triple. + * Understating is safe and overstating is not. A count below the writer's + * true position only widens the reported span, and the client discards what + * its log already holds. A count above it makes the World report less than + * the writer is missing, which is a hole the writer never learns about. * * A batch of writes issued from one snapshot starts from the same * `eventCount`; they land on consecutive slots in whatever order the World diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts index 06a49b973e..9b2c21b610 100644 --- a/packages/world/src/interfaces.ts +++ b/packages/world/src/interfaces.ts @@ -335,22 +335,20 @@ export interface WorldCapabilities { }; /** - * The World enforces the optimistic-concurrency precondition guard: an - * event creation carrying a `stateUpdatedAt` snapshot is rejected with a - * `PreconditionFailedError` (412) when a newer out-of-band event (e.g. a - * received hook) was recorded after that snapshot. Worlds that accept but - * ignore `stateUpdatedAt` must leave this unset so runtime optimizations - * that rely on the 412 fence (see `WORKFLOW_PRECONDITION_GUARD`) are not - * enabled without an actual fence behind them. + * The World fences a stale replay-context write: an event creation whose + * snapshot is behind what the run has already recorded is rejected with a + * `PreconditionFailedError` (412) rather than committed. The runtime's + * response is to abandon the write and replay from a corrected log. * - * A World declaring this should honour the whole snapshot the runtime sends, - * not just the watermark: `stateUpdatedAt`, `stateEventCount` (the count - * fence, which catches an event missing at or below the watermark — the case - * the watermark provably cannot see) and, optionally, `stateCursor` (return - * the missing events on the 412 to save the client a reload). See - * `CreateEventParams` for each field's contract. The runtime does not branch - * on which halves are implemented; a World that ignores the count simply - * fences less. + * Runtime optimizations that are only safe behind that fence read this + * capability, so a World that accepts a snapshot and ignores it must leave + * this unset: sending a snapshot is not the same as one being enforced, and + * enabling those optimizations with nothing behind them makes a stale replay + * commit. + * + * Orthogonal to {@link slotEventIds}, which never rejects — it commits above + * the contention and reports back what the writer missed. A World may + * declare either, both, or neither. */ preconditionGuard?: boolean; @@ -425,8 +423,7 @@ export interface WorldCapabilities { * - **Density.** A run's slots are contiguous from 1, so the number of * events a reader holds *is* the position of the last one. That is what * makes {@link CreateEventParams.eventCount} a complete statement of the - * writer's snapshot, where the `stateUpdatedAt` / `stateEventCount` - * watermark pair could only approximate it. + * writer's snapshot in a single integer. * - **Bump and report.** A create never fails because its requested slot is * taken. The World advances to the next free slot, commits there, and * returns the events occupying the slots it skipped over on the success diff --git a/workbench/sim-world/scenarios/in-flight-before-decision-counted.ts b/workbench/sim-world/scenarios/in-flight-before-decision-counted.ts index 7caea5f29c..827a8aa660 100644 --- a/workbench/sim-world/scenarios/in-flight-before-decision-counted.ts +++ b/workbench/sim-world/scenarios/in-flight-before-decision-counted.ts @@ -13,11 +13,10 @@ export const scenario: ScenarioSpec = { 'ahead of the timeout in log order, re-decides the fork as "arrived", and ' + 'commits the branch the log agrees with. This is the regression test for ' + 'the half of the fence a high-water mark cannot express: same fault, same ' + - 'tempo, one flag apart. This is also the production-shaped one of the ' + - 'pair: since #3145 `@workflow/core` sends the count on every ' + - 'replay-context create and the server checks it by default, so the flag ' + - 'below is now the default rather than an opt-in, and it is the twin above ' + - 'that has to switch it off. ' + + 'tempo, one flag apart. The count the caller sends is the same number ' + + '`@workflow/core` puts on every replay-context create; what differs is ' + + 'what a World does with it. This sim rejects on it, which is why the flag ' + + 'below is the default and the twin above has to switch it off. ' + 'Under an append-only log the 412 still fires and now saves nothing: the ' + 'count is taken at or below the caller’s watermark, and the watermark ' + 'is a millisecond, so a hook that commits after the timeout inside the ' + diff --git a/workbench/sim-world/scenarios/in-flight-before-decision.ts b/workbench/sim-world/scenarios/in-flight-before-decision.ts index 47759e8441..23259f7015 100644 --- a/workbench/sim-world/scenarios/in-flight-before-decision.ts +++ b/workbench/sim-world/scenarios/in-flight-before-decision.ts @@ -15,7 +15,7 @@ export const scenario: ScenarioSpec = { 'holds an event the writer never loaded. The watermark guard is on and ' + 'passes anyway, by construction: the marker moves to the ULID time of the ' + "hook, which sorts at or below the writer's own snapshot, so " + - '`stateUpdatedAt < marker` is false. It corrupts — the same corruption as ' + + '`snapshot.updatedAt < marker` is false. It corrupts — the same corruption as ' + 'the doc-23 pair, reached without a stale read. ' + 'Under an append-only log there is no position to be spoken for: the hook ' + 'commits after the timeout and therefore sorts after it, the log says the ' + diff --git a/workbench/sim-world/scenarios/stale-read-step-count-fork-fenced.ts b/workbench/sim-world/scenarios/stale-read-step-count-fork-fenced.ts index c738f8f977..40e821dbbc 100644 --- a/workbench/sim-world/scenarios/stale-read-step-count-fork-fenced.ts +++ b/workbench/sim-world/scenarios/stale-read-step-count-fork-fenced.ts @@ -5,8 +5,8 @@ export const scenario: ScenarioSpec = { name: 'corrupt: same shape, with the optimistic-concurrency fence armed', description: 'Identical to the count: fork scenario but with preconditionGuard on, so ' + - 'the World rejects a replay-context write whose stateUpdatedAt snapshot ' + - 'predates the newest out-of-band event. Does the 412 fence stop it? It ' + + 'the World rejects a replay-context write whose snapshot predates the ' + + 'newest out-of-band event. Does the 412 fence stop it? It ' + 'does: every rejected write is traced as a `!!` line, and the run ' + 'reconciles instead of diverging.', workflow: 'stepCountForkWorkflow', diff --git a/workbench/sim-world/scenarios/step-vs-step-fork-fenced.ts b/workbench/sim-world/scenarios/step-vs-step-fork-fenced.ts index 34b6bfef49..59925e0a5b 100644 --- a/workbench/sim-world/scenarios/step-vs-step-fork-fenced.ts +++ b/workbench/sim-world/scenarios/step-vs-step-fork-fenced.ts @@ -12,8 +12,8 @@ export const scenario: ScenarioSpec = { 'PreconditionFailedError rejections, and it corrupts identically. The ' + 'reason is the shape of the predicate, not the event type: the fence ' + 'compares the snapshot against a HIGH-WATER MARK of the newest ' + - 'out-of-band write, and rejects only `stateUpdatedAt < marker`. Here the ' + - 'newest such write is the one the reader CAN see (`fast`); the withheld ' + + 'out-of-band write, and rejects only `snapshot.updatedAt < marker`. Here ' + + 'the newest such write is the one the reader CAN see (`fast`); the withheld ' + "one is older, a hole in the middle of the log, so the reader's snapshot " + 'is never strictly older than the mark. Separating the two completions ' + 'in virtual time does not change it — the miss is structural, not a ' + @@ -24,11 +24,13 @@ export const scenario: ScenarioSpec = { 'up in the trace as `!!` lines, unasked for. ' + 'To be precise about which fence: this scenario arms the watermark half ' + 'ALONE, which is why `countGuard` is switched off below against the ' + - 'default. The count half is aimed at exactly this hole and does catch it, ' + - 'and since #3145 it is armed in production too — so what stays red here ' + - 'is the watermark predicate, not a hole anything real still has. See the ' + - 'in-flight trio below, where the two halves are separated and tested one ' + - 'flag apart.', + 'default. The count half is aimed at exactly this hole and does catch it. ' + + 'Neither half models world-vercel any more: a slot-identity run has no ' + + 'stale-snapshot rejection to make, because the World allocates the slot ' + + 'at commit time rather than taking a position from the writer. What the ' + + 'fence still buys is coverage of the 412 path the runtime keeps for ' + + 'Worlds that do fence. See the in-flight trio below, where the two halves ' + + 'are separated and tested one flag apart.', workflow: 'stepVsStepForkWorkflow', input: ['doc-27'], preconditionGuard: true,