diff --git a/internal-packages/run-store/src/index.ts b/internal-packages/run-store/src/index.ts index 3717dc01527..8893975cf12 100644 --- a/internal-packages/run-store/src/index.ts +++ b/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"; diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts index 0c7b4c0720a..a9db5a0790d 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.test.ts +++ b/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: 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); }); diff --git a/internal-packages/run-store/src/snapshotComparator.isolation.test.ts b/internal-packages/run-store/src/snapshotComparator.isolation.test.ts new file mode 100644 index 00000000000..ae71d227b80 --- /dev/null +++ b/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(""); + + 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([]); + }); +}); diff --git a/internal-packages/run-store/src/snapshotComparator.test.ts b/internal-packages/run-store/src/snapshotComparator.test.ts new file mode 100644 index 00000000000..ce61386c85a --- /dev/null +++ b/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 { + 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; + const read: SnapshotRead = { id: "s1", seq: 1, isValid: true, raw: "{}", entry }; + const n = normalizeFromRedis(read) as Record; + + 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" }, + ]); + }); +}); diff --git a/internal-packages/run-store/src/snapshotComparator.ts b/internal-packages/run-store/src/snapshotComparator.ts new file mode 100644 index 00000000000..d9f0407c030 --- /dev/null +++ b/internal-packages/run-store/src/snapshotComparator.ts @@ -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([ + ...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): 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; + 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); + return n; +} + +export function normalizeFromRedis(read: SnapshotRead): NormalizedSnapshot { + const e = read.entry as Record; + 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; +} diff --git a/internal-packages/testcontainers/src/clusterSlot.test.ts b/internal-packages/testcontainers/src/clusterSlot.test.ts new file mode 100644 index 00000000000..a7bec4313fa --- /dev/null +++ b/internal-packages/testcontainers/src/clusterSlot.test.ts @@ -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); + }); +}); diff --git a/internal-packages/testcontainers/src/clusterSlot.ts b/internal-packages/testcontainers/src/clusterSlot.ts new file mode 100644 index 00000000000..ccf74942176 --- /dev/null +++ b/internal-packages/testcontainers/src/clusterSlot.ts @@ -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)}` + ); + } +} diff --git a/internal-packages/testcontainers/src/faultInjection.test.ts b/internal-packages/testcontainers/src/faultInjection.test.ts new file mode 100644 index 00000000000..fbcce7dc955 --- /dev/null +++ b/internal-packages/testcontainers/src/faultInjection.test.ts @@ -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({ 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(); + }); +}); diff --git a/internal-packages/testcontainers/src/faultInjection.ts b/internal-packages/testcontainers/src/faultInjection.ts new file mode 100644 index 00000000000..826cbbb3c36 --- /dev/null +++ b/internal-packages/testcontainers/src/faultInjection.ts @@ -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 = { + 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(opts: { + error: (boundary: TBoundary) => Error; +}): FaultInjector { + const armed = new Map(); + const counts = new Map(); + + 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; + }, + }; +} diff --git a/internal-packages/testcontainers/src/heteroRunOpsWithRedis.test.ts b/internal-packages/testcontainers/src/heteroRunOpsWithRedis.test.ts new file mode 100644 index 00000000000..192fd877a17 --- /dev/null +++ b/internal-packages/testcontainers/src/heteroRunOpsWithRedis.test.ts @@ -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 +); diff --git a/internal-packages/testcontainers/src/index.ts b/internal-packages/testcontainers/src/index.ts index 820ca827aec..8cdaee8571b 100644 --- a/internal-packages/testcontainers/src/index.ts +++ b/internal-packages/testcontainers/src/index.ts @@ -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({ - 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) => { await use(await getWorkerPostgresContainer()); }, - postgresContainer17: async ({}, use) => { + postgresContainer17: async ({}, use: Use) => { await use(await getRunOpsWorkerPostgresContainer17()); }, - uri14: async ({ postgresContainer14 }, use) => { + uri14: async ( + { postgresContainer14 }: { postgresContainer14: StartedPostgreSqlContainer }, + use: Use + ) => { const baseUri = postgresContainer14.getConnectionUri(); const cloneDb = `heteroRunOps14_${pgCloneCounter++}`; await createDatabaseFromTemplate(baseUri, cloneDb); @@ -448,7 +453,10 @@ export const heteroRunOpsPostgresTest = test.extend { + uri17: async ( + { postgresContainer17 }: { postgresContainer17: StartedPostgreSqlContainer }, + use: Use + ) => { const baseUri = postgresContainer17.getConnectionUri(); const cloneDb = `heteroRunOps17_${pgCloneCounter++}`; await createDatabaseFromTemplate(baseUri, cloneDb); @@ -458,7 +466,7 @@ export const heteroRunOpsPostgresTest = test.extend { + prisma14: async ({ uri14 }: { uri14: string }, use: Use) => { const prisma = new PrismaClient({ datasources: { db: { url: uri14 } } }); try { await use(prisma); @@ -466,7 +474,7 @@ export const heteroRunOpsPostgresTest = test.extend { + prisma17: async ({ uri17 }: { uri17: string }, use: Use) => { const prisma = new RunOpsPrismaClient({ datasources: { db: { url: uri17 } } }); try { await use(prisma); @@ -474,6 +482,10 @@ export const heteroRunOpsPostgresTest = test.extend({ + ...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({ + ...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";