From a30e93bde9d1c17a6efd7b1db557ecb694fbfee6 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 17:15:24 +0100 Subject: [PATCH 01/16] test(testcontainers): extract shared cluster-slot assertion helper --- .../run-store/src/redisSnapshotStore.test.ts | 34 ++++--------------- .../testcontainers/src/clusterSlot.test.ts | 28 +++++++++++++++ .../testcontainers/src/clusterSlot.ts | 33 ++++++++++++++++++ internal-packages/testcontainers/src/index.ts | 2 ++ 4 files changed, 69 insertions(+), 28 deletions(-) create mode 100644 internal-packages/testcontainers/src/clusterSlot.test.ts create mode 100644 internal-packages/testcontainers/src/clusterSlot.ts 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/testcontainers/src/clusterSlot.test.ts b/internal-packages/testcontainers/src/clusterSlot.test.ts new file mode 100644 index 00000000000..1626b216061 --- /dev/null +++ b/internal-packages/testcontainers/src/clusterSlot.test.ts @@ -0,0 +1,28 @@ +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("hashes the whole key when the tag is empty or malformed", () => { + expect(slotOf("snap:{}:e")).toBe(slotOf("snap:{}:e")); + expect(slotOf("plain-key")).toBe(slotOf("plain-key")); + }); +}); + +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..06ad7e33563 --- /dev/null +++ b/internal-packages/testcontainers/src/clusterSlot.ts @@ -0,0 +1,33 @@ +// 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. + +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; +} + +/** 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/index.ts b/internal-packages/testcontainers/src/index.ts index 820ca827aec..e88c7717f78 100644 --- a/internal-packages/testcontainers/src/index.ts +++ b/internal-packages/testcontainers/src/index.ts @@ -980,3 +980,5 @@ export const postgresAndMinioTest = withWarmup( await getWorkerPostgresContainer(); } ); + +export { slotOf, expectOneSlot } from "./clusterSlot"; From 88762d46fd4b4e81de2e06015670845d551e9937 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 17:17:23 +0100 Subject: [PATCH 02/16] test(testcontainers): add shared fault-injection harness --- .../testcontainers/src/faultInjection.test.ts | 50 +++++++++++++++++++ .../testcontainers/src/faultInjection.ts | 42 ++++++++++++++++ internal-packages/testcontainers/src/index.ts | 1 + 3 files changed, 93 insertions(+) create mode 100644 internal-packages/testcontainers/src/faultInjection.test.ts create mode 100644 internal-packages/testcontainers/src/faultInjection.ts diff --git a/internal-packages/testcontainers/src/faultInjection.test.ts b/internal-packages/testcontainers/src/faultInjection.test.ts new file mode 100644 index 00000000000..3837b841783 --- /dev/null +++ b/internal-packages/testcontainers/src/faultInjection.test.ts @@ -0,0 +1,50 @@ +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("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..0db06343233 --- /dev/null +++ b/internal-packages/testcontainers/src/faultInjection.ts @@ -0,0 +1,42 @@ +// 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) { + armed.set(boundary, { remaining: o?.times ?? Infinity, 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/index.ts b/internal-packages/testcontainers/src/index.ts index e88c7717f78..9244f07d8e8 100644 --- a/internal-packages/testcontainers/src/index.ts +++ b/internal-packages/testcontainers/src/index.ts @@ -982,3 +982,4 @@ export const postgresAndMinioTest = withWarmup( ); export { slotOf, expectOneSlot } from "./clusterSlot"; +export { createFaultInjector, type FaultInjector } from "./faultInjection"; From 30da522a2fe5224ef04f1f25f0bde47c1b4a12e1 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 17:20:04 +0100 Subject: [PATCH 03/16] test(testcontainers): add combined hetero-Postgres + Redis fixture --- .../src/heteroRunOpsWithRedis.test.ts | 23 ++++++++++ internal-packages/testcontainers/src/index.ts | 42 +++++++++++++++---- 2 files changed, 58 insertions(+), 7 deletions(-) create mode 100644 internal-packages/testcontainers/src/heteroRunOpsWithRedis.test.ts 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 9244f07d8e8..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; From 9b21b0e16c8a91e8894907a03066b9cf064afe16 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 17:21:26 +0100 Subject: [PATCH 04/16] feat(run-store): compare-mode snapshot diff layer (pure, cannot read) --- .../run-store/src/snapshotComparator.test.ts | 105 +++++++++ .../run-store/src/snapshotComparator.ts | 216 ++++++++++++++++++ 2 files changed, 321 insertions(+) create mode 100644 internal-packages/run-store/src/snapshotComparator.test.ts create mode 100644 internal-packages/run-store/src/snapshotComparator.ts 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..a17ea786400 --- /dev/null +++ b/internal-packages/run-store/src/snapshotComparator.test.ts @@ -0,0 +1,105 @@ +import { expect, it, describe } from "vitest"; +import { diffLatest, diffSince, type NormalizedSnapshot } from "./snapshotComparator.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" })]); + }); +}); + +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" }), + ]); + }); +}); diff --git a/internal-packages/run-store/src/snapshotComparator.ts b/internal-packages/run-store/src/snapshotComparator.ts new file mode 100644 index 00000000000..a35460cb24a --- /dev/null +++ b/internal-packages/run-store/src/snapshotComparator.ts @@ -0,0 +1,216 @@ +// 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; +}; + +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"); + +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 wps = (row.completedWaitpoints ?? []) as Array<{ id: string }>; + return { + 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 ?? [])], + waitpointIdSet: [...wps.map((w) => w.id)].sort(), + }; +} + +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(); + return { + 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, + }; +} + +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[] = []; + const known = new Set([ + ...COMPARED_FIELDS, + ...EXCLUDED_FIELDS, + "completedWaitpointOrder", + "waitpointIdSet", + ]); + + 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, + }); + } + for (const k of Object.keys(redis)) { + if (!known.has(k)) out.push({ field: k, class: "unknownField", 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 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; +} From 19bbdac4ddeb007dd619654c327f1082d568ea85 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 17:27:59 +0100 Subject: [PATCH 05/16] feat(run-store): comparator sampler + import-isolation guard --- internal-packages/run-store/src/index.ts | 1 + .../src/snapshotComparator.isolation.test.ts | 50 +++++++++++++++++++ .../run-store/src/snapshotComparator.test.ts | 35 ++++++++++++- .../run-store/src/snapshotComparator.ts | 32 ++++++++++++ 4 files changed, 117 insertions(+), 1 deletion(-) create mode 100644 internal-packages/run-store/src/snapshotComparator.isolation.test.ts 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/snapshotComparator.isolation.test.ts b/internal-packages/run-store/src/snapshotComparator.isolation.test.ts new file mode 100644 index 00000000000..c53d871081f --- /dev/null +++ b/internal-packages/run-store/src/snapshotComparator.isolation.test.ts @@ -0,0 +1,50 @@ +// 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 src = readFileSync(sourcePath, "utf8").replace(/\/\*[\s\S]*?\*\//g, ""); + const out: string[] = []; + + if (/(^|[^.\w])import\s*\(/.test(src)) out.push(""); + + const importRe = /^\s*import\b([\s\S]*?)\bfrom\s*["']([^"']+)["']/gm; + for (let m = importRe.exec(src); m !== null; m = importRe.exec(src)) { + const clause = m[1]; + const spec = m[2]; + if (/^\s*type\b/.test(clause)) continue; // `import type ... from` + const named = clause.match(/\{([\s\S]*?)\}/); + if (named && !/(^|,)\s*[A-Za-z_$]/.test(named[1].replace(/\btype\s+[A-Za-z_$][\w$]*/g, ""))) { + 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(src); m !== null; m = bareRe.exec(src)) 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 index a17ea786400..92886c41ede 100644 --- a/internal-packages/run-store/src/snapshotComparator.test.ts +++ b/internal-packages/run-store/src/snapshotComparator.test.ts @@ -1,5 +1,11 @@ import { expect, it, describe } from "vitest"; -import { diffLatest, diffSince, type NormalizedSnapshot } from "./snapshotComparator.js"; +import { + diffLatest, + diffSince, + SnapshotComparator, + type DivergenceClass, + type NormalizedSnapshot, +} from "./snapshotComparator.js"; function norm(over: Partial = {}): NormalizedSnapshot { const base: NormalizedSnapshot = { @@ -103,3 +109,30 @@ describe("diffSince", () => { ]); }); }); + +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 index a35460cb24a..366bdf5fc47 100644 --- a/internal-packages/run-store/src/snapshotComparator.ts +++ b/internal-packages/run-store/src/snapshotComparator.ts @@ -179,6 +179,38 @@ export function diffLatest( return fieldDivergences(pg, redis); } +export type SnapshotComparatorMetrics = { + recordDivergence(op: string, cls: DivergenceClass): void; + recordSample(op: string): 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: string, 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[]; From 0906112c45475df669a0ed7f9868b64e947e67f7 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 17:28:42 +0100 Subject: [PATCH 06/16] feat(run-store): export cycleKey for the backfill script --- .../run-store/src/redisSnapshotStore.test.ts | 8 ++++++++ internal-packages/run-store/src/redisSnapshotStore.ts | 6 ++++++ 2 files changed, 14 insertions(+) diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts index a9db5a0790d..364a9144d20 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.test.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts @@ -6,6 +6,7 @@ import { createRedisClient } from "@internal/redis"; import { Logger } from "@trigger.dev/core/logger"; import { snapshotKeys, + cycleKey, deriveOrder, isValidFor, RedisSnapshotStore, @@ -1303,6 +1304,13 @@ describe("hash tag and keyPrefix", () => { expect(slots.size).toBe(1); }); + it("cycleKey shares the run's hash tag and matches the Lua-derived key shape", () => { + const k = snapshotKeys("run_1"); + const base = k.e.slice(0, -2); // "snap:{run_1}" + expect(cycleKey("run_1", 3)).toBe(`${base}:wp:3`); + expect(slotOf(`engine:${cycleKey("run_1", 3)}`)).toBe(slotOf(`engine:${k.e}`)); + }); + redisTest("the terminal append expires the PREFIXED cycle keys", async ({ redisOptions }) => { // This is the guard for the trap: ioredis prefixes only the KEYS array, so a cycle key minted // inside Lua would be UNPREFIXED while the client wrote a prefixed one. Deriving it from KEYS[1] diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts index 2964959c4fe..cd064cc1683 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -17,6 +17,12 @@ export function snapshotKeys(runId: string): SnapshotKeys { return { e: `${base}:e`, idx: `${base}:idx`, cur: `${base}:cur`, seq: `${base}:seq` }; } +// The per-wait-cycle key. Derived only inside Lua today (wpKey). Exported so the break-glass backfill +// can name a cycle key without duplicating the frozen keyspace contract. Nothing in the store calls it. +export function cycleKey(runId: string, cycleSeq: number): string { + return `snap:{${runId}}:wp:${cycleSeq}`; +} + export type CompletedWaitpointRef = { id: string; index?: number }; // Reproduces PostgresRunStore.#createExecutionSnapshot's completedWaitpointOrder derivation exactly: From 95ad485ec9e5e2f1c019b17856e491407d6b68a4 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 17:30:53 +0100 Subject: [PATCH 07/16] feat(run-store): backfill keyspace read and pure row mapper --- .../run-store/src/snapshotBackfill.test.ts | 101 +++++++++ .../run-store/src/snapshotBackfill.ts | 211 ++++++++++++++++++ 2 files changed, 312 insertions(+) create mode 100644 internal-packages/run-store/src/snapshotBackfill.test.ts create mode 100644 internal-packages/run-store/src/snapshotBackfill.ts diff --git a/internal-packages/run-store/src/snapshotBackfill.test.ts b/internal-packages/run-store/src/snapshotBackfill.test.ts new file mode 100644 index 00000000000..bffabad02ad --- /dev/null +++ b/internal-packages/run-store/src/snapshotBackfill.test.ts @@ -0,0 +1,101 @@ +import { expect, it, describe } from "vitest"; +import { containerTest } from "@internal/testcontainers"; +import { createRedisClient } from "@internal/redis"; +import { snapshotRowsFromRedis, readRunSnapshotsForBackfill, type RunBackfillData } from "./snapshotBackfill.js"; +import { RedisSnapshotStore } from "./redisSnapshotStore.js"; + +function entry(over: Record = {}) { + return { + id: "s1", engine: "V2", executionStatus: "EXECUTING", description: "d", + runId: "r1", runStatus: "EXECUTING", createdAt: "2026-08-24T00:00:00.000Z", + environmentId: "env", environmentType: "DEVELOPMENT", projectId: "p", organizationId: "o", + ...over, + }; +} + +describe("snapshotRowsFromRedis", () => { + it("maps a valid entry with a cycle to a row + join ids from records (not order)", () => { + const data: RunBackfillData = { + runId: "r1", + entries: [ + { id: "s1", seq: 1, raw: JSON.stringify(entry()), entry: entry(), cycle: { cycleSeq: 1, orderCount: 1 } }, + ], + cycles: new Map([ + [1, { + cycleSeq: 1, + order: ["w_a"], + records: [ + { id: "w_a", friendlyId: "f_a", type: "MANUAL", completedAt: "x", outputType: "application/json", outputIsError: false, output: null }, + { id: "w_b", friendlyId: "f_b", type: "MANUAL", completedAt: "x", outputType: "application/json", outputIsError: false, output: null }, + ], + }], + ]), + }; + const { rows, report } = snapshotRowsFromRedis(data); + expect(rows).toHaveLength(1); + expect(rows[0].row.id).toBe("s1"); + expect(rows[0].row.isValid).toBe(true); + expect(rows[0].row.completedWaitpointOrder).toEqual(["w_a"]); + expect(rows[0].waitpointIds.sort()).toEqual(["w_a", "w_b"]); + expect(report.unreconstructable).toEqual([]); + }); + + it("includes an invalid entry (isValid false)", () => { + const e = entry({ id: "s2", error: "boom" }); + const data: RunBackfillData = { + runId: "r1", + entries: [{ id: "s2", seq: 2, raw: JSON.stringify(e), entry: e }], + cycles: new Map(), + }; + const { rows } = snapshotRowsFromRedis(data); + expect(rows[0].row.isValid).toBe(false); + expect(rows[0].row.error).toBe("boom"); + expect(rows[0].waitpointIds).toEqual([]); + }); + + it("reports a cycle without records as unreconstructable, not a guess", () => { + const e = entry({ id: "s3" }); + const data: RunBackfillData = { + runId: "r1", + entries: [{ id: "s3", seq: 3, raw: JSON.stringify(e), entry: e, cycle: { cycleSeq: 2, orderCount: 1 } }], + cycles: new Map([[2, { cycleSeq: 2, order: ["w_a"], records: null }]]), + }; + const { rows, report } = snapshotRowsFromRedis(data); + expect(rows[0].row.completedWaitpointOrder).toEqual(["w_a"]); + expect(report.unreconstructable).toEqual([ + { runId: "r1", snapshotId: "s3", reason: "cycle-without-records" }, + ]); + }); +}); + +containerTest( + "reads back every entry the store wrote, invalid ones included", + async ({ redisOptions }) => { + const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); + const raw = createRedisClient(redisOptions); + try { + const base = { + engine: "V2" as const, runId: "r1", runStatus: "EXECUTING", environmentId: "env", + environmentType: "DEVELOPMENT", projectId: "p", organizationId: "o", + }; + await store.append({ + entry: { ...base, id: "s1", executionStatus: "RUN_CREATED", description: "birth", createdAt: new Date().toISOString() }, + kind: "birth", isTerminal: false, + }); + await store.append({ + entry: { ...base, id: "s2", executionStatus: "EXECUTING", description: "invalid", error: "boom", createdAt: new Date().toISOString(), previousSnapshotId: "s1" }, + kind: "transition", isTerminal: false, + }); + + const data = await readRunSnapshotsForBackfill(raw, "r1"); + expect(data).not.toBeNull(); + expect(data!.entries.map((e) => e.id).sort()).toEqual(["s1", "s2"]); + const s2 = data!.entries.find((e) => e.id === "s2")!; + expect(s2.seq).toBeGreaterThan(0); // seq came from the #s sidecar, not idx + expect(s2.entry.error).toBe("boom"); + } finally { + await store.quit(); + await raw.quit(); + } + } +); diff --git a/internal-packages/run-store/src/snapshotBackfill.ts b/internal-packages/run-store/src/snapshotBackfill.ts new file mode 100644 index 00000000000..a7532deff2e --- /dev/null +++ b/internal-packages/run-store/src/snapshotBackfill.ts @@ -0,0 +1,211 @@ +// Break-glass Redis -> Postgres reconstruction. Reverse direction only, for a retreat below +// redis-read after Postgres stopped being written. Never expected to run. +import type { Prisma, PrismaClient } from "@trigger.dev/database"; +import type { Redis } from "@internal/redis"; +import { + snapshotKeys, + cycleKey, + isValidFor, + type CompletedWaitpointRecord, +} from "./redisSnapshotStore.js"; + +export type BackfillEntry = { + id: string; + seq: number; + raw: string; + entry: Record; + cycle?: { cycleSeq: number; orderCount: number }; +}; +export type BackfillCycle = { + cycleSeq: number; + order: string[]; + records: CompletedWaitpointRecord[] | null; +}; +export type RunBackfillData = { + runId: string; + entries: BackfillEntry[]; + cycles: Map; +}; +export type BackfillRow = { + row: Prisma.TaskRunExecutionSnapshotUncheckedCreateInput; + waitpointIds: string[]; +}; +export type BackfillReport = { + unreconstructable: Array<{ + runId: string; + snapshotId: string; + reason: "no-cycle-pointer" | "cycle-without-records"; + }>; +}; + +// Reads the whole `e` hash and splits each field on its LAST `#`: a bare name is the entry body, +// `#s` is its seq, `#c` is its cycle pointer ":". Returns null when there is no +// `e` hash — the keyspace expired or never existed, nothing to reconstruct. +export async function readRunSnapshotsForBackfill( + redis: Redis, + runId: string, + keyPrefix = "" +): Promise { + const k = snapshotKeys(runId); + const hash = await redis.hgetall(keyPrefix + k.e); + if (!hash || Object.keys(hash).length === 0) return null; + + const bodies = new Map(); + const seqs = new Map(); + const pointers = new Map(); + for (const [field, value] of Object.entries(hash)) { + const hashIdx = field.lastIndexOf("#"); + if (hashIdx === -1) { + bodies.set(field, value); + continue; + } + const id = field.slice(0, hashIdx); + const suffix = field.slice(hashIdx + 1); + if (suffix === "s") { + seqs.set(id, Number(value)); + } else if (suffix === "c") { + const [cs, oc] = value.split(":"); + pointers.set(id, { cycleSeq: Number(cs), orderCount: Number(oc) }); + } + } + + const entries: BackfillEntry[] = []; + for (const [id, raw] of bodies) { + entries.push({ + id, + seq: seqs.get(id) ?? 0, + raw, + entry: JSON.parse(raw) as Record, + cycle: pointers.get(id), + }); + } + entries.sort((a, b) => a.seq - b.seq); + + const cycles = new Map(); + const wanted = new Set([...pointers.values()].map((p) => p.cycleSeq)); + for (const cs of wanted) { + const wp = await redis.hgetall(keyPrefix + cycleKey(runId, cs)); + if (!wp || Object.keys(wp).length === 0) continue; + cycles.set(cs, { + cycleSeq: cs, + order: wp.order ? (JSON.parse(wp.order) as string[]) : [], + records: wp.records ? (JSON.parse(wp.records) as CompletedWaitpointRecord[]) : null, + }); + } + + return { runId, entries, cycles }; +} + +// Pure. Maps read data to row + join inputs, and reports what cannot be faithfully reconstructed. +export function snapshotRowsFromRedis(data: RunBackfillData): { + rows: BackfillRow[]; + report: BackfillReport; +} { + const rows: BackfillRow[] = []; + const report: BackfillReport = { unreconstructable: [] }; + + for (const e of data.entries) { + const j = e.entry; + const error = (j.error ?? null) as string | null; + let order: string[] = []; + let waitpointIds: string[] = []; + + if (e.cycle) { + const c = data.cycles.get(e.cycle.cycleSeq); + if (!c) { + report.unreconstructable.push({ + runId: data.runId, + snapshotId: e.id, + reason: "no-cycle-pointer", + }); + } else { + order = c.order; + if (c.records === null) { + report.unreconstructable.push({ + runId: data.runId, + snapshotId: e.id, + reason: "cycle-without-records", + }); + waitpointIds = [...new Set(order)]; // best-effort: the index-bearing subset only + } else { + waitpointIds = [...new Set(c.records.map((r) => r.id))]; + } + } + } + + rows.push({ + waitpointIds, + row: { + id: e.id, + engine: "V2", + executionStatus: j.executionStatus as Prisma.TaskRunExecutionSnapshotUncheckedCreateInput["executionStatus"], + description: j.description as string, + isValid: isValidFor(j as { error?: unknown }), + error, + previousSnapshotId: (j.previousSnapshotId ?? null) as string | null, + runId: j.runId as string, + runStatus: j.runStatus as Prisma.TaskRunExecutionSnapshotUncheckedCreateInput["runStatus"], + batchId: (j.batchId ?? null) as string | null, + attemptNumber: (j.attemptNumber ?? null) as number | null, + environmentId: j.environmentId as string, + environmentType: j.environmentType as Prisma.TaskRunExecutionSnapshotUncheckedCreateInput["environmentType"], + projectId: j.projectId as string, + organizationId: j.organizationId as string, + checkpointId: (j.checkpointId ?? null) as string | null, + workerId: (j.workerId ?? null) as string | null, + runnerId: (j.runnerId ?? null) as string | null, + createdAt: new Date(String(j.createdAt)), + completedWaitpointOrder: order, + metadata: (j.metadata ?? undefined) as Prisma.InputJsonValue | undefined, + }, + }); + } + + return { rows, report }; +} + +// Writes rows then links. Links use the production FK-free path, NEVER Prisma `connect`: the +// _completedWaitpoints -> Waitpoint FK was dropped for the run-ops split, so a dangling waitpoint id +// is legal and `connect` would raise P2025 and abort the transaction. Legacy: raw INSERT ... ON +// CONFLICT DO NOTHING (A = snapshotId, B = waitpointId). Dedicated: createMany against the explicit +// CompletedWaitpoint model, since the implicit join table does not exist on that schema. +export async function applyBackfill( + prisma: PrismaClient, + rows: BackfillRow[], + opts: { dryRun: boolean; schemaVariant: "legacy" | "dedicated" } +): Promise<{ written: number; linked: number }> { + if (opts.dryRun || rows.length === 0) return { written: 0, linked: 0 }; + + let written = 0; + let linked = 0; + for (const { row, waitpointIds } of rows) { + const snapshotId = row.id as string; // always set by snapshotRowsFromRedis + await prisma.$transaction(async (tx) => { + await tx.taskRunExecutionSnapshot.create({ data: row }); + written += 1; + if (waitpointIds.length === 0) return; + if (opts.schemaVariant === "dedicated") { + const client = tx as unknown as { + completedWaitpoint: { + createMany(args: { + data: Array<{ snapshotId: string; waitpointId: string }>; + skipDuplicates: boolean; + }): Promise; + }; + }; + await client.completedWaitpoint.createMany({ + data: waitpointIds.map((waitpointId) => ({ snapshotId, waitpointId })), + skipDuplicates: true, + }); + } else { + await tx.$executeRaw` + INSERT INTO "_completedWaitpoints" ("A", "B") + SELECT ${snapshotId}, w.id + FROM unnest(${waitpointIds}::text[]) AS w(id) + ON CONFLICT DO NOTHING`; + } + linked += waitpointIds.length; + }); + } + return { written, linked }; +} From 69a7b149c6ba05a7a07537994a3c76319026e47b Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 17:32:48 +0100 Subject: [PATCH 08/16] =?UTF-8?q?feat(run-store):=20backfill=20apply=20?= =?UTF-8?q?=E2=80=94=20FK-free=20join=20insert,=20schema-variant=20aware?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../run-store/src/snapshotBackfill.test.ts | 155 +++++++++++++++++- 1 file changed, 154 insertions(+), 1 deletion(-) diff --git a/internal-packages/run-store/src/snapshotBackfill.test.ts b/internal-packages/run-store/src/snapshotBackfill.test.ts index bffabad02ad..9be69b9db47 100644 --- a/internal-packages/run-store/src/snapshotBackfill.test.ts +++ b/internal-packages/run-store/src/snapshotBackfill.test.ts @@ -1,8 +1,93 @@ import { expect, it, describe } from "vitest"; import { containerTest } from "@internal/testcontainers"; +import type { PrismaClient } from "@trigger.dev/database"; import { createRedisClient } from "@internal/redis"; -import { snapshotRowsFromRedis, readRunSnapshotsForBackfill, type RunBackfillData } from "./snapshotBackfill.js"; +import { + snapshotRowsFromRedis, + readRunSnapshotsForBackfill, + applyBackfill, + type BackfillRow, + type RunBackfillData, +} from "./snapshotBackfill.js"; import { RedisSnapshotStore } from "./redisSnapshotStore.js"; +import { PostgresRunStore } from "./PostgresRunStore.js"; + +// Minimal legacy seed: the owning rows the kept FKs require, plus a TaskRun (via the store's own +// createRun) so a reconstructed snapshot's runId FK resolves, plus a completed-target waitpoint. +async function seedRunAndWaitpoint(prisma: PrismaClient, suffix: string) { + const organization = await prisma.organization.create({ + data: { title: `Org ${suffix}`, slug: `org-${suffix}` }, + }); + const project = await prisma.project.create({ + data: { + name: `Project ${suffix}`, + slug: `project-${suffix}`, + externalRef: `proj_${suffix}`, + organizationId: organization.id, + }, + }); + const environment = await prisma.runtimeEnvironment.create({ + data: { + type: "DEVELOPMENT", + slug: "dev", + projectId: project.id, + organizationId: organization.id, + apiKey: `tr_dev_${suffix}`, + pkApiKey: `pk_dev_${suffix}`, + shortcode: `short_${suffix}`, + }, + }); + const store = new PostgresRunStore({ + prisma: prisma as never, + readOnlyPrisma: prisma as never, + schemaVariant: "legacy", + }); + await store.createRun({ + data: { + id: `run_${suffix}`, + engine: "V2", + status: "PENDING", + friendlyId: `run_friendly_${suffix}`, + runtimeEnvironmentId: environment.id, + environmentType: "DEVELOPMENT", + organizationId: organization.id, + projectId: project.id, + taskIdentifier: "my-task", + payload: "{}", + payloadType: "application/json", + traceId: `trace_${suffix}`, + spanId: `span_${suffix}`, + queue: "task/my-task", + isTest: false, + taskEventStore: "taskEvent", + depth: 0, + createdAt: new Date("2024-01-01T00:00:00.000Z"), + }, + snapshot: { + engine: "V2", + executionStatus: "RUN_CREATED", + description: "Run was created", + runStatus: "PENDING", + environmentId: environment.id, + environmentType: "DEVELOPMENT", + projectId: project.id, + organizationId: organization.id, + }, + }); + const waitpoint = await prisma.waitpoint.create({ + data: { + id: `wp_${suffix}`, + friendlyId: `wp_friendly_${suffix}`, + type: "MANUAL", + status: "PENDING", + idempotencyKey: `idem_${suffix}`, + userProvidedIdempotencyKey: false, + projectId: project.id, + environmentId: environment.id, + }, + }); + return { organization, project, environment, runId: `run_${suffix}`, waitpointId: waitpoint.id }; +} function entry(over: Record = {}) { return { @@ -99,3 +184,71 @@ containerTest( } } ); + +containerTest( + "applied rows are stored correctly, join links inserted FK-free (legacy)", + async ({ prisma }) => { + const seed = await seedRunAndWaitpoint(prisma, "apply"); + const row: BackfillRow = { + waitpointIds: [seed.waitpointId], + row: { + id: "s_recon", + engine: "V2", + executionStatus: "EXECUTING", + description: "reconstructed", + isValid: true, + error: null, + runId: seed.runId, + runStatus: "EXECUTING", + environmentId: seed.environment.id, + environmentType: "DEVELOPMENT", + projectId: seed.project.id, + organizationId: seed.organization.id, + createdAt: new Date("2026-08-24T00:00:00.000Z"), + completedWaitpointOrder: [seed.waitpointId], + }, + }; + + const result = await applyBackfill(prisma, [row], { dryRun: false, schemaVariant: "legacy" }); + expect(result).toEqual({ written: 1, linked: 1 }); + + const written = await prisma.taskRunExecutionSnapshot.findUniqueOrThrow({ + where: { id: "s_recon" }, + include: { completedWaitpoints: true }, + }); + expect(written.isValid).toBe(true); + expect(written.completedWaitpointOrder).toEqual([seed.waitpointId]); + expect(written.completedWaitpoints.map((w) => w.id)).toEqual([seed.waitpointId]); + } +); + +containerTest("dryRun writes nothing", async ({ prisma }) => { + const before = await prisma.taskRunExecutionSnapshot.count(); + const result = await applyBackfill( + prisma, + [ + { + waitpointIds: [], + row: { + id: "s_dry", + engine: "V2", + executionStatus: "EXECUTING", + description: "d", + isValid: true, + error: null, + runId: "run_dry", + runStatus: "EXECUTING", + environmentId: "env", + environmentType: "DEVELOPMENT", + projectId: "p", + organizationId: "o", + createdAt: new Date(), + completedWaitpointOrder: [], + }, + }, + ], + { dryRun: true, schemaVariant: "legacy" } + ); + expect(result).toEqual({ written: 0, linked: 0 }); + expect(await prisma.taskRunExecutionSnapshot.count()).toBe(before); +}); From 09c60f8e1f063aa3176b4f3717bc6c3c4e34f5c4 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 17:34:23 +0100 Subject: [PATCH 09/16] feat(run-store): break-glass backfill CLI wrapper --- .../run-store/scripts/backfill-snapshots.ts | 72 +++++++++++++++++++ .../run-store/src/snapshotBackfill.ts | 4 +- knip.json | 3 + 3 files changed, 77 insertions(+), 2 deletions(-) create mode 100644 internal-packages/run-store/scripts/backfill-snapshots.ts diff --git a/internal-packages/run-store/scripts/backfill-snapshots.ts b/internal-packages/run-store/scripts/backfill-snapshots.ts new file mode 100644 index 00000000000..5fd3db8530a --- /dev/null +++ b/internal-packages/run-store/scripts/backfill-snapshots.ts @@ -0,0 +1,72 @@ +// Break-glass Redis -> Postgres snapshot reconstruction. DRY-RUN by default; pass --apply to write. +// pnpm exec tsx scripts/backfill-snapshots.ts --run [--run ...] [--apply] [--dedicated] +// Connections come from env: REDIS_HOST/REDIS_PORT/REDIS_PASSWORD and DATABASE_URL. +import { PrismaClient } from "@trigger.dev/database"; +import { createRedisClient } from "@internal/redis"; +import { + readRunSnapshotsForBackfill, + snapshotRowsFromRedis, + applyBackfill, +} from "../src/snapshotBackfill.js"; + +function parseArgs(argv: string[]) { + const runIds: string[] = []; + let apply = false; + let dedicated = false; + for (let i = 0; i < argv.length; i++) { + if (argv[i] === "--run") runIds.push(argv[++i]); + else if (argv[i] === "--apply") apply = true; + else if (argv[i] === "--dedicated") dedicated = true; + } + return { runIds, apply, dedicated }; +} + +async function main() { + const { runIds, apply, dedicated } = parseArgs(process.argv.slice(2)); + if (runIds.length === 0) { + console.error("usage: backfill-snapshots.ts --run [--run ...] [--apply] [--dedicated]"); + process.exit(1); + } + + const redis = createRedisClient({ + host: process.env.REDIS_HOST, + port: process.env.REDIS_PORT ? Number(process.env.REDIS_PORT) : undefined, + password: process.env.REDIS_PASSWORD, + }); + const prisma = new PrismaClient(); + + try { + let totalWritten = 0; + const allUnreconstructable: unknown[] = []; + for (const runId of runIds) { + const data = await readRunSnapshotsForBackfill(redis, runId); + if (!data) { + console.log(`[skip] ${runId}: no keyspace in Redis`); + continue; + } + const { rows, report } = snapshotRowsFromRedis(data); + allUnreconstructable.push(...report.unreconstructable); + const result = await applyBackfill(prisma, rows, { + dryRun: !apply, + schemaVariant: dedicated ? "dedicated" : "legacy", + }); + totalWritten += rows.length; + console.log( + `[${apply ? "apply" : "dry-run"}] ${runId}: ${rows.length} rows, ` + + `${result.written} written, ${result.linked} links` + ); + } + if (allUnreconstructable.length > 0) { + console.warn(`UNRECONSTRUCTABLE (${allUnreconstructable.length}):`); + for (const u of allUnreconstructable) console.warn(" " + JSON.stringify(u)); + } + if (!apply) { + console.log(`\nDRY RUN — nothing written. Re-run with --apply. (${totalWritten} rows would write)`); + } + } finally { + await redis.quit(); + await prisma.$disconnect(); + } +} + +void main(); diff --git a/internal-packages/run-store/src/snapshotBackfill.ts b/internal-packages/run-store/src/snapshotBackfill.ts index a7532deff2e..87615e4b313 100644 --- a/internal-packages/run-store/src/snapshotBackfill.ts +++ b/internal-packages/run-store/src/snapshotBackfill.ts @@ -9,14 +9,14 @@ import { type CompletedWaitpointRecord, } from "./redisSnapshotStore.js"; -export type BackfillEntry = { +type BackfillEntry = { id: string; seq: number; raw: string; entry: Record; cycle?: { cycleSeq: number; orderCount: number }; }; -export type BackfillCycle = { +type BackfillCycle = { cycleSeq: number; order: string[]; records: CompletedWaitpointRecord[] | null; diff --git a/knip.json b/knip.json index c6e8aee8977..906e1639b7d 100644 --- a/knip.json +++ b/knip.json @@ -44,6 +44,9 @@ "internal-packages/run-ops-database": { "ignoreDependencies": ["@prisma/client", "prisma"] }, + "internal-packages/run-store": { + "entry": ["scripts/**/*.{js,mjs,cjs,ts,mts,cts}"] + }, "internal-packages/sdk-compat-tests": { "entry": ["src/fixtures/**/*.{js,mjs,cjs,ts,mts,cts,tsx}"] }, From 0d38baa0028de12c2e3a67b2f17153cba47f3002 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Mon, 24 Aug 2026 17:51:49 +0100 Subject: [PATCH 10/16] revert(run-store): drop break-glass backfill (TRI-13450) Reverts the cycleKey export, snapshotBackfill read/mapper/apply, and the CLI wrapper (Tasks 6-9). Forward self-heal (re-enable snapshot writes) plus Redis drain covers a redis-only rollback's correctness, and the ClickHouse audit tail covers history, so a reverse Redis->Postgres backfill reconstructs rows nothing reads before they age out. Removes the item from the plan's redis-only go/no-go gate pending plan-owner sign-off. Comparator, fixture, and shared test utils (Tasks 1-5) are unaffected. --- .../run-store/scripts/backfill-snapshots.ts | 72 ----- .../run-store/src/redisSnapshotStore.test.ts | 8 - .../run-store/src/redisSnapshotStore.ts | 6 - .../run-store/src/snapshotBackfill.test.ts | 254 ------------------ .../run-store/src/snapshotBackfill.ts | 211 --------------- knip.json | 3 - 6 files changed, 554 deletions(-) delete mode 100644 internal-packages/run-store/scripts/backfill-snapshots.ts delete mode 100644 internal-packages/run-store/src/snapshotBackfill.test.ts delete mode 100644 internal-packages/run-store/src/snapshotBackfill.ts diff --git a/internal-packages/run-store/scripts/backfill-snapshots.ts b/internal-packages/run-store/scripts/backfill-snapshots.ts deleted file mode 100644 index 5fd3db8530a..00000000000 --- a/internal-packages/run-store/scripts/backfill-snapshots.ts +++ /dev/null @@ -1,72 +0,0 @@ -// Break-glass Redis -> Postgres snapshot reconstruction. DRY-RUN by default; pass --apply to write. -// pnpm exec tsx scripts/backfill-snapshots.ts --run [--run ...] [--apply] [--dedicated] -// Connections come from env: REDIS_HOST/REDIS_PORT/REDIS_PASSWORD and DATABASE_URL. -import { PrismaClient } from "@trigger.dev/database"; -import { createRedisClient } from "@internal/redis"; -import { - readRunSnapshotsForBackfill, - snapshotRowsFromRedis, - applyBackfill, -} from "../src/snapshotBackfill.js"; - -function parseArgs(argv: string[]) { - const runIds: string[] = []; - let apply = false; - let dedicated = false; - for (let i = 0; i < argv.length; i++) { - if (argv[i] === "--run") runIds.push(argv[++i]); - else if (argv[i] === "--apply") apply = true; - else if (argv[i] === "--dedicated") dedicated = true; - } - return { runIds, apply, dedicated }; -} - -async function main() { - const { runIds, apply, dedicated } = parseArgs(process.argv.slice(2)); - if (runIds.length === 0) { - console.error("usage: backfill-snapshots.ts --run [--run ...] [--apply] [--dedicated]"); - process.exit(1); - } - - const redis = createRedisClient({ - host: process.env.REDIS_HOST, - port: process.env.REDIS_PORT ? Number(process.env.REDIS_PORT) : undefined, - password: process.env.REDIS_PASSWORD, - }); - const prisma = new PrismaClient(); - - try { - let totalWritten = 0; - const allUnreconstructable: unknown[] = []; - for (const runId of runIds) { - const data = await readRunSnapshotsForBackfill(redis, runId); - if (!data) { - console.log(`[skip] ${runId}: no keyspace in Redis`); - continue; - } - const { rows, report } = snapshotRowsFromRedis(data); - allUnreconstructable.push(...report.unreconstructable); - const result = await applyBackfill(prisma, rows, { - dryRun: !apply, - schemaVariant: dedicated ? "dedicated" : "legacy", - }); - totalWritten += rows.length; - console.log( - `[${apply ? "apply" : "dry-run"}] ${runId}: ${rows.length} rows, ` + - `${result.written} written, ${result.linked} links` - ); - } - if (allUnreconstructable.length > 0) { - console.warn(`UNRECONSTRUCTABLE (${allUnreconstructable.length}):`); - for (const u of allUnreconstructable) console.warn(" " + JSON.stringify(u)); - } - if (!apply) { - console.log(`\nDRY RUN — nothing written. Re-run with --apply. (${totalWritten} rows would write)`); - } - } finally { - await redis.quit(); - await prisma.$disconnect(); - } -} - -void main(); diff --git a/internal-packages/run-store/src/redisSnapshotStore.test.ts b/internal-packages/run-store/src/redisSnapshotStore.test.ts index 364a9144d20..a9db5a0790d 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.test.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.test.ts @@ -6,7 +6,6 @@ import { createRedisClient } from "@internal/redis"; import { Logger } from "@trigger.dev/core/logger"; import { snapshotKeys, - cycleKey, deriveOrder, isValidFor, RedisSnapshotStore, @@ -1304,13 +1303,6 @@ describe("hash tag and keyPrefix", () => { expect(slots.size).toBe(1); }); - it("cycleKey shares the run's hash tag and matches the Lua-derived key shape", () => { - const k = snapshotKeys("run_1"); - const base = k.e.slice(0, -2); // "snap:{run_1}" - expect(cycleKey("run_1", 3)).toBe(`${base}:wp:3`); - expect(slotOf(`engine:${cycleKey("run_1", 3)}`)).toBe(slotOf(`engine:${k.e}`)); - }); - redisTest("the terminal append expires the PREFIXED cycle keys", async ({ redisOptions }) => { // This is the guard for the trap: ioredis prefixes only the KEYS array, so a cycle key minted // inside Lua would be UNPREFIXED while the client wrote a prefixed one. Deriving it from KEYS[1] diff --git a/internal-packages/run-store/src/redisSnapshotStore.ts b/internal-packages/run-store/src/redisSnapshotStore.ts index cd064cc1683..2964959c4fe 100644 --- a/internal-packages/run-store/src/redisSnapshotStore.ts +++ b/internal-packages/run-store/src/redisSnapshotStore.ts @@ -17,12 +17,6 @@ export function snapshotKeys(runId: string): SnapshotKeys { return { e: `${base}:e`, idx: `${base}:idx`, cur: `${base}:cur`, seq: `${base}:seq` }; } -// The per-wait-cycle key. Derived only inside Lua today (wpKey). Exported so the break-glass backfill -// can name a cycle key without duplicating the frozen keyspace contract. Nothing in the store calls it. -export function cycleKey(runId: string, cycleSeq: number): string { - return `snap:{${runId}}:wp:${cycleSeq}`; -} - export type CompletedWaitpointRef = { id: string; index?: number }; // Reproduces PostgresRunStore.#createExecutionSnapshot's completedWaitpointOrder derivation exactly: diff --git a/internal-packages/run-store/src/snapshotBackfill.test.ts b/internal-packages/run-store/src/snapshotBackfill.test.ts deleted file mode 100644 index 9be69b9db47..00000000000 --- a/internal-packages/run-store/src/snapshotBackfill.test.ts +++ /dev/null @@ -1,254 +0,0 @@ -import { expect, it, describe } from "vitest"; -import { containerTest } from "@internal/testcontainers"; -import type { PrismaClient } from "@trigger.dev/database"; -import { createRedisClient } from "@internal/redis"; -import { - snapshotRowsFromRedis, - readRunSnapshotsForBackfill, - applyBackfill, - type BackfillRow, - type RunBackfillData, -} from "./snapshotBackfill.js"; -import { RedisSnapshotStore } from "./redisSnapshotStore.js"; -import { PostgresRunStore } from "./PostgresRunStore.js"; - -// Minimal legacy seed: the owning rows the kept FKs require, plus a TaskRun (via the store's own -// createRun) so a reconstructed snapshot's runId FK resolves, plus a completed-target waitpoint. -async function seedRunAndWaitpoint(prisma: PrismaClient, suffix: string) { - const organization = await prisma.organization.create({ - data: { title: `Org ${suffix}`, slug: `org-${suffix}` }, - }); - const project = await prisma.project.create({ - data: { - name: `Project ${suffix}`, - slug: `project-${suffix}`, - externalRef: `proj_${suffix}`, - organizationId: organization.id, - }, - }); - const environment = await prisma.runtimeEnvironment.create({ - data: { - type: "DEVELOPMENT", - slug: "dev", - projectId: project.id, - organizationId: organization.id, - apiKey: `tr_dev_${suffix}`, - pkApiKey: `pk_dev_${suffix}`, - shortcode: `short_${suffix}`, - }, - }); - const store = new PostgresRunStore({ - prisma: prisma as never, - readOnlyPrisma: prisma as never, - schemaVariant: "legacy", - }); - await store.createRun({ - data: { - id: `run_${suffix}`, - engine: "V2", - status: "PENDING", - friendlyId: `run_friendly_${suffix}`, - runtimeEnvironmentId: environment.id, - environmentType: "DEVELOPMENT", - organizationId: organization.id, - projectId: project.id, - taskIdentifier: "my-task", - payload: "{}", - payloadType: "application/json", - traceId: `trace_${suffix}`, - spanId: `span_${suffix}`, - queue: "task/my-task", - isTest: false, - taskEventStore: "taskEvent", - depth: 0, - createdAt: new Date("2024-01-01T00:00:00.000Z"), - }, - snapshot: { - engine: "V2", - executionStatus: "RUN_CREATED", - description: "Run was created", - runStatus: "PENDING", - environmentId: environment.id, - environmentType: "DEVELOPMENT", - projectId: project.id, - organizationId: organization.id, - }, - }); - const waitpoint = await prisma.waitpoint.create({ - data: { - id: `wp_${suffix}`, - friendlyId: `wp_friendly_${suffix}`, - type: "MANUAL", - status: "PENDING", - idempotencyKey: `idem_${suffix}`, - userProvidedIdempotencyKey: false, - projectId: project.id, - environmentId: environment.id, - }, - }); - return { organization, project, environment, runId: `run_${suffix}`, waitpointId: waitpoint.id }; -} - -function entry(over: Record = {}) { - return { - id: "s1", engine: "V2", executionStatus: "EXECUTING", description: "d", - runId: "r1", runStatus: "EXECUTING", createdAt: "2026-08-24T00:00:00.000Z", - environmentId: "env", environmentType: "DEVELOPMENT", projectId: "p", organizationId: "o", - ...over, - }; -} - -describe("snapshotRowsFromRedis", () => { - it("maps a valid entry with a cycle to a row + join ids from records (not order)", () => { - const data: RunBackfillData = { - runId: "r1", - entries: [ - { id: "s1", seq: 1, raw: JSON.stringify(entry()), entry: entry(), cycle: { cycleSeq: 1, orderCount: 1 } }, - ], - cycles: new Map([ - [1, { - cycleSeq: 1, - order: ["w_a"], - records: [ - { id: "w_a", friendlyId: "f_a", type: "MANUAL", completedAt: "x", outputType: "application/json", outputIsError: false, output: null }, - { id: "w_b", friendlyId: "f_b", type: "MANUAL", completedAt: "x", outputType: "application/json", outputIsError: false, output: null }, - ], - }], - ]), - }; - const { rows, report } = snapshotRowsFromRedis(data); - expect(rows).toHaveLength(1); - expect(rows[0].row.id).toBe("s1"); - expect(rows[0].row.isValid).toBe(true); - expect(rows[0].row.completedWaitpointOrder).toEqual(["w_a"]); - expect(rows[0].waitpointIds.sort()).toEqual(["w_a", "w_b"]); - expect(report.unreconstructable).toEqual([]); - }); - - it("includes an invalid entry (isValid false)", () => { - const e = entry({ id: "s2", error: "boom" }); - const data: RunBackfillData = { - runId: "r1", - entries: [{ id: "s2", seq: 2, raw: JSON.stringify(e), entry: e }], - cycles: new Map(), - }; - const { rows } = snapshotRowsFromRedis(data); - expect(rows[0].row.isValid).toBe(false); - expect(rows[0].row.error).toBe("boom"); - expect(rows[0].waitpointIds).toEqual([]); - }); - - it("reports a cycle without records as unreconstructable, not a guess", () => { - const e = entry({ id: "s3" }); - const data: RunBackfillData = { - runId: "r1", - entries: [{ id: "s3", seq: 3, raw: JSON.stringify(e), entry: e, cycle: { cycleSeq: 2, orderCount: 1 } }], - cycles: new Map([[2, { cycleSeq: 2, order: ["w_a"], records: null }]]), - }; - const { rows, report } = snapshotRowsFromRedis(data); - expect(rows[0].row.completedWaitpointOrder).toEqual(["w_a"]); - expect(report.unreconstructable).toEqual([ - { runId: "r1", snapshotId: "s3", reason: "cycle-without-records" }, - ]); - }); -}); - -containerTest( - "reads back every entry the store wrote, invalid ones included", - async ({ redisOptions }) => { - const store = new RedisSnapshotStore({ redisOptions, completedTtlMs: 60_000 }); - const raw = createRedisClient(redisOptions); - try { - const base = { - engine: "V2" as const, runId: "r1", runStatus: "EXECUTING", environmentId: "env", - environmentType: "DEVELOPMENT", projectId: "p", organizationId: "o", - }; - await store.append({ - entry: { ...base, id: "s1", executionStatus: "RUN_CREATED", description: "birth", createdAt: new Date().toISOString() }, - kind: "birth", isTerminal: false, - }); - await store.append({ - entry: { ...base, id: "s2", executionStatus: "EXECUTING", description: "invalid", error: "boom", createdAt: new Date().toISOString(), previousSnapshotId: "s1" }, - kind: "transition", isTerminal: false, - }); - - const data = await readRunSnapshotsForBackfill(raw, "r1"); - expect(data).not.toBeNull(); - expect(data!.entries.map((e) => e.id).sort()).toEqual(["s1", "s2"]); - const s2 = data!.entries.find((e) => e.id === "s2")!; - expect(s2.seq).toBeGreaterThan(0); // seq came from the #s sidecar, not idx - expect(s2.entry.error).toBe("boom"); - } finally { - await store.quit(); - await raw.quit(); - } - } -); - -containerTest( - "applied rows are stored correctly, join links inserted FK-free (legacy)", - async ({ prisma }) => { - const seed = await seedRunAndWaitpoint(prisma, "apply"); - const row: BackfillRow = { - waitpointIds: [seed.waitpointId], - row: { - id: "s_recon", - engine: "V2", - executionStatus: "EXECUTING", - description: "reconstructed", - isValid: true, - error: null, - runId: seed.runId, - runStatus: "EXECUTING", - environmentId: seed.environment.id, - environmentType: "DEVELOPMENT", - projectId: seed.project.id, - organizationId: seed.organization.id, - createdAt: new Date("2026-08-24T00:00:00.000Z"), - completedWaitpointOrder: [seed.waitpointId], - }, - }; - - const result = await applyBackfill(prisma, [row], { dryRun: false, schemaVariant: "legacy" }); - expect(result).toEqual({ written: 1, linked: 1 }); - - const written = await prisma.taskRunExecutionSnapshot.findUniqueOrThrow({ - where: { id: "s_recon" }, - include: { completedWaitpoints: true }, - }); - expect(written.isValid).toBe(true); - expect(written.completedWaitpointOrder).toEqual([seed.waitpointId]); - expect(written.completedWaitpoints.map((w) => w.id)).toEqual([seed.waitpointId]); - } -); - -containerTest("dryRun writes nothing", async ({ prisma }) => { - const before = await prisma.taskRunExecutionSnapshot.count(); - const result = await applyBackfill( - prisma, - [ - { - waitpointIds: [], - row: { - id: "s_dry", - engine: "V2", - executionStatus: "EXECUTING", - description: "d", - isValid: true, - error: null, - runId: "run_dry", - runStatus: "EXECUTING", - environmentId: "env", - environmentType: "DEVELOPMENT", - projectId: "p", - organizationId: "o", - createdAt: new Date(), - completedWaitpointOrder: [], - }, - }, - ], - { dryRun: true, schemaVariant: "legacy" } - ); - expect(result).toEqual({ written: 0, linked: 0 }); - expect(await prisma.taskRunExecutionSnapshot.count()).toBe(before); -}); diff --git a/internal-packages/run-store/src/snapshotBackfill.ts b/internal-packages/run-store/src/snapshotBackfill.ts deleted file mode 100644 index 87615e4b313..00000000000 --- a/internal-packages/run-store/src/snapshotBackfill.ts +++ /dev/null @@ -1,211 +0,0 @@ -// Break-glass Redis -> Postgres reconstruction. Reverse direction only, for a retreat below -// redis-read after Postgres stopped being written. Never expected to run. -import type { Prisma, PrismaClient } from "@trigger.dev/database"; -import type { Redis } from "@internal/redis"; -import { - snapshotKeys, - cycleKey, - isValidFor, - type CompletedWaitpointRecord, -} from "./redisSnapshotStore.js"; - -type BackfillEntry = { - id: string; - seq: number; - raw: string; - entry: Record; - cycle?: { cycleSeq: number; orderCount: number }; -}; -type BackfillCycle = { - cycleSeq: number; - order: string[]; - records: CompletedWaitpointRecord[] | null; -}; -export type RunBackfillData = { - runId: string; - entries: BackfillEntry[]; - cycles: Map; -}; -export type BackfillRow = { - row: Prisma.TaskRunExecutionSnapshotUncheckedCreateInput; - waitpointIds: string[]; -}; -export type BackfillReport = { - unreconstructable: Array<{ - runId: string; - snapshotId: string; - reason: "no-cycle-pointer" | "cycle-without-records"; - }>; -}; - -// Reads the whole `e` hash and splits each field on its LAST `#`: a bare name is the entry body, -// `#s` is its seq, `#c` is its cycle pointer ":". Returns null when there is no -// `e` hash — the keyspace expired or never existed, nothing to reconstruct. -export async function readRunSnapshotsForBackfill( - redis: Redis, - runId: string, - keyPrefix = "" -): Promise { - const k = snapshotKeys(runId); - const hash = await redis.hgetall(keyPrefix + k.e); - if (!hash || Object.keys(hash).length === 0) return null; - - const bodies = new Map(); - const seqs = new Map(); - const pointers = new Map(); - for (const [field, value] of Object.entries(hash)) { - const hashIdx = field.lastIndexOf("#"); - if (hashIdx === -1) { - bodies.set(field, value); - continue; - } - const id = field.slice(0, hashIdx); - const suffix = field.slice(hashIdx + 1); - if (suffix === "s") { - seqs.set(id, Number(value)); - } else if (suffix === "c") { - const [cs, oc] = value.split(":"); - pointers.set(id, { cycleSeq: Number(cs), orderCount: Number(oc) }); - } - } - - const entries: BackfillEntry[] = []; - for (const [id, raw] of bodies) { - entries.push({ - id, - seq: seqs.get(id) ?? 0, - raw, - entry: JSON.parse(raw) as Record, - cycle: pointers.get(id), - }); - } - entries.sort((a, b) => a.seq - b.seq); - - const cycles = new Map(); - const wanted = new Set([...pointers.values()].map((p) => p.cycleSeq)); - for (const cs of wanted) { - const wp = await redis.hgetall(keyPrefix + cycleKey(runId, cs)); - if (!wp || Object.keys(wp).length === 0) continue; - cycles.set(cs, { - cycleSeq: cs, - order: wp.order ? (JSON.parse(wp.order) as string[]) : [], - records: wp.records ? (JSON.parse(wp.records) as CompletedWaitpointRecord[]) : null, - }); - } - - return { runId, entries, cycles }; -} - -// Pure. Maps read data to row + join inputs, and reports what cannot be faithfully reconstructed. -export function snapshotRowsFromRedis(data: RunBackfillData): { - rows: BackfillRow[]; - report: BackfillReport; -} { - const rows: BackfillRow[] = []; - const report: BackfillReport = { unreconstructable: [] }; - - for (const e of data.entries) { - const j = e.entry; - const error = (j.error ?? null) as string | null; - let order: string[] = []; - let waitpointIds: string[] = []; - - if (e.cycle) { - const c = data.cycles.get(e.cycle.cycleSeq); - if (!c) { - report.unreconstructable.push({ - runId: data.runId, - snapshotId: e.id, - reason: "no-cycle-pointer", - }); - } else { - order = c.order; - if (c.records === null) { - report.unreconstructable.push({ - runId: data.runId, - snapshotId: e.id, - reason: "cycle-without-records", - }); - waitpointIds = [...new Set(order)]; // best-effort: the index-bearing subset only - } else { - waitpointIds = [...new Set(c.records.map((r) => r.id))]; - } - } - } - - rows.push({ - waitpointIds, - row: { - id: e.id, - engine: "V2", - executionStatus: j.executionStatus as Prisma.TaskRunExecutionSnapshotUncheckedCreateInput["executionStatus"], - description: j.description as string, - isValid: isValidFor(j as { error?: unknown }), - error, - previousSnapshotId: (j.previousSnapshotId ?? null) as string | null, - runId: j.runId as string, - runStatus: j.runStatus as Prisma.TaskRunExecutionSnapshotUncheckedCreateInput["runStatus"], - batchId: (j.batchId ?? null) as string | null, - attemptNumber: (j.attemptNumber ?? null) as number | null, - environmentId: j.environmentId as string, - environmentType: j.environmentType as Prisma.TaskRunExecutionSnapshotUncheckedCreateInput["environmentType"], - projectId: j.projectId as string, - organizationId: j.organizationId as string, - checkpointId: (j.checkpointId ?? null) as string | null, - workerId: (j.workerId ?? null) as string | null, - runnerId: (j.runnerId ?? null) as string | null, - createdAt: new Date(String(j.createdAt)), - completedWaitpointOrder: order, - metadata: (j.metadata ?? undefined) as Prisma.InputJsonValue | undefined, - }, - }); - } - - return { rows, report }; -} - -// Writes rows then links. Links use the production FK-free path, NEVER Prisma `connect`: the -// _completedWaitpoints -> Waitpoint FK was dropped for the run-ops split, so a dangling waitpoint id -// is legal and `connect` would raise P2025 and abort the transaction. Legacy: raw INSERT ... ON -// CONFLICT DO NOTHING (A = snapshotId, B = waitpointId). Dedicated: createMany against the explicit -// CompletedWaitpoint model, since the implicit join table does not exist on that schema. -export async function applyBackfill( - prisma: PrismaClient, - rows: BackfillRow[], - opts: { dryRun: boolean; schemaVariant: "legacy" | "dedicated" } -): Promise<{ written: number; linked: number }> { - if (opts.dryRun || rows.length === 0) return { written: 0, linked: 0 }; - - let written = 0; - let linked = 0; - for (const { row, waitpointIds } of rows) { - const snapshotId = row.id as string; // always set by snapshotRowsFromRedis - await prisma.$transaction(async (tx) => { - await tx.taskRunExecutionSnapshot.create({ data: row }); - written += 1; - if (waitpointIds.length === 0) return; - if (opts.schemaVariant === "dedicated") { - const client = tx as unknown as { - completedWaitpoint: { - createMany(args: { - data: Array<{ snapshotId: string; waitpointId: string }>; - skipDuplicates: boolean; - }): Promise; - }; - }; - await client.completedWaitpoint.createMany({ - data: waitpointIds.map((waitpointId) => ({ snapshotId, waitpointId })), - skipDuplicates: true, - }); - } else { - await tx.$executeRaw` - INSERT INTO "_completedWaitpoints" ("A", "B") - SELECT ${snapshotId}, w.id - FROM unnest(${waitpointIds}::text[]) AS w(id) - ON CONFLICT DO NOTHING`; - } - linked += waitpointIds.length; - }); - } - return { written, linked }; -} diff --git a/knip.json b/knip.json index 906e1639b7d..c6e8aee8977 100644 --- a/knip.json +++ b/knip.json @@ -44,9 +44,6 @@ "internal-packages/run-ops-database": { "ignoreDependencies": ["@prisma/client", "prisma"] }, - "internal-packages/run-store": { - "entry": ["scripts/**/*.{js,mjs,cjs,ts,mts,cts}"] - }, "internal-packages/sdk-compat-tests": { "entry": ["src/fixtures/**/*.{js,mjs,cjs,ts,mts,cts,tsx}"] }, From 5db5141904419a91a23e999cfba25b2723a62cd3 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Tue, 25 Aug 2026 09:21:35 +0100 Subject: [PATCH 11/16] chore(run-store): format comparator sources --- .../run-store/src/snapshotComparator.test.ts | 50 ++++++++++++++----- .../run-store/src/snapshotComparator.ts | 25 ++++++++-- 2 files changed, 59 insertions(+), 16 deletions(-) diff --git a/internal-packages/run-store/src/snapshotComparator.test.ts b/internal-packages/run-store/src/snapshotComparator.test.ts index 92886c41ede..eb3b49e159e 100644 --- a/internal-packages/run-store/src/snapshotComparator.test.ts +++ b/internal-packages/run-store/src/snapshotComparator.test.ts @@ -9,13 +9,29 @@ import { 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: [], + 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 }; } @@ -55,17 +71,23 @@ describe("diffLatest", () => { }); it("classifies waitpoint id set differences, order-insensitive", () => { - expect(diffLatest(norm({ waitpointIdSet: ["a", "b"] }), norm({ waitpointIdSet: ["a", "b"] }))).toEqual([]); + 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([]); + 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" })]); + expect(diffLatest(norm(), null)).toEqual([ + expect.objectContaining({ class: "missingInRedis" }), + ]); }); it("missingInPg when the row exists only in Redis", () => { @@ -112,8 +134,12 @@ describe("diffSince", () => { 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); + 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", () => { diff --git a/internal-packages/run-store/src/snapshotComparator.ts b/internal-packages/run-store/src/snapshotComparator.ts index 366bdf5fc47..f69df719b0a 100644 --- a/internal-packages/run-store/src/snapshotComparator.ts +++ b/internal-packages/run-store/src/snapshotComparator.ts @@ -38,10 +38,27 @@ export type NormalizedSnapshot = { // 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", + "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: From 26b6e9a0fb053e30cf719a74fc82c0732faa69fa Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Tue, 25 Aug 2026 11:34:52 +0100 Subject: [PATCH 12/16] fix(run-store,testcontainers): make unknownField fire and harden the shared test utilities Carry unrecognised source keys through snapshot normalization so an unknown field surfaces as a divergence instead of being silently dropped. Bound the comparator metric op to a fixed union. Hash cluster-slot keys as UTF-8 bytes, strip line comments in the import-isolation scan, and validate the fault harness times argument. --- .../src/snapshotComparator.isolation.test.ts | 6 ++- .../run-store/src/snapshotComparator.test.ts | 33 +++++++++++++ .../run-store/src/snapshotComparator.ts | 46 +++++++++++++------ .../testcontainers/src/clusterSlot.test.ts | 20 ++++++-- .../testcontainers/src/clusterSlot.ts | 7 +-- .../testcontainers/src/faultInjection.test.ts | 8 ++++ .../testcontainers/src/faultInjection.ts | 6 ++- 7 files changed, 105 insertions(+), 21 deletions(-) diff --git a/internal-packages/run-store/src/snapshotComparator.isolation.test.ts b/internal-packages/run-store/src/snapshotComparator.isolation.test.ts index c53d871081f..5ebb4cd9912 100644 --- a/internal-packages/run-store/src/snapshotComparator.isolation.test.ts +++ b/internal-packages/run-store/src/snapshotComparator.isolation.test.ts @@ -13,7 +13,11 @@ 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 src = readFileSync(sourcePath, "utf8").replace(/\/\*[\s\S]*?\*\//g, ""); + // Strip block AND line comments so a comment mentioning `import(` or `import ... from` cannot + // produce a false positive. + const src = readFileSync(sourcePath, "utf8") + .replace(/\/\*[\s\S]*?\*\//g, "") + .replace(/\/\/.*$/gm, ""); const out: string[] = []; if (/(^|[^.\w])import\s*\(/.test(src)) out.push(""); diff --git a/internal-packages/run-store/src/snapshotComparator.test.ts b/internal-packages/run-store/src/snapshotComparator.test.ts index eb3b49e159e..8edac1e0aaa 100644 --- a/internal-packages/run-store/src/snapshotComparator.test.ts +++ b/internal-packages/run-store/src/snapshotComparator.test.ts @@ -2,10 +2,12 @@ import { expect, it, describe } from "vitest"; import { diffLatest, diffSince, + normalizeFromRedis, SnapshotComparator, type DivergenceClass, type NormalizedSnapshot, } from "./snapshotComparator.js"; +import type { SnapshotRead } from "./redisSnapshotStore.js"; function norm(over: Partial = {}): NormalizedSnapshot { const base: NormalizedSnapshot = { @@ -98,6 +100,37 @@ describe("diffLatest", () => { 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" }), + ]); + }); }); describe("diffSince", () => { diff --git a/internal-packages/run-store/src/snapshotComparator.ts b/internal-packages/run-store/src/snapshotComparator.ts index f69df719b0a..5d24bd8454d 100644 --- a/internal-packages/run-store/src/snapshotComparator.ts +++ b/internal-packages/run-store/src/snapshotComparator.ts @@ -25,6 +25,10 @@ export type SnapshotDivergence = { 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; @@ -67,6 +71,21 @@ export const EXCLUDED_FIELDS = ["lastHeartbeatAt", "checkpoint", "completedWaitp 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). +function carryUnknownKeys(target: NormalizedSnapshot, source: Record): void { + for (const k of Object.keys(source)) { + if (!KNOWN_KEYS.has(k) && !(k in target)) 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(",")}]`; @@ -79,7 +98,7 @@ export function normalizeFromPg( row: Prisma.TaskRunExecutionSnapshotGetPayload<{ include: { completedWaitpoints: true } }> ): NormalizedSnapshot { const wps = (row.completedWaitpoints ?? []) as Array<{ id: string }>; - return { + const n: NormalizedSnapshot = { id: row.id, engine: row.engine, executionStatus: row.executionStatus, @@ -104,6 +123,8 @@ export function normalizeFromPg( completedWaitpointOrder: [...(row.completedWaitpointOrder ?? [])], waitpointIdSet: [...wps.map((w) => w.id)].sort(), }; + carryUnknownKeys(n, row as unknown as Record); + return n; } export function normalizeFromRedis(read: SnapshotRead): NormalizedSnapshot { @@ -111,7 +132,7 @@ export function normalizeFromRedis(read: SnapshotRead): NormalizedSnapshot { const createdAtMs = new Date(String(e.createdAt)).getTime(); const order = read.completedWaitpointIds?.order ?? []; const idSet = [...(read.completedWaitpointIds?.distinctIds ?? [])].sort(); - return { + const n: NormalizedSnapshot = { id: read.id, engine: (e.engine ?? "V2") as string, executionStatus: e.executionStatus as string, @@ -136,6 +157,8 @@ export function normalizeFromRedis(read: SnapshotRead): NormalizedSnapshot { completedWaitpointOrder: [...order], waitpointIdSet: idSet, }; + carryUnknownKeys(n, e); + return n; } function sameArray(a: string[], b: string[]): boolean { @@ -144,12 +167,6 @@ function sameArray(a: string[], b: string[]): boolean { function fieldDivergences(pg: NormalizedSnapshot, redis: NormalizedSnapshot): SnapshotDivergence[] { const out: SnapshotDivergence[] = []; - const known = new Set([ - ...COMPARED_FIELDS, - ...EXCLUDED_FIELDS, - "completedWaitpointOrder", - "waitpointIdSet", - ]); for (const f of SCALAR_FIELDS) { if (pg[f] !== redis[f]) { @@ -180,8 +197,11 @@ function fieldDivergences(pg: NormalizedSnapshot, redis: NormalizedSnapshot): Sn redis: redis.waitpointIdSet, }); } - for (const k of Object.keys(redis)) { - if (!known.has(k)) out.push({ field: k, class: "unknownField", redis: redis[k] }); + // 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; } @@ -197,8 +217,8 @@ export function diffLatest( } export type SnapshotComparatorMetrics = { - recordDivergence(op: string, cls: DivergenceClass): void; - recordSample(op: string): void; + 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 @@ -222,7 +242,7 @@ export class SnapshotComparator { return this.#rng() * 100 < this.#samplePercent; } - record(op: string, divergences: SnapshotDivergence[]): void { + record(op: SnapshotReadOp, divergences: SnapshotDivergence[]): void { this.#metrics?.recordSample(op); for (const d of divergences) this.#metrics?.recordDivergence(op, d.class); } diff --git a/internal-packages/testcontainers/src/clusterSlot.test.ts b/internal-packages/testcontainers/src/clusterSlot.test.ts index 1626b216061..a7bec4313fa 100644 --- a/internal-packages/testcontainers/src/clusterSlot.test.ts +++ b/internal-packages/testcontainers/src/clusterSlot.test.ts @@ -8,9 +8,23 @@ describe("slotOf", () => { expect(slotOf("engine:snap:{run_2}:e")).toBe(12239); }); - it("hashes the whole key when the tag is empty or malformed", () => { - expect(slotOf("snap:{}:e")).toBe(slotOf("snap:{}:e")); - expect(slotOf("plain-key")).toBe(slotOf("plain-key")); + 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); }); }); diff --git a/internal-packages/testcontainers/src/clusterSlot.ts b/internal-packages/testcontainers/src/clusterSlot.ts index 06ad7e33563..ccf74942176 100644 --- a/internal-packages/testcontainers/src/clusterSlot.ts +++ b/internal-packages/testcontainers/src/clusterSlot.ts @@ -1,10 +1,11 @@ // 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. +// 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 (let i = 0; i < str.length; i++) { - crc ^= str.charCodeAt(i) << 8; + 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; diff --git a/internal-packages/testcontainers/src/faultInjection.test.ts b/internal-packages/testcontainers/src/faultInjection.test.ts index 3837b841783..fbcce7dc955 100644 --- a/internal-packages/testcontainers/src/faultInjection.test.ts +++ b/internal-packages/testcontainers/src/faultInjection.test.ts @@ -41,6 +41,14 @@ describe("createFaultInjector", () => { 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"); diff --git a/internal-packages/testcontainers/src/faultInjection.ts b/internal-packages/testcontainers/src/faultInjection.ts index 0db06343233..826cbbb3c36 100644 --- a/internal-packages/testcontainers/src/faultInjection.ts +++ b/internal-packages/testcontainers/src/faultInjection.ts @@ -20,7 +20,11 @@ export function createFaultInjector(opts: { return { arm(boundary, o) { - armed.set(boundary, { remaining: o?.times ?? Infinity, runId: o?.runId }); + 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(); From 8766d54e0fe617cd54a2d992b7f382c42eb21f78 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Tue, 25 Aug 2026 11:50:11 +0100 Subject: [PATCH 13/16] fix(run-store): guard carryUnknownKeys against prototype keys and scan imports on raw source Use an own-property check and skip __proto__/constructor/prototype so an inherited-name field surfaces as a divergence and no key can pollute the prototype. Scan import statements on the raw source (line-anchored) so comment stripping cannot hide a real import. --- .../src/snapshotComparator.isolation.test.ts | 16 ++++++------- .../run-store/src/snapshotComparator.test.ts | 23 +++++++++++++++++++ .../run-store/src/snapshotComparator.ts | 7 +++++- 3 files changed, 37 insertions(+), 9 deletions(-) diff --git a/internal-packages/run-store/src/snapshotComparator.isolation.test.ts b/internal-packages/run-store/src/snapshotComparator.isolation.test.ts index 5ebb4cd9912..c93b54d1865 100644 --- a/internal-packages/run-store/src/snapshotComparator.isolation.test.ts +++ b/internal-packages/run-store/src/snapshotComparator.isolation.test.ts @@ -13,17 +13,17 @@ 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[] { - // Strip block AND line comments so a comment mentioning `import(` or `import ... from` cannot - // produce a false positive. - const src = readFileSync(sourcePath, "utf8") - .replace(/\/\*[\s\S]*?\*\//g, "") - .replace(/\/\/.*$/gm, ""); + const raw = readFileSync(sourcePath, "utf8"); const out: string[] = []; - if (/(^|[^.\w])import\s*\(/.test(src)) out.push(""); + // 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(src); m !== null; m = importRe.exec(src)) { + 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` @@ -36,7 +36,7 @@ function valueImports(sourcePath: string): string[] { // Bare side-effect imports (`import "x"`) run the module. const bareRe = /^\s*import\s*["']([^"']+)["']/gm; - for (let m = bareRe.exec(src); m !== null; m = bareRe.exec(src)) out.push(m[1]); + for (let m = bareRe.exec(raw); m !== null; m = bareRe.exec(raw)) out.push(m[1]); return out; } diff --git a/internal-packages/run-store/src/snapshotComparator.test.ts b/internal-packages/run-store/src/snapshotComparator.test.ts index 8edac1e0aaa..055e8e96b00 100644 --- a/internal-packages/run-store/src/snapshotComparator.test.ts +++ b/internal-packages/run-store/src/snapshotComparator.test.ts @@ -131,6 +131,29 @@ describe("diffLatest", () => { 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.prototype.hasOwnProperty.call(n, "toString")).toBe(true); // carried despite 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); + }); }); describe("diffSince", () => { diff --git a/internal-packages/run-store/src/snapshotComparator.ts b/internal-packages/run-store/src/snapshotComparator.ts index 5d24bd8454d..ebbb28ca235 100644 --- a/internal-packages/run-store/src/snapshotComparator.ts +++ b/internal-packages/run-store/src/snapshotComparator.ts @@ -80,9 +80,14 @@ const KNOWN_KEYS = new Set([ // 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). +// Uses an own-property check (not `in`, which sees the prototype chain and would hide keys like +// `constructor`), and skips prototype-pollution keys. +const DANGEROUS_KEYS = new Set(["__proto__", "constructor", "prototype"]); +const hasOwn = (o: object, k: string): boolean => Object.prototype.hasOwnProperty.call(o, k); function carryUnknownKeys(target: NormalizedSnapshot, source: Record): void { for (const k of Object.keys(source)) { - if (!KNOWN_KEYS.has(k) && !(k in target)) target[k] = source[k]; + if (DANGEROUS_KEYS.has(k)) continue; + if (!KNOWN_KEYS.has(k) && !hasOwn(target, k)) target[k] = source[k]; } } From c7751d0b2d451ad6369b5dbc179fe7c9dd9b9370 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Tue, 25 Aug 2026 12:02:37 +0100 Subject: [PATCH 14/16] fix(run-store): drop the redundant own-property guard in carryUnknownKeys The normalizer only ever sets known keys, so a non-known source key is never already present; the check was unnecessary and tripped the lint rule against hasOwnProperty. Skipping the prototype-pollution keys still holds. --- .../run-store/src/snapshotComparator.test.ts | 2 +- internal-packages/run-store/src/snapshotComparator.ts | 11 +++++------ 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/internal-packages/run-store/src/snapshotComparator.test.ts b/internal-packages/run-store/src/snapshotComparator.test.ts index 055e8e96b00..885f863a51f 100644 --- a/internal-packages/run-store/src/snapshotComparator.test.ts +++ b/internal-packages/run-store/src/snapshotComparator.test.ts @@ -143,7 +143,7 @@ describe("diffLatest", () => { const read: SnapshotRead = { id: "s1", seq: 1, isValid: true, raw: "{}", entry }; const n = normalizeFromRedis(read) as Record; - expect(Object.prototype.hasOwnProperty.call(n, "toString")).toBe(true); // carried despite inherited name + 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); diff --git a/internal-packages/run-store/src/snapshotComparator.ts b/internal-packages/run-store/src/snapshotComparator.ts index ebbb28ca235..85df6fdf604 100644 --- a/internal-packages/run-store/src/snapshotComparator.ts +++ b/internal-packages/run-store/src/snapshotComparator.ts @@ -78,16 +78,15 @@ const KNOWN_KEYS = new Set([ "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). -// Uses an own-property check (not `in`, which sees the prototype chain and would hide keys like -// `constructor`), and skips prototype-pollution keys. +// 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"]); -const hasOwn = (o: object, k: string): boolean => Object.prototype.hasOwnProperty.call(o, k); 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) && !hasOwn(target, k)) target[k] = source[k]; + if (!KNOWN_KEYS.has(k)) target[k] = source[k]; } } From 960f18c8fa169738bfcd030fb263a3c2466d2a7d Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Tue, 25 Aug 2026 12:27:00 +0100 Subject: [PATCH 15/16] test(run-store): handle aliased inline type imports in the isolation scanner Strip an `as` alias along with the `type Foo` specifier so an aliased type-only import isn't misread as a value import. --- .../run-store/src/snapshotComparator.isolation.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/internal-packages/run-store/src/snapshotComparator.isolation.test.ts b/internal-packages/run-store/src/snapshotComparator.isolation.test.ts index c93b54d1865..ae71d227b80 100644 --- a/internal-packages/run-store/src/snapshotComparator.isolation.test.ts +++ b/internal-packages/run-store/src/snapshotComparator.isolation.test.ts @@ -28,7 +28,10 @@ function valueImports(sourcePath: string): string[] { const spec = m[2]; if (/^\s*type\b/.test(clause)) continue; // `import type ... from` const named = clause.match(/\{([\s\S]*?)\}/); - if (named && !/(^|,)\s*[A-Za-z_$]/.test(named[1].replace(/\btype\s+[A-Za-z_$][\w$]*/g, ""))) { + // 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); From f5590a5f612e8a0f23bcfcd59c60fe8cf5868194 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Tue, 25 Aug 2026 12:40:24 +0100 Subject: [PATCH 16/16] fix(run-store): compare the index-bearing waitpoint set like-for-like Derive the Postgres waitpointIdSet from completedWaitpointOrder (index-bearing) to match the Redis read surface, whose distinctIds is the dedupe of the same ordered set. Comparing it against the full completedWaitpoints relation flagged a spurious divergence for a non-indexed completed waitpoint (a single wait). --- .../run-store/src/snapshotComparator.test.ts | 34 +++++++++++++++++++ .../run-store/src/snapshotComparator.ts | 6 ++-- 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/internal-packages/run-store/src/snapshotComparator.test.ts b/internal-packages/run-store/src/snapshotComparator.test.ts index 885f863a51f..ce61386c85a 100644 --- a/internal-packages/run-store/src/snapshotComparator.test.ts +++ b/internal-packages/run-store/src/snapshotComparator.test.ts @@ -3,6 +3,7 @@ import { diffLatest, diffSince, normalizeFromRedis, + normalizeFromPg, SnapshotComparator, type DivergenceClass, type NormalizedSnapshot, @@ -154,6 +155,39 @@ describe("diffLatest", () => { ); 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", () => { diff --git a/internal-packages/run-store/src/snapshotComparator.ts b/internal-packages/run-store/src/snapshotComparator.ts index 85df6fdf604..d9f0407c030 100644 --- a/internal-packages/run-store/src/snapshotComparator.ts +++ b/internal-packages/run-store/src/snapshotComparator.ts @@ -101,7 +101,6 @@ function canonicalJson(v: unknown): string { export function normalizeFromPg( row: Prisma.TaskRunExecutionSnapshotGetPayload<{ include: { completedWaitpoints: true } }> ): NormalizedSnapshot { - const wps = (row.completedWaitpoints ?? []) as Array<{ id: string }>; const n: NormalizedSnapshot = { id: row.id, engine: row.engine, @@ -125,7 +124,10 @@ export function normalizeFromPg( updatedAt: row.updatedAt.getTime(), metadata: row.metadata ?? null, completedWaitpointOrder: [...(row.completedWaitpointOrder ?? [])], - waitpointIdSet: [...wps.map((w) => w.id)].sort(), + // 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;