diff --git a/.changeset/hold-hook-token-lock.md b/.changeset/hold-hook-token-lock.md new file mode 100644 index 0000000000..e632817759 --- /dev/null +++ b/.changeset/hold-hook-token-lock.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-local': patch +--- + +Hold the Hook token claim lock through the `hook_created` publish so two processes creating the same hook can no longer both publish a `hook_created` event (the adopter could publish at the claimed slot first and the claimer then bumped past it). diff --git a/.changeset/listen-client-error-handler.md b/.changeset/listen-client-error-handler.md new file mode 100644 index 0000000000..accb6773ed --- /dev/null +++ b/.changeset/listen-client-error-handler.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-postgres': patch +--- + +Handle `error` on the dedicated `LISTEN` clients and on the World's own connection pool: a connection the server closes (restart, failover, idle reaper) is now reopened after a backoff instead of surfacing as an uncaught exception that took the host process down. diff --git a/.changeset/quiet-interruption-handler.md b/.changeset/quiet-interruption-handler.md new file mode 100644 index 0000000000..dc5c61953e --- /dev/null +++ b/.changeset/quiet-interruption-handler.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Fix a host-process crash when a run fails before its body starts (an unregistered workflow name, a bundle that fails to evaluate, input that fails to hydrate) while the event log still holds an unconsumed event: the consumer's deferred divergence check rejected an interruption promise nothing had awaited yet, which surfaced as an `unhandledRejection` about 100ms after the flow route had already reported the run as failed. diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 18ca523641..2688ce04a3 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -447,6 +447,13 @@ jobs: TURBO_TEAM: ${{ vars.TURBO_TEAM }} WORKFLOW_PUBLIC_MANIFEST: '1' WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS: '5000' + # Every CLI assertion spawns a full `node` child that boots oclif, + # infers the Vercel project and team, builds a World and talks to the + # API before it can answer. Under the concurrent suite on a 4-core + # runner the 20s default budget was being spent on that boot, and the + # child was killed with SIGTERM mid-request (seen as + # `inspect --withData` dying while still "Inferring vercel project"). + WORKFLOW_E2E_CLI_TIMEOUT_MS: '40000' steps: - name: Checkout Repo uses: actions/checkout@v4 @@ -517,6 +524,9 @@ jobs: WORKFLOW_VERCEL_TEAM: "team_nO2mCG4W8IxPIeKoSsqwAxxB" WORKFLOW_VERCEL_PROJECT: ${{ matrix.app.project-id }} WORKFLOW_VERCEL_PROJECT_SLUG: ${{ matrix.app.project-slug }} + # The CLI children the suite spawns skip `.vercel` folder inference + # when the project id, team and project name are all in the env. + WORKFLOW_VERCEL_PROJECT_NAME: ${{ matrix.app.project-slug }} # The trusted-sources OIDC token (x-vercel-trusted-oidc-idp-token) # is minted on demand inside `scripts/trusted-sources-headers.mjs` # via the runner's ACTIONS_ID_TOKEN_REQUEST_URL/_TOKEN env vars @@ -595,6 +605,13 @@ jobs: TURBO_TEAM: ${{ vars.TURBO_TEAM }} WORKFLOW_PUBLIC_MANIFEST: '1' WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS: '5000' + # Every CLI assertion spawns a full `node` child that boots oclif, + # infers the Vercel project and team, builds a World and talks to the + # API before it can answer. Under the concurrent suite on a 4-core + # runner the 20s default budget was being spent on that boot, and the + # child was killed with SIGTERM mid-request (seen as + # `inspect --withData` dying while still "Inferring vercel project"). + WORKFLOW_E2E_CLI_TIMEOUT_MS: '40000' steps: - name: Checkout Repo uses: actions/checkout@v4 @@ -646,6 +663,7 @@ jobs: WORKFLOW_VERCEL_TEAM: "team_nO2mCG4W8IxPIeKoSsqwAxxB" WORKFLOW_VERCEL_PROJECT: "prj_yjkM7UdHliv8bfxZ1sMJQf1pMpdi" WORKFLOW_VERCEL_PROJECT_SLUG: "example-nextjs-workflow-turbopack" + WORKFLOW_VERCEL_PROJECT_NAME: "example-nextjs-workflow-turbopack" # See the note on e2e-vercel-prod: PRs point at the protected # workflow-server preview; unset on main for production. VERCEL_WORKFLOW_SERVER_URL: ${{ github.ref != 'refs/heads/main' && !startsWith(github.head_ref, 'changeset-release/') && secrets.VERCEL_WORKFLOW_SERVER_URL || '' }} @@ -832,6 +850,9 @@ jobs: WORKFLOW_VERCEL_TEAM: ${{ env.WS_TEAM_ID }} WORKFLOW_VERCEL_PROJECT: ${{ matrix.app.project-id }} WORKFLOW_VERCEL_PROJECT_SLUG: ${{ matrix.app.project-slug }} + # The CLI children the suite spawns skip `.vercel` folder inference + # when the project id, team and project name are all in the env. + WORKFLOW_VERCEL_PROJECT_NAME: ${{ matrix.app.project-slug }} VERCEL_WORKFLOW_SERVER_URL: ${{ github.ref != 'refs/heads/main' && secrets.VERCEL_WORKFLOW_SERVER_URL || '' }} - name: Capture runtime logs on failure @@ -1008,6 +1029,9 @@ jobs: WORKFLOW_VERCEL_TEAM: ${{ env.WS_TEAM_ID }} WORKFLOW_VERCEL_PROJECT: ${{ matrix.app.project-id }} WORKFLOW_VERCEL_PROJECT_SLUG: ${{ matrix.app.project-slug }} + # The CLI children the suite spawns skip `.vercel` folder inference + # when the project id, team and project name are all in the env. + WORKFLOW_VERCEL_PROJECT_NAME: ${{ matrix.app.project-slug }} VERCEL_WORKFLOW_SERVER_URL: ${{ github.ref != 'refs/heads/main' && secrets.VERCEL_WORKFLOW_SERVER_URL || '' }} - name: Capture runtime logs on failure diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index a60b016e89..f938cdc2ce 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -2051,7 +2051,13 @@ describe.concurrent('e2e', () => { test( 'hookCleanupTestWorkflow - hook token reuse after workflow completion', - { timeout: 60_000 }, + // Two full hook round trips in sequence, each paying a `returnValue` + // poll interval (5s in CI) plus a hook registration poll, then two CLI + // inspects. On green lanes this ran at 42-59s locally and 50-74s on + // Vercel against the previous 60s budget, so its retries were budget + // exhaustion, not a race. Same budget as the sibling multi-round-trip + // hook tests (`hookClaimOnlyMutexWorkflow`, `retainedInterleavingWorkflow`). + { timeout: 90_000 }, async () => { const token = Math.random().toString(36).slice(2); const customData = Math.random().toString(36).slice(2); diff --git a/packages/core/src/events-consumer.test.ts b/packages/core/src/events-consumer.test.ts index c8ccead476..7be7af30dd 100644 --- a/packages/core/src/events-consumer.test.ts +++ b/packages/core/src/events-consumer.test.ts @@ -1258,24 +1258,34 @@ describe('EventsConsumer', () => { realEvent('step_started', corr) ), ]; - // Run at the real delay: the point of the test is the difference - // between paying it and not, so shortening it would erase the signal. - vi.unstubAllEnvs(); - const consumer = consumerFor(events); + const onUnconsumedEvent = vi.fn(); + const onDuplicateEvent = vi.fn(); + const consumer = consumerFor(events, { + onUnconsumedEvent, + onDuplicateEvent, + }); - const start = Date.now(); consumer.subscribe(entityConsumer(corr, 'step_completed')); - await vi.waitFor( - () => { - expect(consumer.eventIndex).toBe(events.length); - }, - { interval: 1 } - ); + // `subscribe()` queues one `consume` pass on `process.nextTick`, and a + // pass steps over a straggler synchronously. Deferring one instead ends + // the pass with the walk parked on that straggler until a + // `getDeferredCheckDelayMs()` timer fires, so the walk standing at the + // end of the log on the very next tick is the proof that none of them + // paid the window. That holds however loaded the runner is, where the + // wall-clock bound this replaced (finish inside one 100ms window) did + // not: Windows CI measured 102ms. + await waitForNextTick(); - // Deferring each straggler would cost one window apiece, so the walk - // finishing inside a single window means none of them went through the - // deferred check. - expect(Date.now() - start).toBeLessThan(DEFERRED_CHECK_DELAY_MS); + expect(consumer.eventIndex).toBe(events.length); + // Each straggler left through the skip, not the deferred check. + expect(onDuplicateEvent).toHaveBeenCalledTimes(stragglers); + for (const straggler of events.slice(-stragglers)) { + expect(onDuplicateEvent).toHaveBeenCalledWith( + straggler, + 'step_started' + ); + } + expect(onUnconsumedEvent).not.toHaveBeenCalled(); }); it('reports the first outcome when a repeat decides the class differently', async () => { diff --git a/packages/core/src/runtime/run-return-value-real-world.test.ts b/packages/core/src/runtime/run-return-value-real-world.test.ts index ef47acc25e..39224d0ff9 100644 --- a/packages/core/src/runtime/run-return-value-real-world.test.ts +++ b/packages/core/src/runtime/run-return-value-real-world.test.ts @@ -23,10 +23,18 @@ import { setWorld } from './world.js'; * * So these assert the property a user actually observes — how long after the * run finishes the await resolves — with world-local standing in for "a World - * that can wait". A run finishing 300ms in can only be reported at the ~1s + * that can wait". A run finishing 300ms in can only be reported at the next * tick by interval polling, so the two paths are far apart and the kill switch * is observable rather than merely configured. + * + * The interval is raised from its 1s default so that "far apart" survives a + * loaded runner: at 1s the wait path had 500ms of headroom under its bound, + * and Windows CI spent it (828ms measured against an 800ms bound). At 3s the + * wait path is bounded at half the interval, which forgives over a second of + * stall, while the interval path still cannot report before ~3s. */ +const POLL_INTERVAL_MS = 3_000; + describe('run.returnValue over a real World', () => { const envName = 'WORKFLOW_RETURN_VALUE_LONG_POLL'; const original = process.env[envName]; @@ -34,12 +42,17 @@ describe('run.returnValue over a real World', () => { let world: World; beforeEach(async () => { + vi.stubEnv( + 'WORKFLOW_RETURN_VALUE_POLL_INTERVAL_MS', + String(POLL_INTERVAL_MS) + ); dir = await mkdtemp(join(tmpdir(), 'returnvalue-real-world-')); world = createWorld({ dataDir: dir }) as unknown as World; setWorld(world); }); afterEach(async () => { + vi.unstubAllEnvs(); if (original === undefined) delete process.env[envName]; else process.env[envName] = original; setWorld(undefined as unknown as World); @@ -103,8 +116,8 @@ describe('run.returnValue over a real World', () => { const elapsed = Date.now() - startedAt; await finishing; - // At or above the 1s interval would mean the poll reported it, not the wait. - expect(elapsed).toBeLessThan(800); + // At or above the interval would mean the poll reported it, not the wait. + expect(elapsed).toBeLessThan(POLL_INTERVAL_MS / 2); }, 30_000); it('reports a cancellation without waiting out the interval', async () => { @@ -130,7 +143,7 @@ describe('run.returnValue over a real World', () => { const elapsed = Date.now() - startedAt; await cancelling; - expect(elapsed).toBeLessThan(800); + expect(elapsed).toBeLessThan(POLL_INTERVAL_MS / 2); }, 30_000); it('restores fixed-interval polling under the kill switch', async () => { @@ -145,7 +158,7 @@ describe('run.returnValue over a real World', () => { const elapsed = Date.now() - startedAt; await finishing; - // The run was done at ~300ms, but only the ~1s tick can observe it. - expect(elapsed).toBeGreaterThanOrEqual(900); + // The run was done at ~300ms, but only the interval tick can observe it. + expect(elapsed).toBeGreaterThanOrEqual(POLL_INTERVAL_MS * 0.9); }, 30_000); }); diff --git a/packages/core/src/workflow.test.ts b/packages/core/src/workflow.test.ts index aa7acb780a..ea494a3a77 100644 --- a/packages/core/src/workflow.test.ts +++ b/packages/core/src/workflow.test.ts @@ -1536,6 +1536,74 @@ describe('runWorkflow', () => { ); }); + it('does not leave an unhandled rejection when the workflow is not registered and the log holds an unconsumed event', async () => { + // The structural consumer starts the event walk before the workflow + // function is looked up. With an ordered event in the log that nobody + // will ever claim, the consumer's deferred check fires after + // `WorkflowNotRegisteredError` has already been thrown and calls + // `onWorkflowError`, which rejects the initial interruption promise. + // Nothing was awaiting it yet, so that used to be an unhandled + // rejection: a crashed host process, not a failed run. + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandled); + try { + const ops: Promise[] = []; + const runId = 'test-run-unregistered'; + const workflowRun: WorkflowRun = { + runId, + workflowName: 'value', + status: 'running', + input: await dehydrateWorkflowArguments( + [], + 'wrun_123', + noEncryptionKey, + ops + ), + createdAt: new Date('2024-01-01T00:00:00.000Z'), + updatedAt: new Date('2024-01-01T00:00:00.000Z'), + startedAt: new Date('2024-01-01T00:00:00.000Z'), + deploymentId: 'test-deployment', + }; + const events: Event[] = [ + { + runId, + eventId: 'evnt_run_started', + eventType: 'run_started', + createdAt: new Date('2024-01-01T00:00:00.000Z'), + } as Event, + { + runId, + eventId: 'evnt_orphan_step_created', + eventType: 'step_created', + correlationId: 'step_orphan', + eventData: { stepName: 'orphan' }, + createdAt: new Date('2024-01-01T00:00:01.000Z'), + } as Event, + ]; + + await expect( + runWorkflow( + `const value = "test"${getWorkflowTransformCode()}`, + workflowRun, + events, + noEncryptionKey + ) + ).rejects.toMatchObject({ name: 'WorkflowNotRegisteredError' }); + + // Let the consumer's deferred unconsumed-event check run its course: + // the promise queue drains, deliveries are idle, then the delay. + await new Promise((resolve) => + setTimeout(resolve, DEFERRED_CHECK_DELAY_MS * 4) + ); + expect(unhandled).toEqual([]); + } finally { + process.off('unhandledRejection', onUnhandled); + } + }); + it('should throw user-defined error when workflow code throws an error', async () => { let error: Error | undefined; try { diff --git a/packages/core/src/workflow.ts b/packages/core/src/workflow.ts index faecee4c46..c5f11c068a 100644 --- a/packages/core/src/workflow.ts +++ b/packages/core/src/workflow.ts @@ -416,6 +416,18 @@ async function createWorkflowSessionInner( }); const initialInterruption = withResolvers(); + // Nothing awaits this promise until `waitForExecution` at the bottom of this + // function, but the structural consumer registered below starts walking the + // event log as soon as it subscribes. If anything between here and there + // throws (an unregistered workflow name, a bundle that fails to evaluate, + // input that fails to hydrate), the walk can still reach an ordered event + // nobody claims, and its deferred check then calls `onWorkflowError`, whose + // `running` branch rejects this promise. With no handler attached yet that + // is an `unhandledRejection`, which takes the host process down about + // `DEFERRED_CHECK_DELAY_MS` after the flow route already reported the run + // as failed. Mark it handled up front: `Promise.race` in `waitForExecution` + // attaches its own handler and still observes the rejection. + initialInterruption.promise.catch(() => {}); let state: WorkflowSessionState = { type: 'running', interruption: initialInterruption, diff --git a/packages/world-local/src/storage/events-storage.ts b/packages/world-local/src/storage/events-storage.ts index 1dd52a93e0..a1bc90ff36 100644 --- a/packages/world-local/src/storage/events-storage.ts +++ b/packages/world-local/src/storage/events-storage.ts @@ -74,8 +74,10 @@ import { } from '../fs.js'; import { stripEventDataRefs } from './filters.js'; import { + acquireHookTokenClaimLock, getObjectCreatedAt, type HookTokenClaim, + type HookTokenClaimLockHandle, hookDisposeLockPath, hookRecoveryMarkerPath, hookResumeClaimPath, @@ -90,7 +92,6 @@ import { releaseHookTokenClaimIfOwnedBy, runTerminalMarkerPath, scanRunEventIds, - withHookTokenClaimLock, } from './helpers.js'; import { deleteHookByRunMarker, @@ -947,6 +948,17 @@ export function createEventsStorage( data: AnyEventRequest, params?: CreateEventParams ): Promise { + /** + * The Hook token claim lock this create holds, when it holds one. The + * `hook_created` branch acquires it and it stays held until the whole + * create settles (the publish included), so it is released in + * `createImpl`'s finally rather than at the end of the branch. See the + * branch for why the publish has to be inside the critical section. + * Declared up here, ahead of the dispatch below, because `createImpl` + * is hoisted and runs before anything past those `return`s. + */ + let hookTokenClaimLock: HookTokenClaimLockHandle | undefined; + if ( data.eventType === 'hook_created' && data.eventData.tokenRetentionUntil !== undefined && @@ -1019,6 +1031,27 @@ export function createEventsStorage( return createImpl(); async function createImpl(): Promise { + let settled = false; + try { + const result = await createImplBody(); + settled = true; + return result; + } finally { + const held = hookTokenClaimLock; + hookTokenClaimLock = undefined; + if (held) { + // After a success a failed release is this call's error to + // report; after a failure the original error wins. + if (settled) { + await held.release(); + } else { + await held.release().catch(() => {}); + } + } + } + } + + async function createImplBody(): Promise { // Most paths use the freshly-drawn candidate eventId. The // hook_created dedup-recovery path below may reassign it to // the canonical eventId persisted in the durable token claim @@ -2293,68 +2326,83 @@ export function createEventsStorage( // run cannot race its successor and create a spurious conflict // (issue #2778). Missing claim caches are rebuilt from the event log; // stale owners are removed before the successor is admitted. - const claimResult = await withHookTokenClaimLock( + // + // The lock is held until this create settles, not just across the + // claim decision (it is released in `createImpl`'s finally). The + // claim records this writer's *candidate* eventId, and an unpinned + // publish may bump past that slot if something else took it. A + // second writer that read the claim before this writer's publish + // committed would adopt the recorded slot, publish there first, + // and then the claimer's collision-and-bump publishes a second + // `hook_created` for the same hook. Keeping the lock through the + // publish means an adopter can only read a claim whose event is + // already in the log (and collides into the benign duplicate), or + // one whose writer crashed and left the lock stale (and repairs + // it). Per token, so nothing else serializes behind it. + hookTokenClaimLock = await acquireHookTokenClaimLock( basedir, - hookData.token, - async (signal) => { - let existingClaim = await readHookTokenClaim(constraintPath); - if (!existingClaim) { - // Repair a missing or corrupt claim from the event log. - signal.throwIfAborted(); - await deleteJSON(constraintPath); - await rebuildLiveHookByTokenFromEventLog( - basedir, - hookData.token, - tag - ); - existingClaim = await readHookTokenClaim(constraintPath); - } - - if (!existingClaim) { - signal.throwIfAborted(); - assert(await writeExclusive(constraintPath, claimContent)); - return { status: 'claimed' as const }; - } - if ( - existingClaim.runId === effectiveRunId && - existingClaim.hookId === data.correlationId - ) { - return { status: 'owned' as const, claim: existingClaim }; - } - if ( - !(await isHookTokenClaimReleasable(basedir, existingClaim, tag)) - ) { - return { status: 'conflict' as const, claim: existingClaim }; - } - - // The previous owner committed its release but did not finish - // cleanup. Remove that lifetime before admitting a successor. + hookData.token + ); + const decideClaim = async (signal: AbortSignal) => { + let existingClaim = await readHookTokenClaim(constraintPath); + if (!existingClaim) { + // Repair a missing or corrupt claim from the event log. signal.throwIfAborted(); await deleteJSON(constraintPath); - if (existingClaim.hookId) { - await deleteJSON( - taggedPath(basedir, 'hooks', existingClaim.hookId, tag) - ); - await deleteJSON( - hookRecoveryMarkerPath( - basedir, - hookData.token, - existingClaim.runId, - existingClaim.hookId - ) - ); - await deleteHookByRunMarker( - basedir, - existingClaim.runId, - existingClaim.hookId, - tag - ); - } + await rebuildLiveHookByTokenFromEventLog( + basedir, + hookData.token, + tag + ); + existingClaim = await readHookTokenClaim(constraintPath); + } + + if (!existingClaim) { signal.throwIfAborted(); assert(await writeExclusive(constraintPath, claimContent)); return { status: 'claimed' as const }; } - ); + if ( + existingClaim.runId === effectiveRunId && + existingClaim.hookId === data.correlationId + ) { + return { status: 'owned' as const, claim: existingClaim }; + } + if ( + !(await isHookTokenClaimReleasable(basedir, existingClaim, tag)) + ) { + return { status: 'conflict' as const, claim: existingClaim }; + } + + // The previous owner committed its release but did not finish + // cleanup. Remove that lifetime before admitting a successor. + signal.throwIfAborted(); + await deleteJSON(constraintPath); + if (existingClaim.hookId) { + await deleteJSON( + taggedPath(basedir, 'hooks', existingClaim.hookId, tag) + ); + await deleteJSON( + hookRecoveryMarkerPath( + basedir, + hookData.token, + existingClaim.runId, + existingClaim.hookId + ) + ); + await deleteHookByRunMarker( + basedir, + existingClaim.runId, + existingClaim.hookId, + tag + ); + } + signal.throwIfAborted(); + assert(await writeExclusive(constraintPath, claimContent)); + return { status: 'claimed' as const }; + }; + const claimResult = await decideClaim(hookTokenClaimLock.signal); + hookTokenClaimLock.signal.throwIfAborted(); // The claim and Hook entity are written before `hook_created`, so a // crash can leave either cache without the durable event. A retry by diff --git a/packages/world-local/src/storage/helpers.ts b/packages/world-local/src/storage/helpers.ts index 9584fec930..9a62742e38 100644 --- a/packages/world-local/src/storage/helpers.ts +++ b/packages/world-local/src/storage/helpers.ts @@ -376,15 +376,36 @@ export async function readHookTokenClaim( } } +/** A held Hook token claim lock (see {@link acquireHookTokenClaimLock}). */ +export interface HookTokenClaimLockHandle { + /** + * Aborts if the lock is compromised while held (its lockfile disappeared or + * stopped refreshing, so another process may hold it too). Work done under + * the lock must check it before every write that relies on exclusivity. + */ + readonly signal: AbortSignal; + /** + * Release the lock. Idempotent. A compromised lock is no longer this + * holder's to release, so that case is a no-op. Rejects with + * `WorkflowWorldError` when the lockfile could not be removed. + */ + release(): Promise; +} + /** - * Serializes claim handoffs. Exclusive writes admit the first owner, but - * cannot atomically replace a stale owner across local-world processes. + * Acquire the per-token lock that serializes claim handoffs. Exclusive writes + * admit the first owner, but cannot atomically replace a stale owner across + * local-world processes. + * + * Callers that only need the lock around one block use + * {@link withHookTokenClaimLock}. The handle form exists for the + * `hook_created` publish, which has to keep the lock across a section that + * spans several stages of `events.create` and releases it in a `finally`. */ -export async function withHookTokenClaimLock( +export async function acquireHookTokenClaimLock( basedir: string, - token: string, - fn: (signal: AbortSignal) => Promise -): Promise { + token: string +): Promise { const claimPath = hookTokenClaimPath(basedir, token); await fs.mkdir(path.dirname(claimPath), { recursive: true }); const controller = new AbortController(); @@ -414,24 +435,49 @@ export async function withHookTokenClaimLock( }); } + let released = false; + return { + signal: controller.signal, + async release() { + if (released) return; + released = true; + if (controller.signal.aborted) return; + try { + await release(); + } catch (error) { + throw new WorkflowWorldError( + 'Could not release Hook token claim lock', + { + cause: error, + } + ); + } + }, + }; +} + +/** + * Run `fn` under the per-token claim lock (see + * {@link acquireHookTokenClaimLock}). The lock is released when `fn` settles; + * a compromised lock fails the operation. + */ +export async function withHookTokenClaimLock( + basedir: string, + token: string, + fn: (signal: AbortSignal) => Promise +): Promise { + const held = await acquireHookTokenClaimLock(basedir, token); + let result: T; try { - result = await fn(controller.signal); - controller.signal.throwIfAborted(); + result = await fn(held.signal); + held.signal.throwIfAborted(); } catch (error) { - if (!controller.signal.aborted) { - await release().catch(() => {}); - } + await held.release().catch(() => {}); throw error; } - try { - await release(); - } catch (error) { - throw new WorkflowWorldError('Could not release Hook token claim lock', { - cause: error, - }); - } + await held.release(); return result; } diff --git a/packages/world-postgres/src/index.ts b/packages/world-postgres/src/index.ts index 336daad58e..c5858e8ebd 100644 --- a/packages/world-postgres/src/index.ts +++ b/packages/world-postgres/src/index.ts @@ -63,6 +63,15 @@ export function createWorld( connectionString: config.connectionString || getDefaultConnectionString(), ...(maxPoolSize !== undefined ? { max: maxPoolSize } : {}), }); + if (pool !== config.pool) { + // `pg.Pool` re-emits an idle client's `error` (the server closed the + // connection while it sat in the pool) on the pool itself. With no + // listener that is an uncaught exception that takes the process down; + // with one, the pool simply discards the client and the next checkout + // opens a new connection. A caller-supplied pool keeps the caller's own + // handling. + pool.on('error', () => {}); + } const drizzle = createClient(pool); const queue = createQueue(config, pool); diff --git a/packages/world-postgres/src/reenqueue.test.ts b/packages/world-postgres/src/reenqueue.test.ts index 0ded6200dc..c956ebcac2 100644 --- a/packages/world-postgres/src/reenqueue.test.ts +++ b/packages/world-postgres/src/reenqueue.test.ts @@ -25,6 +25,9 @@ vi.mock('pg', () => ({ return { query: vi.fn(async () => ({ rows: [{ exists: false }] })), end: vi.fn(), + // `createWorld()` registers an `error` listener on a pool it owns (a + // real `pg.Pool` is an EventEmitter). + on: vi.fn(), }; }), })); diff --git a/packages/world-postgres/src/run-status.ts b/packages/world-postgres/src/run-status.ts index a53477e9f0..ed2148dab9 100644 --- a/packages/world-postgres/src/run-status.ts +++ b/packages/world-postgres/src/run-status.ts @@ -106,9 +106,23 @@ export function createRunStatusListener(pool: Pool): RunStatusListener { // run into a connection attempt per poll interval. if (Date.now() < retrySubscribeAfter) return undefined; - subscription = listenChannel(pool, RUN_STATUS_TOPIC, async (payload) => { - if (payload) emitter.emit(`run:${payload}`); - }).catch(() => { + subscription = listenChannel( + pool, + RUN_STATUS_TOPIC, + async (payload) => { + if (payload) emitter.emit(`run:${payload}`); + }, + { + onError: () => { + // The connection died under us. Forget it so the next wait past the + // backoff opens a fresh one; until then waits fall back to the + // backstop re-read. A closed listener stays closed. + if (retrySubscribeAfter === Number.POSITIVE_INFINITY) return; + subscription = undefined; + retrySubscribeAfter = Date.now() + LISTEN_RETRY_BACKOFF_MS; + }, + } + ).catch(() => { // No listener connection available (pool options that don't permit a // second client, a database without LISTEN, a restarting server). Waits // degrade to the backstop re-read (the behavior of a plain poll), and diff --git a/packages/world-postgres/src/streamer.ts b/packages/world-postgres/src/streamer.ts index a6b2cb60b5..52be077433 100644 --- a/packages/world-postgres/src/streamer.ts +++ b/packages/world-postgres/src/streamer.ts @@ -45,21 +45,47 @@ class Rc { } } +/** + * How long a `LISTEN` subscription that lost its connection waits before it + * is opened again. Long enough that a database that is restarting costs one + * connection attempt every few seconds, short enough that live stream reads + * resume well within the time a reader would notice. + */ +export const LISTEN_RECONNECT_BACKOFF_MS = 5_000; + /** * Subscribe to a PostgreSQL NOTIFY channel using a dedicated client created * from the pool's connection options. `channel` must be a trusted identifier. + * + * A `pg.Client` emits `error` when the server ends the connection under it (a + * restart, a failover, `pg_terminate_backend`, an idle-connection reaper). + * With no listener registered that is an uncaught exception, so every + * subscription gets one here. The connection is not usable afterwards: it is + * ended and `onError` is told, and the owner decides whether to subscribe + * again. Notifications published in between are lost, which every consumer + * of this helper already tolerates (they are signals over durable rows). */ export const listenChannel = async ( pool: Pool, channel: string, - onPayload: (payload: string) => Promise + onPayload: (payload: string) => Promise, + options: { onError?: (error: Error) => void } = {} ): Promise<{ close: () => Promise }> => { const client = new Client(pool.options); + let ended = false; + + client.on('error', (error: Error) => { + if (ended) return; + ended = true; + client.end().catch(() => {}); + options.onError?.(error); + }); try { await client.connect(); await client.query(`LISTEN ${channel}`); } catch (err) { + ended = true; await client.end().catch(() => {}); throw err; } @@ -73,6 +99,8 @@ export const listenChannel = async ( return { close: async () => { client.removeListener('notification', onNotification); + if (ended) return; + ended = true; try { await client.query(`UNLISTEN ${channel}`); } finally { @@ -118,7 +146,7 @@ export function createStreamer(pool: Pool, drizzle: Drizzle): PostgresStreamer { const STREAM_TOPIC = 'workflow_event_chunk'; - const listenSubscription = listenChannel(pool, STREAM_TOPIC, async (msg) => { + const onChunkPublished = async (msg: string) => { const parsed = StreamPublishMessage.parse(JSON.parse(msg)); const key = `strm:${parsed.streamId}` as const; @@ -142,7 +170,27 @@ export function createStreamer(pool: Pool, drizzle: Drizzle): PostgresStreamer { const { data, eof } = value; events.emit(key, { id: parsed.chunkId, data, eof }); }); - }); + }; + + // The subscription is re-opened when its connection dies (see + // `listenChannel`); the shared `closed` flag stops that once the streamer + // is shut down. + let reconnectTimer: ReturnType | undefined; + let listenSubscription: Promise<{ close: () => Promise } | undefined>; + const subscribe = () => { + listenSubscription = listenChannel(pool, STREAM_TOPIC, onChunkPublished, { + onError: () => { + if (closed) return; + reconnectTimer = setTimeout(() => { + reconnectTimer = undefined; + if (!closed) subscribe(); + }, LISTEN_RECONNECT_BACKOFF_MS); + // A dead subscription must not keep a process alive on its own. + reconnectTimer.unref?.(); + }, + }).catch(() => undefined); + }; + subscribe(); const notifyStream = async (payload: string) => { await pool.query('SELECT pg_notify($1, $2)', [STREAM_TOPIC, payload]); @@ -492,10 +540,14 @@ export function createStreamer(pool: Pool, drizzle: Drizzle): PostgresStreamer { async close() { closed = true; + if (reconnectTimer) { + clearTimeout(reconnectTimer); + reconnectTimer = undefined; + } for (const abort of [...activeReaders]) { abort(); } - const sub = await listenSubscription.catch(() => undefined); + const sub = await listenSubscription; if (sub) await sub.close(); }, }; diff --git a/vitest.config.ts b/vitest.config.ts index 8a5a069fc0..d5d63883f2 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -3,6 +3,15 @@ import { configDefaults, defineConfig } from 'vitest/config'; export default defineConfig({ test: { testTimeout: 60_000, + // Fixture hooks get the budget of the tests they set up. Vitest's 10s + // default assumes an idle disk: on a Windows runner the world-local suite + // runs its filesystem-heavy files in parallel against one slow temp + // volume, and a `beforeEach` that writes ten small JSON files + // (fs.test.ts) stalled past 10s while a sibling file was pushing a + // thousand events through the same disk. Main's green run had that + // whole 78-test file at 6.7s, so the hook is not what is slow. A test + // body doing the same writes would have had 60s. + hookTimeout: 60_000, // Deployment e2e suites can lose timing races to queue delays, cold // starts, and watcher latency. They always set DEPLOYMENT_URL, so keep // their one visible retry without masking deterministic unit/integration