@@ -25,6 +25,10 @@ export type SnapshotDivergence = {
2525 redis ?: unknown ;
2626} ;
2727
28+ // The read operations the comparator samples. Bounded so the `op` metric attribute cannot become a
29+ // high-cardinality label (a caller cannot pass a run id or other unbounded value).
30+ export type SnapshotReadOp = "getLatest" | "getById" | "getSince" | "getSnapshotWaitpointIds" ;
31+
2832export type NormalizedSnapshot = {
2933 [ k : string ] : unknown ;
3034 id : string ;
@@ -67,6 +71,21 @@ export const EXCLUDED_FIELDS = ["lastHeartbeatAt", "checkpoint", "completedWaitp
6771
6872const SCALAR_FIELDS = COMPARED_FIELDS . filter ( ( f ) => f !== "metadata" ) ;
6973
74+ const KNOWN_KEYS = new Set < string > ( [
75+ ...COMPARED_FIELDS ,
76+ ...EXCLUDED_FIELDS ,
77+ "completedWaitpointOrder" ,
78+ "waitpointIdSet" ,
79+ ] ) ;
80+
81+ // Carry a source key normalization does not recognise onto the normalized object, so the
82+ // unknownField check sees it instead of it being silently dropped (a false clean comparison).
83+ function carryUnknownKeys ( target : NormalizedSnapshot , source : Record < string , unknown > ) : void {
84+ for ( const k of Object . keys ( source ) ) {
85+ if ( ! KNOWN_KEYS . has ( k ) && ! ( k in target ) ) target [ k ] = source [ k ] ;
86+ }
87+ }
88+
7089function canonicalJson ( v : unknown ) : string {
7190 if ( v === null || typeof v !== "object" ) return JSON . stringify ( v ) ?? "null" ;
7291 if ( Array . isArray ( v ) ) return `[${ v . map ( canonicalJson ) . join ( "," ) } ]` ;
@@ -79,7 +98,7 @@ export function normalizeFromPg(
7998 row : Prisma . TaskRunExecutionSnapshotGetPayload < { include : { completedWaitpoints : true } } >
8099) : NormalizedSnapshot {
81100 const wps = ( row . completedWaitpoints ?? [ ] ) as Array < { id : string } > ;
82- return {
101+ const n : NormalizedSnapshot = {
83102 id : row . id ,
84103 engine : row . engine ,
85104 executionStatus : row . executionStatus ,
@@ -104,14 +123,16 @@ export function normalizeFromPg(
104123 completedWaitpointOrder : [ ...( row . completedWaitpointOrder ?? [ ] ) ] ,
105124 waitpointIdSet : [ ...wps . map ( ( w ) => w . id ) ] . sort ( ) ,
106125 } ;
126+ carryUnknownKeys ( n , row as unknown as Record < string , unknown > ) ;
127+ return n ;
107128}
108129
109130export function normalizeFromRedis ( read : SnapshotRead ) : NormalizedSnapshot {
110131 const e = read . entry as Record < string , unknown > ;
111132 const createdAtMs = new Date ( String ( e . createdAt ) ) . getTime ( ) ;
112133 const order = read . completedWaitpointIds ?. order ?? [ ] ;
113134 const idSet = [ ...( read . completedWaitpointIds ?. distinctIds ?? [ ] ) ] . sort ( ) ;
114- return {
135+ const n : NormalizedSnapshot = {
115136 id : read . id ,
116137 engine : ( e . engine ?? "V2" ) as string ,
117138 executionStatus : e . executionStatus as string ,
@@ -136,6 +157,8 @@ export function normalizeFromRedis(read: SnapshotRead): NormalizedSnapshot {
136157 completedWaitpointOrder : [ ...order ] ,
137158 waitpointIdSet : idSet ,
138159 } ;
160+ carryUnknownKeys ( n , e ) ;
161+ return n ;
139162}
140163
141164function sameArray ( a : string [ ] , b : string [ ] ) : boolean {
@@ -144,12 +167,6 @@ function sameArray(a: string[], b: string[]): boolean {
144167
145168function fieldDivergences ( pg : NormalizedSnapshot , redis : NormalizedSnapshot ) : SnapshotDivergence [ ] {
146169 const out : SnapshotDivergence [ ] = [ ] ;
147- const known = new Set < string > ( [
148- ...COMPARED_FIELDS ,
149- ...EXCLUDED_FIELDS ,
150- "completedWaitpointOrder" ,
151- "waitpointIdSet" ,
152- ] ) ;
153170
154171 for ( const f of SCALAR_FIELDS ) {
155172 if ( pg [ f ] !== redis [ f ] ) {
@@ -180,8 +197,11 @@ function fieldDivergences(pg: NormalizedSnapshot, redis: NormalizedSnapshot): Sn
180197 redis : redis . waitpointIdSet ,
181198 } ) ;
182199 }
183- for ( const k of Object . keys ( redis ) ) {
184- if ( ! known . has ( k ) ) out . push ( { field : k , class : "unknownField" , redis : redis [ k ] } ) ;
200+ // Unknown keys on EITHER side, so a new field in either store fails loudly.
201+ for ( const k of new Set ( [ ...Object . keys ( pg ) , ...Object . keys ( redis ) ] ) ) {
202+ if ( ! KNOWN_KEYS . has ( k ) ) {
203+ out . push ( { field : k , class : "unknownField" , pg : pg [ k ] , redis : redis [ k ] } ) ;
204+ }
185205 }
186206 return out ;
187207}
@@ -197,8 +217,8 @@ export function diffLatest(
197217}
198218
199219export type SnapshotComparatorMetrics = {
200- recordDivergence ( op : string , cls : DivergenceClass ) : void ;
201- recordSample ( op : string ) : void ;
220+ recordDivergence ( op : SnapshotReadOp , cls : DivergenceClass ) : void ;
221+ recordSample ( op : SnapshotReadOp ) : void ;
202222} ;
203223
204224// Samples reads and records divergence metrics. Holds no store and returns nothing from record(), so
@@ -222,7 +242,7 @@ export class SnapshotComparator {
222242 return this . #rng( ) * 100 < this . #samplePercent;
223243 }
224244
225- record ( op : string , divergences : SnapshotDivergence [ ] ) : void {
245+ record ( op : SnapshotReadOp , divergences : SnapshotDivergence [ ] ) : void {
226246 this . #metrics?. recordSample ( op ) ;
227247 for ( const d of divergences ) this . #metrics?. recordDivergence ( op , d . class ) ;
228248 }
0 commit comments