diff --git a/.changeset/world-local-hook-staging-slots.md b/.changeset/world-local-hook-staging-slots.md new file mode 100644 index 0000000000..bc51248b27 --- /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 deliver a raced resume exactly once instead of duplicating it or reporting 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..b9e59fc066 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -2734,25 +2734,32 @@ 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; - } - 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 { @@ -2785,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; } @@ -2811,6 +2857,10 @@ export function createEventsStorage( tag ); } + // 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 new file mode 100644 index 0000000000..8f5bb94fba --- /dev/null +++ b/packages/world-local/src/storage/hook-staging-slots.test.ts @@ -0,0 +1,260 @@ +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, 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 +// 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); + promoteGate.armed = false; + promoteGate.releaseFirst = null; + promoteGate.firstParked = null; + }); + + 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('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(); + // 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([ + 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..0e8b256625 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,44 @@ 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. + # 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" ;; + *) 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 +281,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 +292,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 +312,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 +342,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 +380,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 +465,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