|
| 1 | +import { expect, it, describe } from "vitest"; |
| 2 | +import { |
| 3 | + diffLatest, |
| 4 | + diffSince, |
| 5 | + normalizeFromRedis, |
| 6 | + normalizeFromPg, |
| 7 | + SnapshotComparator, |
| 8 | + type DivergenceClass, |
| 9 | + type NormalizedSnapshot, |
| 10 | +} from "./snapshotComparator.js"; |
| 11 | +import type { SnapshotRead } from "./redisSnapshotStore.js"; |
| 12 | + |
| 13 | +function norm(over: Partial<NormalizedSnapshot> = {}): NormalizedSnapshot { |
| 14 | + const base: NormalizedSnapshot = { |
| 15 | + id: "s1", |
| 16 | + engine: "V2", |
| 17 | + executionStatus: "RUN_CREATED", |
| 18 | + description: "d", |
| 19 | + isValid: true, |
| 20 | + error: null, |
| 21 | + previousSnapshotId: null, |
| 22 | + runId: "r1", |
| 23 | + runStatus: "PENDING", |
| 24 | + batchId: null, |
| 25 | + attemptNumber: null, |
| 26 | + environmentId: "env", |
| 27 | + environmentType: "DEVELOPMENT", |
| 28 | + projectId: "p", |
| 29 | + organizationId: "o", |
| 30 | + checkpointId: null, |
| 31 | + workerId: null, |
| 32 | + runnerId: null, |
| 33 | + createdAt: 1000, |
| 34 | + updatedAt: 1000, |
| 35 | + metadata: null, |
| 36 | + completedWaitpointOrder: [], |
| 37 | + waitpointIdSet: [], |
| 38 | + }; |
| 39 | + return { ...base, ...over }; |
| 40 | +} |
| 41 | + |
| 42 | +describe("diffLatest", () => { |
| 43 | + it("no divergence when the two sides match", () => { |
| 44 | + expect(diffLatest(norm(), norm())).toEqual([]); |
| 45 | + }); |
| 46 | + |
| 47 | + it("reports a scalar difference by field", () => { |
| 48 | + expect(diffLatest(norm(), norm({ executionStatus: "EXECUTING" }))).toEqual([ |
| 49 | + { field: "executionStatus", class: "scalar", pg: "RUN_CREATED", redis: "EXECUTING" }, |
| 50 | + ]); |
| 51 | + }); |
| 52 | + |
| 53 | + it("compares createdAt and updatedAt by strict equality", () => { |
| 54 | + expect(diffLatest(norm(), norm({ createdAt: 1001 }))).toEqual([ |
| 55 | + { field: "createdAt", class: "scalar", pg: 1000, redis: 1001 }, |
| 56 | + ]); |
| 57 | + }); |
| 58 | + |
| 59 | + it("classifies a validity mismatch", () => { |
| 60 | + const d = diffLatest(norm({ isValid: true }), norm({ isValid: false, error: "boom" })); |
| 61 | + expect(d.map((x) => x.field).sort()).toEqual(["error", "isValid"]); |
| 62 | + expect(d.find((x) => x.field === "isValid")!.class).toBe("validity"); |
| 63 | + }); |
| 64 | + |
| 65 | + it("classifies completedWaitpointOrder differences as order, repeats significant", () => { |
| 66 | + expect( |
| 67 | + diffLatest( |
| 68 | + norm({ completedWaitpointOrder: ["a", "a", "b"] }), |
| 69 | + norm({ completedWaitpointOrder: ["a", "b"] }) |
| 70 | + ) |
| 71 | + ).toEqual([ |
| 72 | + { field: "completedWaitpointOrder", class: "order", pg: ["a", "a", "b"], redis: ["a", "b"] }, |
| 73 | + ]); |
| 74 | + }); |
| 75 | + |
| 76 | + it("classifies waitpoint id set differences, order-insensitive", () => { |
| 77 | + expect( |
| 78 | + diffLatest(norm({ waitpointIdSet: ["a", "b"] }), norm({ waitpointIdSet: ["a", "b"] })) |
| 79 | + ).toEqual([]); |
| 80 | + const d2 = diffLatest(norm({ waitpointIdSet: ["a", "b"] }), norm({ waitpointIdSet: ["a"] })); |
| 81 | + expect(d2[0]).toMatchObject({ field: "waitpointIdSet", class: "waitpointIdSet" }); |
| 82 | + }); |
| 83 | + |
| 84 | + it("does NOT emit a divergence for a rotated idempotency key — invisible at id-set granularity", () => { |
| 85 | + expect(diffLatest(norm({ waitpointIdSet: ["w1"] }), norm({ waitpointIdSet: ["w1"] }))).toEqual( |
| 86 | + [] |
| 87 | + ); |
| 88 | + }); |
| 89 | + |
| 90 | + it("missingInRedis when the row exists only in Postgres", () => { |
| 91 | + expect(diffLatest(norm(), null)).toEqual([ |
| 92 | + expect.objectContaining({ class: "missingInRedis" }), |
| 93 | + ]); |
| 94 | + }); |
| 95 | + |
| 96 | + it("missingInPg when the row exists only in Redis", () => { |
| 97 | + expect(diffLatest(null, norm())).toEqual([expect.objectContaining({ class: "missingInPg" })]); |
| 98 | + }); |
| 99 | + |
| 100 | + it("raises unknownField for a key on neither the compared nor excluded list", () => { |
| 101 | + const d = diffLatest(norm(), { ...norm(), somethingNew: 1 } as NormalizedSnapshot); |
| 102 | + expect(d).toEqual([expect.objectContaining({ field: "somethingNew", class: "unknownField" })]); |
| 103 | + }); |
| 104 | + |
| 105 | + it("normalizeFromRedis carries an unrecognised entry field, so unknownField fires on real input", () => { |
| 106 | + const read: SnapshotRead = { |
| 107 | + id: "s1", |
| 108 | + seq: 1, |
| 109 | + isValid: true, |
| 110 | + raw: "{}", |
| 111 | + entry: { |
| 112 | + engine: "V2", |
| 113 | + executionStatus: "RUN_CREATED", |
| 114 | + description: "d", |
| 115 | + runId: "r1", |
| 116 | + runStatus: "PENDING", |
| 117 | + createdAt: "2026-08-24T00:00:00.000Z", |
| 118 | + environmentId: "env", |
| 119 | + environmentType: "DEVELOPMENT", |
| 120 | + projectId: "p", |
| 121 | + organizationId: "o", |
| 122 | + mysteryField: "surprise", |
| 123 | + }, |
| 124 | + }; |
| 125 | + const redis = normalizeFromRedis(read); |
| 126 | + expect(redis.mysteryField).toBe("surprise"); // not dropped by normalization |
| 127 | + const d = diffLatest( |
| 128 | + norm({ id: "s1", createdAt: redis.createdAt, updatedAt: redis.updatedAt }), |
| 129 | + redis |
| 130 | + ); |
| 131 | + expect(d).toEqual([ |
| 132 | + expect.objectContaining({ field: "mysteryField", class: "unknownField", redis: "surprise" }), |
| 133 | + ]); |
| 134 | + }); |
| 135 | + |
| 136 | + it("surfaces an inherited-name key and does not pollute the prototype", () => { |
| 137 | + // JSON.parse produces OWN keys for `toString` and `__proto__` (unlike an object literal). |
| 138 | + const entry = JSON.parse( |
| 139 | + '{"engine":"V2","executionStatus":"RUN_CREATED","description":"d","runId":"r1",' + |
| 140 | + '"runStatus":"PENDING","createdAt":"2026-08-24T00:00:00.000Z","environmentId":"env",' + |
| 141 | + '"environmentType":"DEVELOPMENT","projectId":"p","organizationId":"o",' + |
| 142 | + '"toString":"surprise","__proto__":{"polluted":true}}' |
| 143 | + ) as Record<string, unknown>; |
| 144 | + const read: SnapshotRead = { id: "s1", seq: 1, isValid: true, raw: "{}", entry }; |
| 145 | + const n = normalizeFromRedis(read) as Record<string, unknown>; |
| 146 | + |
| 147 | + expect(Object.keys(n)).toContain("toString"); // carried as an own key despite the inherited name |
| 148 | + expect(n["toString"]).toBe("surprise"); |
| 149 | + expect(Object.getPrototypeOf(n)).toBe(Object.prototype); // __proto__ skipped, no pollution |
| 150 | + expect("polluted" in {}).toBe(false); |
| 151 | + |
| 152 | + const d = diffLatest( |
| 153 | + norm({ id: "s1", createdAt: n.createdAt as number, updatedAt: n.updatedAt as number }), |
| 154 | + n as NormalizedSnapshot |
| 155 | + ); |
| 156 | + expect(d.some((x) => x.field === "toString" && x.class === "unknownField")).toBe(true); |
| 157 | + }); |
| 158 | + |
| 159 | + it("normalizeFromPg's waitpointIdSet is index-bearing only, matching the Redis read surface", () => { |
| 160 | + // A non-indexed completed waitpoint is in the relation but not in completedWaitpointOrder; Redis's |
| 161 | + // distinctIds (dedupe of order) does not expose it, so the PG side must not either. |
| 162 | + const row = { |
| 163 | + id: "s1", |
| 164 | + engine: "V2", |
| 165 | + executionStatus: "EXECUTING", |
| 166 | + description: "d", |
| 167 | + isValid: true, |
| 168 | + error: null, |
| 169 | + previousSnapshotId: null, |
| 170 | + runId: "r1", |
| 171 | + runStatus: "EXECUTING", |
| 172 | + batchId: null, |
| 173 | + attemptNumber: null, |
| 174 | + environmentId: "env", |
| 175 | + environmentType: "DEVELOPMENT", |
| 176 | + projectId: "p", |
| 177 | + organizationId: "o", |
| 178 | + checkpointId: null, |
| 179 | + workerId: null, |
| 180 | + runnerId: null, |
| 181 | + createdAt: new Date(1000), |
| 182 | + updatedAt: new Date(1000), |
| 183 | + metadata: null, |
| 184 | + completedWaitpointOrder: ["w_indexed"], |
| 185 | + completedWaitpoints: [{ id: "w_indexed" }, { id: "w_nonindexed" }], |
| 186 | + }; |
| 187 | + // eslint-disable-next-line @typescript-eslint/no-explicit-any |
| 188 | + const n = normalizeFromPg(row as any); |
| 189 | + expect(n.waitpointIdSet).toEqual(["w_indexed"]); |
| 190 | + }); |
| 191 | +}); |
| 192 | + |
| 193 | +describe("diffSince", () => { |
| 194 | + const cursor = { id: "s1", createdAtMs: 1000 }; |
| 195 | + |
| 196 | + it("a Postgres-only entry at the cursor ms is a lost append (missingInRedis), never a tie", () => { |
| 197 | + const pg = [norm({ id: "s2", createdAt: 1000, previousSnapshotId: "s1" })]; |
| 198 | + expect(diffSince({ pg, redis: [], cursor })).toEqual([ |
| 199 | + expect.objectContaining({ field: "s2", class: "missingInRedis" }), |
| 200 | + ]); |
| 201 | + }); |
| 202 | + |
| 203 | + it("a Redis-only chain-boundary surplus at the cursor ms is expected:redisSurplusAtCursorTie", () => { |
| 204 | + const redis = [norm({ id: "s2", createdAt: 1000, previousSnapshotId: "s1" })]; |
| 205 | + expect(diffSince({ pg: [], redis, cursor })).toEqual([ |
| 206 | + expect.objectContaining({ field: "s2", class: "expected:redisSurplusAtCursorTie" }), |
| 207 | + ]); |
| 208 | + }); |
| 209 | + |
| 210 | + it("a Redis-only surplus that is NOT a chain boundary is a real missingInPg", () => { |
| 211 | + const redis = [norm({ id: "s3", createdAt: 1000, previousSnapshotId: "s2" })]; |
| 212 | + expect(diffSince({ pg: [], redis, cursor })).toEqual([ |
| 213 | + expect.objectContaining({ field: "s3", class: "missingInPg" }), |
| 214 | + ]); |
| 215 | + }); |
| 216 | + |
| 217 | + it("a Redis-only surplus above the cursor ms is a real missingInPg", () => { |
| 218 | + const redis = [norm({ id: "s2", createdAt: 1500, previousSnapshotId: "s1" })]; |
| 219 | + expect(diffSince({ pg: [], redis, cursor })).toEqual([ |
| 220 | + expect.objectContaining({ field: "s2", class: "missingInPg" }), |
| 221 | + ]); |
| 222 | + }); |
| 223 | +}); |
| 224 | + |
| 225 | +describe("SnapshotComparator", () => { |
| 226 | + it("shouldSample honours the injected rng and percent", () => { |
| 227 | + expect(new SnapshotComparator({ samplePercent: 10, rng: () => 0.05 }).shouldSample()).toBe( |
| 228 | + true |
| 229 | + ); |
| 230 | + expect(new SnapshotComparator({ samplePercent: 10, rng: () => 0.5 }).shouldSample()).toBe( |
| 231 | + false |
| 232 | + ); |
| 233 | + }); |
| 234 | + |
| 235 | + it("record emits one metric per divergence, tagged by class and op, and returns void", () => { |
| 236 | + const seen: Array<{ op: string; cls: DivergenceClass }> = []; |
| 237 | + const cmp = new SnapshotComparator({ |
| 238 | + samplePercent: 100, |
| 239 | + metrics: { |
| 240 | + recordDivergence: (op, cls) => seen.push({ op, cls }), |
| 241 | + recordSample: () => {}, |
| 242 | + }, |
| 243 | + }); |
| 244 | + const ret = cmp.record("getLatest", [ |
| 245 | + { field: "executionStatus", class: "scalar" }, |
| 246 | + { field: "idempotencyKey", class: "expected:rotatedIdempotencyKey" }, |
| 247 | + ]); |
| 248 | + expect(ret).toBeUndefined(); |
| 249 | + expect(seen).toEqual([ |
| 250 | + { op: "getLatest", cls: "scalar" }, |
| 251 | + { op: "getLatest", cls: "expected:rotatedIdempotencyKey" }, |
| 252 | + ]); |
| 253 | + }); |
| 254 | +}); |
0 commit comments