feat(run-store,testcontainers): execution-snapshot read comparator and shared test utilities (#4772)

## Summary Adds the read comparator for the in-progress migration of the run execution-snapshot log from Postgres to Redis. The comparator samples a single read against both stores, normalizes the two results to one shape, and reports any per-field difference with a tagged metric. It never serves a read itself: the diff layer imports only types, so it cannot hold a store client, and a test enforces that by failing if any value import appears. Also adds a combined Postgres-and-Redis test fixture and two shared test utilities (a cluster-slot assertion and a generic fault-injection harness) that the parallel Redis-store work reuses. Everything here is inert. Nothing constructs the comparator, so merging changes no runtime behavior. It becomes active only when a later change turns on compare mode. ## Notes The divergence classes separate real differences (scalar, ordering, waitpoint id set, validity, missing on one side) from two expected classes that must not be driven to zero: a rotated idempotency key, and a Redis-only surplus at a since-cursor tie. The since comparison is direction sensitive: a Postgres-only entry at the cursor is always a lost write, never an expected tie.
2026-08-25 14:01:24 +01:00
parent 036cf8d2c8
commit 45eaaa7bd7
11 changed files with 850 additions and 35 deletions
@@ -3,3 +3,4 @@ export * from "./PostgresRunStore.js";
 export * from "./runOpsStore.js";
 export * from "./readReplicaClient.js";
 export * from "./redisSnapshotStore.js";
+export * from "./snapshotComparator.js";
@@ -1,7 +1,7 @@
 // Unit suite for the raw Redis execution-snapshot store. Redis-only: the store holds no Prisma
 // reference, so no Postgres container is needed.
 import { expect, describe, vi } from "vitest";
-import { redisTest } from "@internal/testcontainers";
+import { redisTest, slotOf } from "@internal/testcontainers";
 import { createRedisClient } from "@internal/redis";
 import { Logger } from "@trigger.dev/core/logger";
 import {
@@ -1283,45 +1283,23 @@ describe("expectedCur compare-and-set", () => {
  );
 });

-// CRC16/XMODEM over a key's hash tag, per Redis's cluster hashing rule. CLUSTER KEYSLOT is
-// unavailable on this standalone container ("cluster support disabled"), so the slot is computed
-// here instead. Verified against the `cluster-key-slot` package's output for our key shapes.
-function crc16(str: string): number {
-  let crc = 0;
-  for (let i = 0; i < str.length; i++) {
-    crc ^= str.charCodeAt(i) << 8;
-    for (let j = 0; j < 8; j++) {
-      crc = crc & 0x8000 ? (crc << 1) ^ 0x1021 : crc << 1;
-      crc &= 0xffff;
-    }
-  }
-  return crc;
-}
-
-function hashSlot(key: string): number {
-  const start = key.indexOf("{");
-  const end = start === -1 ? -1 : key.indexOf("}", start + 1);
-  const tag = start !== -1 && end !== -1 && end > start + 1 ? key.slice(start + 1, end) : key;
-  return crc16(tag) % 16384;
-}
-
 describe("hash tag and keyPrefix", () => {
  it("every key for one run lands in one cluster slot", () => {
    // Keys come from snapshotKeys() plus the wp:<n> suffix the Lua prelude derives the same way,
    // with a keyPrefix prepended by hand as ioredis would. A dropped hash tag would split the slots.
-    // Pin the helper itself before trusting it: the published XMODEM check value, and two known
+    // Pin the shared helper before trusting it: the published XMODEM check value, and two known
    // slots (one matching cluster-key-slot, one a different run's tag as a negative control --
    // otherwise a constant-valued crc16 would satisfy slots.size === 1 for the wrong reason).
-    expect(crc16("123456789")).toBe(0x31c3);
-    expect(hashSlot("engine:snap:{run_1}:e")).toBe(8108);
-    expect(hashSlot("engine:snap:{run_2}:e")).toBe(12239);
+    expect(slotOf("123456789")).toBe(0x31c3);
+    expect(slotOf("engine:snap:{run_1}:e")).toBe(8108);
+    expect(slotOf("engine:snap:{run_2}:e")).toBe(12239);

    const k = snapshotKeys("run_1");
    const base = k.e.slice(0, -2);
    const keys = [k.e, k.idx, k.cur, k.seq, `${base}:wp:1`, `${base}:wp:2`].map(
      (key) => `engine:${key}`
    );
-    const slots = new Set(keys.map(hashSlot));
+    const slots = new Set(keys.map(slotOf));
    expect(slots.size).toBe(1);
  });

@@ -0,0 +1,57 @@
+// Proves the Frozen rule: the comparator's VALUE-import set is empty. Every import it has is
+// `import type`, erased at runtime, so the compiled module pulls in no Redis or Prisma client and
+// cannot read. Goes red the instant any value import is added — a client, the barrel, or a dynamic
+// import(). The detector is pinned against redisSnapshotStore.ts (which value-imports a client) so
+// this cannot pass as a tautology.
+import { expect, it, describe } from "vitest";
+import { readFileSync } from "node:fs";
+import { fileURLToPath } from "node:url";
+import { dirname, resolve } from "node:path";
+
+const here = dirname(fileURLToPath(import.meta.url));
+
+// Returns the module specifiers a file imports FOR VALUE (i.e. that survive to runtime). `import type`
+// declarations and named blocks whose specifiers are all inline `type` are erased and excluded.
+function valueImports(sourcePath: string): string[] {
+  const raw = readFileSync(sourcePath, "utf8");
+  const out: string[] = [];
+
+  // Statements are scanned on RAW source, anchored to line start (`^\s*import`), so a `//` comment
+  // line never matches and no stripping can hide a real import. Only the mid-line dynamic `import(`
+  // check runs on comment-stripped source. The pin test below guarantees the scan catches a real import.
+  const stripped = raw.replace(/\/\*[\s\S]*?\*\//g, "").replace(/\/\/.*$/gm, "");
+  if (/(^|[^.\w])import\s*\(/.test(stripped)) out.push("<dynamic import()>");
+
+  const importRe = /^\s*import\b([\s\S]*?)\bfrom\s*["']([^"']+)["']/gm;
+  for (let m = importRe.exec(raw); m !== null; m = importRe.exec(raw)) {
+    const clause = m[1];
+    const spec = m[2];
+    if (/^\s*type\b/.test(clause)) continue; // `import type ... from`
+    const named = clause.match(/\{([\s\S]*?)\}/);
+    // Strip inline `type Foo` specifiers, including an `as Bar` alias, before checking whether any
+    // value specifier remains.
+    const inlineType = /\btype\s+[A-Za-z_$][\w$]*(?:\s+as\s+[A-Za-z_$][\w$]*)?/g;
+    if (named && !/(^|,)\s*[A-Za-z_$]/.test(named[1].replace(inlineType, ""))) {
+      continue; // every named specifier is an inline `type` — nothing left for value
+    }
+    out.push(spec);
+  }
+
+  // Bare side-effect imports (`import "x"`) run the module.
+  const bareRe = /^\s*import\s*["']([^"']+)["']/gm;
+  for (let m = bareRe.exec(raw); m !== null; m = bareRe.exec(raw)) out.push(m[1]);
+
+  return out;
+}
+
+describe("comparator read-isolation", () => {
+  it("the detector flags a real value import (pin against the store)", () => {
+    // redisSnapshotStore.ts value-imports @internal/redis, so a working detector MUST see it.
+    const storeImports = valueImports(resolve(here, "redisSnapshotStore.ts"));
+    expect(storeImports).toContain("@internal/redis");
+  });
+
+  it("the comparator has no value imports — it is import-type-only and cannot read", () => {
+    expect(valueImports(resolve(here, "snapshotComparator.ts"))).toEqual([]);
+  });
+});
@@ -0,0 +1,254 @@
+import { expect, it, describe } from "vitest";
+import {
+  diffLatest,
+  diffSince,
+  normalizeFromRedis,
+  normalizeFromPg,
+  SnapshotComparator,
+  type DivergenceClass,
+  type NormalizedSnapshot,
+} from "./snapshotComparator.js";
+import type { SnapshotRead } from "./redisSnapshotStore.js";
+
+function norm(over: Partial<NormalizedSnapshot> = {}): NormalizedSnapshot {
+  const base: NormalizedSnapshot = {
+    id: "s1",
+    engine: "V2",
+    executionStatus: "RUN_CREATED",
+    description: "d",
+    isValid: true,
+    error: null,
+    previousSnapshotId: null,
+    runId: "r1",
+    runStatus: "PENDING",
+    batchId: null,
+    attemptNumber: null,
+    environmentId: "env",
+    environmentType: "DEVELOPMENT",
+    projectId: "p",
+    organizationId: "o",
+    checkpointId: null,
+    workerId: null,
+    runnerId: null,
+    createdAt: 1000,
+    updatedAt: 1000,
+    metadata: null,
+    completedWaitpointOrder: [],
+    waitpointIdSet: [],
+  };
+  return { ...base, ...over };
+}
+
+describe("diffLatest", () => {
+  it("no divergence when the two sides match", () => {
+    expect(diffLatest(norm(), norm())).toEqual([]);
+  });
+
+  it("reports a scalar difference by field", () => {
+    expect(diffLatest(norm(), norm({ executionStatus: "EXECUTING" }))).toEqual([
+      { field: "executionStatus", class: "scalar", pg: "RUN_CREATED", redis: "EXECUTING" },
+    ]);
+  });
+
+  it("compares createdAt and updatedAt by strict equality", () => {
+    expect(diffLatest(norm(), norm({ createdAt: 1001 }))).toEqual([
+      { field: "createdAt", class: "scalar", pg: 1000, redis: 1001 },
+    ]);
+  });
+
+  it("classifies a validity mismatch", () => {
+    const d = diffLatest(norm({ isValid: true }), norm({ isValid: false, error: "boom" }));
+    expect(d.map((x) => x.field).sort()).toEqual(["error", "isValid"]);
+    expect(d.find((x) => x.field === "isValid")!.class).toBe("validity");
+  });
+
+  it("classifies completedWaitpointOrder differences as order, repeats significant", () => {
+    expect(
+      diffLatest(
+        norm({ completedWaitpointOrder: ["a", "a", "b"] }),
+        norm({ completedWaitpointOrder: ["a", "b"] })
+      )
+    ).toEqual([
+      { field: "completedWaitpointOrder", class: "order", pg: ["a", "a", "b"], redis: ["a", "b"] },
+    ]);
+  });
+
+  it("classifies waitpoint id set differences, order-insensitive", () => {
+    expect(
+      diffLatest(norm({ waitpointIdSet: ["a", "b"] }), norm({ waitpointIdSet: ["a", "b"] }))
+    ).toEqual([]);
+    const d2 = diffLatest(norm({ waitpointIdSet: ["a", "b"] }), norm({ waitpointIdSet: ["a"] }));
+    expect(d2[0]).toMatchObject({ field: "waitpointIdSet", class: "waitpointIdSet" });
+  });
+
+  it("does NOT emit a divergence for a rotated idempotency key — invisible at id-set granularity", () => {
+    expect(diffLatest(norm({ waitpointIdSet: ["w1"] }), norm({ waitpointIdSet: ["w1"] }))).toEqual(
+      []
+    );
+  });
+
+  it("missingInRedis when the row exists only in Postgres", () => {
+    expect(diffLatest(norm(), null)).toEqual([
+      expect.objectContaining({ class: "missingInRedis" }),
+    ]);
+  });
+
+  it("missingInPg when the row exists only in Redis", () => {
+    expect(diffLatest(null, norm())).toEqual([expect.objectContaining({ class: "missingInPg" })]);
+  });
+
+  it("raises unknownField for a key on neither the compared nor excluded list", () => {
+    const d = diffLatest(norm(), { ...norm(), somethingNew: 1 } as NormalizedSnapshot);
+    expect(d).toEqual([expect.objectContaining({ field: "somethingNew", class: "unknownField" })]);
+  });
+
+  it("normalizeFromRedis carries an unrecognised entry field, so unknownField fires on real input", () => {
+    const read: SnapshotRead = {
+      id: "s1",
+      seq: 1,
+      isValid: true,
+      raw: "{}",
+      entry: {
+        engine: "V2",
+        executionStatus: "RUN_CREATED",
+        description: "d",
+        runId: "r1",
+        runStatus: "PENDING",
+        createdAt: "2026-08-24T00:00:00.000Z",
+        environmentId: "env",
+        environmentType: "DEVELOPMENT",
+        projectId: "p",
+        organizationId: "o",
+        mysteryField: "surprise",
+      },
+    };
+    const redis = normalizeFromRedis(read);
+    expect(redis.mysteryField).toBe("surprise"); // not dropped by normalization
+    const d = diffLatest(
+      norm({ id: "s1", createdAt: redis.createdAt, updatedAt: redis.updatedAt }),
+      redis
+    );
+    expect(d).toEqual([
+      expect.objectContaining({ field: "mysteryField", class: "unknownField", redis: "surprise" }),
+    ]);
+  });
+
+  it("surfaces an inherited-name key and does not pollute the prototype", () => {
+    // JSON.parse produces OWN keys for `toString` and `__proto__` (unlike an object literal).
+    const entry = JSON.parse(
+      '{"engine":"V2","executionStatus":"RUN_CREATED","description":"d","runId":"r1",' +
+        '"runStatus":"PENDING","createdAt":"2026-08-24T00:00:00.000Z","environmentId":"env",' +
+        '"environmentType":"DEVELOPMENT","projectId":"p","organizationId":"o",' +
+        '"toString":"surprise","__proto__":{"polluted":true}}'
+    ) as Record<string, unknown>;
+    const read: SnapshotRead = { id: "s1", seq: 1, isValid: true, raw: "{}", entry };
+    const n = normalizeFromRedis(read) as Record<string, unknown>;
+
+    expect(Object.keys(n)).toContain("toString"); // carried as an own key despite the inherited name
+    expect(n["toString"]).toBe("surprise");
+    expect(Object.getPrototypeOf(n)).toBe(Object.prototype); // __proto__ skipped, no pollution
+    expect("polluted" in {}).toBe(false);
+
+    const d = diffLatest(
+      norm({ id: "s1", createdAt: n.createdAt as number, updatedAt: n.updatedAt as number }),
+      n as NormalizedSnapshot
+    );
+    expect(d.some((x) => x.field === "toString" && x.class === "unknownField")).toBe(true);
+  });
+
+  it("normalizeFromPg's waitpointIdSet is index-bearing only, matching the Redis read surface", () => {
+    // A non-indexed completed waitpoint is in the relation but not in completedWaitpointOrder; Redis's
+    // distinctIds (dedupe of order) does not expose it, so the PG side must not either.
+    const row = {
+      id: "s1",
+      engine: "V2",
+      executionStatus: "EXECUTING",
+      description: "d",
+      isValid: true,
+      error: null,
+      previousSnapshotId: null,
+      runId: "r1",
+      runStatus: "EXECUTING",
+      batchId: null,
+      attemptNumber: null,
+      environmentId: "env",
+      environmentType: "DEVELOPMENT",
+      projectId: "p",
+      organizationId: "o",
+      checkpointId: null,
+      workerId: null,
+      runnerId: null,
+      createdAt: new Date(1000),
+      updatedAt: new Date(1000),
+      metadata: null,
+      completedWaitpointOrder: ["w_indexed"],
+      completedWaitpoints: [{ id: "w_indexed" }, { id: "w_nonindexed" }],
+    };
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const n = normalizeFromPg(row as any);
+    expect(n.waitpointIdSet).toEqual(["w_indexed"]);
+  });
+});
+
+describe("diffSince", () => {
+  const cursor = { id: "s1", createdAtMs: 1000 };
+
+  it("a Postgres-only entry at the cursor ms is a lost append (missingInRedis), never a tie", () => {
+    const pg = [norm({ id: "s2", createdAt: 1000, previousSnapshotId: "s1" })];
+    expect(diffSince({ pg, redis: [], cursor })).toEqual([
+      expect.objectContaining({ field: "s2", class: "missingInRedis" }),
+    ]);
+  });
+
+  it("a Redis-only chain-boundary surplus at the cursor ms is expected:redisSurplusAtCursorTie", () => {
+    const redis = [norm({ id: "s2", createdAt: 1000, previousSnapshotId: "s1" })];
+    expect(diffSince({ pg: [], redis, cursor })).toEqual([
+      expect.objectContaining({ field: "s2", class: "expected:redisSurplusAtCursorTie" }),
+    ]);
+  });
+
+  it("a Redis-only surplus that is NOT a chain boundary is a real missingInPg", () => {
+    const redis = [norm({ id: "s3", createdAt: 1000, previousSnapshotId: "s2" })];
+    expect(diffSince({ pg: [], redis, cursor })).toEqual([
+      expect.objectContaining({ field: "s3", class: "missingInPg" }),
+    ]);
+  });
+
+  it("a Redis-only surplus above the cursor ms is a real missingInPg", () => {
+    const redis = [norm({ id: "s2", createdAt: 1500, previousSnapshotId: "s1" })];
+    expect(diffSince({ pg: [], redis, cursor })).toEqual([
+      expect.objectContaining({ field: "s2", class: "missingInPg" }),
+    ]);
+  });
+});
+
+describe("SnapshotComparator", () => {
+  it("shouldSample honours the injected rng and percent", () => {
+    expect(new SnapshotComparator({ samplePercent: 10, rng: () => 0.05 }).shouldSample()).toBe(
+      true
+    );
+    expect(new SnapshotComparator({ samplePercent: 10, rng: () => 0.5 }).shouldSample()).toBe(
+      false
+    );
+  });
+
+  it("record emits one metric per divergence, tagged by class and op, and returns void", () => {
+    const seen: Array<{ op: string; cls: DivergenceClass }> = [];
+    const cmp = new SnapshotComparator({
+      samplePercent: 100,
+      metrics: {
+        recordDivergence: (op, cls) => seen.push({ op, cls }),
+        recordSample: () => {},
+      },
+    });
+    const ret = cmp.record("getLatest", [
+      { field: "executionStatus", class: "scalar" },
+      { field: "idempotencyKey", class: "expected:rotatedIdempotencyKey" },
+    ]);
+    expect(ret).toBeUndefined();
+    expect(seen).toEqual([
+      { op: "getLatest", cls: "scalar" },
+      { op: "getLatest", cls: "expected:rotatedIdempotencyKey" },
+    ]);
+  });
+});
@@ -0,0 +1,291 @@
+// Compare-mode read comparator: PURE diff layer. It NEVER serves a read — it takes results the caller
+// already obtained and reports how the two stores disagree, by field, with a class. Type-only imports
+// of client types, so this module holds no Redis or Prisma client (proven by the isolation test).
+import type { Prisma } from "@trigger.dev/database";
+import type { SnapshotRead } from "./redisSnapshotStore.js";
+
+export type DivergenceClass =
+  | "missingInRedis"
+  | "missingInPg"
+  | "scalar"
+  | "order"
+  | "waitpointIdSet"
+  | "validity"
+  | "unknownField"
+  // Reserved shared vocabulary for the payload-comparing layer a later ticket adds. This module
+  // compares id sets and order, not record payloads, so it never emits this — a rotated idempotency
+  // key does not change a waitpoint id. Kept in the union so the metric tag space stays stable.
+  | "expected:rotatedIdempotencyKey"
+  | "expected:redisSurplusAtCursorTie";
+
+export type SnapshotDivergence = {
+  field: string;
+  class: DivergenceClass;
+  pg?: unknown;
+  redis?: unknown;
+};
+
+// The read operations the comparator samples. Bounded so the `op` metric attribute cannot become a
+// high-cardinality label (a caller cannot pass a run id or other unbounded value).
+export type SnapshotReadOp = "getLatest" | "getById" | "getSince" | "getSnapshotWaitpointIds";
+
+export type NormalizedSnapshot = {
+  [k: string]: unknown;
+  id: string;
+  createdAt: number; // ms
+  updatedAt: number; // ms
+  completedWaitpointOrder: string[];
+  waitpointIdSet: string[];
+  previousSnapshotId?: string | null;
+};
+
+// The 22 compared entry columns. EXCLUDED_FIELDS names the columns deliberately not compared; any key
+// on a normalized entry that is on neither list raises `unknownField`, so a new column fails loudly.
+export const COMPARED_FIELDS = [
+  "id",
+  "engine",
+  "executionStatus",
+  "description",
+  "isValid",
+  "error",
+  "previousSnapshotId",
+  "runId",
+  "runStatus",
+  "batchId",
+  "attemptNumber",
+  "environmentId",
+  "environmentType",
+  "projectId",
+  "organizationId",
+  "checkpointId",
+  "workerId",
+  "runnerId",
+  "createdAt",
+  "updatedAt",
+  "metadata",
+] as const;
+
+// lastHeartbeatAt: Postgres-only, never written by the current engine. Waitpoint/checkpoint payloads:
+// expanded by the run-engine resolver, out of this module's scope.
+export const EXCLUDED_FIELDS = ["lastHeartbeatAt", "checkpoint", "completedWaitpoints"] as const;
+
+const SCALAR_FIELDS = COMPARED_FIELDS.filter((f) => f !== "metadata");
+
+const KNOWN_KEYS = new Set<string>([
+  ...COMPARED_FIELDS,
+  ...EXCLUDED_FIELDS,
+  "completedWaitpointOrder",
+  "waitpointIdSet",
+]);
+
+// Carry a source key normalization does not recognise onto the normalized object, so the unknownField
+// check sees it instead of it being silently dropped (a false clean comparison). Skips the
+// prototype-pollution keys. No own-property guard is needed: the normalizer only ever sets KNOWN_KEYS,
+// so a non-known source key is never already present and cannot overwrite a normalized value.
+const DANGEROUS_KEYS = new Set(["__proto__", "constructor", "prototype"]);
+function carryUnknownKeys(target: NormalizedSnapshot, source: Record<string, unknown>): void {
+  for (const k of Object.keys(source)) {
+    if (DANGEROUS_KEYS.has(k)) continue;
+    if (!KNOWN_KEYS.has(k)) target[k] = source[k];
+  }
+}
+
+function canonicalJson(v: unknown): string {
+  if (v === null || typeof v !== "object") return JSON.stringify(v) ?? "null";
+  if (Array.isArray(v)) return `[${v.map(canonicalJson).join(",")}]`;
+  const obj = v as Record<string, unknown>;
+  const keys = Object.keys(obj).sort();
+  return `{${keys.map((k) => `${JSON.stringify(k)}:${canonicalJson(obj[k])}`).join(",")}}`;
+}
+
+export function normalizeFromPg(
+  row: Prisma.TaskRunExecutionSnapshotGetPayload<{ include: { completedWaitpoints: true } }>
+): NormalizedSnapshot {
+  const n: NormalizedSnapshot = {
+    id: row.id,
+    engine: row.engine,
+    executionStatus: row.executionStatus,
+    description: row.description,
+    isValid: row.isValid,
+    error: row.error ?? null,
+    previousSnapshotId: row.previousSnapshotId ?? null,
+    runId: row.runId,
+    runStatus: row.runStatus,
+    batchId: row.batchId ?? null,
+    attemptNumber: row.attemptNumber ?? null,
+    environmentId: row.environmentId,
+    environmentType: row.environmentType,
+    projectId: row.projectId,
+    organizationId: row.organizationId,
+    checkpointId: row.checkpointId ?? null,
+    workerId: row.workerId ?? null,
+    runnerId: row.runnerId ?? null,
+    createdAt: row.createdAt.getTime(),
+    updatedAt: row.updatedAt.getTime(),
+    metadata: row.metadata ?? null,
+    completedWaitpointOrder: [...(row.completedWaitpointOrder ?? [])],
+    // Index-bearing distinct set, from completedWaitpointOrder, to match the Redis read surface
+    // (distinctIds = dedupe of `order`). The full relation holds non-indexed ids Redis does not
+    // expose here (payload-layer, out of scope), so comparing it would fire a spurious divergence.
+    waitpointIdSet: [...new Set(row.completedWaitpointOrder ?? [])].sort(),
+  };
+  carryUnknownKeys(n, row as unknown as Record<string, unknown>);
+  return n;
+}
+
+export function normalizeFromRedis(read: SnapshotRead): NormalizedSnapshot {
+  const e = read.entry as Record<string, unknown>;
+  const createdAtMs = new Date(String(e.createdAt)).getTime();
+  const order = read.completedWaitpointIds?.order ?? [];
+  const idSet = [...(read.completedWaitpointIds?.distinctIds ?? [])].sort();
+  const n: NormalizedSnapshot = {
+    id: read.id,
+    engine: (e.engine ?? "V2") as string,
+    executionStatus: e.executionStatus as string,
+    description: e.description as string,
+    isValid: read.isValid,
+    error: (e.error ?? null) as string | null,
+    previousSnapshotId: (e.previousSnapshotId ?? null) as string | null,
+    runId: e.runId as string,
+    runStatus: e.runStatus as string,
+    batchId: (e.batchId ?? null) as string | null,
+    attemptNumber: (e.attemptNumber ?? null) as number | null,
+    environmentId: e.environmentId as string,
+    environmentType: e.environmentType as string,
+    projectId: e.projectId as string,
+    organizationId: e.organizationId as string,
+    checkpointId: (e.checkpointId ?? null) as string | null,
+    workerId: (e.workerId ?? null) as string | null,
+    runnerId: (e.runnerId ?? null) as string | null,
+    createdAt: createdAtMs,
+    updatedAt: createdAtMs, // write-once row: updatedAt equals createdAt
+    metadata: e.metadata ?? null,
+    completedWaitpointOrder: [...order],
+    waitpointIdSet: idSet,
+  };
+  carryUnknownKeys(n, e);
+  return n;
+}
+
+function sameArray(a: string[], b: string[]): boolean {
+  return a.length === b.length && a.every((x, i) => x === b[i]);
+}
+
+function fieldDivergences(pg: NormalizedSnapshot, redis: NormalizedSnapshot): SnapshotDivergence[] {
+  const out: SnapshotDivergence[] = [];
+
+  for (const f of SCALAR_FIELDS) {
+    if (pg[f] !== redis[f]) {
+      out.push({
+        field: f,
+        class: f === "isValid" ? "validity" : "scalar",
+        pg: pg[f],
+        redis: redis[f],
+      });
+    }
+  }
+  if (canonicalJson(pg.metadata) !== canonicalJson(redis.metadata)) {
+    out.push({ field: "metadata", class: "scalar", pg: pg.metadata, redis: redis.metadata });
+  }
+  if (!sameArray(pg.completedWaitpointOrder, redis.completedWaitpointOrder)) {
+    out.push({
+      field: "completedWaitpointOrder",
+      class: "order",
+      pg: pg.completedWaitpointOrder,
+      redis: redis.completedWaitpointOrder,
+    });
+  }
+  if (!sameArray(pg.waitpointIdSet, redis.waitpointIdSet)) {
+    out.push({
+      field: "waitpointIdSet",
+      class: "waitpointIdSet",
+      pg: pg.waitpointIdSet,
+      redis: redis.waitpointIdSet,
+    });
+  }
+  // Unknown keys on EITHER side, so a new field in either store fails loudly.
+  for (const k of new Set([...Object.keys(pg), ...Object.keys(redis)])) {
+    if (!KNOWN_KEYS.has(k)) {
+      out.push({ field: k, class: "unknownField", pg: pg[k], redis: redis[k] });
+    }
+  }
+  return out;
+}
+
+export function diffLatest(
+  pg: NormalizedSnapshot | null,
+  redis: NormalizedSnapshot | null
+): SnapshotDivergence[] {
+  if (pg && !redis) return [{ field: pg.id, class: "missingInRedis", pg }];
+  if (redis && !pg) return [{ field: redis.id, class: "missingInPg", redis }];
+  if (!pg || !redis) return [];
+  return fieldDivergences(pg, redis);
+}
+
+export type SnapshotComparatorMetrics = {
+  recordDivergence(op: SnapshotReadOp, cls: DivergenceClass): void;
+  recordSample(op: SnapshotReadOp): void;
+};
+
+// Samples reads and records divergence metrics. Holds no store and returns nothing from record(), so
+// it structurally cannot serve a read. samplePercent is injected, never read from env.server.
+export class SnapshotComparator {
+  readonly #samplePercent: number;
+  readonly #metrics?: SnapshotComparatorMetrics;
+  readonly #rng: () => number;
+
+  constructor(opts: {
+    samplePercent: number;
+    metrics?: SnapshotComparatorMetrics;
+    rng?: () => number;
+  }) {
+    this.#samplePercent = opts.samplePercent;
+    this.#metrics = opts.metrics;
+    this.#rng = opts.rng ?? Math.random;
+  }
+
+  shouldSample(): boolean {
+    return this.#rng() * 100 < this.#samplePercent;
+  }
+
+  record(op: SnapshotReadOp, divergences: SnapshotDivergence[]): void {
+    this.#metrics?.recordSample(op);
+    for (const d of divergences) this.#metrics?.recordDivergence(op, d.class);
+  }
+}
+
+export function diffSince(args: {
+  pg: NormalizedSnapshot[];
+  redis: NormalizedSnapshot[];
+  cursor: { id: string; createdAtMs: number };
+}): SnapshotDivergence[] {
+  const { pg, redis, cursor } = args;
+  const byId = (xs: NormalizedSnapshot[]) => new Map(xs.map((x) => [x.id, x]));
+  const pgMap = byId(pg);
+  const redisMap = byId(redis);
+  const out: SnapshotDivergence[] = [];
+
+  // Present on both: field-diff.
+  for (const [id, p] of pgMap) {
+    const r = redisMap.get(id);
+    if (r) out.push(...fieldDivergences(p, r));
+  }
+  // Postgres-only: ALWAYS a lost append. A same-ms tie can never surface here, because Postgres's own
+  // window drops the same-ms entry too. So there is no "expected tie" on this side.
+  for (const [id, p] of pgMap) {
+    if (!redisMap.has(id)) out.push({ field: id, class: "missingInRedis", pg: p });
+  }
+  // Redis-only: expected ONLY when it is a chain boundary sitting exactly on the cursor ms (the
+  // id-cursor getSince path keeps a same-ms entry that Postgres's `> cursor` drops). Anything else is
+  // a real surplus.
+  for (const [id, r] of redisMap) {
+    if (pgMap.has(id)) continue;
+    const isTie = r.createdAt === cursor.createdAtMs && r.previousSnapshotId === cursor.id;
+    out.push({
+      field: id,
+      class: isTie ? "expected:redisSurplusAtCursorTie" : "missingInPg",
+      redis: r,
+    });
+  }
+  return out;
+}
@@ -0,0 +1,42 @@
+import { expect, it, describe } from "vitest";
+import { slotOf, expectOneSlot } from "./clusterSlot";
+
+describe("slotOf", () => {
+  it("matches the published CRC16/XMODEM check value and known slots", () => {
+    expect(slotOf("123456789")).toBe(0x31c3);
+    expect(slotOf("engine:snap:{run_1}:e")).toBe(8108);
+    expect(slotOf("engine:snap:{run_2}:e")).toBe(12239);
+  });
+
+  it("groups keys that share a non-empty tag into one slot", () => {
+    expect(slotOf("a{tag}b")).toBe(slotOf("c{tag}d"));
+  });
+
+  it("hashes the whole key when the tag is empty (not the empty tag)", () => {
+    // If the empty `{}` were used as the tag, these would collide; hashing the whole key keeps them apart.
+    expect(slotOf("a{}b")).not.toBe(slotOf("c{}d"));
+  });
+
+  it("hashes the whole key when a brace is unclosed (malformed tag)", () => {
+    // `b` is not a tag here (no closing brace), so these must not share a slot the way `{b}` would.
+    expect(slotOf("a{b")).not.toBe(slotOf("x{b"));
+  });
+
+  it("hashes UTF-8 bytes, matching Redis for a non-ASCII tag", () => {
+    // Redis (cluster-key-slot) hashes the UTF-8 bytes of `é` to slot 10180.
+    expect(slotOf("{é}")).toBe(10180);
+  });
+});
+
+describe("expectOneSlot", () => {
+  it("passes when every key shares one slot", () => {
+    expect(() => expectOneSlot(["snap:{r}:e", "snap:{r}:idx", "snap:{r}:cur"])).not.toThrow();
+  });
+  it("passes for zero or one key", () => {
+    expect(() => expectOneSlot([])).not.toThrow();
+    expect(() => expectOneSlot(["snap:{r}:e"])).not.toThrow();
+  });
+  it("throws when two keys fall in different slots", () => {
+    expect(() => expectOneSlot(["snap:{run_1}:e", "snap:{run_2}:e"])).toThrow(/slot/i);
+  });
+});
@@ -0,0 +1,34 @@
+// CRC16/XMODEM over a key's hash tag, computed here because CLUSTER KEYSLOT is unavailable on a
+// standalone test container. Pinned against the cluster-key-slot package for our key shapes. Hashes
+// UTF-8 BYTES (as Redis does), not UTF-16 code units, so a non-ASCII key still matches Redis's slot.
+
+function crc16(str: string): number {
+  let crc = 0;
+  for (const byte of Buffer.from(str, "utf8")) {
+    crc ^= byte << 8;
+    for (let j = 0; j < 8; j++) {
+      crc = crc & 0x8000 ? (crc << 1) ^ 0x1021 : crc << 1;
+      crc &= 0xffff;
+    }
+  }
+  return crc;
+}
+
+/** The Redis cluster slot (0–16383) for a key, honouring `{…}` hash-tag extraction. */
+export function slotOf(key: string): number {
+  const start = key.indexOf("{");
+  const end = start === -1 ? -1 : key.indexOf("}", start + 1);
+  const tag = start !== -1 && end !== -1 && end > start + 1 ? key.slice(start + 1, end) : key;
+  return crc16(tag) % 16384;
+}
+
+/** Throws unless every key maps to one slot. A `[]` or single-key input passes. */
+export function expectOneSlot(keys: string[]): void {
+  if (keys.length <= 1) return;
+  const slots = new Set(keys.map(slotOf));
+  if (slots.size !== 1) {
+    throw new Error(
+      `expected all keys in one cluster slot, got ${slots.size}: ${JSON.stringify(keys)}`
+    );
+  }
+}
@@ -0,0 +1,58 @@
+import { expect, it, describe } from "vitest";
+import { createFaultInjector } from "./faultInjection";
+
+type B = "afterPgBeforeRedis" | "midFlushRetry";
+class TestFault extends Error {
+  constructor(readonly boundary: B) {
+    super(`injected at ${boundary}`);
+    this.name = "TestFault";
+  }
+}
+const make = () => createFaultInjector<B>({ error: (b) => new TestFault(b) });
+
+describe("createFaultInjector", () => {
+  it("does not throw when nothing is armed", () => {
+    const f = make();
+    expect(() => f.hook("afterPgBeforeRedis", { runId: "r1" })).not.toThrow();
+    expect(f.fired("afterPgBeforeRedis")).toBe(0);
+  });
+
+  it("throws the injected error while armed, and counts each throw", () => {
+    const f = make();
+    f.arm("afterPgBeforeRedis");
+    expect(() => f.hook("afterPgBeforeRedis")).toThrow(TestFault);
+    expect(f.fired("afterPgBeforeRedis")).toBe(1);
+  });
+
+  it("times limits the number of throws", () => {
+    const f = make();
+    f.arm("midFlushRetry", { times: 2 });
+    expect(() => f.hook("midFlushRetry")).toThrow();
+    expect(() => f.hook("midFlushRetry")).toThrow();
+    expect(() => f.hook("midFlushRetry")).not.toThrow();
+    expect(f.fired("midFlushRetry")).toBe(2);
+  });
+
+  it("runId scopes throws to the matching run only", () => {
+    const f = make();
+    f.arm("afterPgBeforeRedis", { runId: "r1" });
+    expect(() => f.hook("afterPgBeforeRedis", { runId: "r2" })).not.toThrow();
+    expect(() => f.hook("afterPgBeforeRedis", { runId: "r1" })).toThrow();
+    expect(f.fired("afterPgBeforeRedis")).toBe(1);
+  });
+
+  it("rejects a non-integer or negative times, but allows the default (Infinity)", () => {
+    const f = make();
+    expect(() => f.arm("midFlushRetry", { times: Number.NaN })).toThrow(RangeError);
+    expect(() => f.arm("midFlushRetry", { times: 1.5 })).toThrow(RangeError);
+    expect(() => f.arm("midFlushRetry", { times: -1 })).toThrow(RangeError);
+    expect(() => f.arm("midFlushRetry")).not.toThrow(); // unlimited
+  });
+
+  it("disarm clears a boundary", () => {
+    const f = make();
+    f.arm("afterPgBeforeRedis");
+    f.disarm("afterPgBeforeRedis");
+    expect(() => f.hook("afterPgBeforeRedis")).not.toThrow();
+  });
+});
@@ -0,0 +1,46 @@
+// Test-only fault-injection harness, shared by the snapshot decorator (crash-gap) and the waitpoint
+// lane. Generic over the boundary union; the caller passes the error constructor, so this package
+// takes no dependency on @internal/run-store (which would close a dependency cycle). The armed hook
+// is SYNCHRONOUS: a crash at a write boundary must interrupt before the next write.
+
+export type FaultInjector<TBoundary extends string> = {
+  arm(boundary: TBoundary, opts?: { times?: number; runId?: string }): void;
+  disarm(boundary?: TBoundary): void;
+  hook: (boundary: TBoundary, context?: { runId?: string }) => void;
+  fired(boundary: TBoundary): number;
+};
+
+type Armed = { remaining: number; runId?: string };
+
+export function createFaultInjector<TBoundary extends string>(opts: {
+  error: (boundary: TBoundary) => Error;
+}): FaultInjector<TBoundary> {
+  const armed = new Map<TBoundary, Armed>();
+  const counts = new Map<TBoundary, number>();
+
+  return {
+    arm(boundary, o) {
+      const times = o?.times ?? Infinity;
+      if (times !== Infinity && (!Number.isInteger(times) || times < 0)) {
+        throw new RangeError("times must be a non-negative integer or Infinity");
+      }
+      armed.set(boundary, { remaining: times, runId: o?.runId });
+    },
+    disarm(boundary) {
+      if (boundary === undefined) armed.clear();
+      else armed.delete(boundary);
+    },
+    hook: (boundary, context) => {
+      const a = armed.get(boundary);
+      if (!a || a.remaining <= 0) return;
+      if (a.runId !== undefined && a.runId !== context?.runId) return;
+      a.remaining -= 1;
+      if (a.remaining <= 0) armed.delete(boundary);
+      counts.set(boundary, (counts.get(boundary) ?? 0) + 1);
+      throw opts.error(boundary);
+    },
+    fired(boundary) {
+      return counts.get(boundary) ?? 0;
+    },
+  };
+}
@@ -0,0 +1,23 @@
+import { expect } from "vitest";
+import Redis from "ioredis";
+import { heteroRunOpsWithRedisTest } from "./index";
+
+heteroRunOpsWithRedisTest(
+  "provides both Postgres clients and a live Redis",
+  async ({ prisma14, prisma17, redisOptions }) => {
+    const a = await prisma14.$queryRaw`SELECT 1 as ok`;
+    const b = await prisma17.$queryRaw`SELECT 1 as ok`;
+    expect(a).toEqual([{ ok: 1 }]);
+    expect(b).toEqual([{ ok: 1 }]);
+
+    const redis = new Redis(redisOptions);
+    try {
+      expect(await redis.dbsize()).toBe(0);
+      await redis.set("k", "v");
+      expect(await redis.get("k")).toBe("v");
+    } finally {
+      await redis.quit();
+    }
+  },
+  120_000
+);
@@ -431,14 +431,19 @@ type HeteroRunOpsPostgresTestContext = {
 // control-plane schema on PG14 (legacy), prisma17 is a RunOpsPrismaClient over the dedicated SUBSET
 // schema on a SEPARATE PG17 container. Lets a test prove the two sides carry different schemas
 // without disturbing the existing heteroPostgresTest (which keeps the full schema on both sides).
-export const heteroRunOpsPostgresTest = test.extend<HeteroRunOpsPostgresTestContext>({
-  postgresContainer14: async ({}, use) => {
+// The six hetero run-ops fixtures, shared by heteroRunOpsPostgresTest and heteroRunOpsWithRedisTest
+// so the two cannot drift.
+const heteroRunOpsFixtures = {
+  postgresContainer14: async ({}, use: Use<StartedPostgreSqlContainer>) => {
    await use(await getWorkerPostgresContainer());
  },
-  postgresContainer17: async ({}, use) => {
+  postgresContainer17: async ({}, use: Use<StartedPostgreSqlContainer>) => {
    await use(await getRunOpsWorkerPostgresContainer17());
  },
-  uri14: async ({ postgresContainer14 }, use) => {
+  uri14: async (
+    { postgresContainer14 }: { postgresContainer14: StartedPostgreSqlContainer },
+    use: Use<string>
+  ) => {
    const baseUri = postgresContainer14.getConnectionUri();
    const cloneDb = `heteroRunOps14_${pgCloneCounter++}`;
    await createDatabaseFromTemplate(baseUri, cloneDb);
@@ -448,7 +453,10 @@ export const heteroRunOpsPostgresTest = test.extend<HeteroRunOpsPostgresTestCont
      await dropCloneDatabase(baseUri, cloneDb);
    }
  },
-  uri17: async ({ postgresContainer17 }, use) => {
+  uri17: async (
+    { postgresContainer17 }: { postgresContainer17: StartedPostgreSqlContainer },
+    use: Use<string>
+  ) => {
    const baseUri = postgresContainer17.getConnectionUri();
    const cloneDb = `heteroRunOps17_${pgCloneCounter++}`;
    await createDatabaseFromTemplate(baseUri, cloneDb);
@@ -458,7 +466,7 @@ export const heteroRunOpsPostgresTest = test.extend<HeteroRunOpsPostgresTestCont
      await dropCloneDatabase(baseUri, cloneDb);
    }
  },
-  prisma14: async ({ uri14 }, use) => {
+  prisma14: async ({ uri14 }: { uri14: string }, use: Use<PrismaClient>) => {
    const prisma = new PrismaClient({ datasources: { db: { url: uri14 } } });
    try {
      await use(prisma);
@@ -466,7 +474,7 @@ export const heteroRunOpsPostgresTest = test.extend<HeteroRunOpsPostgresTestCont
      await prisma.$disconnect();
    }
  },
-  prisma17: async ({ uri17 }, use) => {
+  prisma17: async ({ uri17 }: { uri17: string }, use: Use<RunOpsPrismaClient>) => {
    const prisma = new RunOpsPrismaClient({ datasources: { db: { url: uri17 } } });
    try {
      await use(prisma);
@@ -474,6 +482,10 @@ export const heteroRunOpsPostgresTest = test.extend<HeteroRunOpsPostgresTestCont
      await prisma.$disconnect();
    }
  },
+} as const;
+
+export const heteroRunOpsPostgresTest = test.extend<HeteroRunOpsPostgresTestContext>({
+  ...heteroRunOpsFixtures,
 });

 type ThreeDbRunOpsPostgresTestContext = {
@@ -635,6 +647,22 @@ const flushRedis = async (
  await use();
 };

+type HeteroRunOpsWithRedisContext = HeteroRunOpsPostgresTestContext & {
+  redisContainer: StartedRedisContainer;
+  resetRedis: void;
+  redisOptions: RedisOptions;
+};
+
+// heteroRunOpsPostgresTest (PG14 + PG17, dedicated-schema run-ops) composed with the WORKER-SCOPED
+// Redis container — boots once per worker, FLUSHALL between tests, matching containerTest. Not
+// postgresAndRedisTest, which boots a container per test and times out under load.
+export const heteroRunOpsWithRedisTest = test.extend<HeteroRunOpsWithRedisContext>({
+  ...heteroRunOpsFixtures,
+  redisContainer: [bootWorkerRedis, { scope: "worker" }],
+  resetRedis: [flushRedis, { auto: true }],
+  redisOptions,
+});
+
 type RedisTestContext = {
  redisContainer: StartedRedisContainer;
  resetRedis: void;
@@ -980,3 +1008,6 @@ export const postgresAndMinioTest = withWarmup(
    await getWorkerPostgresContainer();
  }
 );
+
+export { slotOf, expectOneSlot } from "./clusterSlot";
+export { createFaultInjector, type FaultInjector } from "./faultInjection";