diff --git a/.changeset/drop-precondition-guard-capability.md b/.changeset/drop-precondition-guard-capability.md new file mode 100644 index 0000000000..7b9c2cf27d --- /dev/null +++ b/.changeset/drop-precondition-guard-capability.md @@ -0,0 +1,7 @@ +--- +'@workflow/world': patch +'@workflow/core': patch +'@workflow/world-vercel': patch +--- + +Remove the `preconditionGuard` World capability. A stale replay-context write no longer needs to be rejected: a reader holds a prefix of the log, replay is deterministic on a prefix, and the writer's next write reports the events it was pushed past. diff --git a/.changeset/require-slot-event-ids.md b/.changeset/require-slot-event-ids.md new file mode 100644 index 0000000000..bcb8974eb4 --- /dev/null +++ b/.changeset/require-slot-event-ids.md @@ -0,0 +1,6 @@ +--- +'@workflow/world': patch +'@workflow/core': patch +--- + +Require every event id the runtime reads to be a log position. `requireEventSlot` replaces the lenient decode that returned "no position" for an id that is not a slot. diff --git a/.changeset/resilient-step-dispatch-off.md b/.changeset/resilient-step-dispatch-off.md new file mode 100644 index 0000000000..0f5e4ea322 --- /dev/null +++ b/.changeset/resilient-step-dispatch-off.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Turn resilient step dispatch off by default. Set `WORKFLOW_RESILIENT_STEP_DISPATCH=1` to opt back in. diff --git a/.changeset/slot-ids-are-required.md b/.changeset/slot-ids-are-required.md new file mode 100644 index 0000000000..7ad5b27a3e --- /dev/null +++ b/.changeset/slot-ids-are-required.md @@ -0,0 +1,9 @@ +--- +'@workflow/world': patch +'@workflow/world-testing': patch +'@workflow/world-local': patch +'@workflow/world-postgres': patch +'@workflow/world-vercel': patch +--- + +Slot-numbered event ids are a requirement of the World contract, not a capability. The `slotEventIds` flag is gone, and the conformance suite now fails a World whose event ids are not positions. diff --git a/.changeset/windows-preload-timeout.md b/.changeset/windows-preload-timeout.md deleted file mode 100644 index 2665fe54af..0000000000 --- a/.changeset/windows-preload-timeout.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -'@workflow/world-local': patch ---- - -Raise the timeout on the world-local complete-preload test so it stops failing on the Windows CI runner. diff --git a/.github/scripts/render-event-log-race-repro-results.js b/.github/scripts/render-event-log-race-repro-results.js index bbead520b2..d66152fb51 100644 --- a/.github/scripts/render-event-log-race-repro-results.js +++ b/.github/scripts/render-event-log-race-repro-results.js @@ -9,6 +9,15 @@ let previousCommentPath = ''; let timestamp = new Date().toISOString(); let runAttempt = ''; let check = false; +// Which lane this run belongs to (e.g. `world-local`, `world-postgres`). +// Suffixes the results marker and the heading so each lane owns its own sticky +// comment: the lanes run as parallel jobs, and a shared marker would make one +// lane's "previous comment" fetch pick up another lane's history. The Vercel +// lane passes no label and keeps the original marker, so its comment history +// survives this flag's introduction. Marker matching is exact-substring +// (`` does not match the `-world-local` +// variant because of the closing ` -->`), which is what keeps the lanes apart. +let label = ''; for (let index = 0; index < args.length; index += 1) { const arg = args[index]; @@ -24,6 +33,9 @@ for (let index = 0; index < args.length; index += 1) { } else if (arg === '--run-attempt' && args[index + 1]) { runAttempt = args[index + 1]; index += 1; + } else if (arg === '--label' && args[index + 1]) { + label = args[index + 1]; + index += 1; } else if (arg === '--check') { check = true; } else if (!arg.startsWith('--')) { @@ -432,8 +444,10 @@ function render(resultsFile, previousComment) { '. Rates are still comparable; totals are not.' : ''; - console.log(''); - console.log('## Event Log Race Repro\n'); + console.log( + `` + ); + console.log(`## Event Log Race Repro${label ? ` (${label})` : ''}\n`); console.log( latest.missingResults ? 'No result file was produced by the latest repro job.' diff --git a/.github/workflows/event-log-race-repro.yml b/.github/workflows/event-log-race-repro.yml index 6f78986f2a..b3603070ec 100644 --- a/.github/workflows/event-log-race-repro.yml +++ b/.github/workflows/event-log-race-repro.yml @@ -181,3 +181,131 @@ jobs: - name: Fail on corrupted, failed, or stuck runs if: always() run: node .github/scripts/render-event-log-race-repro-results.js event-log-race-repro-results.json --check + + # The same harness against the self-hosted Worlds, one parallel job per World. + # `scripts/event-log-race-repro-local.sh` owns the whole lifecycle: it builds + # the packages and the workbench app with WORKFLOW_TARGET_WORLD and + # WORKFLOW_PUBLIC_MANIFEST=1 set at *build* time (both are build-time inputs), + # brings up Postgres and applies migrations for `--world postgres`, starts the + # app, runs the harness, and tears the server down. + # + # Unlike the Vercel lane above, these lanes publish numbers instead of a + # verdict. The local storms bite much harder than the Vercel preview does — + # the world-postgres step-storm has red baselines at the default scale (see + # "Event Log Race Repro" in CLAUDE.md) — so a gate here would be red on PRs + # that changed nothing, and a gate that is always red is a gate everyone + # learns to ignore. Each lane posts its own sticky PR comment + # (`world-local` / `world-postgres` suffixed markers, so the three lanes' + # histories never mix), and the number to watch is in the comment. The only + # thing that fails these jobs is plumbing: a harness that produced no result + # file at all. + # + # A soak dispatch that raises `budget_ms` has to raise `timeout-minutes` here + # too, same as the Vercel lane. + event-log-race-repro-local-worlds: + name: Event Log Race Repro (world-${{ matrix.world }}) + runs-on: ubuntu-latest + # The Vercel lane's 25 minutes, plus room for what it never pays for: + # pnpm build, the app build, and (postgres) container startup + migrations. + timeout-minutes: 45 + if: ${{ github.event_name == 'workflow_dispatch' || contains(github.event.pull_request.labels.*.name, 'event-log-race-repro') }} + strategy: + fail-fast: false + matrix: + world: [local, postgres] + permissions: + contents: read + issues: write + pull-requests: write + env: + TURBO_TOKEN: ${{ secrets.TURBO_TOKEN }} + TURBO_TEAM: ${{ vars.TURBO_TEAM }} + + steps: + - name: Checkout Repo + uses: actions/checkout@v4 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + + - name: Setup environment + uses: ./.github/actions/setup-workflow-dev + with: + # The script runs its own `pnpm build` so the packages are built in + # the same environment as the app. + build-packages: 'false' + + - name: Run event log race repro (world-${{ matrix.world }}) + id: repro + # The script exits with the harness' own status, which fails on any + # regression outcome — expected on some baselines, see the job comment. + continue-on-error: true + run: ./scripts/event-log-race-repro-local.sh --world ${{ matrix.world }} + env: + # Deliberately unguarded pass-through: on a `pull_request` event every + # `inputs.*` is empty, and the harness reads an empty variable as unset + # and falls back to its own default. See the note on the inputs above. + EVENT_LOG_RACE_REPRO_STEP_STORM_ATTEMPTS: ${{ inputs.step_storm_attempts }} + EVENT_LOG_RACE_REPRO_HOOK_STORM_ATTEMPTS: ${{ inputs.hook_storm_attempts }} + EVENT_LOG_RACE_REPRO_ATTEMPTS: ${{ inputs.attempts }} + EVENT_LOG_RACE_REPRO_CONCURRENCY: ${{ inputs.concurrency }} + EVENT_LOG_RACE_REPRO_ROUNDS: ${{ inputs.rounds }} + EVENT_LOG_RACE_REPRO_WIDTH: ${{ inputs.width }} + EVENT_LOG_RACE_REPRO_WATCHDOG_MS: ${{ inputs.watchdog_ms }} + EVENT_LOG_RACE_REPRO_STEP_DELAY_MS: ${{ inputs.step_delay_ms }} + EVENT_LOG_RACE_REPRO_HOOK_RESUME_STAGGER_MS: ${{ inputs.hook_resume_stagger_ms }} + EVENT_LOG_RACE_REPRO_POKE_INTERVAL_MS: ${{ inputs.poke_interval_ms }} + EVENT_LOG_RACE_REPRO_RUN_TIMEOUT_MS: ${{ inputs.run_timeout_ms }} + EVENT_LOG_RACE_REPRO_BUDGET_MS: ${{ inputs.budget_ms }} + + - name: Fetch previous repro comment (world-${{ matrix.world }}) + if: always() && github.event_name == 'pull_request' + env: + GH_TOKEN: ${{ github.token }} + run: | + gh api \ + "repos/${{ github.repository }}/issues/${{ github.event.pull_request.number }}/comments?per_page=100" \ + --jq '[.[] | select(.body | contains(""))][-1].body // ""' \ + > event-log-race-repro-previous-comment.md + + - name: Render repro summary + if: always() + run: | + previous_comment_args=() + if [ -f event-log-race-repro-previous-comment.md ]; then + previous_comment_args=(--previous-comment event-log-race-repro-previous-comment.md) + fi + + node .github/scripts/render-event-log-race-repro-results.js \ + event-log-race-repro-results.json \ + --label "world-${{ matrix.world }}" \ + --run-url "${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}" \ + --run-attempt "${{ github.run_attempt }}" \ + --timestamp "$(date -u +'%Y-%m-%dT%H:%M:%SZ')" \ + "${previous_comment_args[@]}" \ + | tee event-log-race-repro-summary.md >> "$GITHUB_STEP_SUMMARY" + + - name: Update PR comment + if: always() && github.event_name == 'pull_request' + uses: marocchino/sticky-pull-request-comment@773744901bac0e8cbb5a0dc842800d45e9b2b405 # v2.9.4 + with: + header: event-log-race-repro-results-world-${{ matrix.world }} + path: event-log-race-repro-summary.md + + - name: Upload repro results + if: always() + uses: actions/upload-artifact@v4 + with: + name: event-log-race-repro-results-world-${{ matrix.world }} + path: | + event-log-race-repro-results.json + event-log-race-repro-summary.md + event-log-race-repro-server.log + retention-days: 7 + if-no-files-found: ignore + + # Plumbing gate only: regressions are reported, not gated (see the job + # comment), but a run that produced no result file at all is a broken + # lane, and a broken lane that stays green never gets fixed. + - name: Fail if the harness produced no results + if: always() + run: test -f event-log-race-repro-results.json diff --git a/.github/workflows/world-sim.yml b/.github/workflows/world-sim.yml index df62b815f7..8587ef4282 100644 --- a/.github/workflows/world-sim.yml +++ b/.github/workflows/world-sim.yml @@ -3,20 +3,20 @@ name: World Sim # Plays the deterministic scenario book (`workbench/sim-world`) against the # runtime in this commit, once per log world, and publishes the two summaries. # -# This lane never blocks a merge, by design. Six scenarios in the book fail on +# This lane never blocks a merge, by design. Three scenarios in the book fail on # purpose: each one is a reproduction of a corruption the runtime can still # produce, stating the outcome its own durable log implies, and staying red # until the runtime gets there. A gate that goes red on every PR is a gate # everyone learns to ignore, so the job publishes numbers instead of verdicts — # and the number to watch is in the comment, not the check mark. # -# mint-ordered (production): 35 passed, 6 failed, 6 violations +# mint-ordered (production): 38 passed, 3 failed, 3 violations # append-only: 41 passed, 0 failed, 0 violations # -# A seventh red is a regression. Five means something got fixed and a scenario +# A fourth red is a regression. Two means something got fixed and a scenario # is ready to retire. The append-only column is the measurement the pair exists -# for: it says which of the six would close if event positions were assigned at -# commit instead of at the handler's mint. +# for: it says which of the three would close if event positions were assigned +# at commit instead of at the handler's mint. on: push: diff --git a/AGENTS.md b/AGENTS.md index 83324adbc3..73640b109d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -211,7 +211,13 @@ In CI the same harness runs from `.github/workflows/event-log-race-repro.yml`, triggered by adding the `event-log-race-repro` label to a PR (or by `workflow_dispatch`, whose inputs are the soak dial — raise `timeout-minutes` in that dispatch's branch if you raise `budget_ms`). Results land in a sticky PR -comment that keeps a history of previous runs and their configs. +comment that keeps a history of previous runs and their configs. Alongside the +Vercel lane, the workflow runs the local script against world-local and +world-postgres as parallel lanes, each with its own sticky comment. Those two +lanes are report-only — the local storms have red baselines at the default +scale (see above), so they publish numbers rather than a verdict and fail only +when the harness produced no result file at all; the Vercel lane remains the +gate. To poke at a run afterwards, the CLI reads the same world from the environment: 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 fc4a994e4f..77edb53eae 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 @@ -10,7 +10,7 @@ related: `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 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. +No world in this repository throws it. A stale replay does not need to be refused: its log is a prefix rather than a prefix with a hole in it, replay is deterministic on a prefix, and the write it makes next comes back carrying the events it was pushed past (see [Stale reads](/docs/configuration/runtime-tuning#stale-reads-and-why-nothing-has-to-be-rejected)). The error and the runtime's handling of it remain for a world that would rather refuse than report — one that allocates positions somewhere other than the commit, and so cannot report a gap reliably. Event creations that carry no position are never rejected with it. 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 352a53337f..0d72640318 100644 --- a/docs/content/docs/v5/configuration/runtime-tuning.mdx +++ b/docs/content/docs/v5/configuration/runtime-tuning.mdx @@ -73,22 +73,21 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL ### `WORKFLOW_RESILIENT_STEP_DISPATCH` -- Default: enabled +- Default: disabled - 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 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. +- It is off by default because the publish races the create's verdict, and a create can come back refused: as a duplicate this replay should stop pursuing, or as a [stale write](#stale-reads-and-why-nothing-has-to-be-rejected) on a World that refuses rather than reports. Either way the message carrying the payload is already out, so the consumer can materialize a step whose create was refused, and nothing orders the verdict before the consumer's redelivery re-ensure. The sequential path is the only one that gives the message a happens-after edge over it. +- Even when enabled, the runtime falls back to the sequential create-then-publish dispatch when the step input is too large to inline on the queue message, or when the run's queue transport cannot carry binary payloads (pre-CBOR spec versions). - 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. +- Set `1` to enable it. -### Stale-write rejection +### Stale reads, and why nothing has to be rejected -- 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. -- 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. +- Not a variable: this is how a replay working from an out-of-date event log stays correct, and why no World needs a precondition guard to make it so. +- Three properties do it together. A reader's log is always a **prefix** of the run's log, never a prefix with a hole in it — positions are allocated by the World at commit, so nothing lands behind a position a reader has already passed. Replay is **deterministic on a prefix**: the same prefix always yields the same decisions, so a shorter log does not mean a different run, only a run that has not caught up. And every write **reports what it missed**: a creation names the position it replayed from (`eventCount`), and the World returns the events occupying the positions it was pushed past. The replay merges those and continues, correcting itself on the write rather than on a read. +- So a stale replay costs a merge, not a rejection. None of the shipped Worlds refuses a write for being stale. +- A World *may* refuse instead, with 412 ([`PreconditionFailedError`](/docs/api-reference/workflow-errors/precondition-failed-error)) — appropriate when it allocates positions somewhere other than the commit and cannot report a gap reliably. The runtime handles that: it 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, because a replay working from a corrected log derives different events. +- A World that does refuse should only ever do so on evidence, and accept the write in every other case. A rejection then always means the position really was stale, while the absence of one proves nothing about currency. +- Two runtime behaviors follow from the properties above rather than from any fence. 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) stays active while the run has an open hook: a `hook_received` missed by the delta window is observed one iteration later, and the next write brings it back. And while a hook is open, inline steps take the await-then-run path even when optimistic inline start is enabled — several invocations race for one step's claim there, and awaiting it means the body runs only for the writer that won. ### `WORKFLOW_SLOT_GAP_CHECK` @@ -101,7 +100,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 an event creation is [rejected as stale](#stale-write-rejection) 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-reads-and-why-nothing-has-to-be-rejected) before it falls back to a re-invocation. No shipped World rejects one, so this budget is reserved for a World that chooses to. - 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 757ac23a7a..fc1024283c 100644 --- a/docs/content/worlds/v5/building-a-world.mdx +++ b/docs/content/worlds/v5/building-a-world.mdx @@ -36,8 +36,6 @@ interface WorldCapabilities { hookRetention?: { active: boolean; }; - slotEventIds?: boolean; - preconditionGuard?: boolean; } interface World extends Storage, Queue, Streamer { @@ -49,7 +47,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 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 optional `capabilities` object advertises additional behavior. Set `hookRetention.active` to `true` only when the World implements Hook token retention. Note what is *not* in there: [slot-numbered event IDs](#event-id-allocation) are a requirement of this contract, not a capability to opt into. 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 @@ -110,20 +108,28 @@ Keep the owning Run available for at least as long as its token remains unavaila ### Event ID Allocation -Your World assigns every event ID. An ID is `evnt_` followed by the event's 1-based position in that run's log, zero-padded to 26 characters, so the run's first event is `evnt_00000000000000000000000001`. Use `slotToEventId()` from `@workflow/world` to format one, and declare `capabilities.slotEventIds`. + + **Required.** Your World assigns every event ID, and every ID must be a position in its run's log. The runtime reads a position out of every ID it loads and fails the run when it cannot, so a World whose IDs are anything else — a ULID, a UUID, a database sequence shared across runs — cannot replay a single workflow. There is no capability flag and no fallback path. + + +An ID is `evnt_` followed by the event's 1-based position in that run's log, zero-padded to 26 characters, so the run's first event is `evnt_00000000000000000000000001`. Use `slotToEventId()` from `@workflow/world` to format one. Two properties have to hold, and both are about what a reader can conclude from the log: - **Uniqueness.** Two writers racing to append must not both take a position. Settle it where the store settles it, with a unique constraint on `(runId, eventId)` or a conditional write, rather than reading the maximum and adding one in your own process. - **Density.** Positions run from 1 with no holes, which is what lets a reader tell a complete log from a truncated one by its length alone. A writer that loses a race must re-derive its position from the store and take the next free one. Incrementing a local number after a loss leaves a permanent hole, and the runtime treats a hole as a log it cannot safely replay across. +Allocate the position **at the commit**, in the same operation that appends the event. That is what makes a reader's log a *prefix* of the run's log rather than a prefix with a hole in it: nothing can land behind a position a reader has already passed. Handing a position out earlier — in a request handler, say — and committing later breaks the property every replay depends on, and is the one case where you may need [a stale-write rejection](#optional-rejecting-a-stale-write) to compensate. + `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. ### Optional: Rejecting a Stale Write 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. -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. +No World in this repository does. Reporting is the better mechanism, and the reason it is *sufficient* rather than merely cheaper is worth stating, because it is what removed the need to reject at all: a reader's log is a prefix of the run's log rather than a prefix with a hole in it, replay is deterministic on a prefix, and the writer's next write brings back what it missed. A shorter log therefore means a run that has not caught up, never a run that will decide differently. + +That leaves rejecting worth implementing in one case — your store allocates positions somewhere other than the commit, so it cannot report the skipped span reliably and refusing a stale write is safer than accepting one out of order. Two rules make this safe: @@ -132,7 +138,7 @@ Two rules make this safe: 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. -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. +The runtime handles a rejection wherever it can arrive: it restarts the replay in-process from a corrected log, and re-invokes with a fresh replay once that budget is spent. It never retries the rejected write as-is, because a replay working from a corrected log derives different events. Nothing has to be declared for that to work — there is no capability to set. ## Queue Interface diff --git a/packages/core/src/retained-vm-loop.test.ts b/packages/core/src/retained-vm-loop.test.ts index 39ff8f64e6..b7a88a979f 100644 --- a/packages/core/src/retained-vm-loop.test.ts +++ b/packages/core/src/retained-vm-loop.test.ts @@ -2,6 +2,7 @@ import { PreconditionFailedError } from '@workflow/errors'; import { type Event, SPEC_VERSION_CURRENT, + slotToEventId, type WorkflowRun, } from '@workflow/world'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; @@ -225,7 +226,7 @@ async function drive( return { run, events }; } const event = { - eventId: `e-${++seq}`, + eventId: slotToEventId(++seq), runId, createdAt: new Date(), ...data, diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index 108b07d870..18bf614eea 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -105,7 +105,7 @@ async function runWorkflowHandlerWithEvents( } const event = { - eventId: `event-${createdEvents.length}`, + eventId: slotToEventId(createdEvents.length), runId: workflowRun.runId, createdAt: new Date(), ...data, @@ -393,7 +393,7 @@ describe('workflowEntrypoint replay guards', () => { createdEvents.push(data); return { event: { - eventId: `event-${createdEvents.length}`, + eventId: slotToEventId(createdEvents.length), runId: 'wrun_schema_validation', createdAt: new Date(), ...data, @@ -492,7 +492,7 @@ describe('workflowEntrypoint replay guards', () => { ? { run: workflowRun } : { event: { - eventId: `event-${createdEvents.length}`, + eventId: slotToEventId(createdEvents.length), runId: workflowRun.runId, createdAt: new Date(), ...data, @@ -590,7 +590,7 @@ describe('workflowEntrypoint replay guards', () => { ? { run: workflowRun } : { event: { - eventId: `event-${createdEvents.length}`, + eventId: slotToEventId(createdEvents.length), runId: workflowRun.runId, createdAt: new Date(), ...data, @@ -670,7 +670,7 @@ describe('workflowEntrypoint replay guards', () => { createdEvents.push(data); return { event: { - eventId: `event-${createdEvents.length}`, + eventId: slotToEventId(createdEvents.length), runId: 'wrun_parse', createdAt: new Date(), ...data, @@ -755,7 +755,7 @@ describe('workflowEntrypoint replay guards', () => { }; const events: Event[] = [ { - eventId: 'event-foreign-failed', + eventId: slotToEventId(1), runId: 'wrun_other', eventType: 'run_failed', eventData: { @@ -798,7 +798,7 @@ describe('workflowEntrypoint replay guards', () => { const events: Event[] = [ { - eventId: 'event-0', + eventId: slotToEventId(1), runId: workflowRun.runId, eventType: 'wait_created', correlationId: 'wait_01HK153X00VFKAJV9XFN9JXXRS', @@ -808,7 +808,7 @@ describe('workflowEntrypoint replay guards', () => { createdAt: new Date('2024-01-01T00:00:00.000Z'), }, { - eventId: 'event-1', + eventId: slotToEventId(2), runId: workflowRun.runId, eventType: 'wait_completed', correlationId: 'wait_01HK153X00VFKAJV9XFN9JXXRS', @@ -841,7 +841,7 @@ describe('workflowEntrypoint replay guards', () => { expect(queueCalls.map((c) => c.message)).toContainEqual( expect.objectContaining({ replayDivergence: { - eventId: 'event-0', + eventId: slotToEventId(1), count: 1, }, }) @@ -958,7 +958,7 @@ describe('workflowEntrypoint replay guards', () => { // matches no hook' below). const events: Event[] = [ { - eventId: 'event-0', + eventId: slotToEventId(1), runId: workflowRun.runId, eventType: 'hook_created', correlationId: 'hook_01HK153X00VFKAJV9XFN9JXXRS', @@ -989,7 +989,7 @@ describe('workflowEntrypoint replay guards', () => { ); expect(queueCalls.map((c) => c.message)).toContainEqual( expect.objectContaining({ - replayDivergence: { eventId: 'event-0', count: 1 }, + replayDivergence: { eventId: slotToEventId(1), count: 1 }, }) ); }); @@ -1018,7 +1018,7 @@ describe('workflowEntrypoint replay guards', () => { // the run. const events: Event[] = [ { - eventId: 'event-0', + eventId: slotToEventId(1), runId: workflowRun.runId, eventType: 'hook_received', correlationId: 'hook_01HK153X00VFKAJV9XFN9JXXRS', @@ -1166,7 +1166,7 @@ describe('workflowEntrypoint replay guards', () => { createdEvents.push(data); return { event: { - eventId: `event-${createdEvents.length}`, + eventId: slotToEventId(createdEvents.length), runId: workflowRun.runId, createdAt: new Date(), ...data, @@ -1276,7 +1276,7 @@ describe('workflowEntrypoint replay guards', () => { createdEvents.push(data); return { event: { - eventId: `event-${createdEvents.length}`, + eventId: slotToEventId(createdEvents.length), runId: workflowRun.runId, createdAt: new Date(), ...data, @@ -1441,7 +1441,7 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { const recordEvent = (data: any): Event => { eventSeq += 1; const created = { - eventId: `event-${eventSeq}`, + eventId: slotToEventId(eventSeq), runId: workflowRun.runId, createdAt: new Date(), ...data, @@ -1699,7 +1699,7 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { const rec = (data: any): Event => { seq += 1; const e = { - eventId: `e-${seq}`, + eventId: slotToEventId(seq), runId: workflowRun.runId, createdAt: new Date(), ...data, @@ -1851,12 +1851,13 @@ describe('workflowEntrypoint resilient step consumption (stepInput re-ensure)', deploymentId: 'test-deployment', }; - let eventSeq = 0; + // Slot 1 is taken by the seeded event below, so recorded events start at 2. + let eventSeq = 1; const durableEvents: Event[] = [ // An unrelated pending step: keeps the run un-replayable so the handler // returns right after the background step completes. { - eventId: 'event-other', + eventId: slotToEventId(1), runId: opts.runId, createdAt: new Date(), eventType: 'step_created', @@ -1868,7 +1869,7 @@ describe('workflowEntrypoint resilient step consumption (stepInput re-ensure)', const recordEvent = (data: any): Event => { eventSeq += 1; const created = { - eventId: `event-${eventSeq}`, + eventId: slotToEventId(eventSeq), runId: opts.runId, createdAt: new Date(), ...data, @@ -2203,7 +2204,7 @@ describe('workflowEntrypoint turbo mode', () => { const rec = (data: any): Event => { seq += 1; const e = { - eventId: `e-${seq}`, + eventId: slotToEventId(seq), runId, createdAt: new Date(), ...data, @@ -2512,7 +2513,7 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => { * World capabilities to declare. Absent by default — capability-gated * fast paths must fail closed without them. */ - capabilities?: { preconditionGuard?: boolean; maxConcurrency?: boolean }; + capabilities?: { maxConcurrency?: boolean }; /** Workflow source to run (defaults to hookAndStepWorkflow). */ source?: string; /** @@ -2653,15 +2654,14 @@ 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 World enforces the precondition guard', async () => { + it('requests the inline delta despite the open hook', async () => { const { res, eventsCreate } = await driveDeltaGate( - 'wrun_delta_gate_guard_on', - { capabilities: { preconditionGuard: true } } + 'wrun_delta_gate_guard_on' ); expect(res.status).toBe(204); - // The suspension created a hook (left open) and one lazy inline step — - // with an enforced guard, a hook_received missed by the delta window is - // fenced by the outside-event marker, so the fast path stays active. + // The suspension created a hook (left open) and one lazy inline step. A + // hook_received missed by the delta window is fenced by the outside-event + // marker, so the fast path stays active. expect(stepCompletedParams(eventsCreate)?.sinceCursor).toBe( 'cursor_delta_gate' ); @@ -2672,16 +2672,6 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => { ); }); - 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' - ); - expect(res.status).toBe(204); - // Without the guard there is no fence for a hook_received landing in the - // delta window, so the conservative gate keeps the fetch path. - 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 () => { // Simulates the interleaving the fence exists for: after step A's // terminal write, an out-of-band hook_received bumps the run's marker; @@ -2690,7 +2680,6 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => { const { res, eventsCreate, queueMock } = await driveDeltaGate( 'wrun_delta_gate_stale_claim', { - capabilities: { preconditionGuard: true }, source: hookAndTwoStepWorkflow, rejectClaimOnce: { stepName: 'deltaGateStepB', @@ -2752,7 +2741,6 @@ describe('workflowEntrypoint inline-delta gate with open hooks', () => { const { res, eventsCreate } = await driveDeltaGate( 'wrun_delta_gate_stale_claim_optimistic', { - capabilities: { preconditionGuard: true }, source: hookAndTwoStepWorkflow, rejectClaimOnce: { stepName: 'deltaGateStepB', @@ -2838,7 +2826,7 @@ describe('workflowEntrypoint latency telemetry (ttfs / stso)', () => { const rec = (data: any): Event => { seq += 1; const e = { - eventId: `e-${seq}`, + eventId: slotToEventId(seq), runId, createdAt: new Date(), ...data, @@ -3122,7 +3110,7 @@ describe('workflowEntrypoint latency telemetry (ttfs / stso)', () => { const attrOccurredAt = new Date(runCreatedAtMs + 7_000); const attrEvent = { ...attrCreates[0], - eventId: 'e-attr-1', + eventId: slotToEventId(1), runId, createdAt: attrOccurredAt, occurredAt: attrOccurredAt, diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 0f2db2550d..0c9bc472d2 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -1325,14 +1325,13 @@ export function workflowEntrypoint( appendEventLog(eventLog, delta); eventLog = { ...eventLog, type: 'ready' }; } else { - // MUST be a full, cursor-less reload. The cursor filters - // by lexicographic event id while a hole is defined by - // ULID *time*: an event in the same millisecond sorts - // either side of the cursor depending on its random - // component, and an event minted in an earlier - // millisecond but committed later always sorts below it. - // An incremental load therefore heals the hole only by - // luck. + // MUST be a full, cursor-less reload. The cursor lists + // forward from the highest event id this replay holds, + // and what a stale snapshot is missing sits below it: a + // slot allocated inside a concurrent commit takes a + // position this replay had already read past. An + // incremental load starts above the hole and never + // returns it. eventLog = { type: 'loadAll' }; // The corrected log inserts the missing events BELOW the // length already scanned for payload prewarming, shifting @@ -3684,14 +3683,15 @@ export function workflowEntrypoint( // siblings queued to background handlers, and no other // inline step writing its own events out of band). // - No pending wait timer from THIS suspension, and no - // open wait in the cumulative log: a concurrent - // `wait_completed` landing after the delta snapshot - // 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 World declares it fences stale writes - // (`capabilities.preconditionGuard`). The - // delta snapshots the log at the step_completed + // open wait in the cumulative log. A `wait_completed` + // is a resolution the replay is waiting on rather + // than an event it can observe one iteration late, + // so consuming a delta that predates it would settle + // the sleep from a view that does not contain its + // completion. + // - An open (or this-suspension-created) hook is + // fine, and that is the gate this used to carry. + // 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 // is absent from the delta and observed one @@ -3699,24 +3699,22 @@ 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); 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 + // and before the suspension's writes), and it is + // bounded by the same thing: what a stale replay + // holds is a *prefix* of the log, never a hole in + // it, and replaying a prefix twice reaches the same + // decisions. Its next write is allocated above + // whatever it missed and comes back carrying it + // (`reportSkippedSlots` on the local worlds, + // commit-time allocation on the Vercel one), so the + // replay corrects itself on the write rather than + // on a read. A World MAY instead refuse the write as + // stale (412), which the runtime handles the same + // way; nothing requires it to. 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 a - // fencing World there is nothing to reject a stale - // write, so keep the conservative gate. + // are subject to the same window. // - 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 @@ -3726,13 +3724,6 @@ 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 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 = - world.capabilities?.preconditionGuard === true; - const requestInlineDelta = typeof eventLog.cursor === 'string' && err.stepCount === 1 && @@ -3741,32 +3732,29 @@ export function workflowEntrypoint( lazyInlineSteps.length === 1 && ownedRecoverySteps.length === 0 && !suspensionResult.waitTimeout && - !openHookWaitState.openWait && - (guardEnforced || - (err.hookCount === 0 && - !openHookWaitState.openHook)); + !openHookWaitState.openWait; // Stale-sensitive batch: a hook is open in the run (or // was created by this suspension, so its hook_received // can land any moment) — an out-of-band event can make - // the view this batch was scheduled from stale. With - // the guard in force, the fence rejects a stale - // claim's durable writes — but it cannot un-run a step - // BODY that optimistic start began before the claim - // settled. Suppress optimistic start for these batches - // (take await-then-run) so a 412-fenced step never - // executes user code at all: the fence then covers - // side effects, not just the event log. Costs one - // claim round-trip per step while a hook is open, only - // on guard-enforcing deployments. Without the guard - // nothing 412s, so suppression would buy nothing — - // stale-view exposure there is the pre-existing - // optimistic-start contract (idempotent side effects). + // the view this batch was scheduled from stale, which + // is when several invocations race for one step's + // claim. Optimistic start begins the body before the + // claim settles, so the writer that LOSES the claim + // (`EntityConflictError` → `skipped`) has already run + // user code it does not own. Suppress it for these + // batches, so the body runs only after the claim came + // back won. Costs one claim round-trip per step while a + // hook is open, and buys the same thing on every World: + // a step body executes once per claim, not once per + // racer. (A World that refuses the claim as stale (412) + // gets the same protection through the same await; + // none of the shipped ones does that for a + // slot-numbered run.) const suppressOptimisticStart = - guardEnforced && - (openHookWaitState.openHook || - err.hookCount > 0 || - suspensionResult.hasHookEvents); + openHookWaitState.openHook || + err.hookCount > 0 || + suspensionResult.hasHookEvents; // Turbo mode forces optimistic inline start for this // batch — but only while the run is still "clean" (a pure diff --git a/packages/core/src/runtime/constants.ts b/packages/core/src/runtime/constants.ts index a6b71eb839..f04647d10b 100644 --- a/packages/core/src/runtime/constants.ts +++ b/packages/core/src/runtime/constants.ts @@ -240,11 +240,18 @@ export const MAX_RESILIENT_STEP_INPUT_BYTES = 128 * 1024; * event if the direct write failed transiently. Mirrors the resilient start * (`runInput`) and resilient hook resume (`hookInput`) patterns. * - * **On by default.** Disable via `WORKFLOW_RESILIENT_STEP_DISPATCH=0` to - * restore the sequential create-then-queue dispatch. + * **Off by default.** Enable via `WORKFLOW_RESILIENT_STEP_DISPATCH=1`. + * + * The queue publish races the create's verdict, and a create can come back + * refused — as a duplicate this replay should stop pursuing, or as a stale + * write on a World that refuses rather than reports. Either way the message + * carrying the payload is already out, so the consumer can materialize a step + * whose create was refused, and nothing orders the verdict before the + * consumer's redelivery re-ensure. Enabling this trades that window for the + * latency the parallel publish saves. */ export function isResilientStepDispatchEnabled(): boolean { - return process.env.WORKFLOW_RESILIENT_STEP_DISPATCH !== '0'; + return process.env.WORKFLOW_RESILIENT_STEP_DISPATCH === '1'; } const warnedMaxEventsValues = new Set(); diff --git a/packages/core/src/runtime/helpers.test.ts b/packages/core/src/runtime/helpers.test.ts index 15b84ba5e0..21852fac4b 100644 --- a/packages/core/src/runtime/helpers.test.ts +++ b/packages/core/src/runtime/helpers.test.ts @@ -1,7 +1,6 @@ import { PreconditionFailedError, WorkflowWorldError } from '@workflow/errors'; import type { Event, World } from '@workflow/world'; import { slotToEventId } from '@workflow/world'; -import { ulid } from 'ulid'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { bytesToBase64, deriveRunKeyPair, seal } from '../sealed-box.js'; import { @@ -589,14 +588,8 @@ describe('loadWorkflowRunEvents', () => { }); }); -const makeUlidEvent = (time: number): Event => - ({ - eventId: `evnt_${ulid(time)}`, - runId: 'wrun_mockidnumber0001', - eventType: 'step_created', - correlationId: 'step_mock', - createdAt: new Date(time), - }) as unknown as Event; +/** An id from the scheme slots replaced: a ULID, which carries no position. */ +const UNPOSITIONED_EVENT_ID = 'evnt_01HF7YATRRC3M0F1K9Q2J8XW5B'; describe('slotSnapshotParams', () => { it('sends the highest slot the loaded log occupies', () => { @@ -627,35 +620,35 @@ describe('slotSnapshotParams', () => { expect(slotSnapshotParams([])).toEqual({}); }); - it('sends nothing for a run whose events are not slot-numbered', () => { - expect(slotSnapshotParams([makeUlidEvent(1_700_000_000_000)])).toEqual({}); - }); - - 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, and a count derived from part of the log would understate - // the writer's position in a way the World cannot detect. + it('throws when any event of the log carries no slot', () => { + // Skipping the id instead would understate the writer's position, and the + // World cannot tell an understated position from an honest one: it would + // hand back the same events on every create for the rest of the run. const events = [ makeEvent(slotToEventId(1)), - makeUlidEvent(1_700_000_000_000), + makeEvent(UNPOSITIONED_EVENT_ID), ]; - expect(slotSnapshotParams(events)).toEqual({}); + expect(() => slotSnapshotParams(events)).toThrow(UNPOSITIONED_EVENT_ID); }); }); describe('maxEventSlot', () => { - it('is undefined for a log with no slot ids', () => { + it('is undefined for an empty log', () => { expect(maxEventSlot([])).toBeUndefined(); - expect(maxEventSlot([makeUlidEvent(1_700_000_000_000)])).toBeUndefined(); + }); + + it('throws rather than ignoring an id that carries no slot', () => { + expect(() => maxEventSlot([makeEvent(UNPOSITIONED_EVENT_ID)])).toThrow( + UNPOSITIONED_EVENT_ID + ); }); }); /** * The hole check a replay runs over its loaded log. It gates whether the run * executes at all, so it is one-sided in the opposite direction from the - * World's density counter: it reports a hole only where the log proves one, and - * says nothing about a log it cannot read as slots. + * World's density counter: it reports a hole only where the log proves one. */ describe('findEventSlotGap', () => { const slotLog = (...slots: number[]) => @@ -703,20 +696,17 @@ describe('findEventSlotGap', () => { }); }); - it('says nothing about a log it cannot read as slots', () => { + it('says nothing about an empty log', () => { expect(findEventSlotGap([])).toBeUndefined(); - expect( - findEventSlotGap([makeUlidEvent(1_700_000_000_000)]) - ).toBeUndefined(); - // A ULID anywhere disarms it: the run is not slot-numbered, and a mixed - // log has no density to measure. - expect( - findEventSlotGap([ - ...slotLog(1, 2), - makeUlidEvent(1_700_000_000_000), - ...slotLog(9), - ]) - ).toBeUndefined(); + }); + + it('throws on a log whose ids carry no position', () => { + // The check is entirely positional. An id it cannot read is a log it + // cannot judge, and passing the run as dense would be a verdict it never + // reached. + expect(() => + findEventSlotGap([...slotLog(1, 2), makeEvent(UNPOSITIONED_EVENT_ID)]) + ).toThrow(UNPOSITIONED_EVENT_ID); }); }); @@ -811,80 +801,41 @@ describe('mergeReportedEvents', () => { expect(mergeReportedEvents(target, [makeEvent(slotToEventId(2))])).toBe(0); expect(target).toHaveLength(2); }); - - it('leaves a ULID log in receipt order', () => { - // Only a slot log has an id order the runtime may impose. A World that - // orders by (createdAt, eventId) would be reordered into a log it never - // produced. - const first = makeUlidEvent(1_700_000_000_000); - const second = makeUlidEvent(1_600_000_000_000); - const target = [first]; - - mergeReportedEvents(target, [second]); - - expect(target.map((e) => e.eventId)).toEqual([ - first.eventId, - second.eventId, - ]); - }); }); describe('appendUniqueEvents', () => { it('appends in receipt order', () => { - const first = makeUlidEvent(1_700_000_000_000); - const second = makeUlidEvent(1_700_000_001_000); - const third = makeUlidEvent(1_700_000_002_000); - const target = [first]; + const target = [makeEvent(slotToEventId(1))]; - appendUniqueEvents(target, [second, third]); - - expect(target.map((e) => e.eventId)).toEqual([ - first.eventId, - second.eventId, - third.eventId, + appendUniqueEvents(target, [ + makeEvent(slotToEventId(2)), + makeEvent(slotToEventId(3)), ]); + + expect(target.map((e) => e.eventId)).toEqual([1, 2, 3].map(slotToEventId)); }); it('preserves the order the World returned, never re-sorting by event id', () => { - // A World's canonical order is its own: world-local orders by - // `(createdAt, eventId)` and re-mints keys so the two diverge, so an - // id-ordered re-sort here would produce an order no load would return. - const older = makeUlidEvent(1_700_000_000_000); - const newer = makeUlidEvent(1_700_000_002_000); - const middle = makeUlidEvent(1_700_000_001_000); - const target = [older, newer]; - - appendUniqueEvents(target, [middle]); - - expect(target.map((e) => e.eventId)).toEqual([ - older.eventId, - newer.eventId, - middle.eventId, - ]); + // Unlike mergeReportedEvents, this appends a page the World handed back as + // a unit. Every source is already in canonical order relative to the tail, + // so a sort could only ever be a wasted pass over the log — the reason + // helpers.ts gives. Asserting the unsorted order is how a sort creeping in + // gets noticed. + const target = [makeEvent(slotToEventId(1)), makeEvent(slotToEventId(3))]; + + appendUniqueEvents(target, [makeEvent(slotToEventId(2))]); + + expect(target.map((e) => e.eventId)).toEqual([1, 3, 2].map(slotToEventId)); }); it('deduplicates by event id', () => { - const first = makeUlidEvent(1_700_000_000_000); - const second = makeUlidEvent(1_700_000_001_000); + const first = makeEvent(slotToEventId(1)); + const second = makeEvent(slotToEventId(2)); const target = [first]; appendUniqueEvents(target, [first, second, second]); - expect(target.map((e) => e.eventId)).toEqual([ - first.eventId, - second.eventId, - ]); - }); - - it('keeps a same-millisecond pair in receipt order', () => { - const time = 1_700_000_000_000; - const a = makeEvent(`evnt_${ulid(time).slice(0, 10)}AAAAAAAAAAAAAAAA`); - const b = makeEvent(`evnt_${ulid(time).slice(0, 10)}ZZZZZZZZZZZZZZZZ`); - const target = [b]; - - appendUniqueEvents(target, [a]); - - expect(target.map((e) => e.eventId)).toEqual([b.eventId, a.eventId]); + expect(target.map((e) => e.eventId)).toEqual([1, 2].map(slotToEventId)); }); it('leaves the snapshot correct even when the merge is not id-ordered', () => { @@ -901,7 +852,7 @@ describe('appendUniqueEvents', () => { }); describe('preconditionEventDelta', () => { - // The run every `makeUlidEvent` belongs to. + // The run every fixture event below belongs to. const RUN_ID = 'wrun_mockidnumber0001'; const delta = (details: unknown) => preconditionEventDelta( @@ -910,7 +861,7 @@ describe('preconditionEventDelta', () => { ); it('returns the decoded events and cursor a World attached to the 412', () => { - const event = makeUlidEvent(1_700_000_000_000); + const event = makeEvent(slotToEventId(1)); expect(delta({ events: [event], cursor: 'eid:next' })).toEqual({ events: [event], @@ -919,7 +870,7 @@ describe('preconditionEventDelta', () => { }); it('returns a null cursor when the World sent events without one', () => { - const event = makeUlidEvent(1_700_000_000_000); + const event = makeEvent(slotToEventId(1)); expect(delta({ events: [event] })).toEqual({ events: [event], @@ -941,9 +892,9 @@ describe('preconditionEventDelta', () => { // The delta is merged straight into the replay's log, so a foreign event // there produces a corrupt log rather than a corrected one: the replay // consumes a correlation id for an event this run does not have. - const mine = makeUlidEvent(1_700_000_000_000); + const mine = makeEvent(slotToEventId(1)); const theirs = { - ...makeUlidEvent(1_700_000_001_000), + ...makeEvent(slotToEventId(2)), runId: 'wrun_someotherrun001', } as Event; diff --git a/packages/core/src/runtime/helpers.ts b/packages/core/src/runtime/helpers.ts index be1d867655..03c08f88c3 100644 --- a/packages/core/src/runtime/helpers.ts +++ b/packages/core/src/runtime/helpers.ts @@ -14,11 +14,11 @@ import type { World, } from '@workflow/world'; import { - eventIdToSlot, FIRST_EVENT_SLOT, getQueueTopicPrefix, HealthCheckPayloadSchema, HOOK_RESUME_INPUT_VERSION, + requireEventSlot, resolveQueueNamespace, SPEC_VERSION_CURRENT, SPEC_VERSION_LEGACY, @@ -489,15 +489,11 @@ function recordRequestedEventCursor( * same array. The set is updated alongside `target`. * * 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. 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 maxEventSlot}. + * re-sorted. 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, and re-sorting here would only cost a pass over the log. + * Nothing downstream may assume the tail is the newest event — see + * {@link maxEventSlot}. */ export function appendUniqueEvents( target: Event[], @@ -528,13 +524,10 @@ export function appendUniqueEvents( * land in `eventId` order — a plain `push` would place a late-committing * 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. + * Lexicographic string order is the log's order: a slot id is a fixed-width + * zero-padded position, so comparing the strings compares the positions. This + * needs no parse of its own for that reason, and the comparison is exact rather + * than a reconstruction. */ export function insertEventByEventId(target: Event[], event: Event): void { // Linear scan from the end: the spliced event is almost always the newest @@ -774,10 +767,8 @@ export function isSlotGapCheckEnabled(): boolean { * * Unlike {@link appendUniqueEvents}, this re-sorts. The reported events occupy * slots *below* the write that reported them, so appending them would put them - * after events they precede — and on a slot-numbered run the id order is the - * World's canonical order, so restoring it is well defined rather than a guess. - * A run that is not slot-numbered cannot produce this report in the first - * place; the sort is skipped rather than applied to ids it cannot order. + * after events they precede. Sorting by id restores the World's canonical order + * rather than guessing at it: a slot id is that order, written down. */ export function mergeReportedEvents( target: Event[], @@ -786,7 +777,7 @@ export function mergeReportedEvents( const before = target.length; appendUniqueEvents(target, events); const added = target.length - before; - if (added > 0 && maxEventSlot(target) !== undefined) { + if (added > 0) { target.sort((a, b) => a.eventId < b.eventId ? -1 : a.eventId > b.eventId ? 1 : 0 ); @@ -845,9 +836,7 @@ export function absorbSkippedSlotReport( } /** - * 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 - * settles it for the whole log. + * The highest slot the loaded log occupies, or `undefined` for an empty log. * * The maximum, not the count, and the two are not interchangeable even though * a healthy log makes them equal. A World hands a position to the insert that @@ -860,14 +849,16 @@ export function absorbSkippedSlotReport( * * A hole below the maximum is therefore a property of the read, not of the log, * which is what lets {@link settleEventSlotGap} re-read instead of giving up. + * + * @throws if any event id carries no slot. Every World the runtime replays + * against numbers events by slot, so an id that does not is a broken log rather + * than an older one, and a maximum derived by skipping it would understate the + * log to every write that reads it. */ -export function maxEventSlot(events: Event[]): number | undefined { +export function maxEventSlot(events: readonly Event[]): number | undefined { let max: number | undefined; for (const event of events) { - const slot = eventIdToSlot(event.eventId); - if (slot === null) { - return undefined; - } + const slot = requireEventSlot(event.eventId); if (max === undefined || slot > max) { max = slot; } @@ -888,8 +879,8 @@ export interface EventSlotGap { /** * The hole in a loaded log, or `undefined` when there is none to find. * - * On a slot-numbered run the World allocates every position, so a log that - * holds `n` events below slot `n` is missing one. That matters before a replay + * The World allocates every position, so a log that holds `n` events below slot + * `n` is missing one. That matters before a replay * and nowhere else: the replay reads the log as the complete record of what has * happened, and an absent position is indistinguishable from an event that * never occurred. The branch it would have decided gets decided the other way, @@ -906,8 +897,10 @@ export interface EventSlotGap { * legitimately begins at the second slot and fills in on its own. Every replay * that races a run's own start would otherwise report a hole. * - * Returns `undefined` for a log this cannot read as slots at all: an empty one, - * or a run numbered by ULID, where positions carry no density to check. + * Returns `undefined` for an empty log, which has no density to check. + * + * @throws if any event id carries no slot, for the reason {@link maxEventSlot} + * gives. */ export function findEventSlotGap( events: readonly Event[] @@ -915,10 +908,7 @@ export function findEventSlotGap( const occupied = new Set(); let maxSlot = 0; for (const event of events) { - const slot = eventIdToSlot(event.eventId); - if (slot === null) { - return undefined; - } + const slot = requireEventSlot(event.eventId); occupied.add(slot); if (slot > maxSlot) { maxSlot = slot; @@ -1014,7 +1004,7 @@ export async function settleEventSlotGap( * why {@link slotSnapshotParams} takes the maximum rather than the count. * * 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. + * and so the empty case spreads to nothing. */ export interface SlotSnapshotParams { eventCount?: number; @@ -1023,15 +1013,17 @@ export interface SlotSnapshotParams { /** * Build the slot snapshot to attach to a replay-context event creation. * - * 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. + * Empty for an empty log, which is the state a `run_created` write is issued + * from: there is no position held yet to name. * * 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 slotSnapshotParams(events: Event[]): SlotSnapshotParams { +export function slotSnapshotParams( + events: readonly Event[] +): SlotSnapshotParams { const eventCount = maxEventSlot(events); return eventCount === undefined ? {} : { eventCount }; } diff --git a/packages/core/src/runtime/resume-hook.consumer-preload.test.ts b/packages/core/src/runtime/resume-hook.consumer-preload.test.ts index 52ee1ca227..6cf8d69322 100644 --- a/packages/core/src/runtime/resume-hook.consumer-preload.test.ts +++ b/packages/core/src/runtime/resume-hook.consumer-preload.test.ts @@ -26,6 +26,7 @@ import { type CreateEventRequest, type Event, SPEC_VERSION_CURRENT, + slotToEventId, type WorkflowRun, type World, } from '@workflow/world'; @@ -182,7 +183,6 @@ async function runResumeConsumerScenario(options: { updatedAt: startedAt, }; - const hostUlid = monotonicFactory(); let eventIndex = 0; const event = (data: CreateEventRequest): Event => { const t = +startedAt + ++eventIndex * 100; @@ -190,7 +190,7 @@ async function runResumeConsumerScenario(options: { ...data, specVersion: data.specVersion ?? SPEC_VERSION_CURRENT, runId, - eventId: `evnt_${hostUlid(t)}`, + eventId: slotToEventId(eventIndex), createdAt: new Date(t), } as Event; }; diff --git a/packages/core/src/runtime/resume-latency.runtime.test.ts b/packages/core/src/runtime/resume-latency.runtime.test.ts index 278d131008..f6c0d14a92 100644 --- a/packages/core/src/runtime/resume-latency.runtime.test.ts +++ b/packages/core/src/runtime/resume-latency.runtime.test.ts @@ -30,6 +30,7 @@ import { type Event, type HookResumeTiming, SPEC_VERSION_CURRENT, + slotToEventId, type WorkflowInvokePayload, type WorkflowRun, type World, @@ -342,7 +343,6 @@ async function runScenario(options: ScenarioOptions = {}) { updatedAt: startedAt, }; - const hostUlid = monotonicFactory(); let eventIndex = 0; const event = (data: CreateEventRequest): Event => { const t = +startedAt + ++eventIndex * 100; @@ -350,7 +350,7 @@ async function runScenario(options: ScenarioOptions = {}) { ...data, specVersion: data.specVersion ?? SPEC_VERSION_CURRENT, runId, - eventId: `evnt_${hostUlid(t)}`, + eventId: slotToEventId(eventIndex), createdAt: new Date(t), } as Event; }; @@ -1060,7 +1060,6 @@ async function runScenarioWithoutPreload() { updatedAt: startedAt, }; - const hostUlid = monotonicFactory(); let eventIndex = 0; const event = (data: CreateEventRequest): Event => { const t = +startedAt + ++eventIndex * 100; @@ -1068,7 +1067,7 @@ async function runScenarioWithoutPreload() { ...data, specVersion: data.specVersion ?? SPEC_VERSION_CURRENT, runId, - eventId: `evnt_${hostUlid(t)}`, + eventId: slotToEventId(eventIndex), createdAt: new Date(t), } as Event; }; diff --git a/packages/core/src/runtime/step-executor.ts b/packages/core/src/runtime/step-executor.ts index 390e315889..63295800ce 100644 --- a/packages/core/src/runtime/step-executor.ts +++ b/packages/core/src/runtime/step-executor.ts @@ -23,7 +23,7 @@ import type { World, } from '@workflow/world'; import { - eventIdToSlot, + requireEventSlot, SPEC_VERSION_CURRENT, SPEC_VERSION_SUPPORTS_COMPRESSION, } from '@workflow/world'; @@ -162,8 +162,8 @@ export interface StepExecutorParams { * 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 for a run that is not - * slot-numbered, or a caller with nothing loaded. + * abandon the batch and restart its replay. Undefined for a caller with + * nothing loaded. */ slotSnapshot?: SlotSnapshotParams; /** @@ -313,11 +313,22 @@ export async function executeStep( let knownSlot = params.slotSnapshot?.eventCount; const observeSlot = (result: { event?: Event; events?: Event[] }): void => { if (knownSlot === undefined) { + // The caller scheduled this step without naming a position, so there is + // no snapshot to advance and the writes below send none. Not the same as + // a run without positions: every run has them, this executor just was not + // told which one it started from. return; } - const committed = result.event ? eventIdToSlot(result.event.eventId) : null; - for (const slot of [committed, maxEventSlot(result.events ?? []) ?? null]) { - if (slot !== null && slot > knownSlot) { + const observed: number[] = []; + if (result.event) { + observed.push(requireEventSlot(result.event.eventId)); + } + const reported = maxEventSlot(result.events ?? []); + if (reported !== undefined) { + observed.push(reported); + } + for (const slot of observed) { + if (slot > knownSlot) { knownSlot = slot; } } diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index 1d6bd7327e..d34814500b 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -12,7 +12,7 @@ import { type WorkflowRun, type World, } from '@workflow/world'; -import { describe, expect, it, vi } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { WorkflowSuspension } from '../global.js'; import { maxEventSlot, stepDispatchIdempotencyKey } from './helpers.js'; import { ReplayRecoveryReporter } from './replay-recovery-reporter.js'; @@ -631,6 +631,21 @@ describe('handleSuspension', () => { describe('resilient step dispatch', () => { const queueName = '__wkf_workflow_test-workflow' as ValidQueueName; + // Opt-in feature, so every test that expects a publish has to ask for it. + // The default-off case is covered by its own test below, which unsets this. + let previousFlag: string | undefined; + beforeEach(() => { + previousFlag = process.env.WORKFLOW_RESILIENT_STEP_DISPATCH; + process.env.WORKFLOW_RESILIENT_STEP_DISPATCH = '1'; + }); + afterEach(() => { + if (previousFlag === undefined) { + delete process.env.WORKFLOW_RESILIENT_STEP_DISPATCH; + } else { + process.env.WORKFLOW_RESILIENT_STEP_DISPATCH = previousFlag; + } + }); + /** A run whose queue transport supports binary payloads (CBOR). */ const cborRun: WorkflowRun = { ...run, specVersion: SPEC_VERSION_CURRENT }; @@ -774,51 +789,13 @@ describe('resilient step dispatch', () => { ).rejects.toThrow('bad request'); }); - it('falls back to create-only when the world enforces the precondition guard', async () => { - const { world, eventsCreate, queue } = createQueueWorld({ - capabilities: { preconditionGuard: true }, - }); - - const result = await handleSuspension({ - suspension: new WorkflowSuspension(fourStepsPending(), globalThis), - world, - run: cborRun, - stepDispatch: stepDispatch(), - }); - - // The guarded create can be 412-rejected; a payload-carrying message - // would let the consumer materialize the rejected step. Sequential path: - // create here, caller dispatches. - expect(queue).not.toHaveBeenCalled(); - expect(eventsCreate).toHaveBeenCalledWith( - run.runId, - expect.objectContaining({ - eventType: 'step_created', - correlationId: 's4', - }), - expect.anything() - ); - expect(result.queuedStepCorrelationIds.size).toBe(0); - expect(result.createdStepCorrelationIds).toContain('s4'); - }); - - it('stays sequential under an enforced guard regardless of other capabilities', async () => { - // The guard gate is deliberately not liftable by backend-side revocation - // bookkeeping: nothing orders a slow guarded create's eventual 412 before - // the consumer's redelivery re-ensure, so no capability may re-enable the - // payload-carrying publish while creates are guarded. - const { world, queue } = createQueueWorld({ - capabilities: { - preconditionGuard: true, - // Unknown/extra capability flags must not lift the gate. - ...({ resilientStepDispatch: true } as Record), - } as World['capabilities'], - }); + it('falls back to create-only when the run predates the CBOR queue transport', async () => { + const { world, queue } = createQueueWorld(); const result = await handleSuspension({ suspension: new WorkflowSuspension(fourStepsPending(), globalThis), world, - run: cborRun, + run: { ...run, specVersion: 2 }, stepDispatch: stepDispatch(), }); @@ -826,13 +803,14 @@ describe('resilient step dispatch', () => { expect(result.queuedStepCorrelationIds.size).toBe(0); }); - it('falls back to create-only when the run predates the CBOR queue transport', async () => { + it('falls back to create-only when WORKFLOW_RESILIENT_STEP_DISPATCH is unset', async () => { + delete process.env.WORKFLOW_RESILIENT_STEP_DISPATCH; const { world, queue } = createQueueWorld(); const result = await handleSuspension({ suspension: new WorkflowSuspension(fourStepsPending(), globalThis), world, - run: { ...run, specVersion: 2 }, + run: cborRun, stepDispatch: stepDispatch(), }); @@ -840,30 +818,6 @@ describe('resilient step dispatch', () => { expect(result.queuedStepCorrelationIds.size).toBe(0); }); - it('falls back to create-only when WORKFLOW_RESILIENT_STEP_DISPATCH=0', async () => { - const prev = process.env.WORKFLOW_RESILIENT_STEP_DISPATCH; - process.env.WORKFLOW_RESILIENT_STEP_DISPATCH = '0'; - try { - const { world, queue } = createQueueWorld(); - - const result = await handleSuspension({ - suspension: new WorkflowSuspension(fourStepsPending(), globalThis), - world, - run: cborRun, - stepDispatch: stepDispatch(), - }); - - expect(queue).not.toHaveBeenCalled(); - expect(result.queuedStepCorrelationIds.size).toBe(0); - } finally { - if (prev === undefined) { - delete process.env.WORKFLOW_RESILIENT_STEP_DISPATCH; - } else { - process.env.WORKFLOW_RESILIENT_STEP_DISPATCH = prev; - } - } - }); - it('never queues from here when no stepDispatch is provided (terminal drain)', async () => { const { world, queue } = createQueueWorld(); diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index 713b0a1157..0c303ad760 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -58,12 +58,13 @@ export interface SuspensionHandlerParams { requestId?: string; /** * The runtime's loaded event log. Every event creation this suspension makes - * is sent with the precondition snapshot derived from it, so a backend that - * has recorded an event the replay did not see rejects the write with a 412 - * instead of accepting a divergent event. The rejection is not retried here: - * the event's correlation id was minted by *this* replay's seeded sequence, - * so re-committing it against a corrected log would persist an event no - * correct replay produces. The caller restarts the replay instead. + * names the position it was derived from, so a backend that has recorded + * events the replay did not see can report them back on the write — or, if + * it would rather refuse than report, reject it with a 412. A rejection is + * not retried here: the event's correlation id was minted by *this* replay's + * seeded sequence, so re-committing it against a corrected log would persist + * an event no correct replay produces. The caller restarts the replay + * instead. */ eventLog?: LoadedEventLog; /** @@ -677,26 +678,23 @@ 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 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 - // rejected. This gate is deliberately NOT liftable by backend-side - // revocation bookkeeping: nothing orders a slow create's eventual 412 - // (which is when the backend learns the dispatch is poisoned) before the - // consumer's redelivery re-ensure, and a best-effort marker that fails - // open cannot carry a correctness property. The sequential path is the - // only thing that gives the message a happens-after edge over its - // create's guard verdict. + // - The feature is enabled (`WORKFLOW_RESILIENT_STEP_DISPATCH` opt-in). + // It is off by default because the publish races the create's verdict, + // and a create can come back refused: as a duplicate the replay should + // stop pursuing, or — on a World that would rather refuse a stale write + // than report what it missed — as a 412. Either way the queue message + // carrying the payload is already out, and the consumer can materialize a + // step whose create was refused. Nothing orders that verdict before the + // consumer's redelivery re-ensure, so no backend-side revocation + // bookkeeping can close the window: a best-effort marker that fails open + // cannot carry a correctness property. The sequential path is the only + // thing that gives the message a happens-after edge over the verdict. // - The run's queue transport preserves binary payloads (CBOR, // specVersion >= 3): `stepInput.input` is the serialized (possibly // encrypted) input bytes, which the JSON transport would mangle. const resilientDispatchEligible = stepDispatch !== undefined && isResilientStepDispatchEnabled() && - 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/core/src/runtime/wait-completion-replay.test.ts b/packages/core/src/runtime/wait-completion-replay.test.ts index 8da3149a36..2666f95a37 100644 --- a/packages/core/src/runtime/wait-completion-replay.test.ts +++ b/packages/core/src/runtime/wait-completion-replay.test.ts @@ -81,12 +81,6 @@ async function runStaleWaitReplayScenario(options: { returnInlineDelta?: boolean; /** Truncate that inline delta (hasMore: true), which must not be absorbed. */ inlineDeltaHasMore?: boolean; - /** - * Number the fake log with slot event ids. That is what makes the handler - * send the slot precondition, so it is the only mode in which a write can - * carry both halves of the World's answer channel. - */ - slotEventIds?: boolean; }) { vi.spyOn(Date, 'now').mockReturnValue(+fixedNow); @@ -131,9 +125,7 @@ async function runStaleWaitReplayScenario(options: { ...data, specVersion: data.specVersion ?? SPEC_VERSION_CURRENT, runId, - eventId: options.slotEventIds - ? slotToEventId(eventIndex) - : `evt_${eventIndex.toString().padStart(3, '0')}`, + eventId: slotToEventId(eventIndex), createdAt, }) as Event; @@ -691,9 +683,8 @@ describe('workflow handler wait completion replay', () => { expectHookBranchQueued(result); }); - it('asks a slot-numbered World for the delta and the skipped slots at once', async () => { - // On a slot-numbered run the write carries both halves of the World's - // answer channel: `sinceCursor` asks for the delta since the handler's + it('asks for the delta and the skipped slots at once', async () => { + // The write carries both halves of the World's answer channel: `sinceCursor` asks for the delta since the handler's // snapshot, and `eventCount` states the slot that snapshot reached so a // bumped write can report what it was decided without. They share // `events`/`cursor`/`hasMore` on the response, so a World that answers @@ -702,7 +693,6 @@ describe('workflow handler wait completion replay', () => { const result = await runStaleWaitReplayScenario({ includePreloadedCursor: true, returnInlineDelta: true, - slotEventIds: true, }); const waitWrite = result.createEvent.mock.calls.find( diff --git a/packages/world-local/src/index.ts b/packages/world-local/src/index.ts index b89fd67af5..dfa56584df 100644 --- a/packages/world-local/src/index.ts +++ b/packages/world-local/src/index.ts @@ -80,10 +80,6 @@ export function createWorld(args?: Partial): LocalWorld { // events-storage.ts `claimHookResume`), so resumeHook()'s parallel fast // path converges on one event in dev exactly as it does on Vercel. hookResumeDedup: true, - // New runs get dense per-run slot event ids. Runs created before this - // keep their ULIDs; the scheme is pinned by a run's own first event id, - // not by this flag, which only says what new runs get. - slotEventIds: true, }, ...queue, ...storage, diff --git a/packages/world-postgres/src/index.ts b/packages/world-postgres/src/index.ts index dae35f39cc..84618fdbc2 100644 --- a/packages/world-postgres/src/index.ts +++ b/packages/world-postgres/src/index.ts @@ -66,10 +66,6 @@ export function createWorld( specVersion: SPEC_VERSION_CURRENT, capabilities: { hookRetention: { active: true }, - // New runs get dense per-run slot event ids. Runs created before this - // keep their ULIDs; the scheme is pinned by whether the run owns a slot - // counter, not by this flag, which only says what new runs get. - slotEventIds: true, }, ...storage, ...streamer, diff --git a/packages/world-sim/DESIGN.md b/packages/world-sim/DESIGN.md index 0da704e759..893fb03fd7 100644 --- a/packages/world-sim/DESIGN.md +++ b/packages/world-sim/DESIGN.md @@ -297,15 +297,15 @@ how a scenario can be run one flag apart from its neighbour. `countGuard` **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 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. +**`preconditionGuard`** rejects a replay-context write whose snapshot predates +the newest externally-originated event. It is a store option here and not a +World capability: the runtime assumes any World may refuse a stale write, so a +scenario can change what the store does about one but never what the runtime +expects. It does not describe what world-vercel does to a run either — 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 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** @@ -452,7 +452,7 @@ interface ScenarioSpec { verifyReplay?: boolean; // default on for runs reaching completed/failed expect?: { status?: ScenarioOutcome; output?: unknown }; // output: deep equality limits?: ScenarioLimits; - preconditionGuard?: boolean; // advertise + enforce the optimistic-concurrency fence + preconditionGuard?: boolean; // enforce the optimistic-concurrency fence countGuard?: boolean; // also enforce its count half appendOnlyLog?: boolean; // position at commit, not at mint; see §5 } @@ -655,7 +655,7 @@ Measured on branch `sim-world`. **Scenarios** — `pnpm sim` in `workbench/sim-world`: ``` -41 scenario(s): 35 passed, 6 failed, 6 consistency violation(s) +41 scenario(s): 38 passed, 3 failed, 3 consistency violation(s) ``` And the same book against an append-only log (`pnpm sim --append-only`): @@ -664,8 +664,8 @@ And the same book against an append-only log (`pnpm sim --append-only`): 41 scenario(s): 41 passed, 0 failed, 0 consistency violation(s) ``` -Both numbers are the intended steady state; see "The six" below for which of -the six violations that second line closes on the merits and which close +Both numbers are the intended steady state; see "The three" below for which of +the three violations that second line closes on the merits and which close because the correct answer itself changes. There was a seventh red until recently, `unclaimed-payload-under-fork`, and it @@ -677,20 +677,20 @@ mistake, and agreed. Only the log disagreeing with itself caught it. #3406 fixed the delivery-barrier ordering and it is now green in both worlds; the scenario stays as that fix's regression test. -With the fence forced off (`pnpm sim --no-fence`), violations go to **8** +With the fence forced off (`pnpm sim --no-fence`), violations go to **5** mint-ordered and stay at **0** append-only — see §5. -Replay verification across the book: **33 `ok`, 6 `MISMATCH`, 2 `skipped`** +Replay verification across the book: **36 `ok`, 3 `MISMATCH`, 2 `skipped`** (skipped where the run did not reach a terminal status). -`run.ts` exits non-zero, and that is the intended steady state. The six +`run.ts` exits non-zero, and that is the intended steady state. The three failures are reproductions of corruptions the runtime can still produce; each states the outcome its own durable log implies and fails until the runtime gets there, so the failure line names both sides (`expected "afterSlow:doc-26", got "afterFast:doc-26"`). -**The number is the thing to watch: six today.** A seventh is a regression; -five means something got fixed and a scenario is ready to retire. +**The number is the thing to watch: three today.** A fourth is a regression; +two means something got fixed and a scenario is ready to retire. That makes the book a poor plain CI gate, which is what `--report-only` is for: it prints every failure and exits 0, so a job can *publish* the book's current @@ -698,54 +698,48 @@ state rather than block on it. `--summary-file` writes one collapsed `
` — a visible line carrying the count and a green or orange dot, the whole table behind it — for a PR comment or `$GITHUB_STEP_SUMMARY`, and `--detail-file` writes the full colour-free trace as an artifact to read when a -number moves. Deliberately nothing above the fold but the count: six are red on -purpose, so a comment that leads with the failures leads with the part that is +number moves. Deliberately nothing above the fold but the count: three are red +on purpose, so a comment that leads with the failures leads with the part that is not news, and grows a wall of text on exactly the PRs that changed nothing. The workbench's `pnpm test` is `--report-only --summary-file`, so a recursive `pnpm -r test` stays green and still says what happened; `pnpm sim` stays strict, so running it by hand fails loudly. -### The six +### The three -The `fix` column names the specific change that closes the scenario. `shown -green by` is the stronger claim: a *passing* scenario that is this one with that -fix armed, same workflow and same tempo, one flag apart. Where it says "none -yet", the fix is identified by argument but nothing in the book proves it. - -| scenario | mechanism | fix | shown green by | -|---|---|---|---| -| `stale-read-step-count-fork` (doc-23) | `withholdNextEvent(1)` + `deliverHook`; hook at `#7`, `wait_completed` at `#8`, no-hook branch at `#9` | `preconditionGuard` — the withheld `hook_received` is the newest out-of-band write and the orchestrator's snapshot predates the sleep, so the watermark fires | `stale-read-step-count-fork-fenced` (doc-24) | -| `stale-read-equal-step-counts` (doc-25) | same fault on a fork whose branches emit one step each | `preconditionGuard`, for the same reason | none yet | -| `step-vs-step-fork` (doc-26) | two of the run's own `step_completed` events, one delivery | `countGuard`. **Not** `preconditionGuard`: the withheld completion is a hole in the middle of the log, which moves no high-water mark (§5) | none yet | -| `step-vs-step-fork-fenced` (doc-27) | same, `preconditionGuard: true`, zero rejections | `countGuard`. This row *is* the proof that the watermark half does not fix doc-26 | none yet | -| `in-flight-before-decision` (doc-29) | `beginHookDelivery`, committed before the decision is written | `countGuard` | `in-flight-before-decision-counted` (doc-30) | -| `in-flight-after-decision` (doc-31) | `beginHookDelivery`, committed after the decision | none in the SDK. Needs an append-tail fence — `assertSlotAboveTail`, `vercel/workflow-server#692` | — | +| scenario | mechanism | fix | +|---|---|---| +| `in-flight-before-decision` (doc-29) | `beginHookDelivery`, committed before the decision is written | none in the SDK. The hole is a live reservation, so re-reading finds it still empty | +| `in-flight-before-decision-counted` (doc-30) | same tempo, count half of the fence armed | same. Mint-ordered the write never reaches the fence | +| `in-flight-after-decision` (doc-31) | `beginHookDelivery`, committed after the decision | none in the SDK. Needs an append-tail fence — `assertSlotAboveTail`, `vercel/workflow-server#692` | Those handles are `ScenarioSpec.id`, and they select: `pnpm sim in-flight-after-decision` plays one row of this table. -So: **two** of the six have their fix demonstrated by a paired green scenario, -**three** have a fix identified but unproven here, and **one** has no fix at all. -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 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) -close on the merits, with the book asking them exactly what it asked before: the -reordering was the fault, and once positions are assigned at commit the withheld -read degrades from a hole to a truncation, which the fence can see. The -remaining two (doc-29, doc-31) close because the branch the run ends on changes. -A hook that commits after the timeout genuinely *is* after it when the tail is -the only place a write can land, so the log records the timeout first and the -run that settled is the run the log describes. +**There used to be six, and slot-numbered event ids closed half of them.** The +four that closed — `stale-read-step-count-fork` (doc-23), +`stale-read-equal-step-counts` (doc-25), `step-vs-step-fork` (doc-26), +`step-vs-step-fork-fenced` (doc-27) — all staged a *read* that was missing an +event the log already held. Under ULIDs that read was indistinguishable from a +complete one, and the fence was the only thing that could have caught it, which +is why their `fix` column used to name a predicate. Under slot ids a missing +event is a gap in a numbered sequence, so the runtime sees it without asking +anyone: it re-reads, gets the full log, and decides the fork the way the log +records it. The fence never has to fire. Those four are now regression tests for +the gap audit rather than open reproductions, and doc-24's pairing with doc-23 +is now a pairing between two green scenarios. + +What is left is the family the audit cannot repair by re-reading, because the +position really is empty at the moment of the read: a writer has reserved it and +has not committed. Mint-ordered, doc-29 and doc-30 now fail *loudly* rather than +silently — the replay refuses a log it cannot follow instead of following it +into the wrong branch — which is a better outcome than the divergence they used +to produce, and still a failure. + +**The append-only log closes all three, and by construction rather than by +catching anything.** Nothing reserves a position, so no read can see a hole, and +a hook that commits after the timeout genuinely *is* after it. The log records +the timeout first and the run that settled is the run the log describes. **No expectation is restated per world, and there is no mechanism to.** The first cut of this had one — an `expectAppendOnly` field on three scenarios, @@ -761,43 +755,44 @@ report the branch in the trace. That costs nothing, because the expectations were never what caught the fault. The load-bearing assertion is the invariant: **the log a run wrote must be a log the runtime can replay back into that same run**. It is world-independent, on by -default (`verifyReplay`), and it is what all six reds trip. Measured, not +default (`verifyReplay`), and it is what all three reds trip. Measured, not assumed: strip every `expect` in the book and the violation counts do not move -— 6 mint-ordered, 0 append-only, the same six by name. (Pass/fail does move by +— 3 mint-ordered, 0 append-only, the same three by name. (Pass/fail does move by one, and only for a bookkeeping reason: `hook-never-arrives` expects `stalled`, and a stall's reason is reported as a problem unless the scenario said it was expecting one.) That also removes the one place where the flag's scoreboard rested on a judgement about what the right answer *is* rather than on something the harness checks on its own. -doc-30 is worth a line because it was the third `expectAppendOnly` and is *not* -one of the six — mint-ordered it already passes, since `countGuard` catches -there what the watermark half misses. Its branch moves under the flag for the -same reason its uncounted twin's does, so the old pinned output would have -turned a green scenario red. What made it distinct from doc-29 was never the -branch anyway; it is that the count half of the fence fires at all. That is now -asserted directly, matched on the guard's own message and true in both worlds — -and it fails if `countGuard` is turned off, which is the check that a bare -`rejections().length > 0` would have missed, since doc-29 rejects too. +doc-30 is worth a line because it was the third `expectAppendOnly`. Its branch +moves under the flag for the same reason its uncounted twin's does, so the old +pinned output would have turned a scenario red for ending on the world's answer +rather than on a fault. What makes it distinct from doc-29 was never the branch +anyway; it is that the fence fires at all. That is asserted directly, matched on +`PreconditionFailedError` rather than on "something was rejected", since doc-29 +rejects too. The assertion is scoped to the append-only world, because +mint-ordered the write never reaches the fence — the reservation ahead of it +makes the log unreadable first. Two details worth keeping: - Hook delivery participates. `beginHookDelivery` still reserves a position at the handler boundary; under the flag the reservation stops being binding and the write re-mints at the tail if anything overtook it (`positionAtCommit`). -- doc-30's 412 still fires, and now saves nothing. The count guard counts events - at or below the caller's watermark, the watermark is a millisecond, and a hook - committing after the timeout within the same virtual millisecond is still "at - or below" it. The reload finds nothing to correct and the run settles anyway. - That false positive is the standing cost of the count half of the fence once - the log is append-only, and doc-30's trace is where to see it. - -Four of the six are hook-driven and two deliberately are not — the pair proves -the corruption needs no out-of-band event type. All the pure hook-timing -scenarios pass: placing a hook precisely is what works. What fails is a hook -that is durable in the log but absent from the read the live pass decided on. - -The last row is the only one with no fix in the SDK: the hole opens *after* the +- doc-30's 412 still fires, and now saves nothing. The hook commits after the + timeout and therefore sorts after it, so the reload finds nothing to correct + and the run settles anyway. That false positive is the standing cost of the + fence once the log is append-only, and doc-30's trace is where to see it. + +All three are hook-driven, and the two that were not (doc-26 and doc-27, two of +the run's own `step_completed` events) are the ones the gap audit closed — so +the book no longer holds an open reproduction that needs no out-of-band event +type. All the pure hook-timing scenarios pass: placing a hook precisely is what +works. What fails is a hook whose position is spoken for but whose write has not +landed when the live pass reads. + +doc-31, the last row, is the one that no fence placed anywhere in the write path +could reach: the hole opens *after* the write that should have fenced it, in the quiescent gap between deliveries where the run makes no writes and so meets no checks. `assertSlotAboveTail` in `vercel/workflow-server#692` is the append-tail fence for it. diff --git a/packages/world-sim/src/ids.test.ts b/packages/world-sim/src/ids.test.ts index 1fcd36a9a8..645aecc5ce 100644 --- a/packages/world-sim/src/ids.test.ts +++ b/packages/world-sim/src/ids.test.ts @@ -19,10 +19,10 @@ describe('deterministic id factory', () => { it('sorts by (time, mint order), which is what event-log ordering relies on', () => { let now = 1_704_067_200_000; const ids = createIdFactory(() => now); - const a = ids.eventId(); - const b = ids.eventId(); + const a = ids.messageId(); + const b = ids.messageId(); now += 1; - const c = ids.eventId(); + const c = ids.messageId(); expect(a < b).toBe(true); expect(b < c).toBe(true); }); @@ -30,7 +30,7 @@ describe('deterministic id factory', () => { it('is a pure function of (clock, counter)', () => { const build = () => { const ids = createIdFactory(() => 1_704_067_200_000); - return [ids.runId(), ids.eventId(), ids.messageId()]; + return [ids.runId(), ids.messageId()]; }; expect(build()).toEqual(build()); }); diff --git a/packages/world-sim/src/ids.ts b/packages/world-sim/src/ids.ts index f7a62b227f..0a26e47b87 100644 --- a/packages/world-sim/src/ids.ts +++ b/packages/world-sim/src/ids.ts @@ -3,16 +3,20 @@ * * Every ID the simulation hands out is a function of (virtual time, a * per-scenario counter) — never of `Math.random()` or the host clock. Two - * runs of the same scenario produce byte-identical run IDs, event IDs and - * message IDs, which is what makes an event-stream dump usable as a golden - * file. + * runs of the same scenario produce byte-identical run IDs and message IDs, + * which is what makes an event-stream dump usable as a golden file. * - * IDs still have to be *real* ULIDs: `@workflow/world` validates run IDs with - * `z.string().ulid()` and decodes their embedded timestamp (both to reject - * clock-skewed clients and to seed the workflow VM's fixed clock), so the - * encoding below is the standard Crockford base32 layout — 10 timestamp + * Run and message IDs have to be *real* ULIDs: `@workflow/world` validates run + * IDs with `z.string().ulid()` and decodes their embedded timestamp (both to + * reject clock-skewed clients and to seed the workflow VM's fixed clock), so + * the encoding below is the standard Crockford base32 layout — 10 timestamp * characters followed by 16 characters of "randomness" that we fill from the * counter instead. + * + * Event IDs are not minted here at all. They are the event's position in its + * run's log (`@workflow/world`'s `slotToEventId`), which only the store can + * assign because only the store knows how much of the log is already spoken + * for. See `SimStore.mintEvent`. */ const CROCKFORD = '0123456789ABCDEFGHJKMNPQRSTVWXYZ'; @@ -28,29 +32,6 @@ function encodeBase32(value: number, length: number): string { return out; } -/** - * Decode the mint time back out of an id, prefixed (`evnt_01H…`) or bare. - * - * Both concurrency guards compare *ULID times*, never row timestamps: the - * client's snapshot is the ULID time of the newest event it loaded, and the - * server's marker is the ULID time of the newest out-of-band event. Anything - * that has to reason about the log's order therefore reads it back out of the - * id, which is why this lives beside the minting and not beside a caller. - * - * Returns `+Infinity` for an id whose time field is not base32 — an id that - * cannot be placed sorts after everything rather than silently landing at 0. - */ -export function ulidTimeOf(id: string): number { - const ulid = id.includes('_') ? id.slice(id.indexOf('_') + 1) : id; - let time = 0; - for (const char of ulid.slice(0, 10)) { - const digit = CROCKFORD.indexOf(char); - if (digit === -1) return Number.POSITIVE_INFINITY; - time = time * 32 + digit; - } - return time; -} - /** * A monotonic ULID source. * @@ -63,8 +44,6 @@ export function ulidTimeOf(id: string): number { export interface IdFactory { /** Mint a bare ULID stamped with the current virtual time. */ ulid(): string; - /** Mint `evnt_`. */ - eventId(): string; /** Mint `wrun_`. */ runId(): string; /** Mint a monotonically increasing message id. */ @@ -92,7 +71,6 @@ export function createIdFactory(now: () => number): IdFactory { return { ulid, - eventId: () => `evnt_${ulid()}`, runId: () => `wrun_${ulid()}`, messageId: () => `msg_${ulid()}`, count: () => counter, diff --git a/packages/world-sim/src/invariants.test.ts b/packages/world-sim/src/invariants.test.ts index 1e9eb79745..bef2dcb778 100644 --- a/packages/world-sim/src/invariants.test.ts +++ b/packages/world-sim/src/invariants.test.ts @@ -1,4 +1,9 @@ -import type { Event, Step, WorkflowRun } from '@workflow/world'; +import { + type Event, + type Step, + slotToEventId, + type WorkflowRun, +} from '@workflow/world'; import { describe, expect, it } from 'vitest'; import { checkInvariants, type InvariantInput } from './invariants.js'; @@ -25,7 +30,7 @@ function event(partial: Partial & Pick): Event { counter++; return { runId: RUN, - eventId: `evnt_${String(counter).padStart(4, '0')}`, + eventId: slotToEventId(counter), createdAt: new Date(BASE.getTime() + counter), specVersion: 5, ...partial, diff --git a/packages/world-sim/src/replay.test.ts b/packages/world-sim/src/replay.test.ts index bc4a56ff73..b199fc78ab 100644 --- a/packages/world-sim/src/replay.test.ts +++ b/packages/world-sim/src/replay.test.ts @@ -10,6 +10,7 @@ import { getWorld } from '@workflow/core/runtime'; import { type Event, SPEC_VERSION_CURRENT, + slotToEventId, type WorkflowRun, } from '@workflow/world'; import { describe, expect, it } from 'vitest'; @@ -26,7 +27,7 @@ function event(partial: Partial & Pick): Event { counter++; return { runId: RUN, - eventId: `evnt_${String(counter).padStart(4, '0')}`, + eventId: slotToEventId(counter), createdAt: new Date(AT.getTime() + counter), specVersion: SPEC_VERSION_CURRENT, ...partial, diff --git a/packages/world-sim/src/scenario.ts b/packages/world-sim/src/scenario.ts index b1566027ac..449fb52c9e 100644 --- a/packages/world-sim/src/scenario.ts +++ b/packages/world-sim/src/scenario.ts @@ -324,7 +324,7 @@ export async function runScenario( // It has to be an out-of-band writer. An inline step's `step_completed` // held the same way would stall the orchestrator that should misread the // log, because the runtime awaits its promise before deciding anything. - const position = world.reservePosition(); + const position = world.reservePosition(runId); return { eventId: position.eventId, async commit() { diff --git a/packages/world-sim/src/store.test.ts b/packages/world-sim/src/store.test.ts index 9041627967..944dea5435 100644 --- a/packages/world-sim/src/store.test.ts +++ b/packages/world-sim/src/store.test.ts @@ -3,11 +3,12 @@ import { HookNotFoundError, RunExpiredError, } from '@workflow/errors'; -import { SPEC_VERSION_CURRENT } from '@workflow/world'; +import { requireEventSlot, SPEC_VERSION_CURRENT } from '@workflow/world'; import { beforeEach, describe, expect, it } from 'vitest'; import { createIdFactory } from './ids.js'; import { createSimStore, + type LoadedSnapshot, type MintedEvent, type SimStore, type SimStoreOptions, @@ -16,6 +17,18 @@ import { const SPEC = SPEC_VERSION_CURRENT; +/** + * The snapshot of a caller that loaded the run's whole log. The store is driven + * directly here, so there is no facade tracking reads to derive one from. + */ +function loadedAll(store: SimStore): LoadedSnapshot { + const events = store.allEvents(RUN); + return { + maxSlot: Math.max(...events.map((e) => requireEventSlot(e.eventId))), + count: events.length, + }; +} + function setup(options?: Omit) { let now = 1_704_067_200_000; const store = createSimStore({ @@ -424,10 +437,7 @@ describe('sim store', () => { }); guarded.tick(10); - const snapshot = { - updatedAt: guarded.nowMs(), - count: guarded.store.allEvents(RUN).length, - }; + const snapshot = loadedAll(guarded.store); guarded.tick(10); // An out-of-band resume: no snapshot, so it advances the marker. await guarded.store.events.create(RUN, { @@ -450,7 +460,7 @@ describe('sim store', () => { ) ).rejects.toThrow(/out of band/); - // An up-to-date snapshot passes — an equal timestamp must not livelock. + // An up-to-date snapshot passes — an equal watermark must not livelock. await expect( guarded.store.events.create( RUN, @@ -460,12 +470,7 @@ describe('sim store', () => { correlationId: 'step_1', eventData: { stepName: 'step//./w//s', input: new Uint8Array() }, }, - { - snapshot: { - updatedAt: guarded.nowMs(), - count: guarded.store.allEvents(RUN).length, - }, - } + { snapshot: loadedAll(guarded.store) } ) ).resolves.toBeTruthy(); }); @@ -488,7 +493,7 @@ describe('sim store', () => { it('is off by default: a held write lands behind one committed sooner', async () => { await createRun(store, RUN); // The handler boundary takes a position; the write is then held. - const minted = store.mintEvent(); + const minted = store.mintEvent(RUN); tick(10); const overtook = await store.events.create(RUN, hook); const held = await store.events.create(RUN, step, heldAt(minted)); @@ -505,7 +510,7 @@ describe('sim store', () => { it('re-mints a write that was overtaken while it was held', async () => { const world = setup({ appendOnlyLog: true }); await createRun(world.store, RUN); - const minted = world.store.mintEvent(); + const minted = world.store.mintEvent(RUN); world.tick(10); const overtook = await world.store.events.create(RUN, hook); const held = await world.store.events.create(RUN, step, heldAt(minted)); @@ -523,7 +528,7 @@ describe('sim store', () => { it('leaves an uncontended write at the position it minted', async () => { const world = setup({ appendOnlyLog: true }); await createRun(world.store, RUN); - const minted = world.store.mintEvent(); + const minted = world.store.mintEvent(RUN); world.tick(10); const uncontended = await world.store.events.create( RUN, diff --git a/packages/world-sim/src/store.ts b/packages/world-sim/src/store.ts index abde8a4193..5244555b42 100644 --- a/packages/world-sim/src/store.ts +++ b/packages/world-sim/src/store.ts @@ -43,14 +43,16 @@ import { type PaginatedResponse, type PaginationOptions, type ResolveData, + requireEventSlot, SPEC_VERSION_CURRENT, type Step, type Storage, + slotToEventId, stripEventDataRefs, type Wait, type WorkflowRun, } from '@workflow/world'; -import { type IdFactory, ulidTimeOf } from './ids.js'; +import type { IdFactory } from './ids.js'; /** Per-run event ceiling reported on run responses, mirroring the other worlds. */ const MAX_EVENTS_PER_RUN = 25_000; @@ -67,20 +69,21 @@ const DEFAULT_PAGE_LIMIT = 20; const RUN_EVENT_INDEX_WINDOW = 16; /** - * A log position, minted before the write that will occupy it commits. + * A log position, taken before the write that will occupy it commits. * - * The event id *is* the log's sort key, and the sim mints it the way - * workflow-server does: in the request handler (`EventId.make()`), not in - * storage — DynamoDB does not generate ids. Everything downstream follows from - * that one fact. A write that is minted and then takes a while to commit keeps - * the earlier position it was given, so the log can gain an event *behind* a - * position a reader has already seen. That is the hole no high-water mark can - * detect, and reproducing it is the reason minting is separable from appending. + * The event id *is* the position: `evnt_` followed by the event's 1-based slot + * in its run's log. This store hands the slot out in the request handler rather + * than at the append, which is the shape a World has whenever it cannot ask + * storage to allocate — and everything downstream follows from that one fact. A + * write that takes a slot and then takes a while to commit keeps the earlier + * slot it was given, so the log can gain an event *behind* a position a reader + * has already seen. That is the hole no high-water mark can detect, and + * reproducing it is the reason taking a position is separable from appending. * - * `createdAt` is the mint instant, not the commit instant: in production it is - * decoded back out of the ULID, so the two can never disagree. Entity rows + * `createdAt` is the mint instant, not the commit instant, matching a World + * that derives the row's timestamp at the handler boundary. Entity rows * (step/run/hook timestamps) still use the commit instant, because those are - * written by the transaction rather than derived from the id. + * written by the transaction rather than carried with the position. */ export interface MintedEvent { eventId: string; @@ -103,13 +106,14 @@ interface SimCreateParams { * 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. + * Reconstructed by the world facade from the pages the writer read rather + * than taken off the wire, because the wire carries only half of it: the + * runtime states the highest slot it holds, and the fence also wants how many + * events it loaded at or below that slot. The reconstruction is the same + * derivation the client made — 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`. */ @@ -118,9 +122,9 @@ interface SimCreateParams { /** 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}. */ + /** Slot of the newest loaded event. */ + maxSlot: number; + /** How many loaded events sit at or below {@link maxSlot}. */ count: number; } @@ -142,10 +146,10 @@ interface RunEventIndex { */ function countRecordedAtOrBelow( index: RunEventIndex, - updatedAt: number + maxSlot: number ): number | null { const above = index.recentEventIds.filter( - (id) => ulidTimeOf(id) > updatedAt + (id) => requireEventSlot(id) > maxSlot ).length; const pruned = index.total > index.recentEventIds.length; if (pruned && above === index.recentEventIds.length) return null; @@ -156,13 +160,16 @@ export interface SimStoreOptions { now(): number; ids: IdFactory; /** - * Enforce the optimistic-concurrency precondition guard described in - * `WorldCapabilities.preconditionGuard`: reject a replay-context write whose - * snapshot predates the newest externally-originated event. + * Reject a replay-context write whose snapshot predates the newest + * externally-originated event, with a `PreconditionFailedError` (412). * - * 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 - * changes which runtime fast paths engage, so it is never implicit. + * No shipped World does this — the runtime does not need it, since a + * reader's log is a prefix and its next write reports what it was pushed + * past. The option stays because the sim is where the *reception* path is + * exercised: the runtime still handles a 412 (restart in place, then + * re-invoke), and a World that allocates positions away from the commit may + * still want to refuse rather than report. Off by default, and never + * implicit, because arming it changes which runtime fast paths engage. */ preconditionGuard?: boolean; /** @@ -246,13 +253,13 @@ export interface SimStore extends Storage { */ seedFromLog(log: readonly Event[]): void; /** - * Mint the next log position, without writing anything. + * Take the run's next log position, without writing anything. * * The world facade calls this at the handler boundary — before any hold can * fire — so a write held mid-flight already owns the position it will * eventually occupy. See {@link MintedEvent}. */ - mintEvent(): MintedEvent; + mintEvent(runId: string): MintedEvent; /** * Hide the *next* event appended from the following `reads` event-log reads. * @@ -414,10 +421,28 @@ export function createSimStore(options: SimStoreOptions): SimStore { /** hookIds that have been explicitly disposed; disposal is permanent. */ const disposedHooks = new Set(); /** - * Per run: ULID time of the newest externally-originated event. Only read - * when `preconditionGuard` is on. See `SimCreateParams.snapshot`. + * Per run: slot of the newest externally-originated event. Only read when + * `preconditionGuard` is on. See `SimCreateParams.snapshot`. */ const externalWriteMarker = new Map(); + /** + * Per run: the highest slot handed out, committed or merely spoken for. + * + * Separate from the committed log because a position is taken at the handler + * boundary: between `mintEvent` and the append, the slot exists and belongs + * to nobody. A write that never commits gives its slot back (see + * `releaseSlot`); a position reserved and then abandoned out of band leaves + * it empty for good. + */ + const highestSlot = new Map(); + /** + * Per run: positions handed out and given back, still unoccupied. + * + * Reused before the range grows, so the log stays dense. What makes that + * safe is that a slot only lands here once its create has returned: nothing + * is still holding it, and nothing ever will. + */ + const freeSlots = new Map(); /** * Per run: the tail of the log, for the count guard. Records *every* event, * replay-origin included — the corruption it guards against is one replay @@ -460,8 +485,69 @@ export function createSimStore(options: SimStoreOptions): SimStore { const waitKey = (runId: string, correlationId: string) => `${runId}:${correlationId}`; - function mintEvent(): MintedEvent { - return { eventId: ids.eventId(), createdAt: new Date(nowMs()) }; + /** Highest slot committed to a run's log, or 0 for a log with no events. */ + function committedSlot(runId: string): number { + let max = 0; + for (const event of events) { + if (event.runId !== runId) continue; + const slot = requireEventSlot(event.eventId); + if (slot > max) max = slot; + } + return max; + } + + function mintEvent(runId: string): MintedEvent { + let slot: number; + const free = appendOnlyLog ? undefined : freeSlots.get(runId); + if (free?.length) { + free.sort((a, b) => a - b); + slot = free.shift() as number; + } else { + slot = (highestSlot.get(runId) ?? 0) + 1; + highestSlot.set(runId, slot); + } + return { eventId: slotToEventId(slot), createdAt: new Date(nowMs()) }; + } + + /** + * Give a slot back when nothing committed at it. + * + * A hole in the log is corruption as far as the runtime is concerned, so a + * create that appends nothing must not consume a position. Two kinds do: one + * the store rejects outright, and one it accepts as a no-op (a second + * `run_started` for a run already started, say). Production reaches the same + * place from the other side, by allocating inside the transaction, after the + * validation, so a write it refuses never had a slot to lose. Reproducing + * *that* difference is not what this store is for: the fault it stages is + * two writes taking positions in one order and committing in another, and a + * rejection leaving a permanent hole would sit on top of every one of those + * scenarios as a second, unrelated corruption. + * + * A slot at the top of the range is dropped rather than recycled, because a + * range that never grew is not a hole to fill. Anything below it goes on the + * free list, since concurrent writers mean the rejected position is not + * always the newest one. + */ + function releaseSlot(runId: string, position: MintedEvent): void { + // Under `appendOnlyLog` the position is decided at the append, which keeps + // the mark on the committed tail; a reservation nothing used was never + // counted in the first place. + if (appendOnlyLog) return; + const occupied = events.some( + (e) => e.runId === runId && e.eventId === position.eventId + ); + if (occupied) return; + const free = freeSlots.get(runId) ?? []; + free.push(requireEventSlot(position.eventId)); + let highest = highestSlot.get(runId) ?? 0; + let index = free.indexOf(highest); + while (index !== -1) { + free.splice(index, 1); + highest--; + index = free.indexOf(highest); + } + highestSlot.set(runId, highest); + freeSlots.set(runId, free); } function recordInIndex(event: Event): void { @@ -483,25 +569,38 @@ export function createSimStore(options: SimStoreOptions): SimStore { /** * The position an event actually commits at. * - * Only `appendOnlyLog` can move one. A write that is still the newest - * position when it arrives keeps the id it was already handed out under, so - * uncontended history is unchanged; one that was overtaken while it was held - * re-mints and takes the tail. + * Only `appendOnlyLog` can move one, and there it moves every write that is + * not already landing on the run's next free slot: the position is whatever + * follows the newest *committed* event, decided here rather than at the + * boundary. A write that was never overtaken is already there, so uncontended + * history is unchanged; one that was overtaken while it was held gives up the + * slot it reserved and takes the tail. * - * Compared as plain strings: the ids are fixed-width ULIDs whose lexical - * order *is* `(createdAt, eventId)` order, because the time field is the - * `createdAt` millisecond and the suffix is a monotonic counter. See - * `createIdFactory`. + * Recomputing rather than comparing also keeps the log dense. A slot the + * boundary handed out and nothing committed at is a permanent hole in the + * default mode; under `appendOnlyLog` nothing consumes a slot until it + * commits, so no reservation can leave one behind. */ function positionAtCommit(event: Event): Event { if (!appendOnlyLog) return event; - const tail = events[events.length - 1]; - if (!tail || event.eventId > tail.eventId) return event; - return { ...event, ...mintEvent() }; + const next = committedSlot(event.runId) + 1; + if (requireEventSlot(event.eventId) === next) return event; + return { + ...event, + eventId: slotToEventId(next), + createdAt: new Date(nowMs()), + }; } function append(incoming: Event): Event { const event = positionAtCommit(incoming); + if (appendOnlyLog) { + // The commit decided the position, so the allocator follows the log + // rather than the other way round. A write that reserved a slot and + // then committed *below* it would otherwise leave the mark above the + // tail, and the next mint would skip the difference. + highestSlot.set(event.runId, requireEventSlot(event.eventId)); + } events.push(event); recordInIndex(event); if (armedWithhold !== undefined) { @@ -797,10 +896,18 @@ export function createSimStore(options: SimStoreOptions): SimStore { } } - async function create( + /** + * Append one event, or refuse to. + * + * `held` carries the position this call is holding out to the wrapper below, + * which hands it back when the call throws. It is a parameter rather than a + * closure variable because two creates can be in flight at once. + */ + async function commitEvent( runIdArg: string | null, data: AnyEventRequest, - params?: CreateEventParams + params: CreateEventParams | undefined, + held: { runId?: string; position?: MintedEvent } ): Promise { // Commit time, for the entity rows the transaction writes. The *event's* // timestamp comes from its minted position instead — see `MintedEvent`. @@ -808,11 +915,6 @@ export function createSimStore(options: SimStoreOptions): SimStore { const internal = params as | (CreateEventParams & SimCreateParams) | undefined; - // Reassigned only by the two paths that write a *synthetic* event ahead of - // the requested one: the synthetic takes the position minted at the - // boundary (production mints it first, for exactly this ordering) and the - // requested event re-mints so it still sorts after. - let position = internal?.minted ?? mintEvent(); const resolveData: ResolveData = params?.resolveData ?? 'all'; const specVersion = data.specVersion ?? SPEC_VERSION_CURRENT; @@ -825,6 +927,14 @@ export function createSimStore(options: SimStoreOptions): SimStore { runId = runIdArg; } + // Reassigned only by the two paths that write a *synthetic* event ahead of + // the requested one: the synthetic takes the position taken at the + // boundary (which is the earlier one, for exactly this ordering) and the + // requested event takes a fresh one so it still sorts after. + let position = internal?.minted ?? mintEvent(runId); + held.runId = runId; + held.position = position; + let currentRun = runs.get(runId); // ---- Resilient start --------------------------------------------------- @@ -854,7 +964,8 @@ export function createSimStore(options: SimStoreOptions): SimStore { append(synthetic); // The synthetic took the boundary-minted position, so the `run_started` // row built below needs a fresh one to sort after it. - position = mintEvent(); + position = mintEvent(runId); + held.position = position; } } @@ -874,7 +985,7 @@ export function createSimStore(options: SimStoreOptions): SimStore { const snapshot = internal?.snapshot; if (options.preconditionGuard && snapshot) { const marker = externalWriteMarker.get(runId); - if (marker !== undefined && snapshot.updatedAt < marker) { + if (marker !== undefined && snapshot.maxSlot < marker) { throw new PreconditionFailedError( `Run "${runId}" changed out of band since the caller's snapshot` ); @@ -890,7 +1001,7 @@ export function createSimStore(options: SimStoreOptions): SimStore { if (options.countGuard) { const index = runEventIndex.get(runId); const recorded = index - ? countRecordedAtOrBelow(index, snapshot.updatedAt) + ? countRecordedAtOrBelow(index, snapshot.maxSlot) : null; if (recorded !== null && recorded > snapshot.count) { throw new PreconditionFailedError( @@ -1160,7 +1271,8 @@ export function createSimStore(options: SimStoreOptions): SimStore { // an early position for either. No scenario needs that yet; the writes // that race for position in practice are the completions. const { input: _dropped, ...rest } = data.eventData; - position = mintEvent(); + position = mintEvent(runId); + held.position = position; event = { ...event, ...position, eventData: rest } as Event; } @@ -1180,10 +1292,10 @@ export function createSimStore(options: SimStoreOptions): SimStore { // Two details are load-bearing, both copied from workflow-server's // `recordOutsideEvent`: // - // - The mark is the event's *own* position time, not the commit instant. It - // 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 mark is the event's *own* slot, not the commit instant. It has to + // be the same derivation as a caller's watermark (the slot 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 @@ -1198,7 +1310,7 @@ export function createSimStore(options: SimStoreOptions): SimStore { const previous = externalWriteMarker.get(runId) ?? 0; externalWriteMarker.set( runId, - Math.max(previous, event.createdAt.getTime()) + Math.max(previous, requireEventSlot(event.eventId)) ); } @@ -1254,6 +1366,19 @@ export function createSimStore(options: SimStoreOptions): SimStore { return deltaPage ? { ...result, ...deltaPage } : result; } + async function create( + runIdArg: string | null, + data: AnyEventRequest, + params?: CreateEventParams + ): Promise { + const held: { runId?: string; position?: MintedEvent } = {}; + try { + return await commitEvent(runIdArg, data, params, held); + } finally { + if (held.runId && held.position) releaseSlot(held.runId, held.position); + } + } + const storage: SimStore = { runs: { async get(id: string, params?: { resolveData?: ResolveData }) { @@ -1424,6 +1549,14 @@ export function createSimStore(options: SimStoreOptions): SimStore { const seeded = clone(event) as Event; events.push(seeded); recordInIndex(seeded); + // A cold start inherits the log's positions, so the next write has to + // continue them. Taking the maximum rather than counting: the log a + // scenario seeds is whatever the previous process committed, holes + // included, and re-issuing a slot it already used would be worse than + // leaving the hole. + const slot = requireEventSlot(seeded.eventId); + const highest = highestSlot.get(seeded.runId) ?? 0; + if (slot > highest) highestSlot.set(seeded.runId, slot); // The event's own position time is the only clock a seeded row can // have: the live one belongs to whenever this world was built. applyEvent(seeded, seeded.createdAt); diff --git a/packages/world-sim/src/world.ts b/packages/world-sim/src/world.ts index 0938126a29..eb56e132fc 100644 --- a/packages/world-sim/src/world.ts +++ b/packages/world-sim/src/world.ts @@ -30,11 +30,12 @@ import { type Event, getQueueTopicPrefix, type QueuePayload, + requireEventSlot, SPEC_VERSION_CURRENT, type World, } from '@workflow/world'; import { createVirtualClock, type VirtualClock } from './clock.js'; -import { createIdFactory, type IdFactory, ulidTimeOf } from './ids.js'; +import { createIdFactory, type IdFactory } from './ids.js'; import { createSimQueue, type DirectHandler, type SimQueue } from './queue.js'; import { createSimStore, @@ -99,10 +100,10 @@ export interface SimWorldOptions { * log is append-only and no read can be contradicted by a later one. See * `SimStoreOptions.appendOnlyLog` for what that buys and what it costs. * - * The boundary mint still happens — `reservePosition` and everything a - * scenario hangs off it work unchanged. It just stops being binding: a held - * write that nothing overtook keeps the position it reserved, and one that was - * overtaken re-mints when it lands. + * The boundary reservation still happens — `reservePosition` and everything + * a scenario hangs off it work unchanged. It just stops being binding: a held + * write that nothing overtook keeps the position it reserved, and one that + * was overtaken takes the tail when it lands. */ appendOnlyLog?: boolean; } @@ -129,15 +130,16 @@ export interface SimWorld extends World { */ asExternal(fn: () => Promise): Promise; /** - * Take a log position now, to be used by a write that happens later. + * Take a log position in `runId` now, to be used by a write that happens + * later. * * The scenario's own calls are not call points (see `fireWatches`), so a script - * cannot hold *itself* between minting and committing the way it holds a - * writer. This pair is how it states the same thing directly: reserve the + * cannot hold *itself* between taking a position and committing the way it + * holds a writer. This pair is how it states the same thing directly: reserve the * position, do whatever should observe the log without it, then run the write * inside `withReservedPosition` so it lands where it was reserved. */ - reservePosition(): MintedEvent; + reservePosition(runId: string): MintedEvent; /** Run `fn` with the next `events.create` taking `position` instead of minting. */ withReservedPosition( position: MintedEvent, @@ -608,12 +610,12 @@ export function createSimWorld(options: SimWorldOptions = {}): SimWorld { if (!runId) return undefined; const set = loadedEvents()?.get(runId); if (!set || set.size === 0) return undefined; - let updatedAt = 0; + let maxSlot = 0; for (const eventId of set) { - const at = ulidTimeOf(eventId); - if (at > updatedAt) updatedAt = at; + const slot = requireEventSlot(eventId); + if (slot > maxSlot) maxSlot = slot; } - return { updatedAt, count: set.size }; + return { maxSlot, count: set.size }; } /** Wrap one world method so it becomes a call point. */ @@ -652,7 +654,18 @@ export function createSimWorld(options: SimWorldOptions = {}): SimWorld { if (call === 'events.create') { // A reserved position wins: it belongs to a write whose handler was // entered earlier and is only now reaching storage. - const minted = reservedPosition ?? store.mintEvent(); + // + // No boundary mint for a `run_created` that names no run + // (`events.create(null, …)`, which the Storage contract allows and the + // store supports by generating the id). There is no run to allocate + // against yet: minting under the null key would hand out slots from a + // bucket shared by every such call, and the generated run's own + // allocator would then start at 1 and collide. The store mints after it + // resolves the id instead. The runtime never takes this path, so this + // is about the failure being impossible rather than merely unobserved. + const minted = + reservedPosition ?? + (runId === undefined ? undefined : store.mintEvent(runId)); reservedPosition = undefined; const params = (args[2] ?? {}) as Record; callArgs = [ @@ -660,13 +673,13 @@ export function createSimWorld(options: SimWorldOptions = {}): SimWorld { args[1], { ...params, - minted, - // 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. + ...(minted ? { minted } : {}), + // The snapshot the fence reads. Reconstructed rather than taken + // off the wire, because the wire carries only the writer's highest + // slot and the count guard also wants how many events it loaded at + // or below it. 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. // // Attached whenever the fence is armed, not just for the count // half: `SimCreateParams.snapshot` is also what marks a write as @@ -777,11 +790,11 @@ export function createSimWorld(options: SimWorldOptions = {}): SimWorld { const world: SimWorld = { specVersion: SPEC_VERSION_CURRENT, - capabilities: { - // Only advertise the fence when the store is actually enforcing it — a - // runtime fast path gated on this capability must never run without one. - ...(options.preconditionGuard ? { preconditionGuard: true } : {}), - }, + // Whether the fence is armed is a store option, not a capability: the + // runtime assumes every World may reject a stale write, so a scenario can + // only change what the store does about one, never what the runtime + // expects. See `SimStoreOptions.preconditionGuard`. + capabilities: {}, getDeploymentId: intercept('getDeploymentId', () => simQueue.getDeploymentId() ), @@ -848,7 +861,7 @@ export function createSimWorld(options: SimWorldOptions = {}): SimWorld { async asExternal(fn) { return externalCtx.run(true, fn); }, - reservePosition: () => store.mintEvent(), + reservePosition: (runId) => store.mintEvent(runId), async withReservedPosition(position, fn) { reservedPosition = position; try { diff --git a/packages/world-testing/src/event-ids.mts b/packages/world-testing/src/event-ids.mts new file mode 100644 index 0000000000..1b26b3bff2 --- /dev/null +++ b/packages/world-testing/src/event-ids.mts @@ -0,0 +1,62 @@ +import { + eventIdToSlot, + FIRST_EVENT_SLOT, + slotToEventId, +} from '@workflow/world'; +import { expect, test, vi } from 'vitest'; +import { createFetcher, startServer } from './util.mjs'; + +/** + * Event ids are positions. + * + * This is the one part of the storage contract a World can get wrong while + * every other test in this suite passes. Nothing here reads an event id, so a + * World that mints ULIDs runs an addition, resumes a hook and completes a run + * exactly as it should — and then fails every replay on the deployment, with + * `Event id is not slot-numbered`, because the runtime reads a position out of + * each id it loads and cannot proceed without one. + * + * Asserting it here is the difference between a conformance failure a World + * author can act on and a production failure nobody can explain. See the + * Event ID Allocation section of the building-a-world guide. + */ +export function eventIds(world: string) { + test('numbers events by position', { timeout: 30_000 }, async () => { + const server = await startServer({ world }).then(createFetcher); + const result = await server.invoke( + 'workflows/addition.ts', + 'addition', + [1, 2] + ); + await vi.waitFor( + async () => { + expect((await server.getRun(result.runId)).status).toBe('completed'); + }, + { interval: 200, timeout: 25_000 } + ); + + const events = await server.getEvents(result.runId); + expect(events.length).toBeGreaterThan(0); + + // Every id decodes to a slot. `eventIdToSlot` answers null for anything + // else, which is what `requireEventSlot` turns into a failed run. + const slots = events.map((event) => ({ + eventId: event.eventId, + slot: eventIdToSlot(event.eventId), + })); + expect(slots.filter((entry) => entry.slot === null)).toEqual([]); + + // Dense from 1, in the order the World returns them. Density is what lets + // a reader tell a complete log from a truncated one by its length, and + // what makes the writer's event count a statement of its position. + expect(slots.map((entry) => entry.slot)).toEqual( + slots.map((_, index) => FIRST_EVENT_SLOT + index) + ); + + // The canonical format, not merely something that decodes: a World that + // pads to a different width sorts its own log wrongly past 10 events. + expect(slots.map((entry) => entry.eventId)).toEqual( + slots.map((entry) => slotToEventId(entry.slot as number)) + ); + }); +} diff --git a/packages/world-testing/src/index.mts b/packages/world-testing/src/index.mts index 04637f882f..aa298bb03f 100644 --- a/packages/world-testing/src/index.mts +++ b/packages/world-testing/src/index.mts @@ -1,5 +1,6 @@ import { addition } from './addition.mjs'; import { errors } from './errors.mjs'; +import { eventIds } from './event-ids.mjs'; import { hooks } from './hooks.mjs'; import { idempotency } from './idempotency.mjs'; import { inlineExecution } from './inline-execution.mjs'; @@ -8,6 +9,7 @@ import { nullByte } from './null-byte.mjs'; export function createTestSuite(pkgName: string) { addition(pkgName); + eventIds(pkgName); idempotency(pkgName); hooks(pkgName); nullByte(pkgName); diff --git a/packages/world-testing/src/server.mts b/packages/world-testing/src/server.mts index 1562fc33ae..b5f24ce602 100644 --- a/packages/world-testing/src/server.mts +++ b/packages/world-testing/src/server.mts @@ -117,7 +117,11 @@ const app = new Hono() .get('/runs/:runId/events', async (ctx) => { const runId = ctx.req.param('runId'); const world = await getWorld(); - const allEvents: { eventType: string; correlationId?: string }[] = []; + const allEvents: { + eventId: string; + eventType: string; + correlationId?: string; + }[] = []; let cursor: string | undefined; while (true) { const page = await world.events.list({ @@ -126,6 +130,7 @@ const app = new Hono() }); for (const e of page.data) { allEvents.push({ + eventId: e.eventId, eventType: e.eventType, correlationId: e.correlationId, }); diff --git a/packages/world-testing/src/util.mts b/packages/world-testing/src/util.mts index c565b30895..17e1ca45ca 100644 --- a/packages/world-testing/src/util.mts +++ b/packages/world-testing/src/util.mts @@ -143,6 +143,32 @@ export function createFetcher(control: Control) { }); return data; }, + /** + * Every event of a run, oldest first, as the World stored it. + * + * Ids included on purpose: they are the one part of the storage contract a + * World can get wrong while every workflow still appears to work, right + * up until a replay reads one (see `eventIds`). + */ + async getEvents(runId: string): Promise< + { + eventId: string; + eventType: string; + correlationId?: string; + }[] + > { + const x = await fetch( + `http://localhost:${control.info.port}/runs/${encodeURIComponent(runId)}/events` + ); + const data = (await x.json()) as { + events: { + eventId: string; + eventType: string; + correlationId?: string; + }[]; + }; + return data.events; + }, async getFlowInvocationCount(runId: string): Promise { const x = await fetch( `http://localhost:${control.info.port}/_flow-invocations/${encodeURIComponent(runId)}` diff --git a/packages/world-vercel/src/index.ts b/packages/world-vercel/src/index.ts index f1fb2cbe53..9af8a5db36 100644 --- a/packages/world-vercel/src/index.ts +++ b/packages/world-vercel/src/index.ts @@ -38,17 +38,6 @@ export function createWorld(config?: APIConfig): World { specVersion: SPEC_VERSION_SUPPORTS_SLOT_IDENTITY, capabilities: { hookRetention: { active: true }, - // 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` // flow topics (see queue.ts and @workflow/builders). @@ -56,11 +45,6 @@ export function createWorld(config?: APIConfig): World { // Vercel deployments are atomic and immutable, so a deployment id names // one fixed build for its whole lifetime. deploymentAffinity: true, - // New runs get dense per-run slot event ids. Runs created before the - // backend adopted them keep their ULIDs; the scheme is pinned by the - // spec version stamped on each run, not by this flag, which only says - // what new runs get. - slotEventIds: true, // NOTE: the backend half of resumeHook()'s parallel fast path — that // the server enforces the `(runId, resumeId)` dedup constraint — is // NO LONGER a static world capability here. It is attested per-lookup by diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index 6373bc3678..63bd5afd74 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -792,15 +792,16 @@ export interface CreateEventParams { * parallelized with the queue publish and may have failed. Only meaningful * for `step_created`. * - * Advisory. The runtime never parallelizes a *guarded* `step_created` with - * its publish (see the eligibility gate in the suspension handler), so in - * correct operation a re-ensure can only correspond to an unguarded create - * — there is no guard verdict for it to bypass. A guard-enforcing backend - * MAY nevertheless use this flag as defense-in-depth: refuse the re-ensure - * (world-vercel surfaces the backend's 410 as `RunExpiredError`, which the - * consumer treats as "nothing left to execute" and acks the message) when - * it has recorded a 412 rejection for this correlation id and no step - * entity exists — hardening against a misbehaving or future client. Worlds + * Advisory. Parallelizing a create with its publish is opt-in and off by + * default (`WORKFLOW_RESILIENT_STEP_DISPATCH`), precisely because a create + * can come back refused while the message carrying its payload is already + * out. A deployment that opts in accepts that window, and a backend MAY use + * this flag to narrow it: refuse the re-ensure (world-vercel surfaces the + * backend's 410 as `RunExpiredError`, which the consumer treats as "nothing + * left to execute" and acks the message) when it has recorded a refusal for + * this correlation id and no step entity exists. Best-effort by nature — a + * marker written at refusal time cannot be ordered before the redelivery it + * is meant to stop — so it hardens, and does not close, the window. Worlds * may ignore this flag entirely. */ viaStepDispatch?: boolean; @@ -816,12 +817,13 @@ export interface CreateEventParams { /** * 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. 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). + * every replay-context create; omitted by callers with no loaded log to be + * stale against. * - * Only meaningful against a World that declares - * `WorldCapabilities.slotEventIds`, where slots are dense and 1-based so a - * count and a position are the same number. Such a World attempts + * A World's slots are dense and 1-based (see `Storage.events`), so a count + * and a position are the same number. An id that is not a position does not + * produce a count here — it throws, since the runtime cannot state a + * snapshot for a log it cannot place. Such a World attempts * `eventCount + 1`, and on contention **bumps** to the next free slot and * commits there anyway — a stale count never rejects a write. What it does * instead is report: when the committed slot is higher than the one asked @@ -988,9 +990,8 @@ export type EventResult = { * log through the canonical `hook_received`, so the lazy hook queue * consumer can skip both the `run_started` write and the initial * `events.list`. - * - On any response from a slot-allocating World (see - * `WorldCapabilities.slotEventIds`) whose committed slot came out - * higher than the one {@link CreateEventParams.eventCount} asked for: + * - On any response whose committed slot came out higher than the one + * {@link CreateEventParams.eventCount} asked for: * the events occupying the slots that were skipped over, in slot * order. This is the "report" half of bump-and-report — the write * succeeded, and these are the events the writer had not seen when it diff --git a/packages/world/src/index.ts b/packages/world/src/index.ts index 4a0d932599..a8325cc44c 100644 --- a/packages/world/src/index.ts +++ b/packages/world/src/index.ts @@ -121,6 +121,7 @@ export { isSlotBody, isSlotEventId, MAX_EVENT_SLOT, + requireEventSlot, slotToEventId, } from './slot-identity.js'; export type { SpecVersion } from './spec-version.js'; diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts index 8d62f746dc..ed76fd0c00 100644 --- a/packages/world/src/interfaces.ts +++ b/packages/world/src/interfaces.ts @@ -263,6 +263,35 @@ export interface Storage { ): Promise>; }; + /** + * The event log, and the one part of this interface with a requirement the + * types cannot express: **the World allocates every event id, and every id + * is a slot** — `evnt_` followed by the event's dense, 1-based position in + * its run's log, zero-padded to 26 characters. Use `slotToEventId()` to + * format one. + * + * Not a capability to opt into. The runtime reads a position out of every id + * it loads (`requireEventSlot`) and fails the run if it cannot, so a World + * whose ids are not positions cannot replay anything at all. Two properties + * are what the runtime actually relies on: + * + * - **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 in a single integer, and what lets a reader tell a + * complete log from a truncated one by its length. + * - **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 + * response (see {@link EventResult.events}). The writer learns its + * snapshot was stale without the write being rejected — which is why no + * World needs a precondition guard. + * + * Allocating at the commit is what makes a reader's log a *prefix* of the + * run's log rather than a prefix with a hole in it. A World that hands a + * position out earlier, and can therefore let an event land behind one a + * reader has already passed, breaks the property every replay depends on. + */ events: { /** * Create a run_created event to start a new workflow run. @@ -342,24 +371,6 @@ export interface WorldCapabilities { active: boolean; }; - /** - * 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. - * - * 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; - /** * The World's queue supports `maxConcurrency`-limited consumption — in * particular the per-run flow topics consumed with `maxConcurrency: 1` @@ -421,29 +432,6 @@ export interface WorldCapabilities { * fail ordinary runs after a version bump. */ deploymentAffinity?: boolean; - - /** - * The World allocates **slot-numbered** event ids: `evnt_` plus the event's - * dense, 1-based position in its run's log, zero-padded to 26 characters - * (see `slot-identity.ts`). Two guarantees come with it, and the runtime - * relies on both: - * - * - **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 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 - * response (see {@link EventResult.events}). The writer learns its - * snapshot was stale without the write being rejected. - * - * A run's scheme is pinned by the run, not by this flag: it is readable off - * the shape of the run's own first event id, so a World that turns slots on - * keeps replaying its existing ULID-numbered runs unchanged. The capability - * only says what *new* runs get. - */ - slotEventIds?: boolean; } /** diff --git a/packages/world/src/slot-identity.ts b/packages/world/src/slot-identity.ts index a42283bba3..29fdb98616 100644 --- a/packages/world/src/slot-identity.ts +++ b/packages/world/src/slot-identity.ts @@ -98,3 +98,27 @@ export function eventIdToSlot(eventId: string): number | null { const slot = Number(body); return Number.isSafeInteger(slot) && slot >= FIRST_EVENT_SLOT ? slot : null; } + +/** + * Reads the slot out of an event id, for a caller that has no answer without + * one. + * + * Separate from {@link eventIdToSlot} because the two failures are different + * problems. A caller that can act on either scheme asks the question and takes + * `null` as an answer; a caller whose whole computation is positional (a + * precondition snapshot, a density audit) has no correct behaviour to fall back + * on, and silently skipping the id would make it report a position it never + * verified. Throwing names the id instead. + * + * @throws if the id is not slot-numbered, i.e. the World minting it does not + * allocate slots. + */ +export function requireEventSlot(eventId: string): number { + const slot = eventIdToSlot(eventId); + if (slot === null) { + throw new Error( + `Event id is not slot-numbered: ${eventId}. This World allocates event positions the runtime cannot read.` + ); + } + return slot; +} diff --git a/workbench/sim-world/README.md b/workbench/sim-world/README.md index ea48165e1d..2998a18f1b 100644 --- a/workbench/sim-world/README.md +++ b/workbench/sim-world/README.md @@ -165,20 +165,18 @@ Two of these are measurements rather than conveniences. **`--append-only`** moves every event's position from its handler's mint to its commit, which is the one change that makes a stale read impossible: the log can be behind, never wrong. Running with and without it is how you tell which of -the reds that change would actually close. Today: **35 pass / 6 violations** +the reds that change would actually close. Today: **38 pass / 3 violations** mint-ordered, **41 pass / 0 violations** append-only. -The one red it does *not* close is `unclaimed-payload-under-fork`, and that is -the point of it: no log position is wrong there, the runtime hands two -resolutions to the workflow in the order the log did not record. It is the only -scenario in the book that is red in both worlds. +It closes all three, and by construction rather than by catching anything: with +nothing reserving a position ahead of a commit, no read can see a hole. **`--no-fence`** turns the fence off everywhere, asking whether anything relies on it. It is a diagnostic, not a world — **read the violation count, not the pass count**, because a scenario whose whole point is that the guard fired asserts exactly that and fails by design when you disarm it (`in-flight-before-decision-counted` is the one that does this today). -Measured: **6 → 8** violations mint-ordered, so it is load-bearing there; +Measured: **3 → 5** violations mint-ordered, so it is load-bearing there; **0 → 0** append-only, so it is dead weight once positions are assigned at commit. 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 827a8aa660..e04a5309bc 100644 --- a/workbench/sim-world/scenarios/in-flight-before-decision-counted.ts +++ b/workbench/sim-world/scenarios/in-flight-before-decision-counted.ts @@ -17,14 +17,19 @@ export const scenario: ScenarioSpec = { '`@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 ' + - 'same virtual millisecond still counts as "at or below" it. The reload ' + - 'sees the hook behind the timeout in log order, re-decides the same way, ' + - 'and settles — a restart with nothing to correct. That is what the ' + - 'fence costs once the log is append-only: false positives at millisecond ' + - 'granularity, in exchange for a hole that can no longer open.', + 'Which half fires is no longer the point, though, and that is what slot ' + + 'positions changed. A watermark used to be a millisecond, so two writes ' + + 'inside one virtual millisecond compared equal and only the count could ' + + 'separate them; a watermark that is a slot is strictly ordered, so the ' + + 'watermark half now rejects this on its own. What the scenario still ' + + 'asserts is that the fence fires at all. ' + + 'Under an append-only log the 412 still fires and saves nothing: the ' + + 'reload sees the hook behind the timeout in log order, re-decides the ' + + 'same way, and settles — a restart with nothing to correct. ' + + 'Mint-ordered there is no fence to reach: the receiver holds a position ' + + 'ahead of everything the orchestrator writes, so the log has a hole in it ' + + 'while the write is in flight, and the next replay refuses the log ' + + 'outright rather than following it into the wrong branch.', workflow: 'stepCountForkWorkflow', input: ['doc-30'], preconditionGuard: true, @@ -39,23 +44,25 @@ export const scenario: ScenarioSpec = { await hook.commit(); await wf.release(); - // The point of the scenario, and the part that is true in both worlds: the - // count half of the fence fires where the watermark half did not. What the - // reload then decides is a different question and belongs to the world — - // mint-ordered it corrects the branch, append-only it re-confirms it — so - // that half is left to the trace. Asserting it here is what forced this - // scenario to carry two expectations, and it was never what distinguished - // it from its uncounted twin. - // Matched on the count half's own message, not on "something was - // rejected". The twin rejects too — its writes hit `RunExpiredError` once - // the corrupted branch has run — so a bare `rejections().length > 0` would - // hold there as well and assert nothing about the guard. - sim.check( - 'the count guard fenced the write the watermark let through', - sim.world - .rejections() - .some((r) => r.message.includes('at or below the caller')) - ); + // Matched on the error, not on "something was rejected". The twin rejects + // too — its writes hit `RunExpiredError` once the corrupted branch has run + // — so a bare `rejections().length > 0` would hold there as well and + // assert nothing about the fence. + // + // Only asserted under an append-only log, because only there does the + // write reach the fence. Mint-ordered, the receiver's reserved position is + // binding, so the log carries a hole for as long as the write is in + // flight, and the replay that reads it refuses the log before any write of + // its own is checked. That refusal is the violation the trace reports; a + // check here would restate it as a second failure. + if (sim.appendOnlyLog) { + sim.check( + 'the fence rejected the write', + sim.world + .rejections() + .some((r) => r.errorName === 'PreconditionFailedError') + ); + } }, // The rejection and the reload show up in the trace as `!!` lines. Whichever // branch the reload lands on, it is the one the durable log implies — so diff --git a/workbench/sim-world/scenarios/in-flight-before-decision.ts b/workbench/sim-world/scenarios/in-flight-before-decision.ts index 23259f7015..14764fc576 100644 --- a/workbench/sim-world/scenarios/in-flight-before-decision.ts +++ b/workbench/sim-world/scenarios/in-flight-before-decision.ts @@ -13,10 +13,11 @@ export const scenario: ScenarioSpec = { 'The receiver commits while the orchestrator is held at the produced ' + 'point of C, so by the time C is checked the hole has closed and the log ' + '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 " + - '`snapshot.updatedAt < marker` is false. It corrupts — the same corruption as ' + - 'the doc-23 pair, reached without a stale read. ' + + 'never gets to speak: the log has a hole in it from the moment the ' + + 'receiver takes its position, so the next replay refuses the log before ' + + 'any write of its own is checked. What the fence would have caught, the ' + + 'gap audit catches earlier and more bluntly — the run fails rather than ' + + 'following a log it cannot follow into the wrong branch. ' + '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 ' + 'timeout won, and the settle branch the run took is the one the log ' + diff --git a/workbench/sim-world/scenarios/stale-read-equal-step-counts.ts b/workbench/sim-world/scenarios/stale-read-equal-step-counts.ts index 0dda465206..dd66ca79aa 100644 --- a/workbench/sim-world/scenarios/stale-read-equal-step-counts.ts +++ b/workbench/sim-world/scenarios/stale-read-equal-step-counts.ts @@ -2,9 +2,9 @@ import type { ScenarioSpec } from '@workflow/world-sim'; export const scenario: ScenarioSpec = { id: 'stale-read-equal-step-counts', - name: 'corrupt: stale event load with EQUAL step counts (is the amplifier needed?)', + name: 'stale event load with EQUAL step counts (is the amplifier needed?)', description: - 'Identical fault to the scenario above, but on the fork whose branches ' + + 'Green since slot-numbered event ids: a read missing an event the log already holds is a gap in a numbered sequence, so the runtime re-reads and decides the fork the way the log records it. Kept as the regression test for that. Identical fault to the scenario above, but on the fork whose branches ' + 'each emit exactly one step. If this corrupts too then step-count ' + 'divergence raises the rate rather than being required.', workflow: 'hookTimeoutForkWorkflow', diff --git a/workbench/sim-world/scenarios/stale-read-step-count-fork.ts b/workbench/sim-world/scenarios/stale-read-step-count-fork.ts index 1281d00ee5..4182d5990f 100644 --- a/workbench/sim-world/scenarios/stale-read-step-count-fork.ts +++ b/workbench/sim-world/scenarios/stale-read-step-count-fork.ts @@ -2,9 +2,9 @@ import type { ScenarioSpec } from '@workflow/world-sim'; export const scenario: ScenarioSpec = { id: 'stale-read-step-count-fork', - name: 'corrupt: stale event load + step-count fork', + name: 'stale event load + step-count fork', description: - 'All three preconditions from PR #3147 at once. The hook is committed ' + + 'Green since slot-numbered event ids: a read missing an event the log already holds is a gap in a numbered sequence, so the runtime re-reads and decides the fork the way the log records it. Kept as the regression test for that. All three preconditions from PR #3147 at once. The hook is committed ' + 'ahead of wait_completed in the log, but withheld from the read the live ' + 'pass uses — so the live pass decides the fork without it, while the ' + 'durable log says the hook came first. The branches differ by step count, ' + 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 59925e0a5b..593de68b10 100644 --- a/workbench/sim-world/scenarios/step-vs-step-fork-fenced.ts +++ b/workbench/sim-world/scenarios/step-vs-step-fork-fenced.ts @@ -2,35 +2,31 @@ import type { ScenarioSpec } from '@workflow/world-sim'; export const scenario: ScenarioSpec = { id: 'step-vs-step-fork-fenced', - name: 'corrupt: two racing STEPS, WITH the precondition fence on', + name: 'two racing STEPS, WITH the precondition fence on', description: - "Tests the fence's predicate. WorldCapabilities.preconditionGuard is " + - 'documented as rejecting a stale write when a newer OUT-OF-BAND event ' + - '(e.g. a received hook) was recorded. So does it fence a write made ' + - "stale by one of the run's OWN step_completed events? Same fault as the " + - 'scenario above, fence enabled. Verified answer: NO — zero ' + - '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 `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 ' + - 'millisecond-granularity tie. Contrast the hook/wait variant above, ' + - 'where the withheld hook IS the newest out-of-band write and the ' + - 'orchestrator carries a pre-sleep snapshot: strictly older, so the same ' + - 'fence rejects twice and the run self-corrects — those rejections show ' + - 'up in the trace as `!!` lines, unasked for. ' + + "Tests the fence's predicate, and is green: a green regression test now " + + 'rather than an open reproduction. Slot-numbered event ids closed the ' + + 'fault — a read missing an event the log already holds is a gap in a ' + + 'numbered sequence, so the runtime re-reads and decides the fork the way ' + + 'the log records it, without anything having to be refused. ' + + 'What the scenario still pins is the predicate, which does NOT catch this ' + + 'shape. The watermark half compares the write against a high-water mark of ' + + 'the newest out-of-band write, and refuses only a snapshot strictly below ' + + 'it. 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 millisecond-granularity tie. Contrast the hook/wait ' + + 'variant above, where the withheld hook IS the newest out-of-band write ' + + 'and the orchestrator carries a pre-sleep snapshot. ' + '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. ' + - '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.', + 'Neither half models a shipped World: none of them refuses a stale write ' + + 'at all, because a reader holds a prefix rather than a hole and its next ' + + 'write comes back carrying what it was pushed past. What arming the fence ' + + 'still buys is coverage of the 412 reception path the runtime keeps for a ' + + 'World that would rather refuse than report.', workflow: 'stepVsStepForkWorkflow', input: ['doc-27'], preconditionGuard: true, diff --git a/workbench/sim-world/scenarios/step-vs-step-fork.ts b/workbench/sim-world/scenarios/step-vs-step-fork.ts index 71e11394b8..91b82f3fcd 100644 --- a/workbench/sim-world/scenarios/step-vs-step-fork.ts +++ b/workbench/sim-world/scenarios/step-vs-step-fork.ts @@ -2,9 +2,9 @@ import type { ScenarioSpec } from '@workflow/world-sim'; export const scenario: ScenarioSpec = { id: 'step-vs-step-fork', - name: 'corrupt: two racing STEPS, no hook anywhere', + name: 'two racing STEPS, no hook anywhere', description: - 'Answers "does this need an out-of-band event type?" — no. The fork is ' + + 'Green since slot-numbered event ids: a read missing an event the log already holds is a gap in a numbered sequence, so the runtime re-reads and decides the fork the way the log records it. Kept as the regression test for that. Answers "does this need an out-of-band event type?" — no. The fork is ' + "decided by two of the run's own step_completed events, and withholding " + 'one of them from the deciding read is enough. Two inline step bodies in ' + 'ONE invocation are already two concurrent writers to the same log; no ' +