From 0161a145cbf093fb8ced949f293b2a3fabd79f87 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Wed, 12 Aug 2026 09:45:49 -0700 Subject: [PATCH 1/2] [world-local] Fix hook-resume slot holes and spurious resume conflicts `hook_received` is the only event that does not publish straight into `events/`: it stages under `.locks` first so a terminal transition can reap it before it becomes reader-visible. That staging path was keyed by the event id alone, and a collision on it bumped the writer to the next slot. Under slot ids a staging collision is not evidence the position is taken. The allocator probes `events/` only, so bumping moves the writer past a position nothing will ever fill, and `scanRunEventIds` is max-based so no later writer backfills it. The runtime reads the missing position as a durable hole and fails the run with CORRUPTED_EVENT_LOG. Two triggers: an attempt killed between staging and promoting leaves its staged file behind (cleanup lives in a `finally` the kill skips, and the only other reaper runs on a terminal transition), and two live writers drawing the same candidate where the stager is later rejected. Staging now carries a nonce, so it is private to one attempt and the slot is arbitrated only where it is actually taken, at the promote. Second fix: when a pinned resume loses the publish, the event at the pinned position is that same resume written by the other taker, which is the convergence the pin exists to force. Return that committed event instead of an EntityConflictError the caller cannot act on. Also re-purposes scripts/event-log-race-repro-local.sh to drive either world with `--world postgres|local`. --- .changeset/world-local-hook-staging-slots.md | 5 + AGENTS.md | 43 ++++- .../world-local/src/storage/events-storage.ts | 55 +++++- .../src/storage/hook-staging-slots.test.ts | 177 ++++++++++++++++++ scripts/event-log-race-repro-local.sh | 150 +++++++++++---- 5 files changed, 378 insertions(+), 52 deletions(-) create mode 100644 .changeset/world-local-hook-staging-slots.md create mode 100644 packages/world-local/src/storage/hook-staging-slots.test.ts diff --git a/.changeset/world-local-hook-staging-slots.md b/.changeset/world-local-hook-staging-slots.md new file mode 100644 index 0000000000..6107b901d4 --- /dev/null +++ b/.changeset/world-local-hook-staging-slots.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-local': patch +--- + +Fix `CORRUPTED_EVENT_LOG` after a hook resume that raced another writer or was interrupted mid-write, and return the committed event to both writers of one resume instead of a conflict. diff --git a/AGENTS.md b/AGENTS.md index b17ffccad2..83324adbc3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -137,20 +137,30 @@ control that provides the calibration baseline. Any outcome other than `completed` fails the run, except `infra`, which means the harness could not reach the deployment. -Run it locally against `@workflow/world-postgres` and a locally started -workbench app — no Vercel deployment, no credentials: +Run it locally against a locally started workbench app — no Vercel deployment, +no credentials: ```bash -pnpm run test:e2e:event-log-race-repro:local +pnpm run test:e2e:event-log-race-repro:local # world-postgres +pnpm run test:e2e:event-log-race-repro:local --world local # world-local ``` -The script (`scripts/event-log-race-repro-local.sh`, `--help` for flags) brings up -the world-postgres container, applies migrations, builds and starts -`workbench/nextjs-turbopack` with `WORKFLOW_TARGET_WORLD` and +The script (`scripts/event-log-race-repro-local.sh`, `--help` for flags) builds +and starts `workbench/nextjs-turbopack` with `WORKFLOW_TARGET_WORLD` and `WORKFLOW_PUBLIC_MANIFEST=1` set **at build time** (both are build-time inputs; -missing either silently yields a world-local app or a 404 manifest), runs the -harness, prints the same summary table CI posts, and tears the server down. -Postgres is left running for the next iteration unless `--teardown` is passed. +missing either silently yields a default-world app or a 404 manifest), runs the +harness, prints the same summary table CI posts, and tears the server down. For +world-postgres it first brings up the container and applies migrations, and +leaves Postgres running for the next iteration unless `--teardown` is passed; +the container flags (`--skip-db-setup`, `--no-docker`, `--teardown`) do nothing +under `--world local`, whose only state is a data directory the script clears +before each run. + +Both worlds are worth running, and neither subsumes the other: world-postgres +arbitrates event slots inside one SQL statement, while world-local arbitrates +them with an exclusive `link(2)` against a directory that two processes (the app +and the harness) both write to. A slot race a transaction closes is not +automatically closed by a filesystem. Scale is controlled entirely by `EVENT_LOG_RACE_REPRO_*` environment variables. Their defaults live only in `event-log-race-repro.test.ts` — neither the CI @@ -185,6 +195,17 @@ one heap saturates GC — measured on a 12-core laptop, all 14 attempts came bac sets `WORKFLOW_POSTGRES_WORKER_CONCURRENCY=10` (override by exporting it) and raises the app's old-space limit (`--heap-mb`). If a local run reports `stuck` rather than `CORRUPTED_EVENT_LOG`, suspect the machine before the SDK. +world-local saturates the same single process from its own in-process queue, +which defaults to 1000 deliveries in flight, so the script holds it at the same +number via `WORKFLOW_LOCAL_QUEUE_CONCURRENCY`. + +world-local's storms come out clean far more often than world-postgres's, so the +default scale says even less there: the corruption it does produce needs a +`hook_received` to be staged and then rejected, which the harness reaches only +in a run's terminal moments. Reach for a unit test in +`packages/world-local/src/storage/` when a suspected filesystem race can be +staged directly — it costs milliseconds and does not depend on the interleaving +showing up. 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 @@ -198,6 +219,10 @@ To poke at a run afterwards, the CLI reads the same world from the environment: WORKFLOW_TARGET_WORLD=@workflow/world-postgres \ WORKFLOW_POSTGRES_URL=postgres://world:world@localhost:5432/world \ pnpm wf inspect + +WORKFLOW_TARGET_WORLD=local \ +WORKFLOW_LOCAL_DATA_DIR=workbench/nextjs-turbopack/.next/workflow-data \ + pnpm wf inspect ``` ### Example App Development diff --git a/packages/world-local/src/storage/events-storage.ts b/packages/world-local/src/storage/events-storage.ts index fae85024e8..4a6ded5d97 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -2734,23 +2734,28 @@ export function createEventsStorage( ); } + // Staging is private to this attempt, so the name carries a + // nonce rather than only the event id. Sharing a name across + // writers would make the staging directory a second, invisible + // claim on the slot: the allocator probes `events/` alone, so a + // staged file is not evidence the position is taken, and bumping + // off it moves this writer past a position no one will ever + // publish. A crashed attempt (its cleanup lives in a `finally` + // the kill skips) would hole the log permanently that way, and + // two live writers drawing the same candidate would hole it + // whenever the stager is later rejected. The slot is arbitrated + // where it is actually taken: the promote below. const stagedPath = pendingHookEventPath( basedir, effectiveRunId, - eventId, + `${eventId}.${monotonicUlid()}`, tag ); const staged = await writeExclusive(stagedPath, serializedEvent); if (!staged) { - // The staging path can be occupied by a previous crashed - // attempt of this very event (which never promoted), or, under - // slot ids, by a concurrent writer holding the same position. - // Both are handled the same way: fall through to the bump - // below, which moves off the position when it can and surfaces - // the conflict when it cannot. - if (await bumpEventSlot(attempt)) { - continue; - } + // A nonced path cannot already exist. Surfacing rather than + // retrying keeps a filesystem fault from looking like a + // publish that did not happen. throw new EntityConflictError( `Event "${eventId}" already exists for run "${effectiveRunId}"` ); @@ -2811,6 +2816,36 @@ export function createEventsStorage( tag ); } + // For a resume, losing this publish is the convergence the pinned + // id exists to force: both takers of one resume adopt the claim's + // position, so the event now at it is this same resume, written by + // the other taker. That is the case `converge` above already + // answers with the committed event — it just could not see it yet, + // because the other taker had not published when this attempt + // read. Answer it the same way rather than reporting a conflict + // the caller cannot act on: the resume IS committed, exactly once, + // and the dedup contract is that both writers return that event. + if (eventIdPinned && data.eventType === 'hook_received') { + const occupant = await readJSONWithFallback( + basedir, + 'events', + `${effectiveRunId}-${eventId}`, + EventSchema, + tag + ); + if ( + occupant && + params?.resumeId && + isResumeEvent(occupant, { + runId: effectiveRunId, + resumeId: params.resumeId, + hookId: data.correlationId, + eventId, + }) + ) { + return { event: occupant }; + } + } throw new EntityConflictError( `Event "${eventId}" already exists for run "${effectiveRunId}"` ); diff --git a/packages/world-local/src/storage/hook-staging-slots.test.ts b/packages/world-local/src/storage/hook-staging-slots.test.ts new file mode 100644 index 0000000000..3240b597af --- /dev/null +++ b/packages/world-local/src/storage/hook-staging-slots.test.ts @@ -0,0 +1,177 @@ +import { promises as fs } from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { + eventIdToSlot, + FIRST_EVENT_SLOT, + SPEC_VERSION_CURRENT, + type Storage, +} from '@workflow/world'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { createHook, createRun } from '../test-helpers.js'; +import { hookResumeClaimPath, pendingHookEventPath } from './helpers.js'; +import { createStorage } from './index.js'; + +// `hook_received` is the one event that does not publish straight into +// `events/`: it stages the file under `.locks` first, so a terminal +// transition can reap it before it ever becomes reader-visible. The staging +// path is the only place a slot can be held OUTSIDE `events/`, and the slot +// allocator only probes `events/`. So a staging collision is not evidence +// that the slot is taken, and treating it as one moves the writer off a +// position nothing will ever fill. +describe('world-local hook_received staging and slot density', () => { + let testDir: string; + let storage: Storage; + + beforeEach(async () => { + testDir = await fs.mkdtemp(path.join(os.tmpdir(), 'hook-staging-test-')); + storage = createStorage(testDir); + }); + + afterEach(async () => { + await fs.rm(testDir, { recursive: true, force: true }); + }); + + async function setup() { + const run = await createRun(storage, { + deploymentId: 'dpl_test', + workflowName: 'test-workflow', + input: new Uint8Array(), + }); + const hook = await createHook(storage, run.runId, { + hookId: 'hook_1', + token: 'order:1', + }); + return { runId: run.runId, hook }; + } + + function resume( + runId: string, + hook: { hookId: string; token: string }, + resumeId: string, + payload: Uint8Array + ) { + return storage.events.create( + runId, + { + eventType: 'hook_received', + specVersion: SPEC_VERSION_CURRENT, + correlationId: hook.hookId, + eventData: { token: hook.token, payload }, + }, + { resumeId, resumePayloadDigest: resumeId } + ); + } + + async function slots(runId: string): Promise { + const { data } = await storage.events.list({ + runId, + pagination: { limit: 1000 }, + }); + return data.map((event) => eventIdToSlot(event.eventId) ?? -1); + } + + it('leaves no hole when a crashed attempt still holds the staging path', async () => { + const { runId, hook } = await setup(); + + // A process killed between staging and promoting leaves its staged file + // behind: the cleanup lives in a `finally` the kill skips, and the only + // other reaper runs on a terminal transition this run has not reached. + // The file names the slot the crashed attempt drew, which is the slot the + // next writer draws too, because `events/` never saw it. + const drawn = (await slots(runId)).length + FIRST_EVENT_SLOT; + const stale = pendingHookEventPath( + testDir, + runId, + `evnt_${String(drawn).padStart(26, '0')}` + ); + await fs.mkdir(path.dirname(stale), { recursive: true }); + await fs.writeFile(stale, '{}'); + + await resume(runId, hook, 'resume_1', new Uint8Array([1])); + + // Slot ids are positions, so the log is only readable if it is dense: the + // runtime reads a missing position as a durable hole and fails the run + // with CORRUPTED_EVENT_LOG. + const published = await slots(runId); + expect(published).toEqual( + published.map((_, index) => index + FIRST_EVENT_SLOT) + ); + }); + + it('returns the committed event to the loser of a raced resume', async () => { + const { runId, hook } = await setup(); + const other = createStorage(testDir); + + // A claim with no event behind it is the crash window the adoption path + // exists for: its writer recorded where it meant to append and died. Both + // takers below therefore adopt that position instead of drawing their own. + const drawn = (await slots(runId)).length + FIRST_EVENT_SLOT; + await fs.mkdir( + path.dirname(hookResumeClaimPath(testDir, runId, 'resume_1')), + { + recursive: true, + } + ); + await fs.writeFile( + hookResumeClaimPath(testDir, runId, 'resume_1'), + JSON.stringify({ + runId, + resumeId: 'resume_1', + hookId: hook.hookId, + eventId: `evnt_${String(drawn).padStart(26, '0')}`, + payloadDigest: 'resume_1', + }) + ); + + // Both writers of one resume adopt the claim's position, so the loser of + // the publish finds the winner's event there. That is the convergence the + // adoption exists to force, and the dedup contract is that both writers + // return the one committed event — reporting a conflict instead leaves + // the caller with an error it cannot act on for a resume that did land. + const [first, second] = await Promise.all([ + resume(runId, hook, 'resume_1', new Uint8Array([1])), + other.events.create( + runId, + { + eventType: 'hook_received', + specVersion: SPEC_VERSION_CURRENT, + correlationId: hook.hookId, + eventData: { token: hook.token, payload: new Uint8Array([1]) }, + }, + { resumeId: 'resume_1', resumePayloadDigest: 'resume_1' } + ), + ]); + + expect(second.event.eventId).toBe(first.event.eventId); + const { data } = await storage.events.list({ runId }); + expect(data.filter((e) => e.eventType === 'hook_received')).toHaveLength(1); + }); + + it('keeps the log dense when two instances resume the same hook at once', async () => { + const { runId, hook } = await setup(); + // A second instance over the same directory is the configuration this + // backend supports for the CLI plus the app. Its allocator watermark is + // its own, so both instances hand out the same candidate position. + const other = createStorage(testDir); + + await Promise.all([ + resume(runId, hook, 'resume_a', new Uint8Array([1])), + other.events.create( + runId, + { + eventType: 'hook_received', + specVersion: SPEC_VERSION_CURRENT, + correlationId: hook.hookId, + eventData: { token: hook.token, payload: new Uint8Array([2]) }, + }, + { resumeId: 'resume_b', resumePayloadDigest: 'resume_b' } + ), + ]); + + const published = await slots(runId); + expect(published).toEqual( + published.map((_, index) => index + FIRST_EVENT_SLOT) + ); + }); +}); diff --git a/scripts/event-log-race-repro-local.sh b/scripts/event-log-race-repro-local.sh index ab12c22ed3..fa9371072d 100755 --- a/scripts/event-log-race-repro-local.sh +++ b/scripts/event-log-race-repro-local.sh @@ -1,7 +1,8 @@ #!/usr/bin/env bash # -# Run the event-log race repro harness locally, against @workflow/world-postgres -# and a workbench app started on this machine. No Vercel deployment, no GitHub +# Run the event-log race repro harness locally, against a workbench app started +# on this machine and either @workflow/world-postgres (the default) or +# @workflow/world-local (--world local). No Vercel deployment, no GitHub # Actions, no VERCEL_* credentials. # # This is the same harness `.github/workflows/event-log-race-repro.yml` runs @@ -16,12 +17,18 @@ # defaults it to `local` when VERCEL_DEPLOYMENT_ID is absent, which bakes the # filesystem backend into the app and leaves the harness talking to a # different world than the app is. -# * The schema has to exist before the app boots, or Graphile Worker starts -# against an unmigrated database. -# * The queue has to be empty before the app boots. Postgres outlives the app -# here, so an interrupted run leaves its unfinished flow messages behind and -# the next boot resumes all of them — thousands of stale jobs starving the -# run you actually care about. See clear_queue below. +# * (postgres) The schema has to exist before the app boots, or Graphile Worker +# starts against an unmigrated database. +# * The backend's pending work has to be empty before the app boots. Both +# backends outlive the app here, so an interrupted run leaves its unfinished +# flow messages behind and the next boot resumes all of them — stale work +# starving the run you actually care about. See clear_pending below. +# +# Both worlds are worth running. They fail differently, and neither subsumes the +# other: world-postgres arbitrates event slots inside one SQL statement, while +# world-local arbitrates them with an exclusive `link(2)` against a directory +# that two processes (the app and this harness) both write to. A slot race that +# a transaction closes is not automatically closed by a filesystem. # # Usage: scripts/event-log-race-repro-local.sh [options] # Run with --help for options. @@ -39,6 +46,8 @@ SKIP_DB_SETUP="0" USE_DOCKER="1" TEARDOWN="0" KEEP_QUEUE="0" +# Which World the app and the harness both talk to: `postgres` or `local`. +WORLD="postgres" # In CI every replay of a run lands in its own Fluid invocation; here they all # land in one Next.js process, and a storm run has tens of replays in flight at # once, each holding a VM sandbox and its own copy of the event log. That is @@ -57,6 +66,11 @@ SERVER_HEAP_MB="8192" # replays per process does not weaken the repro: the race is between a handful of # concurrent replays of one run, not a throughput effect. LOCAL_WORKER_CONCURRENCY="10" +# The world-local equivalent. Its queue is in-process timers driving HTTP +# deliveries, and it defaults to 1000 in flight, which saturates the one +# Next.js process the same way the Graphile default does. Held at the same +# number as the postgres lane so the two are comparable. +LOCAL_QUEUE_CONCURRENCY="10" COMPOSE_FILE="packages/world-postgres/docker-compose.yaml" # Matches the compose file and the `e2e-local-postgres` job in tests.yml. @@ -67,9 +81,18 @@ RENDERER=".github/scripts/render-event-log-race-repro-results.js" usage() { cat <<'EOF' -Run the event-log race repro harness against world-postgres + a local workbench app. +Run the event-log race repro harness against a local workbench app, backed by +either world-postgres (default) or world-local. Options: + --world NAME Backend World: `postgres` (default) or `local`. + `local` needs no Docker and no database: the app and this + harness share the app's filesystem data directory, and + --skip-db-setup / --no-docker / --teardown are ignored. + The two worlds arbitrate event slots by different means + (one SQL statement vs. an exclusive link(2) between two + processes), so a clean run on one says nothing about the + other. --app NAME Workbench app to drive (default: nextjs-turbopack). The repro workflow fixtures (101_hook_sleep_repro.ts, 103_event_log_corruption_repro.ts) only exist in @@ -84,17 +107,18 @@ Options: Faster to iterate on workflow fixtures; less like CI. --skip-build Skip `pnpm build` and the app build. Use when only the harness or the driver changed. - --skip-db-setup Skip applying migrations (schema already set up). - --no-docker Do not manage Postgres. Point WORKFLOW_POSTGRES_URL at your - own instance; the schema still needs to exist. - --keep-queue Leave queued Graphile Worker jobs in place. By default they - are deleted before the app boots: an interrupted earlier run - leaves its flow messages queued, and resuming those abandoned - runs saturates the app so this run reports `stuck` for - reasons that have nothing to do with the event log. Only + --skip-db-setup (postgres) Skip applying migrations (schema already set up). + --no-docker (postgres) Do not manage Postgres. Point WORKFLOW_POSTGRES_URL + at your own instance; the schema still needs to exist. + --keep-queue Leave the backend's pending work in place. By default it is + discarded before the app boots: an interrupted earlier run + leaves its flow messages queued (postgres) or its runs + pending in the data directory (local), and resuming those + abandoned runs saturates the app so this run reports `stuck` + for reasons that have nothing to do with the event log. Only useful if you are inspecting the leftovers themselves. - --teardown Stop and delete the Postgres container on exit. Off by - default so repeat runs skip container startup. + --teardown (postgres) Stop and delete the Postgres container on exit. + Off by default so repeat runs skip container startup. -h, --help Show this help. Scale knobs are read from the environment by the harness itself, so anything @@ -111,16 +135,23 @@ This script deliberately sets none of them. Their defaults — and the full list live in packages/core/e2e/event-log-race-repro.test.ts, which is the single source of truth the CI workflow also defers to. -The one knob this script does set is WORKFLOW_POSTGRES_WORKER_CONCURRENCY, which -caps how many replays the app and the harness each run at once. world-postgres -defaults it to 50, i.e. ~100 replays sharing one Next.js process, which on a -12-core laptop saturates GC and reports all 14 attempts as `stuck`. It is set to -10 here instead, matching the pool size. Export your own value to override. +The one knob this script does set is the backend's replay concurrency, which +caps how many replays the app and the harness each run at once. Both backends +default it far too high for one Next.js process on one machine (50 for +world-postgres, i.e. ~100 replays in flight; 1000 for world-local), which on a +12-core laptop saturates GC and reports every attempt as `stuck`. Both are set +to 10 here. Export your own value to override: + + WORKFLOW_POSTGRES_WORKER_CONCURRENCY (--world postgres) + WORKFLOW_LOCAL_QUEUE_CONCURRENCY (--world local) Examples: # Default scale (a regression check, a few minutes). scripts/event-log-race-repro-local.sh + # Same, against world-local. No Docker, no database. + scripts/event-log-race-repro-local.sh --world local + # One run per scenario, to smoke the plumbing. EVENT_LOG_RACE_REPRO_STEP_STORM_ATTEMPTS=1 \ EVENT_LOG_RACE_REPRO_HOOK_STORM_ATTEMPTS=1 \ @@ -139,6 +170,7 @@ EOF while [ $# -gt 0 ]; do case "$1" in + --world) WORLD="${2:?--world needs a value}"; shift 2 ;; --app) APP_NAME="${2:?--app needs a value}"; shift 2 ;; --port) PORT="${2:?--port needs a value}"; shift 2 ;; --heap-mb) SERVER_HEAP_MB="${2:?--heap-mb needs a value}"; shift 2 ;; @@ -167,10 +199,39 @@ for fixture in 101_hook_sleep_repro.ts 103_event_log_corruption_repro.ts; do die "$APP_DIR/workflows/$fixture is missing — the repro fixtures live in workbench/nextjs-turbopack." done -export WORKFLOW_POSTGRES_URL="${WORKFLOW_POSTGRES_URL:-$DEFAULT_POSTGRES_URL}" -export WORKFLOW_TARGET_WORLD="@workflow/world-postgres" +case "$WORLD" in + postgres|local) ;; + *) die "Unknown --world: $WORLD (expected \`postgres\` or \`local\`)" ;; +esac + export WORKFLOW_PUBLIC_MANIFEST="1" -export WORKFLOW_POSTGRES_WORKER_CONCURRENCY="${WORKFLOW_POSTGRES_WORKER_CONCURRENCY:-$LOCAL_WORKER_CONCURRENCY}" + +if [ "$WORLD" = "postgres" ]; then + export WORKFLOW_POSTGRES_URL="${WORKFLOW_POSTGRES_URL:-$DEFAULT_POSTGRES_URL}" + export WORKFLOW_TARGET_WORLD="@workflow/world-postgres" + export WORKFLOW_POSTGRES_WORKER_CONCURRENCY="${WORKFLOW_POSTGRES_WORKER_CONCURRENCY:-$LOCAL_WORKER_CONCURRENCY}" +else + # Postgres is a service both processes address by URL; world-local is a + # directory both processes address by path, so the path has to be pinned + # explicitly and identically or the two silently use different backends. + # + # `withWorkflow()` pairs its `local` default with `.next/workflow-data`, but + # only when WORKFLOW_TARGET_WORLD is unset — naming the world here suppresses + # the data-dir half of that pair, and the app would fall back to + # `.workflow-data` while the harness (`setupWorld` in packages/core/e2e/utils.ts) + # went on using `.next/workflow-data`. Both halves are therefore set here, as + # an absolute path so the app's cwd does not enter into it. `setupWorld` + # recomputes the same path for the harness process. + case "$APP_NAME" in + nextjs*|next-*) DATA_DIR_NAME=".next/workflow-data" ;; + *) DATA_DIR_NAME=".workflow-data" ;; + esac + DATA_DIR="$REPO_ROOT/$APP_DIR/$DATA_DIR_NAME" + export WORKFLOW_TARGET_WORLD="local" + export WORKFLOW_LOCAL_DATA_DIR="$DATA_DIR" + export WORKFLOW_LOCAL_QUEUE_CONCURRENCY="${WORKFLOW_LOCAL_QUEUE_CONCURRENCY:-$LOCAL_QUEUE_CONCURRENCY}" +fi + export PORT="$PORT" DEPLOYMENT_URL="http://localhost:$PORT" MANIFEST_URL="$DEPLOYMENT_URL/.well-known/workflow/v1/manifest.json" @@ -215,7 +276,7 @@ cleanup() { [ -z "$(port_pids)" ] || log "Port $PORT is still held by $(port_pids | tr '\n' ' ')— the next run needs --port or a manual kill." fi - if [ "$TEARDOWN" = "1" ] && [ "$USE_DOCKER" = "1" ]; then + if [ "$WORLD" = "postgres" ] && [ "$TEARDOWN" = "1" ] && [ "$USE_DOCKER" = "1" ]; then log "Removing the Postgres container" docker compose -f "$COMPOSE_FILE" down -v >/dev/null 2>&1 fi @@ -226,7 +287,9 @@ trap cleanup EXIT # --- postgres ----------------------------------------------------------------- -if [ "$USE_DOCKER" = "1" ]; then +if [ "$WORLD" != "postgres" ]; then + log "Backend: world-local at $WORKFLOW_LOCAL_DATA_DIR" +elif [ "$USE_DOCKER" = "1" ]; then command -v docker >/dev/null 2>&1 || die "docker not found. Install Docker, or use --no-docker with your own Postgres." log "Starting Postgres ($COMPOSE_FILE)" docker compose -f "$COMPOSE_FILE" up -d @@ -244,7 +307,7 @@ else log "Using the Postgres at WORKFLOW_POSTGRES_URL (not managed by this script)" fi -# --- queue --------------------------------------------------------------------- +# --- pending work --------------------------------------------------------------- # Runs a single statement as the `world` user. Prefers the container's own psql so # no local Postgres client is required. @@ -274,8 +337,29 @@ clear_queue() { log "Discarded $count queued job(s) left over from an earlier run ($table)" } +# world-local's queue is in-process, so nothing survives the app exiting — but +# the runs do. `start()` re-enqueues every pending/running run it finds in the +# data directory on boot (see resolveRecoverActiveRuns), so an interrupted +# earlier run resumes its abandoned storms alongside this one. Deleting the +# directory is the equivalent of emptying the jobs table, and it also keeps the +# per-run event files from accumulating across runs, which slows every +# directory scan the slot allocator makes. +clear_data_dir() { + [ -d "$WORKFLOW_LOCAL_DATA_DIR" ] || return 0 + local runs + runs="$(find "$WORKFLOW_LOCAL_DATA_DIR/runs" -name '*.json' 2>/dev/null | wc -l | tr -d ' ')" + rm -rf "$WORKFLOW_LOCAL_DATA_DIR" || + die "Could not remove $WORKFLOW_LOCAL_DATA_DIR. Pass --keep-queue to skip this." + [ "${runs:-0}" -gt 0 ] || return 0 + # Loud on purpose: a backlog here means the previous run was interrupted, which + # is worth knowing when comparing results between runs. + log "Discarded $runs run(s) left over from an earlier run ($WORKFLOW_LOCAL_DATA_DIR)" +} + if [ "$KEEP_QUEUE" = "0" ]; then - if [ "$USE_DOCKER" = "0" ] && ! command -v psql >/dev/null 2>&1; then + if [ "$WORLD" = "local" ]; then + clear_data_dir + elif [ "$USE_DOCKER" = "0" ] && ! command -v psql >/dev/null 2>&1; then log "psql not found — leaving the queue as it is. Stale jobs from an interrupted run will compete with this one." else clear_queue @@ -291,7 +375,7 @@ if [ "$SKIP_BUILD" = "0" ]; then pnpm build fi -if [ "$SKIP_DB_SETUP" = "0" ]; then +if [ "$WORLD" = "postgres" ] && [ "$SKIP_DB_SETUP" = "0" ]; then log "Applying the world-postgres schema" ./packages/world-postgres/bin/setup.js fi @@ -376,7 +460,7 @@ else log "No $RESULTS_FILE was written — the harness died before its first checkpoint." fi -if [ "$USE_DOCKER" = "1" ] && [ "$TEARDOWN" = "0" ]; then +if [ "$WORLD" = "postgres" ] && [ "$USE_DOCKER" = "1" ] && [ "$TEARDOWN" = "0" ]; then log "Postgres is still running. Stop it with: docker compose -f $COMPOSE_FILE down -v" fi From c6bf782a93e83b9f383573a90c80142304a22806 Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Wed, 12 Aug 2026 12:46:09 -0700 Subject: [PATCH 2/2] [world-local] Converge a raced resume before bumping, not after the loop Only one of the two takers of a resume claim is pinned. The taker that writes the claim keeps its own id, unpinned, because a slot is a position another instance also hands out for unrelated events and refusing to move would fail the append outright. The taker that adopts an existing claim is pinned to the claimed position. So the loser of the promote can be the unpinned owner, and a loser that bumps publishes a second `hook_received` for one resumeId. The log stays dense and both callers report success, but replay delivers the resume twice, which is what the dedup contract forbids. Run the occupant convergence inside the publish loop, before the bump, for any `hook_received` carrying a resumeId. An occupant that is not this resume is the unrelated-event collision the bump is for and still bumps, or conflicts when pinned. The post-loop pinned check is now redundant. Also: - staging-path faults raise WorkflowWorldError rather than EntityConflictError, which the runtime absorbs as a benign duplicate. - the repro script's app-name case mirrors `setupWorld`'s substring match so a future `example-nextjs` cannot split-brain the data dir. - the live-vs-live density test says what it actually guards: it passes without the fix, because a bumped-off position still gets published by the stager that won it. --- .changeset/world-local-hook-staging-slots.md | 2 +- .../world-local/src/storage/events-storage.ts | 85 ++++++++++------- .../src/storage/hook-staging-slots.test.ts | 93 ++++++++++++++++++- scripts/event-log-race-repro-local.sh | 7 +- 4 files changed, 145 insertions(+), 42 deletions(-) diff --git a/.changeset/world-local-hook-staging-slots.md b/.changeset/world-local-hook-staging-slots.md index 6107b901d4..bc51248b27 100644 --- a/.changeset/world-local-hook-staging-slots.md +++ b/.changeset/world-local-hook-staging-slots.md @@ -2,4 +2,4 @@ '@workflow/world-local': patch --- -Fix `CORRUPTED_EVENT_LOG` after a hook resume that raced another writer or was interrupted mid-write, and return the committed event to both writers of one resume instead of a conflict. +Fix `CORRUPTED_EVENT_LOG` after a hook resume that raced another writer or was interrupted mid-write, and deliver a raced resume exactly once instead of duplicating it or reporting a conflict. diff --git a/packages/world-local/src/storage/events-storage.ts b/packages/world-local/src/storage/events-storage.ts index 4a6ded5d97..b9e59fc066 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -2753,11 +2753,13 @@ export function createEventsStorage( ); const staged = await writeExclusive(stagedPath, serializedEvent); if (!staged) { - // A nonced path cannot already exist. Surfacing rather than - // retrying keeps a filesystem fault from looking like a - // publish that did not happen. - throw new EntityConflictError( - `Event "${eventId}" already exists for run "${effectiveRunId}"` + // A nonced path cannot already exist, so this is a filesystem + // fault rather than a lost race. It is deliberately NOT an + // `EntityConflictError`: the runtime reads that as a benign + // duplicate publish and carries on, which would absorb infra + // trouble as "someone else already wrote it". + throw new WorkflowWorldError( + `Failed to stage event "${eventId}" for run "${effectiveRunId}": staging path already exists` ); } try { @@ -2790,6 +2792,45 @@ export function createEventsStorage( notePublishedSlot(effectiveRunId, eventId); break; } + // Losing this publish to THIS SAME resume is the convergence the + // claim exists to force, and it has to be answered before the bump + // rather than after the loop, because only one of the two takers of + // a claim is pinned. The taker that wrote the claim keeps its own + // (unpinned) id, since a slot is a position another instance hands + // out for unrelated events too; the taker that adopts the claim is + // pinned to the claimed position. So whichever one loses the + // promote, the loser may be the unpinned owner, and bumping it + // publishes a SECOND hook_received for one resumeId: the log stays + // dense, and replay delivers the resume twice. + // + // `converge` above already answers this case with the committed + // event — it just could not see it yet, because the other taker had + // not published when this attempt read. Answer it the same way. An + // occupant that is NOT this resume is the unrelated-event collision + // the bump is for, and still bumps (or conflicts, when pinned). + if (data.eventType === 'hook_received' && params?.resumeId) { + const occupant = await readJSONWithFallback( + basedir, + 'events', + `${effectiveRunId}-${eventId}`, + EventSchema, + tag + ); + if ( + occupant && + isResumeEvent(occupant, { + runId: effectiveRunId, + resumeId: params.resumeId, + hookId: data.correlationId, + eventId, + }) + ) { + // The claim already names this position, so there is no claim + // rewrite to do: the resume is committed, exactly once, and + // both writers return that one event. + return { event: occupant }; + } + } if (!(await bumpEventSlot(attempt))) { break; } @@ -2816,36 +2857,10 @@ export function createEventsStorage( tag ); } - // For a resume, losing this publish is the convergence the pinned - // id exists to force: both takers of one resume adopt the claim's - // position, so the event now at it is this same resume, written by - // the other taker. That is the case `converge` above already - // answers with the committed event — it just could not see it yet, - // because the other taker had not published when this attempt - // read. Answer it the same way rather than reporting a conflict - // the caller cannot act on: the resume IS committed, exactly once, - // and the dedup contract is that both writers return that event. - if (eventIdPinned && data.eventType === 'hook_received') { - const occupant = await readJSONWithFallback( - basedir, - 'events', - `${effectiveRunId}-${eventId}`, - EventSchema, - tag - ); - if ( - occupant && - params?.resumeId && - isResumeEvent(occupant, { - runId: effectiveRunId, - resumeId: params.resumeId, - hookId: data.correlationId, - eventId, - }) - ) { - return { event: occupant }; - } - } + // A resume that lost its publish to its own committed event already + // returned it from inside the loop above, so reaching here means the + // occupant is an unrelated event and this is a duplicate publish the + // runtime's concurrent-replay catch path handles. throw new EntityConflictError( `Event "${eventId}" already exists for run "${effectiveRunId}"` ); diff --git a/packages/world-local/src/storage/hook-staging-slots.test.ts b/packages/world-local/src/storage/hook-staging-slots.test.ts index 3240b597af..8f5bb94fba 100644 --- a/packages/world-local/src/storage/hook-staging-slots.test.ts +++ b/packages/world-local/src/storage/hook-staging-slots.test.ts @@ -7,11 +7,43 @@ import { SPEC_VERSION_CURRENT, type Storage, } from '@workflow/world'; -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { createHook, createRun } from '../test-helpers.js'; import { hookResumeClaimPath, pendingHookEventPath } from './helpers.js'; import { createStorage } from './index.js'; +// Holds the FIRST caller to reach the promote until a LATER one has linked, +// which is the interleaving that decides which taker of a resume claim wins +// the position. Disarmed by default so every other test runs unmocked. +const promoteGate: { + armed: boolean; + releaseFirst: (() => void) | null; + firstParked: Promise | null; +} = { armed: false, releaseFirst: null, firstParked: null }; + +vi.mock('../fs.js', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + promoteExclusive: async (stagedPath: string, filePath: string) => { + if (!promoteGate.armed) { + return actual.promoteExclusive(stagedPath, filePath); + } + if (promoteGate.firstParked === null) { + promoteGate.firstParked = new Promise((resolve) => { + promoteGate.releaseFirst = resolve; + }); + await promoteGate.firstParked; + return actual.promoteExclusive(stagedPath, filePath); + } + const result = await actual.promoteExclusive(stagedPath, filePath); + promoteGate.armed = false; + promoteGate.releaseFirst?.(); + return result; + }, + }; +}); + // `hook_received` is the one event that does not publish straight into // `events/`: it stages the file under `.locks` first, so a terminal // transition can reap it before it ever becomes reader-visible. The staging @@ -26,6 +58,9 @@ describe('world-local hook_received staging and slot density', () => { beforeEach(async () => { testDir = await fs.mkdtemp(path.join(os.tmpdir(), 'hook-staging-test-')); storage = createStorage(testDir); + promoteGate.armed = false; + promoteGate.releaseFirst = null; + promoteGate.firstParked = null; }); afterEach(async () => { @@ -148,11 +183,59 @@ describe('world-local hook_received staging and slot density', () => { expect(data.filter((e) => e.eventType === 'hook_received')).toHaveLength(1); }); - it('keeps the log dense when two instances resume the same hook at once', async () => { + it('writes one event when the claim owner loses the position to an adopter', async () => { + const { runId, hook } = await setup(); + const other = createStorage(testDir); + + // Only one of the two takers of a claim is pinned. The taker that WRITES + // the claim keeps its own id, unpinned, because a slot is a position that + // another instance also hands out for unrelated events, and refusing to + // move would fail this resume's append outright. The taker that ADOPTS an + // existing claim is pinned to the claimed position. + // + // So the loser of the promote can be the unpinned owner, and a loser that + // bumps publishes a second `hook_received` for one resumeId. Nothing in + // the log looks wrong afterwards (it stays dense, both callers report + // success) but the resume is delivered twice on replay. + // + // The gate parks whichever caller reaches the promote first until the + // other has linked. The owner gets there first on its own: the adopter + // reads the claim and scans for a committed event before it stages. + promoteGate.armed = true; + const results = await Promise.all([ + resume(runId, hook, 'resume_1', new Uint8Array([1])), + other.events.create( + runId, + { + eventType: 'hook_received', + specVersion: SPEC_VERSION_CURRENT, + correlationId: hook.hookId, + eventData: { token: hook.token, payload: new Uint8Array([1]) }, + }, + { resumeId: 'resume_1', resumePayloadDigest: 'resume_1' } + ), + ]); + + const { data } = await storage.events.list({ runId }); + expect(data.filter((e) => e.eventType === 'hook_received')).toHaveLength(1); + // Both takers answer with the one committed event, which is the dedup + // contract the caller relies on to treat a redelivery as a no-op. + expect(results[1].event.eventId).toBe(results[0].event.eventId); + }); + + it('keeps the log dense under live contention on one position', async () => { const { runId, hook } = await setup(); - // A second instance over the same directory is the configuration this - // backend supports for the CLI plus the app. Its allocator watermark is - // its own, so both instances hand out the same candidate position. + // Density here is not a regression guard: with two LIVE stagers and no + // terminal transition, the pre-fix code also ended dense, because the + // writer it bumped off the position was the one that went on to publish + // it. The hole needs the stager to be rejected or killed, which is what + // the crashed-attempt test above stages. + // + // What this does cover is that arbitrating at the promote (rather than at + // the staging write) still resolves two instances drawing one position, + // which is the configuration this backend supports for the CLI plus the + // app: each instance's allocator watermark is its own, so both hand out + // the same candidate. const other = createStorage(testDir); await Promise.all([ diff --git a/scripts/event-log-race-repro-local.sh b/scripts/event-log-race-repro-local.sh index fa9371072d..0e8b256625 100755 --- a/scripts/event-log-race-repro-local.sh +++ b/scripts/event-log-race-repro-local.sh @@ -222,8 +222,13 @@ else # went on using `.next/workflow-data`. Both halves are therefore set here, as # an absolute path so the app's cwd does not enter into it. `setupWorld` # recomputes the same path for the harness process. + # Substring matches, not prefixes: `setupWorld` selects on + # `appName.includes('nextjs') || appName.includes('next-')`, so an app named + # `example-nextjs` has to land on the same branch here. A prefix glob would + # send the app to `.workflow-data` and the harness to `.next/workflow-data`, + # which is the split-brain this block exists to prevent. case "$APP_NAME" in - nextjs*|next-*) DATA_DIR_NAME=".next/workflow-data" ;; + *nextjs*|*next-*) DATA_DIR_NAME=".next/workflow-data" ;; *) DATA_DIR_NAME=".workflow-data" ;; esac DATA_DIR="$REPO_ROOT/$APP_DIR/$DATA_DIR_NAME"