perf(run-engine): bound the envelope read's concurrency

The store arm launched one HMGET per waitpoint through a single Promise.all, so a 1000-item
batch fan-in would issue 1000 concurrent commands at once. Each is small, but the burst is
the cost.

Chunks at 100, matching the bound the snapshot hydration already uses for the same shape of
read. Stays one command per key, so nothing can span two cluster slots and the single-slot
assertion is still unnecessary here.

Two tests cover the chunk boundary at 250 and a mixed complete/pending set at 150, since an
increment that disagrees with the slice is the realistic failure of hand-rolled chunking and
is invisible in a single-chunk test.

Author: d-cs

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

@@ -1973,3 +1973,53 @@ describe("readCompletionEnvelopes", () => {
     }
   });
 });
+
+// The read is chunked so a large fan-in cannot burst. This asserts correctness ACROSS the chunk
+// boundary, which is where an off-by-one in the slice would show: a count either side of 100.
+describe("readCompletionEnvelopes chunking", () => {
+  redisTest("returns every envelope across a chunk boundary", async ({ redisOptions }) => {
+    const store = coordinator(redisOptions);
+    try {
+      const ids = Array.from({ length: 250 }, (_, i) => `w_chunk_${i}`);
+      for (const id of ids) {
+        await store.createIfAbsent({ record: record(id), status: "PENDING" });
+        await store.complete({ waitpointId: id, completion: completion() });
+      }
+
+      const envelopes = await store.readCompletionEnvelopes({
+        runId: "run_chunked",
+        waitpointIds: ids,
+      });
+
+      expect(envelopes).toHaveLength(250);
+      expect(new Set(envelopes.map((e) => e.id)).size).toBe(250);
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("omits only the incomplete ones in a mixed chunked read", async ({ redisOptions }) => {
+    const store = coordinator(redisOptions);
+    try {
+      const ids = Array.from({ length: 150 }, (_, i) => `w_mix_${i}`);
+      for (const [i, id] of ids.entries()) {
+        await store.createIfAbsent({ record: record(id), status: "PENDING" });
+        // Leave every third pending, including ids either side of the chunk boundary.
+        if (i % 3 !== 0) {
+          await store.complete({ waitpointId: id, completion: completion() });
+        }
+      }
+
+      const envelopes = await store.readCompletionEnvelopes({
+        runId: "run_chunked",
+        waitpointIds: ids,
+      });
+
+      const expected = ids.filter((_, i) => i % 3 !== 0).length;
+      expect(envelopes).toHaveLength(expected);
+      expect(envelopes.every((e) => e.id.startsWith("w_mix_"))).toBe(true);
+    } finally {
+      await store.quit();
+    }
+  });
+});

internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.ts

@@ -163,6 +163,12 @@ function parseJson<T>(raw: string | undefined): T | undefined {
   return raw ? (JSON.parse(raw) as T) : undefined;
 }
 
+/**
+ * How many envelope reads run concurrently. Matches the chunk the snapshot hydration uses for the
+ * same shape of read, so a large fan-in bounds Redis client pressure instead of bursting.
+ */
+const ENVELOPE_READ_CHUNK_SIZE = 100;
+
 export class WaitpointStoreCoordinator {
   private readonly redis: Redis;
   private readonly logger: Logger;
@@ -527,9 +533,18 @@ export class WaitpointStoreCoordinator {
       return [];
     }
 
-    const halves = await Promise.all(
-      waitpointIds.map((id) => this.redis.hmget(waitpointKeys(id).record, "r", "c"))
-    );
+    // Chunked, not one Promise.all over the whole set. A 1000-item batch fan-in would otherwise
+    // launch 1000 concurrent HMGETs at once, and the burst is the cost even though each command
+    // is small. The bound mirrors the snapshot hydration's own WAITPOINT_CHUNK_SIZE, and stays
+    // per-command so nothing can span two cluster slots.
+    const halves: (string | null)[][] = [];
+    for (let i = 0; i < waitpointIds.length; i += ENVELOPE_READ_CHUNK_SIZE) {
+      const chunk = waitpointIds.slice(i, i + ENVELOPE_READ_CHUNK_SIZE);
+      const settled = await Promise.all(
+        chunk.map((id) => this.redis.hmget(waitpointKeys(id).record, "r", "c"))
+      );
+      halves.push(...settled);
+    }
 
     const out: CompletionEnvelopeSource[] = [];