Skip to main content

faucet_core/
diff.rs

1//! Content verification primitives for `faucet verify` (#701).
2//!
3//! A pipeline can report green while its destination quietly diverges from the
4//! source: a missed change event, a hand edit downstream, a retried page that
5//! landed twice. Row counts (`reconcile:`) cannot see a row that exists on both
6//! sides with different values. This module holds the **pure** building blocks
7//! the verifier composes:
8//!
9//! - [`KeyRange`] — a half-open slice of an integer key space, bisectable, and
10//!   convertible into the PK-range [`ShardSpec`] every SQL source already
11//!   understands (so a range read needs no new connector code);
12//! - [`Normalizer`] + [`row_hash`] — a canonical, type-tolerant fingerprint of
13//!   one row's compared columns, so `1` and `1.0`, or two spellings of the same
14//!   instant, hash alike;
15//! - [`DigestAccumulator`] / [`ContentDigest`] — an order-independent digest of
16//!   a set of rows (count + folded row hashes + observed key bounds), computed
17//!   client-side while streaming;
18//! - [`ServerDigest`] — the same shape computed *inside* the backend (only
19//!   comparable when both sides report the same `algorithm`);
20//! - [`diff_rows`] — the row-level comparison of two leaf ranges, keyed.
21//!
22//! Nothing here performs I/O; the CLI (`cli/src/verify/`) drives the sources.
23
24use crate::shard::ShardSpec;
25use serde::{Deserialize, Serialize};
26use serde_json::{Map, Value};
27use std::collections::{BTreeMap, HashMap};
28
29/// A half-open integer key range `[lo, hi)`; `None` = unbounded on that side.
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
31pub struct KeyRange {
32    pub lo: Option<i64>,
33    pub hi: Option<i64>,
34}
35
36impl KeyRange {
37    /// The whole key space.
38    pub const ALL: KeyRange = KeyRange { lo: None, hi: None };
39
40    /// Whether `k` falls inside this range.
41    pub fn contains(&self, k: i64) -> bool {
42        self.lo.is_none_or(|lo| k >= lo) && self.hi.is_none_or(|hi| k < hi)
43    }
44
45    /// Number of integer keys the range spans, when both ends are bounded.
46    pub fn width(&self) -> Option<u128> {
47        match (self.lo, self.hi) {
48            (Some(lo), Some(hi)) if hi > lo => Some((hi as i128 - lo as i128) as u128),
49            (Some(_), Some(_)) => Some(0),
50            _ => None,
51        }
52    }
53
54    /// The PK-range shard descriptor a SQL source's `apply_shard` narrows to —
55    /// the same shape [`plan_pk_shards`](crate::shard::plan_pk_shards) emits, so
56    /// the range predicate is built by one shared, tested routine. NULL keys are
57    /// owned by the unbounded-above range, so a NULL-key row is read exactly
58    /// once across a partition of the key space.
59    pub fn to_shard(&self, key: &str) -> ShardSpec {
60        ShardSpec::new(
61            format!(
62                "{}..{}",
63                self.lo
64                    .map(|v| v.to_string())
65                    .unwrap_or_else(|| "-inf".into()),
66                self.hi
67                    .map(|v| v.to_string())
68                    .unwrap_or_else(|| "+inf".into())
69            ),
70            serde_json::json!({
71                "key": key,
72                "lo": self.lo.unwrap_or(0),
73                "hi": self.hi.unwrap_or(0),
74                "lo_unbounded": self.lo.is_none(),
75                "hi_unbounded": self.hi.is_none(),
76                "include_null": self.hi.is_none(),
77            }),
78        )
79    }
80
81    /// Split the range in two at its midpoint. An unbounded side is first
82    /// clamped to the observed key bounds (`obs`, the union of both sides'
83    /// `key_min..=key_max`), because a digest can only tell us where rows *are*.
84    /// Returns `None` when the range cannot be split further (one key wide, or
85    /// no observed bounds to clamp with) — the caller then treats it as a leaf.
86    pub fn bisect(&self, obs: Option<(i64, i64)>) -> Option<(KeyRange, KeyRange)> {
87        let lo = self.lo.or(obs.map(|(min, _)| min))?;
88        let hi = self.hi.or(obs.map(|(_, max)| max.saturating_add(1)))?;
89        let (lo, hi) = (i128::from(lo), i128::from(hi));
90        if hi - lo < 2 {
91            return None;
92        }
93        let mid = (lo + (hi - lo) / 2) as i64;
94        // Keep the outer edges as they were (possibly unbounded) so the two
95        // halves still tile exactly the parent range.
96        Some((
97            KeyRange {
98                lo: self.lo,
99                hi: Some(mid),
100            },
101            KeyRange {
102                lo: Some(mid),
103                hi: self.hi,
104            },
105        ))
106    }
107}
108
109impl std::fmt::Display for KeyRange {
110    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
111        match (self.lo, self.hi) {
112            (None, None) => write!(f, "all"),
113            (Some(lo), None) => write!(f, "[{lo}, +inf)"),
114            (None, Some(hi)) => write!(f, "(-inf, {hi})"),
115            (Some(lo), Some(hi)) => write!(f, "[{lo}, {hi})"),
116        }
117    }
118}
119
120/// Split `[min, max]` into `target` contiguous ranges tiling the whole key
121/// space (the first is unbounded below, the last unbounded above), via the same
122/// planner PK-range sharding uses.
123pub fn plan_ranges(min: i64, max: i64, target: usize) -> Vec<KeyRange> {
124    crate::shard::plan_pk_shards("k", min, max, target)
125        .iter()
126        .filter_map(crate::shard::PkShardBounds::from_spec)
127        .map(|b| KeyRange {
128            lo: (!b.lo_unbounded).then_some(b.lo),
129            hi: (!b.hi_unbounded).then_some(b.hi),
130        })
131        .collect()
132}
133
134/// How values are canonicalised before hashing and comparing, so two backends'
135/// spellings of the same value agree.
136#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema, PartialEq)]
137#[serde(deny_unknown_fields)]
138pub struct Normalizer {
139    /// Round floats to this absolute tolerance before comparing (`0` = exact).
140    #[serde(default)]
141    pub float_tolerance: f64,
142    /// Parse strings that look like timestamps (RFC 3339, or
143    /// `YYYY-MM-DD HH:MM:SS[.fff]`) and compare them as UTC microseconds.
144    #[serde(default = "default_true")]
145    pub timestamps: bool,
146    /// Treat a string holding a number (`"42"`, `"1.5"`) as that number.
147    /// Key columns always do, so a key typed differently on each backend
148    /// still matches.
149    #[serde(default)]
150    pub numeric_strings: bool,
151}
152
153fn default_true() -> bool {
154    true
155}
156
157impl Default for Normalizer {
158    fn default() -> Self {
159        Self {
160            float_tolerance: 0.0,
161            timestamps: true,
162            numeric_strings: false,
163        }
164    }
165}
166
167impl Normalizer {
168    /// Fail-fast validation: a finite, non-negative tolerance.
169    pub fn validate(&self) -> Result<(), crate::FaucetError> {
170        if !self.float_tolerance.is_finite() || self.float_tolerance < 0.0 {
171            return Err(crate::FaucetError::Config(format!(
172                "verify: float_tolerance must be a finite number >= 0, got {}",
173                self.float_tolerance
174            )));
175        }
176        Ok(())
177    }
178
179    /// The canonical text of one value. Missing and `null` both canonicalise to
180    /// the same token, so a column absent on one side equals `null` on the other.
181    pub fn canonical(&self, v: Option<&Value>) -> String {
182        match v {
183            None | Some(Value::Null) => "\u{0}null".into(),
184            Some(Value::Bool(b)) => (if *b { "t" } else { "f" }).into(),
185            Some(Value::Number(n)) => self.canonical_number(n.as_f64(), n.as_i64(), n.as_u64()),
186            Some(Value::String(s)) => self.canonical_string(s),
187            Some(Value::Array(a)) => {
188                // Length-prefixed, so an element's text can never read as a
189                // separator (`["a","b"]` vs `["a,sb"]`).
190                let parts: Vec<String> = a
191                    .iter()
192                    .map(|x| length_prefixed(&self.canonical(Some(x))))
193                    .collect();
194                format!("[{}]", parts.concat())
195            }
196            Some(Value::Object(o)) => {
197                // Sorted keys: object key order is not content.
198                let sorted: BTreeMap<&String, &Value> = o.iter().collect();
199                let parts: Vec<String> = sorted
200                    .into_iter()
201                    .map(|(k, x)| {
202                        format!(
203                            "{}{}",
204                            length_prefixed(k),
205                            length_prefixed(&self.canonical(Some(x)))
206                        )
207                    })
208                    .collect();
209                format!("{{{}}}", parts.concat())
210            }
211        }
212    }
213
214    /// The normalizer key columns compare under: a key typed as a number on
215    /// one backend and as text on the other is still one key, or a keyed
216    /// repair would upsert a row and then delete it as extra (#789 CLI-91).
217    fn for_key(&self) -> Normalizer {
218        Normalizer {
219            numeric_strings: true,
220            ..self.clone()
221        }
222    }
223
224    /// Canonical text of column `name`, under the key normalizer when it is
225    /// a key column.
226    fn canonical_column(&self, name: &str, key: &[String], v: Option<&Value>) -> String {
227        if key.iter().any(|k| k == name) {
228            self.for_key().canonical(v)
229        } else {
230            self.canonical(v)
231        }
232    }
233
234    fn canonical_number(&self, f: Option<f64>, i: Option<i64>, u: Option<u64>) -> String {
235        if let Some(i) = i {
236            return format!("n{i}");
237        }
238        if let Some(u) = u {
239            return format!("n{u}");
240        }
241        let Some(f) = f else {
242            return "n?".into();
243        };
244        // An integral float is the same number as the integer (`1.0` == `1`).
245        if f.fract() == 0.0 && f.abs() < 9.0e15 {
246            return format!("n{}", f as i64);
247        }
248        if self.float_tolerance > 0.0 {
249            let rounded = (f / self.float_tolerance).round() * self.float_tolerance;
250            return format!("f{rounded:e}");
251        }
252        format!("f{f:e}")
253    }
254
255    fn canonical_string(&self, s: &str) -> String {
256        if self.timestamps
257            && let Some(micros) = parse_timestamp_micros(s)
258        {
259            return format!("ts{micros}");
260        }
261        if self.numeric_strings {
262            let t = s.trim();
263            if let Ok(i) = t.parse::<i64>() {
264                return format!("n{i}");
265            }
266            if let Ok(f) = t.parse::<f64>()
267                && f.is_finite()
268            {
269                return self.canonical_number(Some(f), None, None);
270            }
271        }
272        format!("s{s}")
273    }
274}
275
276fn length_prefixed(s: &str) -> String {
277    format!("{}:{s}", s.len())
278}
279
280/// UTC microseconds for a timestamp string in RFC 3339 form or the
281/// `YYYY-MM-DD HH:MM:SS[.fff][Z|±hh:mm]` form SQL backends emit (a naive
282/// timestamp is read as UTC). `None` when the string is not a timestamp.
283pub fn parse_timestamp_micros(s: &str) -> Option<i64> {
284    let s = s.trim();
285    // Cheap pre-check: every timestamp we accept starts with a 4-digit year and
286    // a dash, so ordinary strings never pay for the parse attempts.
287    let b = s.as_bytes();
288    if b.len() < 10 || !b[..4].iter().all(u8::is_ascii_digit) || b[4] != b'-' {
289        return None;
290    }
291    if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(s) {
292        return Some(dt.timestamp_micros());
293    }
294    let with_space = s.replacen(' ', "T", 1);
295    if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(&with_space) {
296        return Some(dt.timestamp_micros());
297    }
298    if let Ok(naive) = chrono::NaiveDateTime::parse_from_str(&with_space, "%Y-%m-%dT%H:%M:%S%.f") {
299        return Some(naive.and_utc().timestamp_micros());
300    }
301    if let Ok(date) = chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d") {
302        return Some(date.and_hms_opt(0, 0, 0)?.and_utc().timestamp_micros());
303    }
304    None
305}
306
307/// FNV-1a over `bytes` with a caller-chosen offset basis, so two independent
308/// 64-bit hashes of the same text can be folded into one 128-bit fingerprint.
309pub(crate) fn fnv1a_64(bytes: &[u8], basis: u64) -> u64 {
310    let mut h = basis;
311    for b in bytes {
312        h ^= u64::from(*b);
313        h = h.wrapping_mul(0x0000_0100_0000_01b3);
314    }
315    h
316}
317
318pub(crate) const FNV_BASIS_A: u64 = 0xcbf2_9ce4_8422_2325;
319pub(crate) const FNV_BASIS_B: u64 = 0x84222325_cbf29ce4;
320
321/// A 128-bit fingerprint of `record`'s compared columns: the key columns plus
322/// `columns` (every non-excluded field when `columns` is `None`), each
323/// canonicalised through `norm`. Column order does not matter — the fields are
324/// hashed in sorted name order. Without a column list a null field is skipped,
325/// so a column one side omits and the other carries as null hashes alike (as
326/// [`diff_rows`] compares them).
327pub fn row_hash(
328    record: &Value,
329    key: &[String],
330    columns: Option<&[String]>,
331    exclude: &[String],
332    norm: &Normalizer,
333) -> u128 {
334    let obj = record.as_object();
335    let mut text = String::new();
336    let mut names: Vec<&str> = match columns {
337        Some(cols) => cols.iter().map(String::as_str).collect(),
338        None => obj
339            .map(|o| {
340                o.iter()
341                    .filter(|(_, v)| !v.is_null())
342                    .map(|(k, _)| k.as_str())
343                    .collect()
344            })
345            .unwrap_or_default(),
346    };
347    for k in key {
348        if !names.contains(&k.as_str()) {
349            names.push(k.as_str());
350        }
351    }
352    names.retain(|n| !is_excluded(n, exclude));
353    names.sort_unstable();
354    names.dedup();
355    for name in names {
356        text.push_str(name);
357        text.push('=');
358        text.push_str(&norm.canonical_column(name, key, obj.and_then(|o| o.get(name))));
359        text.push('\u{1f}');
360    }
361    let a = fnv1a_64(text.as_bytes(), FNV_BASIS_A);
362    let b = fnv1a_64(text.as_bytes(), FNV_BASIS_B);
363    (u128::from(a) << 64) | u128::from(b)
364}
365
366/// Whether a column is excluded: an exact name, or a `prefix*` glob.
367pub fn is_excluded(name: &str, exclude: &[String]) -> bool {
368    exclude.iter().any(|e| match e.strip_suffix('*') {
369        Some(prefix) => name.starts_with(prefix),
370        None => e == name,
371    })
372}
373
374/// The canonical text of a record's key tuple (`k1=…;k2=…`), used to match
375/// rows across the two sides.
376pub fn key_text(record: &Value, key: &[String], norm: &Normalizer) -> String {
377    let obj = record.as_object();
378    let norm = norm.for_key();
379    let mut out = String::new();
380    for k in key {
381        out.push_str(k);
382        out.push('=');
383        out.push_str(&norm.canonical(obj.and_then(|o| o.get(k))));
384        out.push(';');
385    }
386    out
387}
388
389/// The integer value of a single-column key, when it is one.
390pub fn key_int(record: &Value, key: &[String]) -> Option<i64> {
391    if key.len() != 1 {
392        return None;
393    }
394    match record.get(&key[0])? {
395        Value::Number(n) => n.as_i64(),
396        Value::String(s) => s.trim().parse().ok(),
397        _ => None,
398    }
399}
400
401/// An order-independent digest of a set of rows.
402#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
403pub struct ContentDigest {
404    pub rows: u64,
405    /// Wrapping sum of the row fingerprints.
406    pub sum: u128,
407    /// XOR of the row fingerprints (catches a sum collision from swapped rows).
408    pub xor: u128,
409    /// Smallest / largest integer key seen (single integer keys only).
410    pub key_min: Option<i64>,
411    pub key_max: Option<i64>,
412}
413
414impl ContentDigest {
415    /// Whether two digests describe the same row set.
416    pub fn same(&self, other: &ContentDigest) -> bool {
417        self.rows == other.rows && self.sum == other.sum && self.xor == other.xor
418    }
419
420    /// The union of the observed key bounds of two digests.
421    pub fn bounds_union(&self, other: &ContentDigest) -> Option<(i64, i64)> {
422        let min = match (self.key_min, other.key_min) {
423            (Some(a), Some(b)) => Some(a.min(b)),
424            (a, b) => a.or(b),
425        }?;
426        let max = match (self.key_max, other.key_max) {
427            (Some(a), Some(b)) => Some(a.max(b)),
428            (a, b) => a.or(b),
429        }?;
430        Some((min, max))
431    }
432}
433
434/// Streams rows into a [`ContentDigest`].
435#[derive(Debug, Default)]
436pub struct DigestAccumulator {
437    digest: ContentDigest,
438}
439
440impl DigestAccumulator {
441    pub fn new() -> Self {
442        Self::default()
443    }
444
445    /// Fold one row in.
446    pub fn add(&mut self, hash: u128, key: Option<i64>) {
447        self.digest.rows += 1;
448        self.digest.sum = self.digest.sum.wrapping_add(hash);
449        self.digest.xor ^= hash;
450        if let Some(k) = key {
451            self.digest.key_min = Some(self.digest.key_min.map_or(k, |m| m.min(k)));
452            self.digest.key_max = Some(self.digest.key_max.map_or(k, |m| m.max(k)));
453        }
454    }
455
456    pub fn finish(self) -> ContentDigest {
457        self.digest
458    }
459}
460
461/// A digest a backend computed itself, without shipping rows. Two server
462/// digests are comparable only when their `algorithm` ids match — each backend
463/// hashes its own text rendering of a row, so a Postgres digest says nothing
464/// about a SQLite table. When the algorithms differ the verifier streams both
465/// sides and uses [`ContentDigest`] instead.
466#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
467pub struct ServerDigest {
468    pub algorithm: String,
469    pub rows: u64,
470    /// Opaque digest text (a decimal sum, a hex string — whatever the
471    /// algorithm defines).
472    pub digest: String,
473    pub key_min: Option<i64>,
474    pub key_max: Option<i64>,
475}
476
477impl ServerDigest {
478    /// Whether two server digests agree. Different algorithms never agree —
479    /// the caller must not treat that as a difference but as "not comparable".
480    pub fn comparable(&self, other: &ServerDigest) -> bool {
481        self.algorithm == other.algorithm
482    }
483
484    pub fn same(&self, other: &ServerDigest) -> bool {
485        self.comparable(other) && self.rows == other.rows && self.digest == other.digest
486    }
487
488    pub fn bounds_union(&self, other: &ServerDigest) -> Option<(i64, i64)> {
489        let min = match (self.key_min, other.key_min) {
490            (Some(a), Some(b)) => Some(a.min(b)),
491            (a, b) => a.or(b),
492        }?;
493        let max = match (self.key_max, other.key_max) {
494            (Some(a), Some(b)) => Some(a.max(b)),
495            (a, b) => a.or(b),
496        }?;
497        Some((min, max))
498    }
499}
500
501/// Why one key differs between the two sides.
502#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
503#[serde(tag = "kind", rename_all = "snake_case")]
504pub enum DifferenceKind {
505    /// The source has the key, the destination does not.
506    MissingInDest,
507    /// The destination has the key, the source does not.
508    ExtraInDest,
509    /// Both have the key; these columns differ.
510    Changed { columns: Vec<String> },
511    /// The key appears more than once on one side, so rows cannot be matched.
512    Duplicate { side: Side, count: u64 },
513}
514
515#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
516#[serde(rename_all = "snake_case")]
517pub enum Side {
518    Source,
519    Destination,
520}
521
522/// One differing key.
523#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
524pub struct Difference {
525    /// The key columns and their values (from whichever side has the row).
526    pub key: Value,
527    #[serde(flatten)]
528    pub kind: DifferenceKind,
529}
530
531impl Difference {
532    /// Whether repairing this difference means writing the source row.
533    pub fn needs_upsert(&self) -> bool {
534        matches!(
535            self.kind,
536            DifferenceKind::MissingInDest | DifferenceKind::Changed { .. }
537        )
538    }
539
540    /// Whether repairing this difference means deleting the destination row.
541    pub fn needs_delete(&self) -> bool {
542        matches!(self.kind, DifferenceKind::ExtraInDest)
543    }
544}
545
546/// The columns whose canonical values differ between two rows.
547fn changed_columns(
548    a: &Value,
549    b: &Value,
550    key: &[String],
551    columns: Option<&[String]>,
552    exclude: &[String],
553    norm: &Normalizer,
554) -> Vec<String> {
555    let names: Vec<String> = match columns {
556        Some(cols) => cols.to_vec(),
557        None => {
558            let mut set: Vec<String> = Vec::new();
559            for side in [a, b] {
560                if let Some(o) = side.as_object() {
561                    for k in o.keys() {
562                        if !set.contains(k) {
563                            set.push(k.clone());
564                        }
565                    }
566                }
567            }
568            set
569        }
570    };
571    let mut out: Vec<String> = names
572        .into_iter()
573        .filter(|n| !is_excluded(n, exclude))
574        .filter(|n| {
575            norm.canonical_column(n, key, a.get(n.as_str()))
576                != norm.canonical_column(n, key, b.get(n.as_str()))
577        })
578        .collect();
579    out.sort();
580    out
581}
582
583fn key_object(record: &Value, key: &[String]) -> Value {
584    let mut m = Map::new();
585    for k in key {
586        m.insert(k.clone(), record.get(k).cloned().unwrap_or(Value::Null));
587    }
588    Value::Object(m)
589}
590
591/// Compare two leaf ranges row by row. Returns every differing key, ordered by
592/// key text so the report is stable. Rows on either side are matched on the
593/// canonical key text; a key present more than once on a side is reported as a
594/// [`DifferenceKind::Duplicate`] and not matched.
595pub fn diff_rows(
596    source: &[Value],
597    dest: &[Value],
598    key: &[String],
599    columns: Option<&[String]>,
600    exclude: &[String],
601    norm: &Normalizer,
602) -> Vec<Difference> {
603    let mut by_key: HashMap<String, (u64, &Value)> = HashMap::with_capacity(dest.len());
604    for d in dest {
605        let e = by_key.entry(key_text(d, key, norm)).or_insert((0, d));
606        e.0 += 1;
607    }
608    // A key repeated on the source side cannot be matched row-for-row: it is
609    // reported once as a duplicate and takes no part in the comparison.
610    let mut source_counts: HashMap<String, u64> = HashMap::with_capacity(source.len());
611    for s in source {
612        *source_counts.entry(key_text(s, key, norm)).or_insert(0) += 1;
613    }
614    let mut out: Vec<(String, Difference)> = Vec::new();
615    let mut reported: std::collections::HashSet<String> = std::collections::HashSet::new();
616    for s in source {
617        let kt = key_text(s, key, norm);
618        let count = source_counts[&kt];
619        if count > 1 {
620            if reported.insert(kt.clone()) {
621                by_key.remove(&kt);
622                out.push((
623                    kt,
624                    Difference {
625                        key: key_object(s, key),
626                        kind: DifferenceKind::Duplicate {
627                            side: Side::Source,
628                            count,
629                        },
630                    },
631                ));
632            }
633            continue;
634        }
635        match by_key.remove(&kt) {
636            None => out.push((
637                kt,
638                Difference {
639                    key: key_object(s, key),
640                    kind: DifferenceKind::MissingInDest,
641                },
642            )),
643            Some((count, _)) if count > 1 => out.push((
644                kt,
645                Difference {
646                    key: key_object(s, key),
647                    kind: DifferenceKind::Duplicate {
648                        side: Side::Destination,
649                        count,
650                    },
651                },
652            )),
653            Some((_, d)) => {
654                let cols = changed_columns(s, d, key, columns, exclude, norm);
655                if !cols.is_empty() {
656                    out.push((
657                        kt,
658                        Difference {
659                            key: key_object(s, key),
660                            kind: DifferenceKind::Changed { columns: cols },
661                        },
662                    ));
663                }
664            }
665        }
666    }
667    for (kt, (count, d)) in by_key {
668        out.push((
669            kt,
670            Difference {
671                key: key_object(d, key),
672                kind: if count > 1 {
673                    DifferenceKind::Duplicate {
674                        side: Side::Destination,
675                        count,
676                    }
677                } else {
678                    DifferenceKind::ExtraInDest
679                },
680            },
681        ));
682    }
683    out.sort_by(|a, b| a.0.cmp(&b.0));
684    out.into_iter().map(|(_, d)| d).collect()
685}
686
687/// Summary of one verification, serialisable for `--json` and the HTTP API.
688#[derive(Debug, Clone, Default, Serialize, Deserialize)]
689pub struct VerifyReport {
690    /// Ranges compared by digest (both passes).
691    pub ranges_compared: u64,
692    /// Ranges whose digests disagreed and were bisected/fetched.
693    pub ranges_differing: u64,
694    /// Rows the source side produced across the leaf fetches.
695    pub rows_fetched_source: u64,
696    /// Rows the destination side produced across the leaf fetches.
697    pub rows_fetched_dest: u64,
698    /// Whether digests were computed inside the backends (no rows shipped for
699    /// matching ranges) rather than by streaming.
700    pub server_digests: bool,
701    pub differences: Vec<Difference>,
702    /// The scan stopped at `max_rows_scanned`; `differences` is a lower bound.
703    pub truncated: bool,
704    /// Rows written back by `--repair` (upserts), if repair ran.
705    #[serde(skip_serializing_if = "Option::is_none")]
706    pub repaired_upserts: Option<u64>,
707    /// Rows deleted by `--repair --allow-delete`, if repair ran.
708    #[serde(skip_serializing_if = "Option::is_none")]
709    pub repaired_deletes: Option<u64>,
710}
711
712impl VerifyReport {
713    /// Counts per difference kind, for the human summary.
714    pub fn tally(&self) -> (usize, usize, usize, usize) {
715        let mut missing = 0;
716        let mut extra = 0;
717        let mut changed = 0;
718        let mut dup = 0;
719        for d in &self.differences {
720            match d.kind {
721                DifferenceKind::MissingInDest => missing += 1,
722                DifferenceKind::ExtraInDest => extra += 1,
723                DifferenceKind::Changed { .. } => changed += 1,
724                DifferenceKind::Duplicate { .. } => dup += 1,
725            }
726        }
727        (missing, extra, changed, dup)
728    }
729
730    /// Whether the two sides matched (no differences and nothing skipped).
731    pub fn equal(&self) -> bool {
732        self.differences.is_empty() && !self.truncated
733    }
734
735    /// Whether a repair re-synced every reported difference (and the report is
736    /// complete), so the destination now matches.
737    pub fn healed(&self) -> bool {
738        if self.truncated || self.differences.is_empty() {
739            return false;
740        }
741        let repaired = self.repaired_upserts.unwrap_or(0) + self.repaired_deletes.unwrap_or(0);
742        repaired >= self.differences.len() as u64
743    }
744}
745
746#[cfg(test)]
747mod tests {
748    use super::*;
749    use serde_json::json;
750
751    fn norm() -> Normalizer {
752        Normalizer::default()
753    }
754
755    #[test]
756    fn ranges_tile_the_key_space_and_bisect_to_leaves() {
757        let ranges = plan_ranges(1, 100, 4);
758        assert_eq!(ranges.len(), 4);
759        assert_eq!(ranges[0].lo, None, "first is unbounded below");
760        assert_eq!(ranges[3].hi, None, "last is unbounded above");
761        for k in [-5, 1, 50, 100, 1_000] {
762            assert_eq!(ranges.iter().filter(|r| r.contains(k)).count(), 1, "{k}");
763        }
764        let (a, b) = KeyRange {
765            lo: Some(0),
766            hi: Some(10),
767        }
768        .bisect(None)
769        .unwrap();
770        assert_eq!(
771            (a.lo, a.hi, b.lo, b.hi),
772            (Some(0), Some(5), Some(5), Some(10))
773        );
774        // An unbounded side clamps to the observed bounds.
775        let (a, b) = KeyRange::ALL.bisect(Some((10, 19))).unwrap();
776        assert_eq!((a.lo, a.hi), (None, Some(15)));
777        assert_eq!((b.lo, b.hi), (Some(15), None));
778        assert!(KeyRange::ALL.bisect(None).is_none());
779        let (a, b) = KeyRange::ALL.bisect(Some((i64::MIN, i64::MAX))).unwrap();
780        assert_eq!((a.lo, a.hi), (None, Some(-1)));
781        assert_eq!((b.lo, b.hi), (Some(-1), None));
782        assert!(
783            KeyRange {
784                lo: Some(3),
785                hi: Some(4)
786            }
787            .bisect(None)
788            .is_none(),
789            "one key wide is a leaf"
790        );
791        assert_eq!(
792            KeyRange {
793                lo: Some(3),
794                hi: Some(4)
795            }
796            .width(),
797            Some(1)
798        );
799        assert_eq!(KeyRange::ALL.width(), None);
800        assert_eq!(KeyRange::ALL.to_string(), "all");
801        assert_eq!(
802            KeyRange {
803                lo: Some(1),
804                hi: None
805            }
806            .to_string(),
807            "[1, +inf)"
808        );
809        assert_eq!(
810            KeyRange {
811                lo: None,
812                hi: Some(9)
813            }
814            .to_string(),
815            "(-inf, 9)"
816        );
817    }
818
819    #[test]
820    fn range_shard_is_the_pk_range_shape_sql_sources_parse() {
821        let shard = KeyRange {
822            lo: Some(5),
823            hi: None,
824        }
825        .to_shard("id");
826        let bounds = crate::shard::PkShardBounds::from_spec(&shard).unwrap();
827        assert_eq!(
828            (bounds.lo, bounds.hi_unbounded, bounds.include_null),
829            (5, true, true)
830        );
831        assert_eq!(shard.id, "5..+inf");
832        let sql = bounds.wrap("SELECT * FROM t", |s| format!("\"{s}\""));
833        assert!(
834            sql.contains("\"id\" >= 5") && sql.contains("IS NULL"),
835            "{sql}"
836        );
837    }
838
839    #[test]
840    fn canonical_values_agree_across_spellings() {
841        let n = norm();
842        assert_eq!(n.canonical(Some(&json!(1))), n.canonical(Some(&json!(1.0))));
843        assert_eq!(n.canonical(None), n.canonical(Some(&json!(null))));
844        assert_ne!(n.canonical(Some(&json!("1"))), n.canonical(Some(&json!(1))));
845        assert_eq!(
846            n.canonical(Some(&json!("2026-09-25T10:00:00Z"))),
847            n.canonical(Some(&json!("2026-09-25 12:00:00+02:00")))
848        );
849        assert_eq!(
850            n.canonical(Some(&json!("2026-09-25 10:00:00.000"))),
851            n.canonical(Some(&json!("2026-09-25T10:00:00Z")))
852        );
853        assert_eq!(
854            n.canonical(Some(&json!("2026-09-25"))),
855            n.canonical(Some(&json!("2026-09-25T00:00:00Z")))
856        );
857        assert_eq!(
858            n.canonical(Some(&json!({"b": 1, "a": [true, null]}))),
859            n.canonical(Some(&json!({"a": [true, null], "b": 1})))
860        );
861        // Plain strings never pay for a timestamp parse and stay themselves.
862        assert_eq!(n.canonical(Some(&json!("hello"))), "shello");
863        assert_eq!(n.canonical(Some(&json!("2026-x"))), "s2026-x");
864        assert!(parse_timestamp_micros("not a date").is_none());
865    }
866
867    #[test]
868    fn array_and_object_text_cannot_be_forged_by_element_contents() {
869        let n = norm();
870        assert_ne!(
871            n.canonical(Some(&json!(["a", "b"]))),
872            n.canonical(Some(&json!(["a,sb"])))
873        );
874        assert_ne!(
875            n.canonical(Some(&json!({"a": "x", "b": "y"}))),
876            n.canonical(Some(&json!({"a": "x,b:sy"})))
877        );
878    }
879
880    #[test]
881    fn numeric_and_text_keys_match_and_repair_never_deletes_them() {
882        let n = norm();
883        let key = vec!["id".to_string()];
884        let src = vec![json!({"id": 7, "v": "a"}), json!({"id": 8, "v": "b"})];
885        let dst = vec![json!({"id": "7", "v": "a"}), json!({"id": "8", "v": "x"})];
886        let diffs = diff_rows(&src, &dst, &key, None, &[], &n);
887        assert_eq!(diffs.len(), 1, "{diffs:?}");
888        assert!(matches!(
889            &diffs[0].kind,
890            DifferenceKind::Changed { columns } if columns == &vec!["v".to_string()]
891        ));
892        assert_eq!(
893            row_hash(&src[0], &key, None, &[], &n),
894            row_hash(&dst[0], &key, None, &[], &n)
895        );
896        assert_ne!(
897            n.canonical(Some(&json!("7"))),
898            n.canonical(Some(&json!(7))),
899            "non-key columns keep numeric_strings off"
900        );
901    }
902
903    #[test]
904    fn float_tolerance_and_numeric_strings_are_opt_in() {
905        let exact = norm();
906        assert_ne!(
907            exact.canonical(Some(&json!(1.00000001))),
908            exact.canonical(Some(&json!(1.00000002)))
909        );
910        let loose = Normalizer {
911            float_tolerance: 1e-6,
912            ..norm()
913        };
914        assert_eq!(
915            loose.canonical(Some(&json!(1.00000001))),
916            loose.canonical(Some(&json!(1.00000002)))
917        );
918        let strings = Normalizer {
919            numeric_strings: true,
920            ..norm()
921        };
922        assert_eq!(
923            strings.canonical(Some(&json!("42"))),
924            strings.canonical(Some(&json!(42)))
925        );
926        assert_eq!(
927            strings.canonical(Some(&json!("1.5"))),
928            strings.canonical(Some(&json!(1.5)))
929        );
930        assert_eq!(strings.canonical(Some(&json!("abc"))), "sabc");
931        assert!(loose.validate().is_ok());
932        assert!(
933            Normalizer {
934                float_tolerance: -1.0,
935                ..norm()
936            }
937            .validate()
938            .is_err()
939        );
940        assert!(
941            Normalizer {
942                float_tolerance: f64::NAN,
943                ..norm()
944            }
945            .validate()
946            .is_err()
947        );
948    }
949
950    #[test]
951    fn row_hash_treats_absent_and_null_columns_alike() {
952        let key = vec!["id".to_string()];
953        let src = json!({"id": 1, "name": "a"});
954        let dst = json!({"id": 1, "name": "a", "note": null});
955        assert_eq!(
956            row_hash(&src, &key, None, &[], &norm()),
957            row_hash(&dst, &key, None, &[], &norm())
958        );
959        assert!(diff_rows(std::slice::from_ref(&src), &[dst], &key, None, &[], &norm()).is_empty());
960        assert_ne!(
961            row_hash(&src, &key, None, &[], &norm()),
962            row_hash(
963                &json!({"id": 1, "name": "a", "note": "x"}),
964                &key,
965                None,
966                &[],
967                &norm()
968            )
969        );
970    }
971
972    #[test]
973    fn row_hash_ignores_field_order_and_excluded_columns() {
974        let key = vec!["id".to_string()];
975        let a = json!({"id": 1, "name": "x", "_faucet_run_id": "r1"});
976        let b = json!({"name": "x", "id": 1, "_faucet_run_id": "r2"});
977        let ex = vec!["_faucet_*".to_string()];
978        assert_eq!(
979            row_hash(&a, &key, None, &ex, &norm()),
980            row_hash(&b, &key, None, &ex, &norm())
981        );
982        assert_ne!(
983            row_hash(&a, &key, None, &[], &norm()),
984            row_hash(&b, &key, None, &[], &norm()),
985            "without the exclusion the run id differs"
986        );
987        // An explicit column list hashes only those (plus the key).
988        let c = json!({"id": 1, "name": "y", "other": 5});
989        let cols = vec!["other".to_string()];
990        assert_eq!(
991            row_hash(&a, &key, Some(&cols), &[], &norm()),
992            row_hash(
993                &json!({"id": 1, "name": "zzz"}),
994                &key,
995                Some(&cols),
996                &[],
997                &norm()
998            )
999        );
1000        assert_ne!(
1001            row_hash(&c, &key, Some(&cols), &[], &norm()),
1002            row_hash(&a, &key, Some(&cols), &[], &norm())
1003        );
1004        assert!(is_excluded("_faucet_loaded_at", &ex));
1005        assert!(!is_excluded("faucet", &ex));
1006        assert!(is_excluded("name", &["name".to_string()]));
1007        assert_eq!(key_int(&a, &key), Some(1));
1008        assert_eq!(key_int(&json!({"id": "7"}), &key), Some(7));
1009        assert_eq!(key_int(&a, &["id".into(), "name".into()]), None);
1010        assert_eq!(key_int(&json!({"id": true}), &key), None);
1011        assert_eq!(key_text(&a, &key, &norm()), "id=n1;");
1012    }
1013
1014    #[test]
1015    fn digest_is_order_independent_and_tracks_bounds() {
1016        let key = vec!["id".to_string()];
1017        let rows = [json!({"id": 3, "v": "a"}), json!({"id": 1, "v": "b"})];
1018        let mut fwd = DigestAccumulator::new();
1019        let mut rev = DigestAccumulator::new();
1020        for r in &rows {
1021            fwd.add(row_hash(r, &key, None, &[], &norm()), key_int(r, &key));
1022        }
1023        for r in rows.iter().rev() {
1024            rev.add(row_hash(r, &key, None, &[], &norm()), key_int(r, &key));
1025        }
1026        let (f, r) = (fwd.finish(), rev.finish());
1027        assert!(f.same(&r));
1028        assert_eq!((f.rows, f.key_min, f.key_max), (2, Some(1), Some(3)));
1029        let mut other = DigestAccumulator::new();
1030        other.add(
1031            row_hash(&json!({"id": 3, "v": "a"}), &key, None, &[], &norm()),
1032            Some(3),
1033        );
1034        let o = other.finish();
1035        assert!(!f.same(&o));
1036        assert_eq!(f.bounds_union(&o), Some((1, 3)));
1037        assert_eq!(
1038            ContentDigest::default().bounds_union(&ContentDigest::default()),
1039            None
1040        );
1041        assert_eq!(ContentDigest::default().bounds_union(&o), Some((3, 3)));
1042    }
1043
1044    #[test]
1045    fn server_digests_compare_only_within_one_algorithm() {
1046        let pg = ServerDigest {
1047            algorithm: "postgres:v1".into(),
1048            rows: 2,
1049            digest: "12".into(),
1050            key_min: Some(1),
1051            key_max: Some(5),
1052        };
1053        let same = ServerDigest {
1054            key_min: Some(2),
1055            key_max: Some(9),
1056            ..pg.clone()
1057        };
1058        let other_algo = ServerDigest {
1059            algorithm: "sqlite:v1".into(),
1060            ..pg.clone()
1061        };
1062        assert!(pg.same(&same));
1063        assert!(!pg.comparable(&other_algo));
1064        assert!(!pg.same(&other_algo));
1065        assert!(!pg.same(&ServerDigest {
1066            digest: "13".into(),
1067            ..pg.clone()
1068        }));
1069        assert_eq!(pg.bounds_union(&same), Some((1, 9)));
1070        assert_eq!(
1071            pg.bounds_union(&ServerDigest {
1072                key_min: None,
1073                key_max: None,
1074                ..pg.clone()
1075            }),
1076            Some((1, 5))
1077        );
1078    }
1079
1080    #[test]
1081    fn diff_rows_reports_every_kind_in_key_order() {
1082        let key = vec!["id".to_string()];
1083        let source = vec![
1084            json!({"id": 1, "v": "a"}),
1085            json!({"id": 2, "v": "b"}),
1086            json!({"id": 3, "v": "c"}),
1087            json!({"id": 5, "v": "e"}),
1088            json!({"id": 5, "v": "e2"}),
1089        ];
1090        let dest = vec![
1091            json!({"id": 1, "v": "a", "_faucet_run_id": "r"}),
1092            json!({"id": 2, "v": "B"}),
1093            json!({"id": 4, "v": "d"}),
1094            json!({"id": 6, "v": "f"}),
1095            json!({"id": 6, "v": "f"}),
1096        ];
1097        let ex = vec!["_faucet_*".to_string()];
1098        let diffs = diff_rows(&source, &dest, &key, None, &ex, &norm());
1099        let kinds: Vec<String> = diffs
1100            .iter()
1101            .map(|d| format!("{}:{:?}", d.key["id"], d.kind))
1102            .collect();
1103        assert_eq!(diffs.len(), 5, "{kinds:?}");
1104        assert_eq!(
1105            diffs[0].kind,
1106            DifferenceKind::Changed {
1107                columns: vec!["v".into()]
1108            }
1109        );
1110        assert_eq!(diffs[0].key, json!({"id": 2}));
1111        assert_eq!(diffs[1].kind, DifferenceKind::MissingInDest);
1112        assert_eq!(diffs[1].key, json!({"id": 3}));
1113        assert_eq!(diffs[2].kind, DifferenceKind::ExtraInDest);
1114        assert_eq!(diffs[2].key, json!({"id": 4}));
1115        assert_eq!(
1116            diffs[3].kind,
1117            DifferenceKind::Duplicate {
1118                side: Side::Source,
1119                count: 2
1120            }
1121        );
1122        assert_eq!(
1123            diffs[4].kind,
1124            DifferenceKind::Duplicate {
1125                side: Side::Destination,
1126                count: 2
1127            }
1128        );
1129        assert!(diffs[0].needs_upsert() && diffs[1].needs_upsert());
1130        assert!(diffs[2].needs_delete() && !diffs[2].needs_upsert());
1131        assert!(!diffs[3].needs_upsert() && !diffs[3].needs_delete());
1132        // Identical sides: nothing.
1133        assert!(diff_rows(&source[..1], &dest[..1], &key, None, &ex, &norm()).is_empty());
1134        // A key repeated on both sides is one report, on the source side, with
1135        // the source's count; it is never also "missing" or "extra".
1136        let three = vec![json!({"id": 9}), json!({"id": 9}), json!({"id": 9})];
1137        let d = diff_rows(
1138            &three,
1139            &[json!({"id": 9}), json!({"id": 9})],
1140            &key,
1141            None,
1142            &[],
1143            &norm(),
1144        );
1145        assert_eq!(d.len(), 1, "a duplicated key is reported once: {d:?}");
1146        assert_eq!(
1147            d[0].kind,
1148            DifferenceKind::Duplicate {
1149                side: Side::Source,
1150                count: 3
1151            }
1152        );
1153    }
1154
1155    #[test]
1156    fn diff_rows_with_explicit_columns_ignores_the_rest() {
1157        let key = vec!["id".to_string()];
1158        let cols = vec!["amount".to_string()];
1159        let s = vec![json!({"id": 1, "amount": 10, "note": "x"})];
1160        let d = vec![json!({"id": 1, "amount": 10, "note": "y"})];
1161        assert!(diff_rows(&s, &d, &key, Some(&cols), &[], &norm()).is_empty());
1162        let d2 = vec![json!({"id": 1, "amount": 11})];
1163        let diffs = diff_rows(&s, &d2, &key, Some(&cols), &[], &norm());
1164        assert_eq!(
1165            diffs[0].kind,
1166            DifferenceKind::Changed {
1167                columns: vec!["amount".into()]
1168            }
1169        );
1170    }
1171
1172    #[test]
1173    fn range_width_display_and_number_canonical_forms() {
1174        let inverted = KeyRange {
1175            lo: Some(5),
1176            hi: Some(5),
1177        };
1178        assert_eq!(inverted.width(), Some(0));
1179        assert_eq!(format!("{inverted}"), "[5, 5)");
1180        assert_eq!(format!("{}", KeyRange::ALL), "all");
1181        let norm = Normalizer::default();
1182        assert_eq!(
1183            norm.canonical(Some(&json!(u64::MAX))),
1184            format!("n{}", u64::MAX)
1185        );
1186        assert_eq!(norm.canonical(Some(&json!(-3))), "n-3");
1187        assert_eq!(norm.canonical(Some(&json!(2.0))), "n2");
1188    }
1189
1190    #[test]
1191    fn timestamps_parse_every_supported_spelling_or_none() {
1192        assert!(parse_timestamp_micros("2026-01-01T00:00:00Z").is_some());
1193        assert!(parse_timestamp_micros("2026-01-01 00:00:00.5").is_some());
1194        assert!(parse_timestamp_micros("2026-01-01 00:00:00+02:00").is_some());
1195        assert!(parse_timestamp_micros("2026-01-01").is_some());
1196        assert_eq!(parse_timestamp_micros("2026-13-99"), None);
1197        assert_eq!(parse_timestamp_micros("not a date"), None);
1198        assert_eq!(parse_timestamp_micros("20260101"), None);
1199    }
1200
1201    #[test]
1202    fn bounds_union_needs_both_ends() {
1203        let a = ContentDigest {
1204            key_min: Some(1),
1205            ..Default::default()
1206        };
1207        let b = ContentDigest::default();
1208        assert_eq!(a.bounds_union(&b), None, "no max anywhere");
1209        let sd = ServerDigest {
1210            algorithm: "x".into(),
1211            rows: 0,
1212            digest: "0".into(),
1213            key_min: None,
1214            key_max: Some(9),
1215        };
1216        assert_eq!(sd.bounds_union(&sd), None, "no min anywhere");
1217        let other = ServerDigest {
1218            key_min: Some(2),
1219            ..sd.clone()
1220        };
1221        assert_eq!(sd.bounds_union(&other), Some((2, 9)));
1222    }
1223
1224    #[test]
1225    fn duplicate_destination_keys_are_reported_once_on_that_side() {
1226        let source = vec![json!({"id": 1, "v": "a"})];
1227        let dest = vec![json!({"id": 1, "v": "a"}), json!({"id": 1, "v": "b"})];
1228        let d = diff_rows(
1229            &source,
1230            &dest,
1231            &["id".into()],
1232            None,
1233            &[],
1234            &Normalizer::default(),
1235        );
1236        assert_eq!(d.len(), 1, "{d:?}");
1237        assert_eq!(
1238            d[0].kind,
1239            DifferenceKind::Duplicate {
1240                side: Side::Destination,
1241                count: 2
1242            }
1243        );
1244        assert!(!d[0].needs_upsert() && !d[0].needs_delete());
1245    }
1246
1247    #[test]
1248    fn report_healed_only_when_every_difference_was_repaired() {
1249        let mut r = VerifyReport::default();
1250        assert!(!r.healed(), "nothing to heal");
1251        r.differences.push(Difference {
1252            key: json!({"id": 1}),
1253            kind: DifferenceKind::MissingInDest,
1254        });
1255        r.differences.push(Difference {
1256            key: json!({"id": 2}),
1257            kind: DifferenceKind::ExtraInDest,
1258        });
1259        assert!(!r.healed(), "no repair ran");
1260        r.repaired_upserts = Some(1);
1261        r.repaired_deletes = Some(0);
1262        assert!(!r.healed(), "the extra row was not deleted");
1263        r.repaired_deletes = Some(1);
1264        assert!(r.healed());
1265        r.truncated = true;
1266        assert!(!r.healed(), "a truncated report cannot claim to be healed");
1267    }
1268
1269    #[test]
1270    fn report_tallies_and_equality() {
1271        let mut r = VerifyReport::default();
1272        assert!(r.equal());
1273        r.differences.push(Difference {
1274            key: json!({"id": 1}),
1275            kind: DifferenceKind::MissingInDest,
1276        });
1277        r.differences.push(Difference {
1278            key: json!({"id": 2}),
1279            kind: DifferenceKind::ExtraInDest,
1280        });
1281        r.differences.push(Difference {
1282            key: json!({"id": 3}),
1283            kind: DifferenceKind::Changed { columns: vec![] },
1284        });
1285        r.differences.push(Difference {
1286            key: json!({"id": 4}),
1287            kind: DifferenceKind::Duplicate {
1288                side: Side::Source,
1289                count: 2,
1290            },
1291        });
1292        assert_eq!(r.tally(), (1, 1, 1, 1));
1293        assert!(!r.equal());
1294        let truncated = VerifyReport {
1295            truncated: true,
1296            ..Default::default()
1297        };
1298        assert!(!truncated.equal(), "a truncated scan proves nothing");
1299        let json = serde_json::to_value(&r).unwrap();
1300        assert_eq!(json["differences"][0]["kind"], "missing_in_dest");
1301        assert_eq!(json["differences"][2]["kind"], "changed");
1302    }
1303}