fix(run-store): clear a stale records field when a new wait cycle reuses a key
The seq counter can vanish under maxmemory eviction while a wp:<n> key survives. A birth does not check seq, unlike a transition, so the counter restarts and re-mints a cycleSeq whose key still holds another cycle's records. order and count are both overwritten, so they stay consistent with each other and the mismatch check cannot see the drift. A resolver would then read a previous cycle's records paired with a new order.
This commit is contained in:
@@ -278,6 +278,54 @@ describe("append", () => {
|
||||
}
|
||||
);
|
||||
|
||||
redisTest(
|
||||
"a recordless new cycle clears another cycle's records off a reused key",
|
||||
async ({ redisOptions }) => {
|
||||
const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 1000 });
|
||||
const raw = createRedisClient(redisOptions);
|
||||
try {
|
||||
await store.append({
|
||||
entry: entry({ id: "snap_1" }),
|
||||
kind: "birth",
|
||||
isTerminal: false,
|
||||
cycle: {
|
||||
kind: "new",
|
||||
completedWaitpoints: [{ id: "w_a", index: 0 }],
|
||||
records: [
|
||||
{
|
||||
id: "w_a",
|
||||
friendlyId: "waitpoint_a",
|
||||
type: "MANUAL",
|
||||
completedAt: "2026-01-01T00:00:00.000Z",
|
||||
outputType: "application/json",
|
||||
outputIsError: false,
|
||||
output: { inline: "stale" },
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
expect(await raw.hget("snap:{run_1}:wp:1", "records")).not.toBeNull();
|
||||
|
||||
// Only the counter is lost, as under maxmemory eviction. A birth does not check seq, so
|
||||
// the next new cycle re-mints cycleSeq 1 onto the surviving key.
|
||||
await raw.del("snap:{run_1}:seq");
|
||||
|
||||
await store.append({
|
||||
entry: entry({ id: "snap_2" }),
|
||||
kind: "birth",
|
||||
isTerminal: false,
|
||||
cycle: { kind: "new", completedWaitpoints: [{ id: "w_b", index: 0 }] },
|
||||
});
|
||||
|
||||
expect(await raw.hget("snap:{run_1}:wp:1", "order")).toBe(JSON.stringify(["w_b"]));
|
||||
expect(await raw.hget("snap:{run_1}:wp:1", "records")).toBeNull();
|
||||
} finally {
|
||||
raw.disconnect();
|
||||
await store.quit();
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
redisTest(
|
||||
"reports a duplicate id without overwriting the original entry",
|
||||
async ({ redisOptions }) => {
|
||||
|
||||
@@ -571,6 +571,11 @@ export class RedisSnapshotStore {
|
||||
redis.call('HSET', wpKey(cycleSeq), 'order', orderJson, 'count', orderCount)
|
||||
if records ~= '' then
|
||||
redis.call('HSET', wpKey(cycleSeq), 'records', records)
|
||||
else
|
||||
-- A new cycle owns the whole key: a lost seq counter can re-mint a cycleSeq whose key
|
||||
-- still holds another cycle's records, and order/count stay mutually consistent so the
|
||||
-- mismatch check cannot see it. No-op on a fresh key.
|
||||
redis.call('HDEL', wpKey(cycleSeq), 'records')
|
||||
end
|
||||
elseif cycleMode == 'carry' then
|
||||
cycleSeq = cycleSeqIn
|
||||
|
||||
Reference in New Issue
Block a user