diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts index d3be34b19..e279e4f80 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.test.ts +++ b/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(); + } + }); +}); diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts index 5aa7e8b97..35d828b29 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/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 { + 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 ): Result; + readSnapshotWaitpointIds( + eKey: string, + idxKey: string, + curKey: string, + seqKey: string, + id: string, + callback?: Callback + ): Result; } }