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

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

@@ -3,3 +3,4 @@ export * from "./PostgresRunStore.js";
 export * from "./runOpsStore.js";
 export * from "./readReplicaClient.js";
 export * from "./redisSnapshotStore.js";
+export * from "./snapshotComparator.js";

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

@@ -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);
   });
 

internal-packages/run-store/src/snapshotComparator.isolation.test.ts

@@ -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([]);
+  });
+});

internal-packages/run-store/src/snapshotComparator.test.ts

@@ -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" },
+    ]);
+  });
+});