feat(run-engine): bounded waitpoint completion fanout and lifecycle cleanup

Internal groundwork for Redis-backed waitpoints, inactive in production
by default. It moves wide completion fanout out of the foreground
request and into a bounded, recoverable worker, while adding explicit
lifecycle rules for active and terminal coordination state. Existing
waitpoint behavior remains on the Postgres path until later routing work
enables this subsystem.

Mono-RevId: c334b81ad8131769a503dab7460b4f003e8cc3b6

Author: d-cs

internal-packages/run-engine/harness/waitpointFanout.ts

@@ -39,15 +39,17 @@
     "@internal/testcontainers": "workspace:*",
     "@opentelemetry/sdk-metrics": "2.7.1",
     "@types/seedrandom": "^3.0.8",
-    "rimraf": "6.0.1"
+    "rimraf": "6.0.1",
+    "tsx": "4.17.0"
   },
   "scripts": {
     "clean": "rimraf dist",
-    "typecheck": "tsc --noEmit -p tsconfig.build.json && tsc --noEmit -p tsconfig.freeze-test.json",
+    "typecheck": "tsc --noEmit -p tsconfig.build.json && tsc --noEmit -p tsconfig.freeze-test.json && tsc --noEmit -p tsconfig.harness.json",
     "test": "vitest --sequence.concurrent=false --no-file-parallelism",
     "test:coverage": "vitest --sequence.concurrent=false --no-file-parallelism --coverage.enabled",
     "build": "pnpm run clean && tsc -p tsconfig.build.json",
     "dev": "tsc --watch  -p tsconfig.build.json",
-    "test:bench": "vitest --config ./vitest.bench.config.ts --run"
+    "test:bench": "vitest --config ./vitest.bench.config.ts --run",
+    "harness:waitpoint-fanout": "tsx harness/waitpointFanout.ts"
   }
 }

internal-packages/run-engine/package.json

@@ -8,9 +8,12 @@
  *     `COUNT(*) ... WHERE status='PENDING'`, over the same population. Like for like.
  *  2. Read amplification — the store's `readBlockState` against a full-payload `SELECT`
  *     of the same waitpoints. Like for like.
- *  3. Store-only write paths — block+complete+deliver and K-watcher fan-out. Absolute
- *     numbers with NO Postgres counterpart: no single statement on the previous path
- *     corresponds to a Redis round trip that both blocks a run and delivers to watchers.
+ *  3. Store-only write paths — block+complete+drain, then foreground completion versus
+ *     worker drain at several fan-out widths. Absolute numbers with NO Postgres
+ *     counterpart: no single statement on the previous path corresponds to a Redis round
+ *     trip that both blocks a run and delivers to watchers. The width sweep is the
+ *     evidence for the flat-foreground claim — completion should not track K, the drain
+ *     should.
  *  4. Register cost versus edge count — `registerBlocks` registers each edge with its own
  *     round trip before the single absorb. This measures whether that serial loop is a
  *     real cost at a wide fan-in, or a non-issue, at several fan-in widths.
@@ -19,10 +22,11 @@
  * empty table measures nothing.
  *
  * Knobs: BENCH_WP_ITERATIONS, BENCH_WP_FANIN, BENCH_WP_WATCHERS, BENCH_WP_REGISTER_WIDTHS,
- * BENCH_WP_REGISTER_SAMPLES.
+ * BENCH_WP_REGISTER_SAMPLES, BENCH_WP_FANOUT_WIDTHS.
  */
 import { containerTest } from "@internal/testcontainers";
 import type { PrismaClient } from "@trigger.dev/database";
+import { WaitpointFanoutWorker } from "../waitpointCoordinator/fanoutWorker.js";
 import {
   WaitpointStoreCoordinator,
   type BlockEdge,
@@ -40,6 +44,13 @@ const REGISTER_WIDTHS = (process.env.BENCH_WP_REGISTER_WIDTHS ?? "1,10,100,1001"
   .map((raw) => Number(raw.trim()))
   .filter((width) => Number.isFinite(width) && width > 0);
 const REGISTER_SAMPLES = Number(process.env.BENCH_WP_REGISTER_SAMPLES ?? 20);
+const FANOUT_WIDTHS = (process.env.BENCH_WP_FANOUT_WIDTHS ?? `1,10,100,${WATCHERS}`)
+  .split(",")
+  .map((raw) => Number(raw.trim()))
+  .filter(
+    (width, index, all) => Number.isFinite(width) && width > 0 && all.indexOf(width) === index
+  )
+  .sort((a, b) => a - b);
 const NOW = new Date().toISOString();
 
 type Sample = { label: string; count: number; p50: number; p99: number; totalMs: number };
@@ -121,6 +132,7 @@ containerTest(
   async ({ prisma, redisOptions }) => {
     const env = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
     const store = new WaitpointStoreCoordinator({ redisOptions });
+    const worker = new WaitpointFanoutWorker({ coordinator: store, enabled: true });
     const samples: Sample[] = [];
     const registerCost: Array<{
       width: number;
@@ -143,13 +155,18 @@ containerTest(
       }
       await store.registerBlocks({
         runId: "bench_run_fanin",
+        blockId: "bench_blk_fanin",
         edges: ids.map((id, index) => edge(id, index)),
       });
 
       // --- group 1: the pending-count gate, like for like ---
       samples.push(
         await measure("store.pendingCount", ITERATIONS, async () => {
-          await store.absorbBlockers({ runId: "bench_run_fanin", edges: [] });
+          await store.absorbBlockers({
+            runId: "bench_run_fanin",
+            blockId: "bench_blk_fanin",
+            edges: [],
+          });
         })
       );
       samples.push(
@@ -179,40 +196,50 @@ containerTest(
             record: record(id, env.id, env.project.id),
             status: "PENDING",
           });
-          await store.registerBlocks({ runId: `bench_run_${i}`, edges: [edge(id)] });
-          const done = await store.complete({ waitpointId: id, completion });
-          for (const watcher of done.watchers) {
-            await store.deliverCompletion({
-              runId: watcher.runId,
-              waitpointId: id,
-              completion: done.completion!,
-            });
-          }
+          await store.registerBlocks({
+            runId: `bench_run_${i}`,
+            blockId: `bench_blk_${i}`,
+            edges: [edge(id)],
+          });
+          await store.complete({ waitpointId: id, completion });
+          await worker.visit(id);
         })
       );
 
-      const fanOutId = "bench_fanout_w";
-      await store.createIfAbsent({
-        record: record(fanOutId, env.id, env.project.id),
-        status: "PENDING",
-      });
-      for (let i = 0; i < WATCHERS; i++) {
-        await store.registerBlocks({ runId: `bench_watcher_${i}`, edges: [edge(fanOutId)] });
+      // Foreground completion versus watcher count. The claim under test is that the
+      // first number is FLAT in K while the second grows with it — that is what "bounded
+      // foreground completion" means in practice, and a single width could not show it.
+      for (const width of FANOUT_WIDTHS) {
+        const fanOutId = `bench_fanout_w_${width}`;
+        await store.createIfAbsent({
+          record: record(fanOutId, env.id, env.project.id),
+          status: "PENDING",
+        });
+        for (let i = 0; i < width; i++) {
+          await store.registerBlocks({
+            runId: `bench_watcher_${width}_${i}`,
+            blockId: `bench_blk_watcher_${width}_${i}`,
+            edges: [edge(fanOutId)],
+          });
+        }
+
+        samples.push(
+          await measure(`store.complete(foreground, watchers=${width})`, 1, async () => {
+            await store.complete({ waitpointId: fanOutId, completion });
+          })
+        );
+        samples.push(
+          await measure(`worker.drain(watchers=${width})`, 1, async () => {
+            let visits = 0;
+            // The visit page budget is bounded, so a wide fan-out takes several visits.
+            for (;;) {
+              const summary = await worker.visit(fanOutId);
+              visits++;
+              if (summary.outcome !== "more" || visits > 1_000) break;
+            }
+          })
+        );
       }
-      samples.push(
-        await measure(`store.complete+deliver(watchers=${WATCHERS})`, 1, async () => {
-          const done = await store.complete({ waitpointId: fanOutId, completion });
-          // Serial on purpose: this is the worst case, and it is the number that says
-          // whether delivery needs to pipeline.
-          for (const watcher of done.watchers) {
-            await store.deliverCompletion({
-              runId: watcher.runId,
-              waitpointId: fanOutId,
-              completion: done.completion!,
-            });
-          }
-        })
-      );
 
       // --- group 4: register cost versus edge count ---
       // registerBlocks registers each edge with its own round trip, serially, before the
@@ -236,7 +263,12 @@ containerTest(
           `store.registerBlocks(edges=${width})`,
           REGISTER_SAMPLES,
           async () => {
-            await store.registerBlocks({ runId: `bench_register_${width}_${call++}`, edges });
+            const n = call++;
+            await store.registerBlocks({
+              runId: `bench_register_${width}_${n}`,
+              blockId: `bench_blk_register_${width}_${n}`,
+              edges,
+            });
           }
         );
         samples.push(sample);

internal-packages/run-engine/src/engine/bench/waitpointCoordinator.bench.test.ts

@@ -0,0 +1,148 @@
+/**
+ * The inertness gate: the ordinary application, booted the way production boots it, with
+ * store-resident waitpoint minting off.
+ *
+ * The store coordinator and its fanout worker are exported but nothing in `RunEngine`
+ * constructs either one, so a full legacy block-and-complete cycle must leave the Waitpoint
+ * namespace completely untouched — no records, no watcher queues, no fanout entries and no
+ * partition index — while behaving exactly as it did before. Asserting the keyspace rather
+ * than the wiring is deliberate: it fails if a later change starts writing there, however
+ * it is introduced.
+ */
+import { createRedisClient } from "@internal/redis";
+import { assertNonNullable, containerTest } from "@internal/testcontainers";
+import { trace } from "@internal/tracing";
+import { setTimeout } from "node:timers/promises";
+import { expect } from "vitest";
+import { FANOUT_PARTITION_COUNT, fanoutIndexKeys } from "../waitpointCoordinator/keys.js";
+import { RunEngine } from "../index.js";
+import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js";
+
+vi.setConfig({ testTimeout: 60_000 });
+
+describe("Waitpoint fanout stays inert while store-resident minting is disabled", () => {
+  containerTest(
+    "a legacy block and complete cycle writes nothing into the Waitpoint namespace",
+    async ({ prisma, redisOptions }) => {
+      const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const probe = createRedisClient(redisOptions);
+
+      const engine = new RunEngine({
+        prisma,
+        worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
+        queue: {
+          redis: redisOptions,
+          masterQueueConsumersDisabled: true,
+          processWorkerQueueDebounceMs: 50,
+        },
+        runLock: { redis: redisOptions },
+        machines: {
+          defaultMachine: "small-1x",
+          machines: {
+            "small-1x": { name: "small-1x" as const, cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 },
+          },
+          baseCostInCents: 0.0001,
+        },
+        tracer: trace.getTracer("test", "0.0.0"),
+      });
+
+      try {
+        // Positive control first: an assertion that can only ever pass is not a guard, so
+        // prove the scan finds a `wp:` key when one exists before relying on it finding none.
+        await probe.set("wp:{sentinel}", "1");
+        expect(await waitpointNamespaceKeys()).toEqual(["wp:{sentinel}"]);
+        await probe.del("wp:{sentinel}");
+        expect(await waitpointNamespaceKeys()).toEqual([]);
+
+        const taskIdentifier = "test-task";
+        await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier);
+
+        const run = await engine.trigger(
+          {
+            number: 1,
+            friendlyId: "run_pinert1",
+            environment: authenticatedEnvironment,
+            taskIdentifier,
+            payload: "{}",
+            payloadType: "application/json",
+            context: {},
+            traceContext: {},
+            traceId: "t_inert",
+            spanId: "s_inert",
+            workerQueue: "main",
+            queue: `task/${taskIdentifier}`,
+            isTest: false,
+            tags: [],
+          },
+          prisma
+        );
+
+        await setTimeout(500);
+        const dequeued = await engine.dequeueFromWorkerQueue({
+          consumerId: "test_inert",
+          workerQueue: "main",
+        });
+        await engine.startRunAttempt({
+          runId: dequeued[0]!.run.id,
+          snapshotId: dequeued[0]!.snapshot.id,
+        });
+
+        const created = await engine.createManualWaitpoint({
+          environmentId: authenticatedEnvironment.id,
+          projectId: authenticatedEnvironment.projectId,
+        });
+
+        await engine.blockRunWithWaitpoint({
+          runId: run.id,
+          waitpoints: created.waitpoint.id,
+          projectId: authenticatedEnvironment.projectId,
+          organizationId: authenticatedEnvironment.organizationId,
+        });
+
+        expect(
+          (await engine.getRunExecutionData({ runId: run.id }))?.snapshot.executionStatus
+        ).toBe("EXECUTING_WITH_WAITPOINTS");
+        // Blocked, and the block is recorded where it always was.
+        const blocking = await prisma.taskRunWaitpoint.findFirst({ where: { taskRunId: run.id } });
+        assertNonNullable(blocking);
+        expect(blocking.waitpointId).toBe(created.waitpoint.id);
+
+        // Mid-cycle: a blocked run is exactly when watcher and fanout state would exist.
+        expect(await waitpointNamespaceKeys()).toEqual([]);
+
+        await engine.completeWaitpoint({ id: created.waitpoint.id });
+        await setTimeout(200);
+
+        // Unchanged legacy behaviour: the run resumes and its block row is gone.
+        expect(
+          (await engine.getRunExecutionData({ runId: run.id }))?.snapshot.executionStatus
+        ).toBe("EXECUTING");
+        expect(
+          await prisma.taskRunWaitpoint.findFirst({ where: { taskRunId: run.id } })
+        ).toBeNull();
+
+        expect(await waitpointNamespaceKeys()).toEqual([]);
+        for (let partition = 0; partition < FANOUT_PARTITION_COUNT; partition++) {
+          expect(await probe.zcard(fanoutIndexKeys(partition).due)).toBe(0);
+          expect(await probe.zcard(fanoutIndexKeys(partition).quarantine)).toBe(0);
+        }
+      } finally {
+        probe.disconnect();
+        await engine.quit();
+      }
+
+      // The engine's own subsystems use their configured key prefixes, so an unprefixed
+      // `wp:` key can only have come from the waitpoint store coordinator.
+      async function waitpointNamespaceKeys(): Promise<string[]> {
+        const found: string[] = [];
+        let cursor = "0";
+        do {
+          const [next, batch] = await probe.scan(cursor, "MATCH", "wp:*", "COUNT", 1_000);
+          found.push(...batch);
+          cursor = next;
+        } while (cursor !== "0");
+        return found.sort();
+      }
+    }
+  );
+});