fix(run-store): stop serving a short history from a keyspace that lost an append

The repair restores the head but not the entries lost with it, so a
keyspace ends up holed with a correct head. At dual-write that is
invisible and harmless. At redis-read the window read serves a range
straight from Redis, and its guards see a miss and a dangling cycle but
cannot see a HOLE, so a window that should hold eight entries returns
four with nothing logged. A history that is short rather than wrong is
the harder kind to notice.

The keyspace now records that its history is untrustworthy and both
window commands refuse, which routes the caller through its existing
miss path to Postgres. Point reads are left alone, because the repair
does guarantee the head converges and refusing those would send every
transition of a once-forked run to Postgres for the rest of its life.

Backfilling instead would be worse. A late append takes a fresh
sequence number, and the window scripts walk the index in sequence order
as though it were time order, so an old entry with a high sequence
truncates the window harder than the hole does.

A fork sets the marker itself, and so do the repair's early exits. The
head converging on its own is exactly the case that hid this: four
entries against eight, with a matching head, and nothing to say so. The
marker is a field on the seq hash, so the keyspace expiry governs it and
it is only ever set on a keyspace that already exists.

Author: d-cs

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

@@ -0,0 +1,179 @@
+// A5. The repair restores the head but not the entries lost in the fork window, so a keyspace ends up
+// with a hole in the middle and a correct head. That is invisible at dual-write, where Postgres is
+// authoritative and the engine reads the head. It is not invisible at redis-read: the window read
+// serves a since-createdAt range straight from Redis, and its guards (a miss, a dangling cycle)
+// cannot see a HOLE, so a window that should hold eight entries returns four with nothing logged. A
+// history that is short rather than wrong is the harder kind to notice.
+//
+// Backfilling the lost entries is NOT the fix and would be worse. A late append takes a fresh seq
+// from HINCRBY, and both window scripts walk the index in seq order treating it as time order,
+// stopping at the first entry past the cursor. A backfilled old entry with a high seq would truncate
+// the window harder than the hole does.
+//
+// So the keyspace records that its history is untrustworthy and windows refuse, which routes the
+// caller through its existing miss path to Postgres. Point reads stay Redis-served, because the
+// repair does guarantee the head converges.
+import { describe, expect } from "vitest";
+import { redisTest } from "@internal/testcontainers";
+import { RedisSnapshotStore, snapshotKeys } from "./redisSnapshotStore.js";
+import type { SnapshotEntryInput } from "./redisSnapshotStore.js";
+import { createRedisClient } from "@internal/redis";
+
+const COMPLETED_TTL_MS = 72 * 60 * 60 * 1000;
+
+function entry(runId: string, id: string, createdAt: string): SnapshotEntryInput {
+  return {
+    id,
+    engine: "V2",
+    executionStatus: "EXECUTING",
+    description: "d",
+    runId,
+    runStatus: "EXECUTING",
+    createdAt,
+    environmentId: "env_1",
+    environmentType: "DEVELOPMENT",
+    projectId: "proj_1",
+    organizationId: "org_1",
+  };
+}
+
+const at = (seconds: number) => new Date(Date.UTC(2026, 0, 1, 0, 0, seconds)).toISOString();
+
+async function seed(store: RedisSnapshotStore, runId: string, count: number): Promise<void> {
+  for (let n = 0; n < count; n++) {
+    await store.append({
+      entry: entry(runId, `snap_${n}`, at(n)),
+      kind: n === 0 ? "birth" : "transition",
+      isTerminal: false,
+    });
+  }
+}
+
+describe("the gaps marker", () => {
+  redisTest(
+    "is unset on a healthy keyspace, which still serves its window",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS });
+      try {
+        const runId = "run_healthy";
+        await seed(store, runId, 5);
+
+        expect(await store.hasGaps(runId)).toBe(false);
+        expect((await store.getSinceCreatedAt(runId, at(1))).kind).toBe("hit");
+        expect((await store.getSince(runId, "snap_1")).kind).toBe("hit");
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+
+  redisTest("makes BOTH window reads refuse, so each falls back", async ({ redisOptions }) => {
+    const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS });
+    try {
+      const runId = "run_holed";
+      await seed(store, runId, 5);
+
+      // What a repair does when it lands: the head is right, the window is not to be trusted.
+      await store.markGaps(runId);
+      expect(await store.hasGaps(runId)).toBe(true);
+
+      // Both window commands, because a caller that fell back on one and not the other would still
+      // serve a short history through the second.
+      expect((await store.getSinceCreatedAt(runId, at(1))).kind).toBe("miss");
+      expect((await store.getSince(runId, "snap_1")).kind).toBe("miss");
+    } finally {
+      await store.quit();
+    }
+  });
+
+  redisTest(
+    "leaves point reads alone, because the repair converges the head",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS });
+      try {
+        const runId = "run_point";
+        await seed(store, runId, 5);
+        await store.markGaps(runId);
+
+        // The head is the engine's hot read and the repair guarantees it. Refusing it would send every
+        // transition of a once-forked run to Postgres for the rest of its life.
+        const head = await store.getLatest(runId);
+        expect(head?.entry.id).toBe("snap_4");
+
+        const byId = await store.getById(runId, "snap_2");
+        expect(byId?.entry.id).toBe("snap_2");
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+
+  redisTest(
+    "is set by a fork, which is direct evidence of divergence",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS });
+      try {
+        const runId = "run_forked";
+        await seed(store, runId, 3);
+
+        const result = await store.append({
+          entry: entry(runId, "snap_late", at(9)),
+          kind: "transition",
+          isTerminal: false,
+          expectedCur: "snap_wrong",
+        });
+
+        expect(result.outcome).toBe("forked");
+        // A fork means this keyspace and Postgres already disagree about the head, so whatever the
+        // repair does later, the window between them is not trustworthy now.
+        expect(await store.hasGaps(runId)).toBe(true);
+        expect((await store.getSinceCreatedAt(runId, at(1))).kind).toBe("miss");
+      } finally {
+        await store.quit();
+      }
+    }
+  );
+
+  redisTest(
+    "dies with the keyspace rather than needing its own expiry",
+    async ({ redisOptions }) => {
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS });
+      const probe = createRedisClient(redisOptions, { onError: () => {} });
+      try {
+        const runId = "run_ttl";
+        await seed(store, runId, 3);
+        await store.markGaps(runId);
+
+        // The marker is a field on the seq hash, so the completion expiry that governs the keyspace
+        // governs it too. No second lifetime to get wrong.
+        expect(await probe.hget(snapshotKeys(runId).seq, "g")).toBe("1");
+        await store.dropRun(runId);
+        expect(await probe.exists(snapshotKeys(runId).seq)).toBe(0);
+      } finally {
+        await Promise.all([store.quit(), probe.quit().catch(() => {})]);
+      }
+    }
+  );
+  redisTest(
+    "a lone seq key is never created for a run that has no keyspace",
+    async ({ redisOptions }) => {
+      // markGaps writes a field on the seq hash, and HSET creates the hash if it is absent. For a
+      // run with no keyspace that would leave a stray seq key holding only the marker: keyspaceAlive
+      // stays false so no read is affected, but the sweeper scans on the entry hash and would never
+      // discover it, so it would never be reaped either. An unbounded leak with no reader.
+      const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS });
+      const probe = createRedisClient(redisOptions, { onError: () => {} });
+      try {
+        const runId = "run_never_born";
+        expect(await store.markGapsIfResident(runId)).toBe(false);
+        expect(await probe.exists(snapshotKeys(runId).seq)).toBe(0);
+
+        await seed(store, runId, 2);
+        expect(await store.markGapsIfResident(runId)).toBe(true);
+        expect(await store.hasGaps(runId)).toBe(true);
+      } finally {
+        await Promise.all([store.quit(), probe.quit().catch(() => {})]);
+      }
+    }
+  );
+});

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

@@ -344,6 +344,36 @@ export class RedisSnapshotStore {
     return this.#breaker.state;
   }
 
+  /**
+   * Records that this run's Redis history has a hole, so window reads must not serve it. Separate
+   * from the append path because a repair can conclude the head is already current and still know
+   * that entries were lost.
+   */
+  async markGaps(runId: string): Promise<void> {
+    await this.redis.hset(snapshotKeys(runId).seq, "g", "1");
+  }
+
+  /**
+   * Marks only a keyspace that exists, and reports whether it did.
+   *
+   * The unconditional form must not be used on a run whose residency is unknown: HSET creates the
+   * hash, so a non-resident run would be left holding a lone `seq` key with nothing but the marker.
+   * `keyspaceAlive` would stay false so no read would be affected, but the sweeper discovers
+   * keyspaces by scanning for the ENTRY hash, so it would never find that key either. An unbounded
+   * leak with no reader is the one outcome worse than the hole this marker exists to report.
+   */
+  async markGapsIfResident(runId: string): Promise<boolean> {
+    const k = snapshotKeys(runId);
+    return this.#timed("markGapsIfResident", async () => {
+      const marked = await this.redis.markSnapshotGaps(k.e, k.seq);
+      return marked === 1;
+    });
+  }
+
+  async hasGaps(runId: string): Promise<boolean> {
+    return (await this.redis.hget(snapshotKeys(runId).seq, "g")) === "1";
+  }
+
   /** Test seam. */
   residencyFor(runId: string): "resident" | "non-resident" | undefined {
     return this.#residency.get(runId);
@@ -365,6 +395,11 @@ export class RedisSnapshotStore {
     kind: "birth" | "transition";
     isTerminal: boolean;
     expectedCur?: string;
+    /**
+     * Marks the keyspace as holed, so window reads refuse and fall back to Postgres. Set by the
+     * repair, which only runs because an append was lost.
+     */
+    markGaps?: boolean;
     cycle?:
       | {
           kind: "new";
@@ -452,7 +487,8 @@ export class RedisSnapshotStore {
         orderCount,
         args.expectedCur ?? "",
         args.expectedCur !== undefined ? "1" : "0",
-        distinctJson
+        distinctJson,
+        args.markGaps ? "1" : "0"
       )) as string[];
 
       return this.#interpretAppend(reply, raw, orderJson, records, args.entry.runId);
@@ -819,6 +855,21 @@ export class RedisSnapshotStore {
       end
     `;
 
+    this.redis.defineCommand("markSnapshotGaps", {
+      numberOfKeys: 2,
+      lua: `
+        local eKey = KEYS[1]
+        local seqKey = KEYS[2]
+        -- Both anchors, the same pair keyspaceAlive uses. Marking on the strength of one of them
+        -- would create the other.
+        if redis.call('EXISTS', eKey) == 0 or redis.call('EXISTS', seqKey) == 0 then
+          return 0
+        end
+        redis.call('HSET', seqKey, 'g', '1')
+        return 1
+      `,
+    });
+
     this.redis.defineCommand("appendSnapshotEntry", {
       numberOfKeys: 4,
       lua: `
@@ -839,6 +890,9 @@ export class RedisSnapshotStore {
         -- The COMPLETE distinct id set. Not the order deduped: order omits every id with no batch
         -- index, and those ids still have to come back on a read.
         local distinctJson = ARGV[14]
+        -- Set by the repair. A repair exists BECAUSE an append was lost, so whatever it manages to
+        -- put back, the entries between are gone and the window is short.
+        local markGaps    = ARGV[15] == '1' 
 
         -- Checking e alone would let a late transition recreate seq with no TTL and restart it at 1
         -- beside a surviving idx. A birth always creates both in this same script, so this never
@@ -860,10 +914,19 @@ export class RedisSnapshotStore {
         if casEnabled then
           local actual = redis.call('GET', curKey)
           if (actual or '') ~= expectedCur then
+            -- The one mutation a refused append makes, and it is not part of the append. A fork means
+            -- this keyspace and Postgres already disagree about the head, so its history cannot be
+            -- served as a window until something re-establishes that it can. The entry itself is
+            -- still not written.
+            redis.call('HSET', seqKey, 'g', '1')
             return { '${FORKED}', actual or '' }
           end
         end
 
+        if markGaps then
+          redis.call('HSET', seqKey, 'g', '1')
+        end
+
         local seq = redis.call('HINCRBY', seqKey, 'e', 1)
 
         local cycleSeq = 0
@@ -1019,6 +1082,11 @@ export class RedisSnapshotStore {
         -- on every poll for the rest of the run's life, with Postgres holding the transitions.
         if not keyspaceAlive() or redis.call('EXISTS', idxKey) == 0 then return nil end
 
+        -- A keyspace that lost an append has a hole, and no guard downstream can see one: a window
+        -- that should hold eight entries would return four and look complete. Refuse, and the
+        -- caller's existing miss path asks Postgres, which still holds the whole log.
+        if redis.call('HGET', seqKey, 'g') == '1' then return nil end
+
         -- STRICTLY greater than the cursor, and same-millisecond entries are dropped. Postgres
         -- serves this window with createdAt > cursor and drops them too; a Redis read that is more
         -- correct than the Postgres read shows up as divergence in compare mode.
@@ -1086,6 +1154,10 @@ export class RedisSnapshotStore {
         -- so the caller stops asking Postgres for a window Postgres alone still holds.
         if not keyspaceAlive() or redis.call('EXISTS', idxKey) == 0 then return nil end
 
+        -- And the same hole gate, for the same reason. A caller that fell back on one window command
+        -- and not the other would still serve a short history through the second.
+        if redis.call('HGET', seqKey, 'g') == '1' then return nil end
+
         -- The index holds valid entries only, so an invalid since id misses ZSCORE. Its seq is still
         -- on its own '#s' field, which keeps the id resolvable without indexing invalid rows.
         local score = redis.call('ZSCORE', idxKey, sinceId)
@@ -1165,6 +1237,11 @@ declare module "@internal/redis" {
       seqKey: string,
       callback?: Callback<number>
     ): Result<number, Context>;
+    markSnapshotGaps(
+      eKey: string,
+      seqKey: string,
+      callback?: Callback<number>
+    ): Result<number, Context>;
     appendSnapshotEntry(
       eKey: string,
       idxKey: string,
@@ -1184,6 +1261,7 @@ declare module "@internal/redis" {
       expectedCur: string,
       casEnabled: string,
       distinctJson: string,
+      markGaps: string,
       callback?: Callback<string[]>
     ): Result<string[], Context>;
     readSnapshotById(

internal-packages/run-store/src/taskRunExecutionSnapshotStore.repair.test.ts

@@ -101,6 +101,14 @@ class RecordingRedis {
   async dropRun(): Promise<void> {
     this.calls.push("dropRun");
   }
+
+  gapsMarked = 0;
+
+  async markGapsIfResident(): Promise<boolean> {
+    this.calls.push("markGapsIfResident");
+    this.gapsMarked += 1;
+    return true;
+  }
 }
 
 function redisHead(id: string, createdAt: Date): SnapshotRead {
@@ -284,7 +292,7 @@ describe("repairRedisHead", () => {
     expect(redis.appends[0]!.kind).toBe("transition");
   });
 
-  it("does nothing when the mirror already holds the Postgres head", async () => {
+  it("appends nothing when the mirror already holds the Postgres head, but still marks the hole", async () => {
     const redis = new RecordingRedis(redisHead("snap_lost", new Date("2026-01-01T00:00:10.000Z")));
     const store = harness({
       pg: pgHead({
@@ -298,9 +306,13 @@ describe("repairRedisHead", () => {
 
     await expect(store.repairRedisHead("run_1", "snap_lost")).resolves.toBe("alreadyCurrent");
     expect(redis.appends).toHaveLength(0);
+    // The head converged on its own, but the repair only ran because an append was lost, so entries
+    // behind it can still be missing. Observed live: four entries in Redis against eight in
+    // Postgres, with a matching head. Without the mark, that keyspace serves short windows as whole.
+    expect(redis.gapsMarked).toBe(1);
   });
 
-  it("refuses to append behind a mirror head that is newer than the Postgres head", async () => {
+  it("refuses to append behind a newer mirror head, and marks the hole", async () => {
     // Appending an older entry at the tail would leave the chain claiming a state the run has left.
     const redis = new RecordingRedis(redisHead("snap_newer", new Date("2026-01-01T00:00:20.000Z")));
     const store = harness({
@@ -315,6 +327,8 @@ describe("repairRedisHead", () => {
 
     await expect(store.repairRedisHead("run_1", "snap_lost")).resolves.toBe("redisAhead");
     expect(redis.appends).toHaveLength(0);
+    // Divergence either way round is still divergence.
+    expect(redis.gapsMarked).toBe(1);
   });
 
   it("heals the head even when the run has transitioned past the lost snapshot", async () => {