feat(run-store): wait-cycle waitpoint id reads

internal-packages/run-store/src/redisSnapshotStore.test.ts

@@ -243,3 +243,104 @@ describe("append", () => {
     }
   );
 });
+
+describe("cycle keys", () => {
+  redisTest(
+    "mints an increasing cycleSeq across successive new cycles",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      try {
+        await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false });
+        const a = await store.append({
+          entry: entry({ id: "snap_2" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: { kind: "new", completedWaitpoints: [{ id: "w_a", index: 0 }] },
+        });
+        const b = await store.append({
+          entry: entry({ id: "snap_3" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: { kind: "new", completedWaitpoints: [{ id: "w_b", index: 0 }] },
+        });
+        expect(a).toMatchObject({ cycleSeq: 1 });
+        expect(b).toMatchObject({ cycleSeq: 2 });
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+
+  redisTest(
+    "a carry-forward reuses the cycle and does not rewrite it",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+      try {
+        await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false });
+        await store.append({
+          entry: entry({ id: "snap_2" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: {
+            kind: "new",
+            completedWaitpoints: [
+              { id: "w_a", index: 0 },
+              { id: "w_a", index: 1 },
+            ],
+          },
+        });
+        const carried = await store.append({
+          entry: entry({ id: "snap_3" }),
+          kind: "transition",
+          isTerminal: false,
+          cycle: { kind: "carryForward", cycleSeq: 1 },
+        });
+        expect(carried).toMatchObject({ cycleSeq: 1, cycleMismatch: false });
+
+        // Both entries resolve to the SAME cycle contents, written once.
+        const first = await store.getSnapshotWaitpointIds("run_1", "snap_2");
+        const second = await store.getSnapshotWaitpointIds("run_1", "snap_3");
+        expect(first.order).toEqual(["w_a", "w_a"]);
+        expect(first.distinctIds).toEqual(["w_a"]);
+        expect(second).toEqual(first);
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+
+  redisTest("a carry-forward naming a missing cycle still appends", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false });
+      const r = await store.append({
+        entry: entry({ id: "snap_2" }),
+        kind: "transition",
+        isTerminal: false,
+        cycle: { kind: "carryForward", cycleSeq: 99 },
+      });
+      expect(r).toMatchObject({ outcome: "written", cycleMismatch: true });
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest("reports presence and emptiness separately", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
+    try {
+      await store.append({ entry: entry({ id: "snap_1" }), kind: "birth", isTerminal: false });
+      expect(await store.getSnapshotWaitpointIds("run_1", "nope")).toEqual({
+        present: false,
+        distinctIds: [],
+        order: [],
+      });
+      expect(await store.getSnapshotWaitpointIds("run_1", "snap_1")).toEqual({
+        present: true,
+        distinctIds: [],
+        order: [],
+      });
+    } finally {
+      await store.quit();
+    }
+  });
+});

internal-packages/run-store/src/redisSnapshotStore.ts

@@ -272,6 +272,18 @@ export class RedisSnapshotStore {
     });
   }
 
+  // Returns all three shapes the Postgres surface needs from one read: `distinctIds` matches the
+  // deduped join that findSnapshotCompletedWaitpointIds returns, `present` serves the WithPresence
+  // variant (which distinguishes "no waitpoints" from "snapshot not visible"), and `order` keeps the
+  // repeats that the engine expands into one CompletedWaitpoint per position.
+  async getSnapshotWaitpointIds(runId: string, snapshotId: string): Promise<WaitpointIds> {
+    return this.#timed("getSnapshotWaitpointIds", async () => {
+      const k = snapshotKeys(runId);
+      const reply = await this.redis.readSnapshotWaitpointIds(k.e, k.idx, k.cur, k.seq, snapshotId);
+      return decodeWaitpointIds(reply[0] === "1", reply[1] ?? "");
+    });
+  }
+
   // [id, raw, seq, pointer, order] -> SnapshotRead. The environment compare is app-side, per the
   // plan: the store returns null for a foreign environment and the 404 throw stays in the engine.
   #decode(reply: string[] | null, environmentId?: string): SnapshotRead | null {
@@ -437,6 +449,19 @@ export class RedisSnapshotStore {
         return { cur, vals[1], vals[2] or '', vals[3] or '', orderFor(vals[3]) }
       `,
     });
+
+    this.redis.defineCommand("readSnapshotWaitpointIds", {
+      numberOfKeys: 4,
+      lua: `
+        ${PRELUDE}
+        local id = ARGV[1]
+        if redis.call('HEXISTS', eKey, id) == 0 then
+          return { '0', '' }
+        end
+        local pointer = redis.call('HGET', eKey, id .. '#c')
+        return { '1', orderFor(pointer) }
+      `,
+    });
   }
 }
 
@@ -482,5 +507,13 @@ declare module "@internal/redis" {
       seqKey: string,
       callback?: Callback<string[] | null>
     ): Result<string[] | null, Context>;
+    readSnapshotWaitpointIds(
+      eKey: string,
+      idxKey: string,
+      curKey: string,
+      seqKey: string,
+      id: string,
+      callback?: Callback<string[]>
+    ): Result<string[], Context>;
   }
 }