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.
325pub fn row_hash(
326    record: &Value,
327    key: &[String],
328    columns: Option<&[String]>,
329    exclude: &[String],
330    norm: &Normalizer,
331) -> u128 {
332    let obj = record.as_object();
333    let mut text = String::new();
334    let mut names: Vec<&str> = match columns {
335        Some(cols) => cols.iter().map(String::as_str).collect(),
336        None => obj
337            .map(|o| o.keys().map(String::as_str).collect())
338            .unwrap_or_default(),
339    };
340    for k in key {
341        if !names.contains(&k.as_str()) {
342            names.push(k.as_str());
343        }
344    }
345    names.retain(|n| !is_excluded(n, exclude));
346    names.sort_unstable();
347    names.dedup();
348    for name in names {
349        text.push_str(name);
350        text.push('=');
351        text.push_str(&norm.canonical_column(name, key, obj.and_then(|o| o.get(name))));
352        text.push('\u{1f}');
353    }
354    let a = fnv1a_64(text.as_bytes(), FNV_BASIS_A);
355    let b = fnv1a_64(text.as_bytes(), FNV_BASIS_B);
356    (u128::from(a) << 64) | u128::from(b)
357}
358
359/// Whether a column is excluded: an exact name, or a `prefix*` glob.
360pub fn is_excluded(name: &str, exclude: &[String]) -> bool {
361    exclude.iter().any(|e| match e.strip_suffix('*') {
362        Some(prefix) => name.starts_with(prefix),
363        None => e == name,
364    })
365}
366
367/// The canonical text of a record's key tuple (`k1=…;k2=…`), used to match
368/// rows across the two sides.
369pub fn key_text(record: &Value, key: &[String], norm: &Normalizer) -> String {
370    let obj = record.as_object();
371    let norm = norm.for_key();
372    let mut out = String::new();
373    for k in key {
374        out.push_str(k);
375        out.push('=');
376        out.push_str(&norm.canonical(obj.and_then(|o| o.get(k))));
377        out.push(';');
378    }
379    out
380}
381
382/// The integer value of a single-column key, when it is one.
383pub fn key_int(record: &Value, key: &[String]) -> Option<i64> {
384    if key.len() != 1 {
385        return None;
386    }
387    match record.get(&key[0])? {
388        Value::Number(n) => n.as_i64(),
389        Value::String(s) => s.trim().parse().ok(),
390        _ => None,
391    }
392}
393
394/// An order-independent digest of a set of rows.
395#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
396pub struct ContentDigest {
397    pub rows: u64,
398    /// Wrapping sum of the row fingerprints.
399    pub sum: u128,
400    /// XOR of the row fingerprints (catches a sum collision from swapped rows).
401    pub xor: u128,
402    /// Smallest / largest integer key seen (single integer keys only).
403    pub key_min: Option<i64>,
404    pub key_max: Option<i64>,
405}
406
407impl ContentDigest {
408    /// Whether two digests describe the same row set.
409    pub fn same(&self, other: &ContentDigest) -> bool {
410        self.rows == other.rows && self.sum == other.sum && self.xor == other.xor
411    }
412
413    /// The union of the observed key bounds of two digests.
414    pub fn bounds_union(&self, other: &ContentDigest) -> Option<(i64, i64)> {
415        let min = match (self.key_min, other.key_min) {
416            (Some(a), Some(b)) => Some(a.min(b)),
417            (a, b) => a.or(b),
418        }?;
419        let max = match (self.key_max, other.key_max) {
420            (Some(a), Some(b)) => Some(a.max(b)),
421            (a, b) => a.or(b),
422        }?;
423        Some((min, max))
424    }
425}
426
427/// Streams rows into a [`ContentDigest`].
428#[derive(Debug, Default)]
429pub struct DigestAccumulator {
430    digest: ContentDigest,
431}
432
433impl DigestAccumulator {
434    pub fn new() -> Self {
435        Self::default()
436    }
437
438    /// Fold one row in.
439    pub fn add(&mut self, hash: u128, key: Option<i64>) {
440        self.digest.rows += 1;
441        self.digest.sum = self.digest.sum.wrapping_add(hash);
442        self.digest.xor ^= hash;
443        if let Some(k) = key {
444            self.digest.key_min = Some(self.digest.key_min.map_or(k, |m| m.min(k)));
445            self.digest.key_max = Some(self.digest.key_max.map_or(k, |m| m.max(k)));
446        }
447    }
448
449    pub fn finish(self) -> ContentDigest {
450        self.digest
451    }
452}
453
454/// A digest a backend computed itself, without shipping rows. Two server
455/// digests are comparable only when their `algorithm` ids match — each backend
456/// hashes its own text rendering of a row, so a Postgres digest says nothing
457/// about a SQLite table. When the algorithms differ the verifier streams both
458/// sides and uses [`ContentDigest`] instead.
459#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
460pub struct ServerDigest {
461    pub algorithm: String,
462    pub rows: u64,
463    /// Opaque digest text (a decimal sum, a hex string — whatever the
464    /// algorithm defines).
465    pub digest: String,
466    pub key_min: Option<i64>,
467    pub key_max: Option<i64>,
468}
469
470impl ServerDigest {
471    /// Whether two server digests agree. Different algorithms never agree —
472    /// the caller must not treat that as a difference but as "not comparable".
473    pub fn comparable(&self, other: &ServerDigest) -> bool {
474        self.algorithm == other.algorithm
475    }
476
477    pub fn same(&self, other: &ServerDigest) -> bool {
478        self.comparable(other) && self.rows == other.rows && self.digest == other.digest
479    }
480
481    pub fn bounds_union(&self, other: &ServerDigest) -> Option<(i64, i64)> {
482        let min = match (self.key_min, other.key_min) {
483            (Some(a), Some(b)) => Some(a.min(b)),
484            (a, b) => a.or(b),
485        }?;
486        let max = match (self.key_max, other.key_max) {
487            (Some(a), Some(b)) => Some(a.max(b)),
488            (a, b) => a.or(b),
489        }?;
490        Some((min, max))
491    }
492}
493
494/// Why one key differs between the two sides.
495#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
496#[serde(tag = "kind", rename_all = "snake_case")]
497pub enum DifferenceKind {
498    /// The source has the key, the destination does not.
499    MissingInDest,
500    /// The destination has the key, the source does not.
501    ExtraInDest,
502    /// Both have the key; these columns differ.
503    Changed { columns: Vec<String> },
504    /// The key appears more than once on one side, so rows cannot be matched.
505    Duplicate { side: Side, count: u64 },
506}
507
508#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
509#[serde(rename_all = "snake_case")]
510pub enum Side {
511    Source,
512    Destination,
513}
514
515/// One differing key.
516#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
517pub struct Difference {
518    /// The key columns and their values (from whichever side has the row).
519    pub key: Value,
520    #[serde(flatten)]
521    pub kind: DifferenceKind,
522}
523
524impl Difference {
525    /// Whether repairing this difference means writing the source row.
526    pub fn needs_upsert(&self) -> bool {
527        matches!(
528            self.kind,
529            DifferenceKind::MissingInDest | DifferenceKind::Changed { .. }
530        )
531    }
532
533    /// Whether repairing this difference means deleting the destination row.
534    pub fn needs_delete(&self) -> bool {
535        matches!(self.kind, DifferenceKind::ExtraInDest)
536    }
537}
538
539/// The columns whose canonical values differ between two rows.
540fn changed_columns(
541    a: &Value,
542    b: &Value,
543    key: &[String],
544    columns: Option<&[String]>,
545    exclude: &[String],
546    norm: &Normalizer,
547) -> Vec<String> {
548    let names: Vec<String> = match columns {
549        Some(cols) => cols.to_vec(),
550        None => {
551            let mut set: Vec<String> = Vec::new();
552            for side in [a, b] {
553                if let Some(o) = side.as_object() {
554                    for k in o.keys() {
555                        if !set.contains(k) {
556                            set.push(k.clone());
557                        }
558                    }
559                }
560            }
561            set
562        }
563    };
564    let mut out: Vec<String> = names
565        .into_iter()
566        .filter(|n| !is_excluded(n, exclude))
567        .filter(|n| {
568            norm.canonical_column(n, key, a.get(n.as_str()))
569                != norm.canonical_column(n, key, b.get(n.as_str()))
570        })
571        .collect();
572    out.sort();
573    out
574}
575
576fn key_object(record: &Value, key: &[String]) -> Value {
577    let mut m = Map::new();
578    for k in key {
579        m.insert(k.clone(), record.get(k).cloned().unwrap_or(Value::Null));
580    }
581    Value::Object(m)
582}
583
584/// Compare two leaf ranges row by row. Returns every differing key, ordered by
585/// key text so the report is stable. Rows on either side are matched on the
586/// canonical key text; a key present more than once on a side is reported as a
587/// [`DifferenceKind::Duplicate`] and not matched.
588pub fn diff_rows(
589    source: &[Value],
590    dest: &[Value],
591    key: &[String],
592    columns: Option<&[String]>,
593    exclude: &[String],
594    norm: &Normalizer,
595) -> Vec<Difference> {
596    let mut by_key: HashMap<String, (u64, &Value)> = HashMap::with_capacity(dest.len());
597    for d in dest {
598        let e = by_key.entry(key_text(d, key, norm)).or_insert((0, d));
599        e.0 += 1;
600    }
601    // A key repeated on the source side cannot be matched row-for-row: it is
602    // reported once as a duplicate and takes no part in the comparison.
603    let mut source_counts: HashMap<String, u64> = HashMap::with_capacity(source.len());
604    for s in source {
605        *source_counts.entry(key_text(s, key, norm)).or_insert(0) += 1;
606    }
607    let mut out: Vec<(String, Difference)> = Vec::new();
608    let mut reported: std::collections::HashSet<String> = std::collections::HashSet::new();
609    for s in source {
610        let kt = key_text(s, key, norm);
611        let count = source_counts[&kt];
612        if count > 1 {
613            if reported.insert(kt.clone()) {
614                by_key.remove(&kt);
615                out.push((
616                    kt,
617                    Difference {
618                        key: key_object(s, key),
619                        kind: DifferenceKind::Duplicate {
620                            side: Side::Source,
621                            count,
622                        },
623                    },
624                ));
625            }
626            continue;
627        }
628        match by_key.remove(&kt) {
629            None => out.push((
630                kt,
631                Difference {
632                    key: key_object(s, key),
633                    kind: DifferenceKind::MissingInDest,
634                },
635            )),
636            Some((count, _)) if count > 1 => out.push((
637                kt,
638                Difference {
639                    key: key_object(s, key),
640                    kind: DifferenceKind::Duplicate {
641                        side: Side::Destination,
642                        count,
643                    },
644                },
645            )),
646            Some((_, d)) => {
647                let cols = changed_columns(s, d, key, columns, exclude, norm);
648                if !cols.is_empty() {
649                    out.push((
650                        kt,
651                        Difference {
652                            key: key_object(s, key),
653                            kind: DifferenceKind::Changed { columns: cols },
654                        },
655                    ));
656                }
657            }
658        }
659    }
660    for (kt, (count, d)) in by_key {
661        out.push((
662            kt,
663            Difference {
664                key: key_object(d, key),
665                kind: if count > 1 {
666                    DifferenceKind::Duplicate {
667                        side: Side::Destination,
668                        count,
669                    }
670                } else {
671                    DifferenceKind::ExtraInDest
672                },
673            },
674        ));
675    }
676    out.sort_by(|a, b| a.0.cmp(&b.0));
677    out.into_iter().map(|(_, d)| d).collect()
678}
679
680/// Summary of one verification, serialisable for `--json` and the HTTP API.
681#[derive(Debug, Clone, Default, Serialize, Deserialize)]
682pub struct VerifyReport {
683    /// Ranges compared by digest (both passes).
684    pub ranges_compared: u64,
685    /// Ranges whose digests disagreed and were bisected/fetched.
686    pub ranges_differing: u64,
687    /// Rows the source side produced across the leaf fetches.
688    pub rows_fetched_source: u64,
689    /// Rows the destination side produced across the leaf fetches.
690    pub rows_fetched_dest: u64,
691    /// Whether digests were computed inside the backends (no rows shipped for
692    /// matching ranges) rather than by streaming.
693    pub server_digests: bool,
694    pub differences: Vec<Difference>,
695    /// The scan stopped at `max_rows_scanned`; `differences` is a lower bound.
696    pub truncated: bool,
697    /// Rows written back by `--repair` (upserts), if repair ran.
698    #[serde(skip_serializing_if = "Option::is_none")]
699    pub repaired_upserts: Option<u64>,
700    /// Rows deleted by `--repair --allow-delete`, if repair ran.
701    #[serde(skip_serializing_if = "Option::is_none")]
702    pub repaired_deletes: Option<u64>,
703}
704
705impl VerifyReport {
706    /// Counts per difference kind, for the human summary.
707    pub fn tally(&self) -> (usize, usize, usize, usize) {
708        let mut missing = 0;
709        let mut extra = 0;
710        let mut changed = 0;
711        let mut dup = 0;
712        for d in &self.differences {
713            match d.kind {
714                DifferenceKind::MissingInDest => missing += 1,
715                DifferenceKind::ExtraInDest => extra += 1,
716                DifferenceKind::Changed { .. } => changed += 1,
717                DifferenceKind::Duplicate { .. } => dup += 1,
718            }
719        }
720        (missing, extra, changed, dup)
721    }
722
723    /// Whether the two sides matched (no differences and nothing skipped).
724    pub fn equal(&self) -> bool {
725        self.differences.is_empty() && !self.truncated
726    }
727
728    /// Whether a repair re-synced every reported difference (and the report is
729    /// complete), so the destination now matches.
730    pub fn healed(&self) -> bool {
731        if self.truncated || self.differences.is_empty() {
732            return false;
733        }
734        let repaired = self.repaired_upserts.unwrap_or(0) + self.repaired_deletes.unwrap_or(0);
735        repaired >= self.differences.len() as u64
736    }
737}
738
739#[cfg(test)]
740mod tests {
741    use super::*;
742    use serde_json::json;
743
744    fn norm() -> Normalizer {
745        Normalizer::default()
746    }
747
748    #[test]
749    fn ranges_tile_the_key_space_and_bisect_to_leaves() {
750        let ranges = plan_ranges(1, 100, 4);
751        assert_eq!(ranges.len(), 4);
752        assert_eq!(ranges[0].lo, None, "first is unbounded below");
753        assert_eq!(ranges[3].hi, None, "last is unbounded above");
754        for k in [-5, 1, 50, 100, 1_000] {
755            assert_eq!(ranges.iter().filter(|r| r.contains(k)).count(), 1, "{k}");
756        }
757        let (a, b) = KeyRange {
758            lo: Some(0),
759            hi: Some(10),
760        }
761        .bisect(None)
762        .unwrap();
763        assert_eq!(
764            (a.lo, a.hi, b.lo, b.hi),
765            (Some(0), Some(5), Some(5), Some(10))
766        );
767        // An unbounded side clamps to the observed bounds.
768        let (a, b) = KeyRange::ALL.bisect(Some((10, 19))).unwrap();
769        assert_eq!((a.lo, a.hi), (None, Some(15)));
770        assert_eq!((b.lo, b.hi), (Some(15), None));
771        assert!(KeyRange::ALL.bisect(None).is_none());
772        let (a, b) = KeyRange::ALL.bisect(Some((i64::MIN, i64::MAX))).unwrap();
773        assert_eq!((a.lo, a.hi), (None, Some(-1)));
774        assert_eq!((b.lo, b.hi), (Some(-1), None));
775        assert!(
776            KeyRange {
777                lo: Some(3),
778                hi: Some(4)
779            }
780            .bisect(None)
781            .is_none(),
782            "one key wide is a leaf"
783        );
784        assert_eq!(
785            KeyRange {
786                lo: Some(3),
787                hi: Some(4)
788            }
789            .width(),
790            Some(1)
791        );
792        assert_eq!(KeyRange::ALL.width(), None);
793        assert_eq!(KeyRange::ALL.to_string(), "all");
794        assert_eq!(
795            KeyRange {
796                lo: Some(1),
797                hi: None
798            }
799            .to_string(),
800            "[1, +inf)"
801        );
802        assert_eq!(
803            KeyRange {
804                lo: None,
805                hi: Some(9)
806            }
807            .to_string(),
808            "(-inf, 9)"
809        );
810    }
811
812    #[test]
813    fn range_shard_is_the_pk_range_shape_sql_sources_parse() {
814        let shard = KeyRange {
815            lo: Some(5),
816            hi: None,
817        }
818        .to_shard("id");
819        let bounds = crate::shard::PkShardBounds::from_spec(&shard).unwrap();
820        assert_eq!(
821            (bounds.lo, bounds.hi_unbounded, bounds.include_null),
822            (5, true, true)
823        );
824        assert_eq!(shard.id, "5..+inf");
825        let sql = bounds.wrap("SELECT * FROM t", |s| format!("\"{s}\""));
826        assert!(
827            sql.contains("\"id\" >= 5") && sql.contains("IS NULL"),
828            "{sql}"
829        );
830    }
831
832    #[test]
833    fn canonical_values_agree_across_spellings() {
834        let n = norm();
835        assert_eq!(n.canonical(Some(&json!(1))), n.canonical(Some(&json!(1.0))));
836        assert_eq!(n.canonical(None), n.canonical(Some(&json!(null))));
837        assert_ne!(n.canonical(Some(&json!("1"))), n.canonical(Some(&json!(1))));
838        assert_eq!(
839            n.canonical(Some(&json!("2026-09-25T10:00:00Z"))),
840            n.canonical(Some(&json!("2026-09-25 12:00:00+02:00")))
841        );
842        assert_eq!(
843            n.canonical(Some(&json!("2026-09-25 10:00:00.000"))),
844            n.canonical(Some(&json!("2026-09-25T10:00:00Z")))
845        );
846        assert_eq!(
847            n.canonical(Some(&json!("2026-09-25"))),
848            n.canonical(Some(&json!("2026-09-25T00:00:00Z")))
849        );
850        assert_eq!(
851            n.canonical(Some(&json!({"b": 1, "a": [true, null]}))),
852            n.canonical(Some(&json!({"a": [true, null], "b": 1})))
853        );
854        // Plain strings never pay for a timestamp parse and stay themselves.
855        assert_eq!(n.canonical(Some(&json!("hello"))), "shello");
856        assert_eq!(n.canonical(Some(&json!("2026-x"))), "s2026-x");
857        assert!(parse_timestamp_micros("not a date").is_none());
858    }
859
860    #[test]
861    fn array_and_object_text_cannot_be_forged_by_element_contents() {
862        let n = norm();
863        assert_ne!(
864            n.canonical(Some(&json!(["a", "b"]))),
865            n.canonical(Some(&json!(["a,sb"])))
866        );
867        assert_ne!(
868            n.canonical(Some(&json!({"a": "x", "b": "y"}))),
869            n.canonical(Some(&json!({"a": "x,b:sy"})))
870        );
871    }
872
873    #[test]
874    fn numeric_and_text_keys_match_and_repair_never_deletes_them() {
875        let n = norm();
876        let key = vec!["id".to_string()];
877        let src = vec![json!({"id": 7, "v": "a"}), json!({"id": 8, "v": "b"})];
878        let dst = vec![json!({"id": "7", "v": "a"}), json!({"id": "8", "v": "x"})];
879        let diffs = diff_rows(&src, &dst, &key, None, &[], &n);
880        assert_eq!(diffs.len(), 1, "{diffs:?}");
881        assert!(matches!(
882            &diffs[0].kind,
883            DifferenceKind::Changed { columns } if columns == &vec!["v".to_string()]
884        ));
885        assert_eq!(
886            row_hash(&src[0], &key, None, &[], &n),
887            row_hash(&dst[0], &key, None, &[], &n)
888        );
889        assert_ne!(
890            n.canonical(Some(&json!("7"))),
891            n.canonical(Some(&json!(7))),
892            "non-key columns keep numeric_strings off"
893        );
894    }
895
896    #[test]
897    fn float_tolerance_and_numeric_strings_are_opt_in() {
898        let exact = norm();
899        assert_ne!(
900            exact.canonical(Some(&json!(1.00000001))),
901            exact.canonical(Some(&json!(1.00000002)))
902        );
903        let loose = Normalizer {
904            float_tolerance: 1e-6,
905            ..norm()
906        };
907        assert_eq!(
908            loose.canonical(Some(&json!(1.00000001))),
909            loose.canonical(Some(&json!(1.00000002)))
910        );
911        let strings = Normalizer {
912            numeric_strings: true,
913            ..norm()
914        };
915        assert_eq!(
916            strings.canonical(Some(&json!("42"))),
917            strings.canonical(Some(&json!(42)))
918        );
919        assert_eq!(
920            strings.canonical(Some(&json!("1.5"))),
921            strings.canonical(Some(&json!(1.5)))
922        );
923        assert_eq!(strings.canonical(Some(&json!("abc"))), "sabc");
924        assert!(loose.validate().is_ok());
925        assert!(
926            Normalizer {
927                float_tolerance: -1.0,
928                ..norm()
929            }
930            .validate()
931            .is_err()
932        );
933        assert!(
934            Normalizer {
935                float_tolerance: f64::NAN,
936                ..norm()
937            }
938            .validate()
939            .is_err()
940        );
941    }
942
943    #[test]
944    fn row_hash_ignores_field_order_and_excluded_columns() {
945        let key = vec!["id".to_string()];
946        let a = json!({"id": 1, "name": "x", "_faucet_run_id": "r1"});
947        let b = json!({"name": "x", "id": 1, "_faucet_run_id": "r2"});
948        let ex = vec!["_faucet_*".to_string()];
949        assert_eq!(
950            row_hash(&a, &key, None, &ex, &norm()),
951            row_hash(&b, &key, None, &ex, &norm())
952        );
953        assert_ne!(
954            row_hash(&a, &key, None, &[], &norm()),
955            row_hash(&b, &key, None, &[], &norm()),
956            "without the exclusion the run id differs"
957        );
958        // An explicit column list hashes only those (plus the key).
959        let c = json!({"id": 1, "name": "y", "other": 5});
960        let cols = vec!["other".to_string()];
961        assert_eq!(
962            row_hash(&a, &key, Some(&cols), &[], &norm()),
963            row_hash(
964                &json!({"id": 1, "name": "zzz"}),
965                &key,
966                Some(&cols),
967                &[],
968                &norm()
969            )
970        );
971        assert_ne!(
972            row_hash(&c, &key, Some(&cols), &[], &norm()),
973            row_hash(&a, &key, Some(&cols), &[], &norm())
974        );
975        assert!(is_excluded("_faucet_loaded_at", &ex));
976        assert!(!is_excluded("faucet", &ex));
977        assert!(is_excluded("name", &["name".to_string()]));
978        assert_eq!(key_int(&a, &key), Some(1));
979        assert_eq!(key_int(&json!({"id": "7"}), &key), Some(7));
980        assert_eq!(key_int(&a, &["id".into(), "name".into()]), None);
981        assert_eq!(key_int(&json!({"id": true}), &key), None);
982        assert_eq!(key_text(&a, &key, &norm()), "id=n1;");
983    }
984
985    #[test]
986    fn digest_is_order_independent_and_tracks_bounds() {
987        let key = vec!["id".to_string()];
988        let rows = [json!({"id": 3, "v": "a"}), json!({"id": 1, "v": "b"})];
989        let mut fwd = DigestAccumulator::new();
990        let mut rev = DigestAccumulator::new();
991        for r in &rows {
992            fwd.add(row_hash(r, &key, None, &[], &norm()), key_int(r, &key));
993        }
994        for r in rows.iter().rev() {
995            rev.add(row_hash(r, &key, None, &[], &norm()), key_int(r, &key));
996        }
997        let (f, r) = (fwd.finish(), rev.finish());
998        assert!(f.same(&r));
999        assert_eq!((f.rows, f.key_min, f.key_max), (2, Some(1), Some(3)));
1000        let mut other = DigestAccumulator::new();
1001        other.add(
1002            row_hash(&json!({"id": 3, "v": "a"}), &key, None, &[], &norm()),
1003            Some(3),
1004        );
1005        let o = other.finish();
1006        assert!(!f.same(&o));
1007        assert_eq!(f.bounds_union(&o), Some((1, 3)));
1008        assert_eq!(
1009            ContentDigest::default().bounds_union(&ContentDigest::default()),
1010            None
1011        );
1012        assert_eq!(ContentDigest::default().bounds_union(&o), Some((3, 3)));
1013    }
1014
1015    #[test]
1016    fn server_digests_compare_only_within_one_algorithm() {
1017        let pg = ServerDigest {
1018            algorithm: "postgres:v1".into(),
1019            rows: 2,
1020            digest: "12".into(),
1021            key_min: Some(1),
1022            key_max: Some(5),
1023        };
1024        let same = ServerDigest {
1025            key_min: Some(2),
1026            key_max: Some(9),
1027            ..pg.clone()
1028        };
1029        let other_algo = ServerDigest {
1030            algorithm: "sqlite:v1".into(),
1031            ..pg.clone()
1032        };
1033        assert!(pg.same(&same));
1034        assert!(!pg.comparable(&other_algo));
1035        assert!(!pg.same(&other_algo));
1036        assert!(!pg.same(&ServerDigest {
1037            digest: "13".into(),
1038            ..pg.clone()
1039        }));
1040        assert_eq!(pg.bounds_union(&same), Some((1, 9)));
1041        assert_eq!(
1042            pg.bounds_union(&ServerDigest {
1043                key_min: None,
1044                key_max: None,
1045                ..pg.clone()
1046            }),
1047            Some((1, 5))
1048        );
1049    }
1050
1051    #[test]
1052    fn diff_rows_reports_every_kind_in_key_order() {
1053        let key = vec!["id".to_string()];
1054        let source = vec![
1055            json!({"id": 1, "v": "a"}),
1056            json!({"id": 2, "v": "b"}),
1057            json!({"id": 3, "v": "c"}),
1058            json!({"id": 5, "v": "e"}),
1059            json!({"id": 5, "v": "e2"}),
1060        ];
1061        let dest = vec![
1062            json!({"id": 1, "v": "a", "_faucet_run_id": "r"}),
1063            json!({"id": 2, "v": "B"}),
1064            json!({"id": 4, "v": "d"}),
1065            json!({"id": 6, "v": "f"}),
1066            json!({"id": 6, "v": "f"}),
1067        ];
1068        let ex = vec!["_faucet_*".to_string()];
1069        let diffs = diff_rows(&source, &dest, &key, None, &ex, &norm());
1070        let kinds: Vec<String> = diffs
1071            .iter()
1072            .map(|d| format!("{}:{:?}", d.key["id"], d.kind))
1073            .collect();
1074        assert_eq!(diffs.len(), 5, "{kinds:?}");
1075        assert_eq!(
1076            diffs[0].kind,
1077            DifferenceKind::Changed {
1078                columns: vec!["v".into()]
1079            }
1080        );
1081        assert_eq!(diffs[0].key, json!({"id": 2}));
1082        assert_eq!(diffs[1].kind, DifferenceKind::MissingInDest);
1083        assert_eq!(diffs[1].key, json!({"id": 3}));
1084        assert_eq!(diffs[2].kind, DifferenceKind::ExtraInDest);
1085        assert_eq!(diffs[2].key, json!({"id": 4}));
1086        assert_eq!(
1087            diffs[3].kind,
1088            DifferenceKind::Duplicate {
1089                side: Side::Source,
1090                count: 2
1091            }
1092        );
1093        assert_eq!(
1094            diffs[4].kind,
1095            DifferenceKind::Duplicate {
1096                side: Side::Destination,
1097                count: 2
1098            }
1099        );
1100        assert!(diffs[0].needs_upsert() && diffs[1].needs_upsert());
1101        assert!(diffs[2].needs_delete() && !diffs[2].needs_upsert());
1102        assert!(!diffs[3].needs_upsert() && !diffs[3].needs_delete());
1103        // Identical sides: nothing.
1104        assert!(diff_rows(&source[..1], &dest[..1], &key, None, &ex, &norm()).is_empty());
1105        // A key repeated on both sides is one report, on the source side, with
1106        // the source's count; it is never also "missing" or "extra".
1107        let three = vec![json!({"id": 9}), json!({"id": 9}), json!({"id": 9})];
1108        let d = diff_rows(
1109            &three,
1110            &[json!({"id": 9}), json!({"id": 9})],
1111            &key,
1112            None,
1113            &[],
1114            &norm(),
1115        );
1116        assert_eq!(d.len(), 1, "a duplicated key is reported once: {d:?}");
1117        assert_eq!(
1118            d[0].kind,
1119            DifferenceKind::Duplicate {
1120                side: Side::Source,
1121                count: 3
1122            }
1123        );
1124    }
1125
1126    #[test]
1127    fn diff_rows_with_explicit_columns_ignores_the_rest() {
1128        let key = vec!["id".to_string()];
1129        let cols = vec!["amount".to_string()];
1130        let s = vec![json!({"id": 1, "amount": 10, "note": "x"})];
1131        let d = vec![json!({"id": 1, "amount": 10, "note": "y"})];
1132        assert!(diff_rows(&s, &d, &key, Some(&cols), &[], &norm()).is_empty());
1133        let d2 = vec![json!({"id": 1, "amount": 11})];
1134        let diffs = diff_rows(&s, &d2, &key, Some(&cols), &[], &norm());
1135        assert_eq!(
1136            diffs[0].kind,
1137            DifferenceKind::Changed {
1138                columns: vec!["amount".into()]
1139            }
1140        );
1141    }
1142
1143    #[test]
1144    fn range_width_display_and_number_canonical_forms() {
1145        let inverted = KeyRange {
1146            lo: Some(5),
1147            hi: Some(5),
1148        };
1149        assert_eq!(inverted.width(), Some(0));
1150        assert_eq!(format!("{inverted}"), "[5, 5)");
1151        assert_eq!(format!("{}", KeyRange::ALL), "all");
1152        let norm = Normalizer::default();
1153        assert_eq!(
1154            norm.canonical(Some(&json!(u64::MAX))),
1155            format!("n{}", u64::MAX)
1156        );
1157        assert_eq!(norm.canonical(Some(&json!(-3))), "n-3");
1158        assert_eq!(norm.canonical(Some(&json!(2.0))), "n2");
1159    }
1160
1161    #[test]
1162    fn timestamps_parse_every_supported_spelling_or_none() {
1163        assert!(parse_timestamp_micros("2026-01-01T00:00:00Z").is_some());
1164        assert!(parse_timestamp_micros("2026-01-01 00:00:00.5").is_some());
1165        assert!(parse_timestamp_micros("2026-01-01 00:00:00+02:00").is_some());
1166        assert!(parse_timestamp_micros("2026-01-01").is_some());
1167        assert_eq!(parse_timestamp_micros("2026-13-99"), None);
1168        assert_eq!(parse_timestamp_micros("not a date"), None);
1169        assert_eq!(parse_timestamp_micros("20260101"), None);
1170    }
1171
1172    #[test]
1173    fn bounds_union_needs_both_ends() {
1174        let a = ContentDigest {
1175            key_min: Some(1),
1176            ..Default::default()
1177        };
1178        let b = ContentDigest::default();
1179        assert_eq!(a.bounds_union(&b), None, "no max anywhere");
1180        let sd = ServerDigest {
1181            algorithm: "x".into(),
1182            rows: 0,
1183            digest: "0".into(),
1184            key_min: None,
1185            key_max: Some(9),
1186        };
1187        assert_eq!(sd.bounds_union(&sd), None, "no min anywhere");
1188        let other = ServerDigest {
1189            key_min: Some(2),
1190            ..sd.clone()
1191        };
1192        assert_eq!(sd.bounds_union(&other), Some((2, 9)));
1193    }
1194
1195    #[test]
1196    fn duplicate_destination_keys_are_reported_once_on_that_side() {
1197        let source = vec![json!({"id": 1, "v": "a"})];
1198        let dest = vec![json!({"id": 1, "v": "a"}), json!({"id": 1, "v": "b"})];
1199        let d = diff_rows(
1200            &source,
1201            &dest,
1202            &["id".into()],
1203            None,
1204            &[],
1205            &Normalizer::default(),
1206        );
1207        assert_eq!(d.len(), 1, "{d:?}");
1208        assert_eq!(
1209            d[0].kind,
1210            DifferenceKind::Duplicate {
1211                side: Side::Destination,
1212                count: 2
1213            }
1214        );
1215        assert!(!d[0].needs_upsert() && !d[0].needs_delete());
1216    }
1217
1218    #[test]
1219    fn report_healed_only_when_every_difference_was_repaired() {
1220        let mut r = VerifyReport::default();
1221        assert!(!r.healed(), "nothing to heal");
1222        r.differences.push(Difference {
1223            key: json!({"id": 1}),
1224            kind: DifferenceKind::MissingInDest,
1225        });
1226        r.differences.push(Difference {
1227            key: json!({"id": 2}),
1228            kind: DifferenceKind::ExtraInDest,
1229        });
1230        assert!(!r.healed(), "no repair ran");
1231        r.repaired_upserts = Some(1);
1232        r.repaired_deletes = Some(0);
1233        assert!(!r.healed(), "the extra row was not deleted");
1234        r.repaired_deletes = Some(1);
1235        assert!(r.healed());
1236        r.truncated = true;
1237        assert!(!r.healed(), "a truncated report cannot claim to be healed");
1238    }
1239
1240    #[test]
1241    fn report_tallies_and_equality() {
1242        let mut r = VerifyReport::default();
1243        assert!(r.equal());
1244        r.differences.push(Difference {
1245            key: json!({"id": 1}),
1246            kind: DifferenceKind::MissingInDest,
1247        });
1248        r.differences.push(Difference {
1249            key: json!({"id": 2}),
1250            kind: DifferenceKind::ExtraInDest,
1251        });
1252        r.differences.push(Difference {
1253            key: json!({"id": 3}),
1254            kind: DifferenceKind::Changed { columns: vec![] },
1255        });
1256        r.differences.push(Difference {
1257            key: json!({"id": 4}),
1258            kind: DifferenceKind::Duplicate {
1259                side: Side::Source,
1260                count: 2,
1261            },
1262        });
1263        assert_eq!(r.tally(), (1, 1, 1, 1));
1264        assert!(!r.equal());
1265        let truncated = VerifyReport {
1266            truncated: true,
1267            ..Default::default()
1268        };
1269        assert!(!truncated.equal(), "a truncated scan proves nothing");
1270        let json = serde_json::to_value(&r).unwrap();
1271        assert_eq!(json["differences"][0]["kind"], "missing_in_dest");
1272        assert_eq!(json["differences"][2]["kind"], "changed");
1273    }
1274}