Fix three crash-class defects behind this week's flaky CI and tighten the e2e harness

.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).

.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.

.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.

.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

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);

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 () => {

packages/core/src/runtime/run-return-value-real-world.test.ts

@@ -23,23 +23,36 @@ 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];
   let dir: string;
   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);
 });

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<any>[] = [];
+        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 {

packages/core/src/workflow.ts

@@ -416,6 +416,18 @@ async function createWorkflowSessionInner(
   });
 
   const initialInterruption = withResolvers<never>();
+  // 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,