From a5ad85a36bb18cec3f3a29af3ba162db439b081e Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Fri, 21 Aug 2026 14:10:22 +0100 Subject: [PATCH] fix(run-store): pair the head waitpoint order with the surviving head row --- .../run-store/src/redisSnapshotStore.test.ts | 36 +++++++++++++++++++ .../run-store/src/redisSnapshotStore.ts | 14 +++++--- 2 files changed, 45 insertions(+), 5 deletions(-) diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts index 00615482d..72f67a463 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.test.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts @@ -688,6 +688,42 @@ describe("getSince", () => { } ); + redisTest( + "does not donate the evicted head's waitpoints to the surviving head", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 }); + const raw = createRedisClient(redisOptions); + try { + await store.append({ entry: entry({ id: "s0" }), kind: "birth", isTerminal: false }); + await store.append({ + entry: entry({ id: "s1" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_old", index: 0 }] }, + }); + await store.append({ + entry: entry({ id: "s2" }), + kind: "transition", + isTerminal: false, + cycle: { kind: "new", completedWaitpoints: [{ id: "w_new", index: 0 }] }, + }); + + // s2 is the newest and its body is gone. s1 must come back with ITS OWN waitpoints, + // never s2's -- a dropped row must not donate its cycle data to the next one. + await raw.hdel("snap:{run_1}:e", "s2"); + + const r = await store.getSince("run_1", "s0"); + expect(r.kind).toBe("hit"); + if (r.kind !== "hit") throw new Error("unreachable"); + expect(r.entries.map((e) => e.id)).toEqual(["s1"]); + expect(r.headWaitpointIds.order).toEqual(["w_old"]); + } finally { + await raw.quit(); + await store.quit(); + } + } + ); + redisTest( "hits with zero entries when scoped to the since entry's own environment", async ({ redisOptions }) => { diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts index e76963ef3..b0190c3da 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -542,21 +542,25 @@ export class RedisSnapshotStore { local ids = redis.call('ZREVRANGEBYSCORE', idxKey, '+inf', '(' .. score, 'LIMIT', 0, limit) if #ids == 0 then return { sinceRaw, '' } end - -- The head is the newest entry, and it is the ONLY one whose cycle key is read. That makes - -- head-only hydration structural: the tail's cycle keys are never touched. - local out = { sinceRaw, orderFor(redis.call('HGET', eKey, ids[1] .. '#c')) } + -- The head is the newest SURVIVING entry, and it is the only one whose cycle key is read. + -- Deriving the order after the loop keeps it paired with the row it is attached to: a row + -- dropped for a missing body must not donate its cycle data to the next one. + local out = { sinceRaw, '' } + local headId = nil for i = 1, #ids do local id = ids[i] local vals = redis.call('HMGET', eKey, id, id .. '#s', id .. '#c') - -- A nil body (e survived only partially, eg. idx outlived e) must drop the row, not emit - -- an unparseable '' that would throw out of JSON.parse in #decode. if vals[1] then + if not headId then headId = id end out[#out + 1] = id out[#out + 1] = vals[1] out[#out + 1] = vals[2] or '' out[#out + 1] = vals[3] or '' end end + if headId then + out[2] = orderFor(redis.call('HGET', eKey, headId .. '#c')) + end return out `, });