feat(run-engine): Redis waitpoint store coordinator, Lua protocol, and waitpoint ids

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

@@ -0,0 +1,266 @@
+/**
+ * Waitpoint coordination benchmark. Reports numbers; asserts nothing — on a shared runner
+ * the timings swing far more than any threshold worth gating on.
+ *
+ * Four groups, and only the first two are pairs:
+ *
+ *  1. Pending count — the store's SCARD gate against the previous path's
+ *     `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.
+ *  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.
+ *
+ * Every Postgres measurement here runs against rows this file inserts. A baseline over an
+ * empty table measures nothing.
+ *
+ * Knobs: BENCH_WP_ITERATIONS, BENCH_WP_FANIN, BENCH_WP_WATCHERS, BENCH_WP_REGISTER_WIDTHS,
+ * BENCH_WP_REGISTER_SAMPLES.
+ */
+import { containerTest } from "@internal/testcontainers";
+import type { PrismaClient } from "@trigger.dev/database";
+import {
+  WaitpointStoreCoordinator,
+  type BlockEdge,
+  type WaitpointRecordInput,
+} from "../waitpointCoordinator/storeCoordinator.js";
+import { setupAuthenticatedEnvironment } from "../tests/setup.js";
+
+vi.setConfig({ testTimeout: 900_000 });
+
+const ITERATIONS = Number(process.env.BENCH_WP_ITERATIONS ?? 100);
+const FANIN = Number(process.env.BENCH_WP_FANIN ?? 1001);
+const WATCHERS = Number(process.env.BENCH_WP_WATCHERS ?? 100);
+const REGISTER_WIDTHS = (process.env.BENCH_WP_REGISTER_WIDTHS ?? "1,10,100,1001")
+  .split(",")
+  .map((raw) => Number(raw.trim()))
+  .filter((width) => Number.isFinite(width) && width > 0);
+const REGISTER_SAMPLES = Number(process.env.BENCH_WP_REGISTER_SAMPLES ?? 20);
+const NOW = new Date().toISOString();
+
+type Sample = { label: string; count: number; p50: number; p99: number; totalMs: number };
+
+function percentile(sorted: number[], p: number): number {
+  if (sorted.length === 0) return 0;
+  return sorted[Math.min(sorted.length - 1, Math.floor((p / 100) * sorted.length))]!;
+}
+
+async function measure(label: string, count: number, run: (i: number) => Promise<void>) {
+  const durations: number[] = [];
+  const started = Date.now();
+  for (let i = 0; i < count; i++) {
+    const t0 = performance.now();
+    await run(i);
+    durations.push(performance.now() - t0);
+  }
+  durations.sort((a, b) => a - b);
+  const sample: Sample = {
+    label,
+    count,
+    p50: percentile(durations, 50),
+    p99: percentile(durations, 99),
+    totalMs: Date.now() - started,
+  };
+  console.log(
+    `[bench] ${sample.label} n=${sample.count} p50=${sample.p50.toFixed(2)}ms ` +
+      `p99=${sample.p99.toFixed(2)}ms total=${sample.totalMs}ms`
+  );
+  return sample;
+}
+
+function record(id: string, environmentId: string, projectId: string): WaitpointRecordInput {
+  return {
+    id,
+    friendlyId: `waitpoint_${id}`,
+    type: "MANUAL",
+    environmentId,
+    projectId,
+    createdAt: NOW,
+    updatedAt: NOW,
+    userProvidedIdempotencyKey: false,
+    tags: [],
+  };
+}
+
+const completion = {
+  completedAt: NOW,
+  outputType: "application/json",
+  outputIsError: false,
+  output: { inline: '{"ok":true}' },
+};
+
+function edge(waitpointId: string, batchIndex?: number): BlockEdge {
+  return { waitpointId, batchIndex, createdAt: NOW, type: "MANUAL" };
+}
+
+async function insertWaitpoints(
+  prisma: PrismaClient,
+  ids: string[],
+  environmentId: string,
+  projectId: string
+) {
+  await prisma.waitpoint.createMany({
+    data: ids.map((id) => ({
+      id,
+      friendlyId: `waitpoint_${id}`,
+      type: "MANUAL" as const,
+      idempotencyKey: id,
+      userProvidedIdempotencyKey: false,
+      projectId,
+      environmentId,
+    })),
+  });
+}
+
+containerTest(
+  "waitpoint coordination: pending count, read amplification, store write paths, register cost",
+  async ({ prisma, redisOptions }) => {
+    const env = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+    const store = new WaitpointStoreCoordinator({ redisOptions });
+    const samples: Sample[] = [];
+    const registerCost: Array<{
+      width: number;
+      p50Ms: number;
+      p99Ms: number;
+      perEdgeMsP50: number;
+    }> = [];
+
+    try {
+      const ids = Array.from({ length: FANIN }, (_, i) => `bench_w_${i}`);
+
+      // Both stores get the SAME population. A Postgres baseline over an empty table
+      // measures an index probe against nothing.
+      await insertWaitpoints(prisma, ids, env.id, env.project.id);
+      for (const id of ids) {
+        await store.createIfAbsent({
+          record: record(id, env.id, env.project.id),
+          status: "PENDING",
+        });
+      }
+      await store.registerBlocks({
+        runId: "bench_run_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: [] });
+        })
+      );
+      samples.push(
+        await measure("postgres.pendingCount", ITERATIONS, async () => {
+          await prisma.$queryRaw`SELECT COUNT(*) FROM "Waitpoint" WHERE id = ANY(${ids}::text[]) AND status = 'PENDING'`;
+        })
+      );
+
+      // --- group 2: read amplification, like for like ---
+      samples.push(
+        await measure("store.readBlockState", ITERATIONS, async () => {
+          await store.readBlockState("bench_run_fanin");
+        })
+      );
+      samples.push(
+        await measure("postgres.hydrateFullPayload", ITERATIONS, async () => {
+          // Every column of every waitpoint — the amplification the store removes.
+          await prisma.waitpoint.findMany({ where: { id: { in: ids } } });
+        })
+      );
+
+      // --- group 3: store-only write paths, no Postgres counterpart ---
+      samples.push(
+        await measure("store.block+complete+deliver", ITERATIONS, async (i) => {
+          const id = `bench_cycle_${i}`;
+          await store.createIfAbsent({
+            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!,
+            });
+          }
+        })
+      );
+
+      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)] });
+      }
+      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
+      // single absorb. A review flagged that a wide fan-in therefore serializes one round
+      // trip per edge. This measures the real cost at several widths rather than predicting
+      // it, so the decision about bounded concurrency is made against a number.
+      const registerPoolWidth = Math.max(0, ...REGISTER_WIDTHS);
+      const registerIds = Array.from({ length: registerPoolWidth }, (_, i) => `bench_reg_w_${i}`);
+      await insertWaitpoints(prisma, registerIds, env.id, env.project.id);
+      for (const id of registerIds) {
+        await store.createIfAbsent({
+          record: record(id, env.id, env.project.id),
+          status: "PENDING",
+        });
+      }
+
+      for (const width of REGISTER_WIDTHS) {
+        const edges = registerIds.slice(0, width).map((id, index) => edge(id, index));
+        let call = 0;
+        const sample = await measure(
+          `store.registerBlocks(edges=${width})`,
+          REGISTER_SAMPLES,
+          async () => {
+            await store.registerBlocks({ runId: `bench_register_${width}_${call++}`, edges });
+          }
+        );
+        samples.push(sample);
+        registerCost.push({
+          width,
+          p50Ms: sample.p50,
+          p99Ms: sample.p99,
+          perEdgeMsP50: sample.p50 / width,
+        });
+        console.log(
+          `[bench] store.registerBlocks(edges=${width}) implied per-edge cost ` +
+            `p50=${(sample.p50 / width).toFixed(3)}ms p99=${(sample.p99 / width).toFixed(3)}ms`
+        );
+      }
+
+      console.log(
+        `[bench] groups 1 and 2 are like-for-like pairs. Group 3 and the register-cost ` +
+          `group (4) have no Postgres counterpart: no single statement on the previous ` +
+          `path corresponds to a Redis round trip that blocks, completes and delivers, ` +
+          `or to a serial per-edge register loop.`
+      );
+      console.log(`[bench] summary\n${JSON.stringify({ samples, registerCost }, null, 2)}`);
+    } finally {
+      await store.quit();
+    }
+  }
+);

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

@@ -0,0 +1,132 @@
+import { describe, expect, it } from "vitest";
+import {
+  WaitpointKeyTagError,
+  assertSingleSlot,
+  edgeField,
+  idempotencyKey,
+  runBlockKeys,
+  waitpointIdFromEdgeField,
+  waitpointKeys,
+  watcherField,
+} from "./keys.js";
+
+describe("waitpointKeys", () => {
+  it("puts the record and its watchers under one hash tag", () => {
+    const k = waitpointKeys("abc123w");
+    expect(k.record).toBe("wp:{abc123w}");
+    expect(k.watchers).toBe("wp:{abc123w}:w");
+  });
+});
+
+describe("runBlockKeys", () => {
+  it("puts all three run keys under one hash tag", () => {
+    const k = runBlockKeys("run_abc");
+    expect(k.pend).toBe("wp:run:{run_abc}:pend");
+    expect(k.done).toBe("wp:run:{run_abc}:done");
+    expect(k.edge).toBe("wp:run:{run_abc}:edge");
+  });
+});
+
+describe("idempotencyKey", () => {
+  it("tags by environment, so one environment's reservations share a slot", () => {
+    expect(idempotencyKey("env_1", "my-key")).toBe("wp:idem:{env_1}:my-key");
+  });
+});
+
+describe("edgeField", () => {
+  it("keys by waitpoint id and batch index, matching the Postgres unique key", () => {
+    expect(edgeField("w_a", 3)).toBe("w_a#3");
+  });
+
+  it("collapses a null or absent batch index onto one field", () => {
+    expect(edgeField("w_a")).toBe("w_a#");
+    expect(edgeField("w_a", null)).toBe("w_a#");
+  });
+
+  it("distinguishes index 0 from an absent index", () => {
+    expect(edgeField("w_a", 0)).not.toBe(edgeField("w_a"));
+  });
+});
+
+describe("waitpointIdFromEdgeField", () => {
+  it("round-trips back to the waitpoint id", () => {
+    for (const index of [undefined, null, 0, 7]) {
+      expect(waitpointIdFromEdgeField(edgeField("w_a", index))).toBe("w_a");
+    }
+  });
+
+  it("returns undefined for a field with no separator", () => {
+    expect(waitpointIdFromEdgeField("nope")).toBeUndefined();
+  });
+
+  it("splits on the last separator, tolerating a '#' inside the waitpoint id", () => {
+    expect(waitpointIdFromEdgeField("a#b#3")).toBe("a#b");
+  });
+});
+
+describe("watcherField", () => {
+  it("keys by run id and batch index, so one run can watch at several indexes", () => {
+    expect(watcherField("run_a", 2)).toBe("run_a#2");
+    expect(watcherField("run_a")).toBe("run_a#");
+    expect(watcherField("run_a", 0)).not.toBe(watcherField("run_a"));
+  });
+});
+
+describe("assertSingleSlot", () => {
+  it("accepts keys that share one tag", () => {
+    const k = runBlockKeys("run_abc");
+    expect(() => assertSingleSlot("runReadBlockState", [k.pend, k.done, k.edge])).not.toThrow();
+  });
+
+  it("accepts a single tagged key", () => {
+    expect(() => assertSingleSlot("wpIdemReserve", [idempotencyKey("env_1", "k")])).not.toThrow();
+  });
+
+  it("accepts an empty key list", () => {
+    expect(() => assertSingleSlot("noKeys", [])).not.toThrow();
+  });
+
+  it("rejects keys from two different tags", () => {
+    const wp = waitpointKeys("w_a");
+    const run = runBlockKeys("run_abc");
+    expect(() => assertSingleSlot("bad", [wp.record, run.pend])).toThrow(WaitpointKeyTagError);
+  });
+
+  it("rejects an untagged key", () => {
+    expect(() => assertSingleSlot("bad", ["wp:no-tag"])).toThrow(WaitpointKeyTagError);
+  });
+
+  it("rejects an empty tag", () => {
+    expect(() => assertSingleSlot("bad", ["wp:{}"])).toThrow(WaitpointKeyTagError);
+  });
+
+  it("rejects an empty first pair, matching Redis rather than skipping to a later one", () => {
+    // Redis stops at the first `{`/`}` pair. An empty one means no tag at all, so it hashes
+    // the whole key. A regex would have found `a` here and wrongly claimed a shared slot.
+    expect(() => assertSingleSlot("bad", ["wp:{}{a}", "wp:{}{a}"])).toThrow(WaitpointKeyTagError);
+  });
+
+  it("takes the first pair when several are present", () => {
+    expect(() => assertSingleSlot("ok", ["wp:{a}{b}", "wp:{a}:w"])).not.toThrow();
+    expect(() => assertSingleSlot("bad", ["wp:{a}{b}", "wp:{b}:w"])).toThrow(WaitpointKeyTagError);
+  });
+
+  it("does not degrade on a key made of many opening braces", () => {
+    const started = performance.now();
+    expect(() => assertSingleSlot("bad", ["{".repeat(50_000)])).toThrow(WaitpointKeyTagError);
+    expect(performance.now() - started).toBeLessThan(1_000);
+  });
+
+  it("names the operation and the offending key in the error", () => {
+    const wp = waitpointKeys("w_a");
+    const run = runBlockKeys("run_abc");
+    try {
+      assertSingleSlot("myOperation", [wp.record, run.pend]);
+      throw new Error("should have thrown");
+    } catch (error) {
+      expect(error).toBeInstanceOf(WaitpointKeyTagError);
+      expect((error as Error).message).toContain("myOperation");
+      expect((error as Error).message).toContain(run.pend);
+    }
+  });
+});