fix(run-engine): fail loudly on a missing edge reply slot, guard empty-array clear

Removes the one ?? on a Lua reply decode in readBlockState — a missing edge slot now throws instead of silently decoding an empty BlockEdge, matching the convention everywhere else in this file. clearBlockState now distinguishes an explicitly empty edgeIds array (a no-op) from an omitted one (the terminal clear-everything), since both previously collapsed onto the Lua's n === 0 clear-everything branch. Also: BlockStateEdge no longer advertises a reported field it can never carry, adds a test where pendingOfRequested and storePendingTotal diverge for a reason other than an empty request list, and trims/renames a few comments per review.
2026-08-21 20:38:59 +01:00
parent 77f0e4dd2b
commit 7e1142a187
2 changed files with 80 additions and 14 deletions
@@ -360,9 +360,10 @@ describe("complete", () => {
  });
 });

-// No coordinator method calls runAbsorbBlockers/runClear/wpIdemReserve yet — a later task
-// wires those in. Registered directly on a raw client so the Lua itself is exercised now.
-describe("runAbsorbBlockers, runClear and wpIdemReserve (direct Lua)", () => {
+// A coordinator method now drives each of these scripts, but this block stays: it is the
+// only place asserting the RAW reply shape, so a Lua/TypeScript framing change made on
+// both sides at once would still fail here even though every class-level test passed.
+describe("reply framing (direct Lua — pins the wire shape the coordinator decodes)", () => {
  const envelope = JSON.stringify(completion());

  redisTest(
@@ -780,8 +781,6 @@ describe("absorbBlockers", () => {
        expect(result.pendingOfRequested).toBe(1);
        expect(result.storePendingTotal).toBe(1);

-        // The edges themselves stay distinct — that multiplicity is what produces the
-        // repeats in the cycle's ordered id list.
        const state = await store.readBlockState(RUN_ID);
        expect(state.edges).toHaveLength(2);
        expect(state.edges.map((e) => e.batchIndex).sort()).toEqual([0, 2]);
@@ -830,7 +829,6 @@ describe("absorbBlockers", () => {
  redisTest("lets a delivery that raced ahead of the absorb win", async ({ redisOptions }) => {
    const store = coordinator(redisOptions);
    try {
-      // The completion landed between register and absorb, so it is already delivered.
      await store.deliverCompletion({
        runId: RUN_ID,
        waitpointId: "w_a",
@@ -896,6 +894,31 @@ describe("absorbBlockers", () => {
    }
  });

+  redisTest(
+    "reports a smaller pendingOfRequested than storePendingTotal when an unrelated blocker is already pending",
+    async ({ redisOptions }) => {
+      const store = coordinator(redisOptions);
+      try {
+        // w_x is a live blocker from an earlier absorb, unrelated to this call's request.
+        await store.absorbBlockers({ runId: RUN_ID, edges: [edge("w_x")] });
+
+        const result = await store.absorbBlockers({
+          runId: RUN_ID,
+          edges: [edge("w_a", { reported: completion() })],
+        });
+
+        // Nothing THIS call requested is pending (w_a arrived already delivered), but the
+        // run's whole store-resident set still holds w_x — a divergence for a different
+        // reason than an empty request list, so a reply[0]/reply[1] swap or a
+        // re-derived-in-TypeScript pendingOfRequested would both be caught here too.
+        expect(result.pendingOfRequested).toBe(0);
+        expect(result.storePendingTotal).toBe(1);
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+
  redisTest("sets no TTL on any run key", async ({ redisOptions }) => {
    const store = coordinator(redisOptions);
    const probe = createRedisClient(redisOptions);
@@ -1117,4 +1140,25 @@ describe("clearBlockState", () => {
      await store.quit();
    }
  });
+
+  redisTest(
+    "is a no-op for an explicitly empty edge id list, unlike an omitted one",
+    async ({ redisOptions }) => {
+      const store = coordinator(redisOptions);
+      try {
+        await store.absorbBlockers({ runId: RUN_ID, edges: [edge("w_a"), edge("w_b")] });
+
+        // Omitting edgeIds reaches the Lua's n === 0 branch and clears everything (proven
+        // above). A caller-computed EMPTY array must not collapse onto that: it means
+        // "nothing to drain", not "clear the run".
+        expect((await store.clearBlockState({ runId: RUN_ID, edgeIds: [] })).outcome).toBe("noop");
+
+        const state = await store.readBlockState(RUN_ID);
+        expect(state.edges.map((e) => e.waitpointId).sort()).toEqual(["w_a", "w_b"]);
+        expect(state.pendingIds.sort()).toEqual(["w_a", "w_b"]);
+      } finally {
+        await store.quit();
+      }
+    }
+  );
 });
@@ -96,7 +96,10 @@ export type CompleteResult = {
  watchers: WatcherEntry[];
 };

-/** One run-to-waitpoint edge. The metadata a frozen return type needs travels here. */
+/**
+ * One run-to-waitpoint edge. The metadata a frozen return type — an existing API response
+ * shape this store must keep reproducing — needs travels here.
+ */
 export type BlockEdge = {
  waitpointId: string;
  batchIndex?: number | null;
@@ -126,7 +129,10 @@ export type AbsorbResult = {
  alreadyDelivered: Array<{ waitpointId: string; completion?: WaitpointCompletion }>;
 };

-export type BlockStateEdge = BlockEdge & { edgeId: string };
+// absorbBlockers strips `reported` before writing the edge blob, so a value read back
+// here can never carry it — Omit says so instead of inheriting a field that is always
+// undefined.
+export type BlockStateEdge = Omit<BlockEdge, "reported"> & { edgeId: string };

 export type BlockState = {
  pendingIds: string[];
@@ -423,7 +429,17 @@ export class WaitpointStoreCoordinator {
    const edges: BlockStateEdge[] = [];
    for (let i = 0; i < edgeCount; i += 2) {
      const edgeId = reply[cursor + i]!;
-      const stored = JSON.parse(reply[cursor + i + 1] ?? "{}") as BlockEdge;
+      // An edge value is always a non-empty JSON.stringify, so a missing slot here means
+      // the cursor walked off the end of the reply. That must fail loudly, not decode a
+      // BlockEdge with no waitpointId — the exact off-by-one this task's arithmetic guards
+      // against.
+      const edgeJson = reply[cursor + i + 1];
+      if (!edgeJson) {
+        throw new Error(
+          `readBlockState(${runId}): missing edge payload at reply index ${cursor + i + 1}`
+        );
+      }
+      const stored = JSON.parse(edgeJson) as BlockEdge;
      edges.push({ ...stored, edgeId });
    }

@@ -433,15 +449,21 @@ export class WaitpointStoreCoordinator {
  /**
   * Drain one cycle's edges, or clear the run entirely when no edge ids are given.
   *
-   * The selective form RECONCILES: after the named edges go, any pending or delivered
-   * entry that no surviving edge references goes too. That is wider than deleting the
-   * named ids, and it has to be — a delivery is written unconditionally, so the window
-   * between register and absorb can leave a delivered entry with no edge at all.
+   * The selective form RECONCILES: any pending or delivered entry that no surviving edge
+   * references goes too, not only the named ones. See runClear in scripts.ts for why.
   */
  async clearBlockState(args: {
    runId: string;
    edgeIds?: string[];
-  }): Promise<{ outcome: "cleared" | "drained" }> {
+  }): Promise<{ outcome: "cleared" | "drained" | "noop" }> {
+    // `omitted` and `explicitly empty` must not collapse onto each other: the Lua's
+    // n === 0 means "clear the whole run", so an omitted edgeIds stays the terminal clear,
+    // but a caller that computed zero edges to drain gets a genuine no-op that never
+    // reaches Redis.
+    if (args.edgeIds && args.edgeIds.length === 0) {
+      return { outcome: "noop" };
+    }
+
    const keys = runBlockKeys(args.runId);
    const edgeIds = args.edgeIds ?? [];