1use crate::shard::ShardSpec;
25use serde::{Deserialize, Serialize};
26use serde_json::{Map, Value};
27use std::collections::{BTreeMap, HashMap};
28
29#[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 pub const ALL: KeyRange = KeyRange { lo: None, hi: None };
39
40 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 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 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 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 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
120pub 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#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema, PartialEq)]
137#[serde(deny_unknown_fields)]
138pub struct Normalizer {
139 #[serde(default)]
141 pub float_tolerance: f64,
142 #[serde(default = "default_true")]
145 pub timestamps: bool,
146 #[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 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 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 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 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 fn for_key(&self) -> Normalizer {
218 Normalizer {
219 numeric_strings: true,
220 ..self.clone()
221 }
222 }
223
224 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 if f.fract() == 0.0 && f >= i64::MIN as f64 && f < i64::MAX as f64 {
246 return format!("n{}", f as i64);
247 }
248 if f.fract() == 0.0 && f >= 0.0 && f < u64::MAX as f64 {
249 return format!("n{}", f as u64);
250 }
251 if self.float_tolerance > 0.0 {
252 let rounded = (f / self.float_tolerance).round() * self.float_tolerance;
253 return format!("f{rounded:e}");
254 }
255 format!("f{f:e}")
256 }
257
258 fn within_tolerance(&self, a: Option<&Value>, b: Option<&Value>) -> bool {
262 if self.float_tolerance <= 0.0 {
263 return false;
264 }
265 match (self.number_of(a), self.number_of(b)) {
266 (Some(x), Some(y)) => (x - y).abs() <= self.float_tolerance,
267 _ => false,
268 }
269 }
270
271 fn number_of(&self, v: Option<&Value>) -> Option<f64> {
272 match v? {
273 Value::Number(n) => n.as_f64(),
274 Value::String(s) if self.numeric_strings => {
275 s.trim().parse::<f64>().ok().filter(|f| f.is_finite())
276 }
277 _ => None,
278 }
279 }
280
281 fn canonical_string(&self, s: &str) -> String {
282 if self.timestamps
283 && let Some(micros) = parse_timestamp_micros(s)
284 {
285 return format!("ts{micros}");
286 }
287 if self.numeric_strings {
288 let t = s.trim();
289 if let Ok(i) = t.parse::<i64>() {
290 return format!("n{i}");
291 }
292 if let Ok(f) = t.parse::<f64>()
293 && f.is_finite()
294 {
295 return self.canonical_number(Some(f), None, None);
296 }
297 }
298 format!("s{s}")
299 }
300}
301
302fn length_prefixed(s: &str) -> String {
303 format!("{}:{s}", s.len())
304}
305
306pub fn parse_timestamp_micros(s: &str) -> Option<i64> {
310 let s = s.trim();
311 let b = s.as_bytes();
314 if b.len() < 10 || !b[..4].iter().all(u8::is_ascii_digit) || b[4] != b'-' {
315 return None;
316 }
317 if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(s) {
318 return Some(dt.timestamp_micros());
319 }
320 let with_space = s.replacen(' ', "T", 1);
321 if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(&with_space) {
322 return Some(dt.timestamp_micros());
323 }
324 if let Ok(naive) = chrono::NaiveDateTime::parse_from_str(&with_space, "%Y-%m-%dT%H:%M:%S%.f") {
325 return Some(naive.and_utc().timestamp_micros());
326 }
327 if let Ok(date) = chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d") {
328 return Some(date.and_hms_opt(0, 0, 0)?.and_utc().timestamp_micros());
329 }
330 None
331}
332
333pub(crate) fn fnv1a_64(bytes: &[u8], basis: u64) -> u64 {
336 let mut h = basis;
337 for b in bytes {
338 h ^= u64::from(*b);
339 h = h.wrapping_mul(0x0000_0100_0000_01b3);
340 }
341 h
342}
343
344pub(crate) const FNV_BASIS_A: u64 = 0xcbf2_9ce4_8422_2325;
345pub(crate) const FNV_BASIS_B: u64 = 0x84222325_cbf29ce4;
346
347pub fn row_hash(
354 record: &Value,
355 key: &[String],
356 columns: Option<&[String]>,
357 exclude: &[String],
358 norm: &Normalizer,
359) -> u128 {
360 let obj = record.as_object();
361 let mut text = String::new();
362 let mut names: Vec<&str> = match columns {
363 Some(cols) => cols.iter().map(String::as_str).collect(),
364 None => obj
365 .map(|o| {
366 o.iter()
367 .filter(|(_, v)| !v.is_null())
368 .map(|(k, _)| k.as_str())
369 .collect()
370 })
371 .unwrap_or_default(),
372 };
373 for k in key {
374 if !names.contains(&k.as_str()) {
375 names.push(k.as_str());
376 }
377 }
378 names.retain(|n| !is_excluded(n, exclude));
379 names.sort_unstable();
380 names.dedup();
381 for name in names {
382 text.push_str(name);
383 text.push('=');
384 text.push_str(&norm.canonical_column(name, key, obj.and_then(|o| o.get(name))));
385 text.push('\u{1f}');
386 }
387 let a = fnv1a_64(text.as_bytes(), FNV_BASIS_A);
388 let b = fnv1a_64(text.as_bytes(), FNV_BASIS_B);
389 (u128::from(a) << 64) | u128::from(b)
390}
391
392pub fn is_excluded(name: &str, exclude: &[String]) -> bool {
394 exclude.iter().any(|e| match e.strip_suffix('*') {
395 Some(prefix) => name.starts_with(prefix),
396 None => e == name,
397 })
398}
399
400pub fn key_text(record: &Value, key: &[String], norm: &Normalizer) -> String {
403 let obj = record.as_object();
404 let norm = norm.for_key();
405 let mut out = String::new();
406 for k in key {
407 out.push_str(k);
408 out.push('=');
409 out.push_str(&norm.canonical(obj.and_then(|o| o.get(k))));
410 out.push(';');
411 }
412 out
413}
414
415pub fn key_int(record: &Value, key: &[String]) -> Option<i64> {
417 if key.len() != 1 {
418 return None;
419 }
420 match record.get(&key[0])? {
421 Value::Number(n) => n.as_i64(),
422 Value::String(s) => s.trim().parse().ok(),
423 _ => None,
424 }
425}
426
427#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
429pub struct ContentDigest {
430 pub rows: u64,
431 pub sum: u128,
433 pub xor: u128,
435 pub key_min: Option<i64>,
437 pub key_max: Option<i64>,
438}
439
440impl ContentDigest {
441 pub fn same(&self, other: &ContentDigest) -> bool {
443 self.rows == other.rows && self.sum == other.sum && self.xor == other.xor
444 }
445
446 pub fn bounds_union(&self, other: &ContentDigest) -> Option<(i64, i64)> {
448 let min = match (self.key_min, other.key_min) {
449 (Some(a), Some(b)) => Some(a.min(b)),
450 (a, b) => a.or(b),
451 }?;
452 let max = match (self.key_max, other.key_max) {
453 (Some(a), Some(b)) => Some(a.max(b)),
454 (a, b) => a.or(b),
455 }?;
456 Some((min, max))
457 }
458}
459
460#[derive(Debug, Default)]
462pub struct DigestAccumulator {
463 digest: ContentDigest,
464}
465
466impl DigestAccumulator {
467 pub fn new() -> Self {
468 Self::default()
469 }
470
471 pub fn add(&mut self, hash: u128, key: Option<i64>) {
473 self.digest.rows += 1;
474 self.digest.sum = self.digest.sum.wrapping_add(hash);
475 self.digest.xor ^= hash;
476 if let Some(k) = key {
477 self.digest.key_min = Some(self.digest.key_min.map_or(k, |m| m.min(k)));
478 self.digest.key_max = Some(self.digest.key_max.map_or(k, |m| m.max(k)));
479 }
480 }
481
482 pub fn finish(self) -> ContentDigest {
483 self.digest
484 }
485}
486
487#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
493pub struct ServerDigest {
494 pub algorithm: String,
495 pub rows: u64,
496 pub digest: String,
499 pub key_min: Option<i64>,
500 pub key_max: Option<i64>,
501}
502
503impl ServerDigest {
504 pub fn comparable(&self, other: &ServerDigest) -> bool {
507 self.algorithm == other.algorithm
508 }
509
510 pub fn same(&self, other: &ServerDigest) -> bool {
511 self.comparable(other) && self.rows == other.rows && self.digest == other.digest
512 }
513
514 pub fn bounds_union(&self, other: &ServerDigest) -> Option<(i64, i64)> {
515 let min = match (self.key_min, other.key_min) {
516 (Some(a), Some(b)) => Some(a.min(b)),
517 (a, b) => a.or(b),
518 }?;
519 let max = match (self.key_max, other.key_max) {
520 (Some(a), Some(b)) => Some(a.max(b)),
521 (a, b) => a.or(b),
522 }?;
523 Some((min, max))
524 }
525}
526
527#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
529#[serde(tag = "kind", rename_all = "snake_case")]
530pub enum DifferenceKind {
531 MissingInDest,
533 ExtraInDest,
535 Changed { columns: Vec<String> },
537 Duplicate { side: Side, count: u64 },
539}
540
541#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
542#[serde(rename_all = "snake_case")]
543pub enum Side {
544 Source,
545 Destination,
546}
547
548#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
550pub struct Difference {
551 pub key: Value,
553 #[serde(flatten)]
554 pub kind: DifferenceKind,
555}
556
557impl Difference {
558 pub fn needs_upsert(&self) -> bool {
560 matches!(
561 self.kind,
562 DifferenceKind::MissingInDest | DifferenceKind::Changed { .. }
563 )
564 }
565
566 pub fn needs_delete(&self) -> bool {
568 matches!(self.kind, DifferenceKind::ExtraInDest)
569 }
570}
571
572fn changed_columns(
574 a: &Value,
575 b: &Value,
576 key: &[String],
577 columns: Option<&[String]>,
578 exclude: &[String],
579 norm: &Normalizer,
580) -> Vec<String> {
581 let names: Vec<String> = match columns {
582 Some(cols) => cols.to_vec(),
583 None => {
584 let mut set: Vec<String> = Vec::new();
585 for side in [a, b] {
586 if let Some(o) = side.as_object() {
587 for k in o.keys() {
588 if !set.contains(k) {
589 set.push(k.clone());
590 }
591 }
592 }
593 }
594 set
595 }
596 };
597 let mut out: Vec<String> = names
598 .into_iter()
599 .filter(|n| !is_excluded(n, exclude))
600 .filter(|n| {
601 let (x, y) = (a.get(n.as_str()), b.get(n.as_str()));
602 norm.canonical_column(n, key, x) != norm.canonical_column(n, key, y)
603 && !norm.within_tolerance(x, y)
604 })
605 .collect();
606 out.sort();
607 out
608}
609
610fn key_object(record: &Value, key: &[String]) -> Value {
611 let mut m = Map::new();
612 for k in key {
613 m.insert(k.clone(), record.get(k).cloned().unwrap_or(Value::Null));
614 }
615 Value::Object(m)
616}
617
618pub fn diff_rows(
623 source: &[Value],
624 dest: &[Value],
625 key: &[String],
626 columns: Option<&[String]>,
627 exclude: &[String],
628 norm: &Normalizer,
629) -> Vec<Difference> {
630 let mut by_key: HashMap<String, (u64, &Value)> = HashMap::with_capacity(dest.len());
631 for d in dest {
632 let e = by_key.entry(key_text(d, key, norm)).or_insert((0, d));
633 e.0 += 1;
634 }
635 let mut source_counts: HashMap<String, u64> = HashMap::with_capacity(source.len());
638 for s in source {
639 *source_counts.entry(key_text(s, key, norm)).or_insert(0) += 1;
640 }
641 let mut out: Vec<(String, Difference)> = Vec::new();
642 let mut reported: std::collections::HashSet<String> = std::collections::HashSet::new();
643 for s in source {
644 let kt = key_text(s, key, norm);
645 let count = source_counts[&kt];
646 if count > 1 {
647 if reported.insert(kt.clone()) {
648 by_key.remove(&kt);
649 out.push((
650 kt,
651 Difference {
652 key: key_object(s, key),
653 kind: DifferenceKind::Duplicate {
654 side: Side::Source,
655 count,
656 },
657 },
658 ));
659 }
660 continue;
661 }
662 match by_key.remove(&kt) {
663 None => out.push((
664 kt,
665 Difference {
666 key: key_object(s, key),
667 kind: DifferenceKind::MissingInDest,
668 },
669 )),
670 Some((count, _)) if count > 1 => out.push((
671 kt,
672 Difference {
673 key: key_object(s, key),
674 kind: DifferenceKind::Duplicate {
675 side: Side::Destination,
676 count,
677 },
678 },
679 )),
680 Some((_, d)) => {
681 let cols = changed_columns(s, d, key, columns, exclude, norm);
682 if !cols.is_empty() {
683 out.push((
684 kt,
685 Difference {
686 key: key_object(s, key),
687 kind: DifferenceKind::Changed { columns: cols },
688 },
689 ));
690 }
691 }
692 }
693 }
694 for (kt, (count, d)) in by_key {
695 out.push((
696 kt,
697 Difference {
698 key: key_object(d, key),
699 kind: if count > 1 {
700 DifferenceKind::Duplicate {
701 side: Side::Destination,
702 count,
703 }
704 } else {
705 DifferenceKind::ExtraInDest
706 },
707 },
708 ));
709 }
710 out.sort_by(|a, b| a.0.cmp(&b.0));
711 out.into_iter().map(|(_, d)| d).collect()
712}
713
714#[derive(Debug, Clone, Default, Serialize, Deserialize)]
716pub struct VerifyReport {
717 pub ranges_compared: u64,
719 pub ranges_differing: u64,
721 pub rows_fetched_source: u64,
723 pub rows_fetched_dest: u64,
725 pub server_digests: bool,
728 pub differences: Vec<Difference>,
729 pub truncated: bool,
731 #[serde(default, skip_serializing_if = "Option::is_none")]
735 pub differences_found: Option<u64>,
736 #[serde(skip_serializing_if = "Option::is_none")]
738 pub repaired_upserts: Option<u64>,
739 #[serde(skip_serializing_if = "Option::is_none")]
741 pub repaired_deletes: Option<u64>,
742}
743
744impl VerifyReport {
745 pub fn tally(&self) -> (usize, usize, usize, usize) {
747 let mut missing = 0;
748 let mut extra = 0;
749 let mut changed = 0;
750 let mut dup = 0;
751 for d in &self.differences {
752 match d.kind {
753 DifferenceKind::MissingInDest => missing += 1,
754 DifferenceKind::ExtraInDest => extra += 1,
755 DifferenceKind::Changed { .. } => changed += 1,
756 DifferenceKind::Duplicate { .. } => dup += 1,
757 }
758 }
759 (missing, extra, changed, dup)
760 }
761
762 pub fn equal(&self) -> bool {
764 self.differences.is_empty() && !self.truncated
765 }
766
767 pub fn differing_keys(&self) -> u64 {
770 self.differences_found
771 .unwrap_or(self.differences.len() as u64)
772 .max(self.differences.len() as u64)
773 }
774
775 pub fn healed(&self) -> bool {
779 let found = self.differing_keys();
780 if self.truncated || found == 0 {
781 return false;
782 }
783 let repaired = self.repaired_upserts.unwrap_or(0) + self.repaired_deletes.unwrap_or(0);
784 repaired >= found
785 }
786}
787
788#[cfg(test)]
789mod tests {
790 use super::*;
791 use serde_json::json;
792
793 fn norm() -> Normalizer {
794 Normalizer::default()
795 }
796
797 #[test]
798 fn ranges_tile_the_key_space_and_bisect_to_leaves() {
799 let ranges = plan_ranges(1, 100, 4);
800 assert_eq!(ranges.len(), 4);
801 assert_eq!(ranges[0].lo, None, "first is unbounded below");
802 assert_eq!(ranges[3].hi, None, "last is unbounded above");
803 for k in [-5, 1, 50, 100, 1_000] {
804 assert_eq!(ranges.iter().filter(|r| r.contains(k)).count(), 1, "{k}");
805 }
806 let (a, b) = KeyRange {
807 lo: Some(0),
808 hi: Some(10),
809 }
810 .bisect(None)
811 .unwrap();
812 assert_eq!(
813 (a.lo, a.hi, b.lo, b.hi),
814 (Some(0), Some(5), Some(5), Some(10))
815 );
816 let (a, b) = KeyRange::ALL.bisect(Some((10, 19))).unwrap();
818 assert_eq!((a.lo, a.hi), (None, Some(15)));
819 assert_eq!((b.lo, b.hi), (Some(15), None));
820 assert!(KeyRange::ALL.bisect(None).is_none());
821 let (a, b) = KeyRange::ALL.bisect(Some((i64::MIN, i64::MAX))).unwrap();
822 assert_eq!((a.lo, a.hi), (None, Some(-1)));
823 assert_eq!((b.lo, b.hi), (Some(-1), None));
824 assert!(
825 KeyRange {
826 lo: Some(3),
827 hi: Some(4)
828 }
829 .bisect(None)
830 .is_none(),
831 "one key wide is a leaf"
832 );
833 assert_eq!(
834 KeyRange {
835 lo: Some(3),
836 hi: Some(4)
837 }
838 .width(),
839 Some(1)
840 );
841 assert_eq!(KeyRange::ALL.width(), None);
842 assert_eq!(KeyRange::ALL.to_string(), "all");
843 assert_eq!(
844 KeyRange {
845 lo: Some(1),
846 hi: None
847 }
848 .to_string(),
849 "[1, +inf)"
850 );
851 assert_eq!(
852 KeyRange {
853 lo: None,
854 hi: Some(9)
855 }
856 .to_string(),
857 "(-inf, 9)"
858 );
859 }
860
861 #[test]
862 fn range_shard_is_the_pk_range_shape_sql_sources_parse() {
863 let shard = KeyRange {
864 lo: Some(5),
865 hi: None,
866 }
867 .to_shard("id");
868 let bounds = crate::shard::PkShardBounds::from_spec(&shard).unwrap();
869 assert_eq!(
870 (bounds.lo, bounds.hi_unbounded, bounds.include_null),
871 (5, true, true)
872 );
873 assert_eq!(shard.id, "5..+inf");
874 let sql = bounds.wrap("SELECT * FROM t", |s| format!("\"{s}\""));
875 assert!(
876 sql.contains("\"id\" >= 5") && sql.contains("IS NULL"),
877 "{sql}"
878 );
879 }
880
881 #[test]
882 fn canonical_values_agree_across_spellings() {
883 let n = norm();
884 assert_eq!(n.canonical(Some(&json!(1))), n.canonical(Some(&json!(1.0))));
885 assert_eq!(n.canonical(None), n.canonical(Some(&json!(null))));
886 assert_ne!(n.canonical(Some(&json!("1"))), n.canonical(Some(&json!(1))));
887 assert_eq!(
888 n.canonical(Some(&json!("2026-09-25T10:00:00Z"))),
889 n.canonical(Some(&json!("2026-09-25 12:00:00+02:00")))
890 );
891 assert_eq!(
892 n.canonical(Some(&json!("2026-09-25 10:00:00.000"))),
893 n.canonical(Some(&json!("2026-09-25T10:00:00Z")))
894 );
895 assert_eq!(
896 n.canonical(Some(&json!("2026-09-25"))),
897 n.canonical(Some(&json!("2026-09-25T00:00:00Z")))
898 );
899 assert_eq!(
900 n.canonical(Some(&json!({"b": 1, "a": [true, null]}))),
901 n.canonical(Some(&json!({"a": [true, null], "b": 1})))
902 );
903 assert_eq!(n.canonical(Some(&json!("hello"))), "shello");
905 assert_eq!(n.canonical(Some(&json!("2026-x"))), "s2026-x");
906 assert!(parse_timestamp_micros("not a date").is_none());
907 }
908
909 #[test]
910 fn array_and_object_text_cannot_be_forged_by_element_contents() {
911 let n = norm();
912 assert_ne!(
913 n.canonical(Some(&json!(["a", "b"]))),
914 n.canonical(Some(&json!(["a,sb"])))
915 );
916 assert_ne!(
917 n.canonical(Some(&json!({"a": "x", "b": "y"}))),
918 n.canonical(Some(&json!({"a": "x,b:sy"})))
919 );
920 }
921
922 #[test]
923 fn numeric_and_text_keys_match_and_repair_never_deletes_them() {
924 let n = norm();
925 let key = vec!["id".to_string()];
926 let src = vec![json!({"id": 7, "v": "a"}), json!({"id": 8, "v": "b"})];
927 let dst = vec![json!({"id": "7", "v": "a"}), json!({"id": "8", "v": "x"})];
928 let diffs = diff_rows(&src, &dst, &key, None, &[], &n);
929 assert_eq!(diffs.len(), 1, "{diffs:?}");
930 assert!(matches!(
931 &diffs[0].kind,
932 DifferenceKind::Changed { columns } if columns == &vec!["v".to_string()]
933 ));
934 assert_eq!(
935 row_hash(&src[0], &key, None, &[], &n),
936 row_hash(&dst[0], &key, None, &[], &n)
937 );
938 assert_ne!(
939 n.canonical(Some(&json!("7"))),
940 n.canonical(Some(&json!(7))),
941 "non-key columns keep numeric_strings off"
942 );
943 }
944
945 #[test]
946 fn float_tolerance_compares_pairs_across_a_rounding_boundary() {
947 let loose = Normalizer {
948 float_tolerance: 0.01,
949 ..norm()
950 };
951 let key = vec!["id".to_string()];
952 let src = vec![json!({"id": 1, "amount": 1.004999})];
953 let dst = vec![json!({"id": 1, "amount": 1.005001})];
954 assert!(diff_rows(&src, &dst, &key, None, &[], &loose).is_empty());
955 let far = vec![json!({"id": 1, "amount": 1.02})];
956 assert_eq!(diff_rows(&src, &far, &key, None, &[], &loose).len(), 1);
957 assert_eq!(
958 diff_rows(&src, &dst, &key, None, &[], &norm()).len(),
959 1,
960 "exact by default"
961 );
962 }
963
964 #[test]
965 fn large_integral_floats_equal_the_integer() {
966 let n = norm();
967 assert_eq!(
968 n.canonical(Some(&json!(1e16))),
969 n.canonical(Some(&json!(10_000_000_000_000_000_i64)))
970 );
971 assert_eq!(
972 n.canonical(Some(&json!(1e19))),
973 n.canonical(Some(&json!(10_000_000_000_000_000_000_u64)))
974 );
975 }
976
977 #[test]
978 fn float_tolerance_and_numeric_strings_are_opt_in() {
979 let exact = norm();
980 assert_ne!(
981 exact.canonical(Some(&json!(1.00000001))),
982 exact.canonical(Some(&json!(1.00000002)))
983 );
984 let loose = Normalizer {
985 float_tolerance: 1e-6,
986 ..norm()
987 };
988 assert_eq!(
989 loose.canonical(Some(&json!(1.00000001))),
990 loose.canonical(Some(&json!(1.00000002)))
991 );
992 let strings = Normalizer {
993 numeric_strings: true,
994 ..norm()
995 };
996 assert_eq!(
997 strings.canonical(Some(&json!("42"))),
998 strings.canonical(Some(&json!(42)))
999 );
1000 assert_eq!(
1001 strings.canonical(Some(&json!("1.5"))),
1002 strings.canonical(Some(&json!(1.5)))
1003 );
1004 assert_eq!(strings.canonical(Some(&json!("abc"))), "sabc");
1005 assert!(loose.validate().is_ok());
1006 assert!(
1007 Normalizer {
1008 float_tolerance: -1.0,
1009 ..norm()
1010 }
1011 .validate()
1012 .is_err()
1013 );
1014 assert!(
1015 Normalizer {
1016 float_tolerance: f64::NAN,
1017 ..norm()
1018 }
1019 .validate()
1020 .is_err()
1021 );
1022 }
1023
1024 #[test]
1025 fn row_hash_treats_absent_and_null_columns_alike() {
1026 let key = vec!["id".to_string()];
1027 let src = json!({"id": 1, "name": "a"});
1028 let dst = json!({"id": 1, "name": "a", "note": null});
1029 assert_eq!(
1030 row_hash(&src, &key, None, &[], &norm()),
1031 row_hash(&dst, &key, None, &[], &norm())
1032 );
1033 assert!(diff_rows(std::slice::from_ref(&src), &[dst], &key, None, &[], &norm()).is_empty());
1034 assert_ne!(
1035 row_hash(&src, &key, None, &[], &norm()),
1036 row_hash(
1037 &json!({"id": 1, "name": "a", "note": "x"}),
1038 &key,
1039 None,
1040 &[],
1041 &norm()
1042 )
1043 );
1044 }
1045
1046 #[test]
1047 fn row_hash_ignores_field_order_and_excluded_columns() {
1048 let key = vec!["id".to_string()];
1049 let a = json!({"id": 1, "name": "x", "_faucet_run_id": "r1"});
1050 let b = json!({"name": "x", "id": 1, "_faucet_run_id": "r2"});
1051 let ex = vec!["_faucet_*".to_string()];
1052 assert_eq!(
1053 row_hash(&a, &key, None, &ex, &norm()),
1054 row_hash(&b, &key, None, &ex, &norm())
1055 );
1056 assert_ne!(
1057 row_hash(&a, &key, None, &[], &norm()),
1058 row_hash(&b, &key, None, &[], &norm()),
1059 "without the exclusion the run id differs"
1060 );
1061 let c = json!({"id": 1, "name": "y", "other": 5});
1063 let cols = vec!["other".to_string()];
1064 assert_eq!(
1065 row_hash(&a, &key, Some(&cols), &[], &norm()),
1066 row_hash(
1067 &json!({"id": 1, "name": "zzz"}),
1068 &key,
1069 Some(&cols),
1070 &[],
1071 &norm()
1072 )
1073 );
1074 assert_ne!(
1075 row_hash(&c, &key, Some(&cols), &[], &norm()),
1076 row_hash(&a, &key, Some(&cols), &[], &norm())
1077 );
1078 assert!(is_excluded("_faucet_loaded_at", &ex));
1079 assert!(!is_excluded("faucet", &ex));
1080 assert!(is_excluded("name", &["name".to_string()]));
1081 assert_eq!(key_int(&a, &key), Some(1));
1082 assert_eq!(key_int(&json!({"id": "7"}), &key), Some(7));
1083 assert_eq!(key_int(&a, &["id".into(), "name".into()]), None);
1084 assert_eq!(key_int(&json!({"id": true}), &key), None);
1085 assert_eq!(key_text(&a, &key, &norm()), "id=n1;");
1086 }
1087
1088 #[test]
1089 fn digest_is_order_independent_and_tracks_bounds() {
1090 let key = vec!["id".to_string()];
1091 let rows = [json!({"id": 3, "v": "a"}), json!({"id": 1, "v": "b"})];
1092 let mut fwd = DigestAccumulator::new();
1093 let mut rev = DigestAccumulator::new();
1094 for r in &rows {
1095 fwd.add(row_hash(r, &key, None, &[], &norm()), key_int(r, &key));
1096 }
1097 for r in rows.iter().rev() {
1098 rev.add(row_hash(r, &key, None, &[], &norm()), key_int(r, &key));
1099 }
1100 let (f, r) = (fwd.finish(), rev.finish());
1101 assert!(f.same(&r));
1102 assert_eq!((f.rows, f.key_min, f.key_max), (2, Some(1), Some(3)));
1103 let mut other = DigestAccumulator::new();
1104 other.add(
1105 row_hash(&json!({"id": 3, "v": "a"}), &key, None, &[], &norm()),
1106 Some(3),
1107 );
1108 let o = other.finish();
1109 assert!(!f.same(&o));
1110 assert_eq!(f.bounds_union(&o), Some((1, 3)));
1111 assert_eq!(
1112 ContentDigest::default().bounds_union(&ContentDigest::default()),
1113 None
1114 );
1115 assert_eq!(ContentDigest::default().bounds_union(&o), Some((3, 3)));
1116 }
1117
1118 #[test]
1119 fn server_digests_compare_only_within_one_algorithm() {
1120 let pg = ServerDigest {
1121 algorithm: "postgres:v1".into(),
1122 rows: 2,
1123 digest: "12".into(),
1124 key_min: Some(1),
1125 key_max: Some(5),
1126 };
1127 let same = ServerDigest {
1128 key_min: Some(2),
1129 key_max: Some(9),
1130 ..pg.clone()
1131 };
1132 let other_algo = ServerDigest {
1133 algorithm: "sqlite:v1".into(),
1134 ..pg.clone()
1135 };
1136 assert!(pg.same(&same));
1137 assert!(!pg.comparable(&other_algo));
1138 assert!(!pg.same(&other_algo));
1139 assert!(!pg.same(&ServerDigest {
1140 digest: "13".into(),
1141 ..pg.clone()
1142 }));
1143 assert_eq!(pg.bounds_union(&same), Some((1, 9)));
1144 assert_eq!(
1145 pg.bounds_union(&ServerDigest {
1146 key_min: None,
1147 key_max: None,
1148 ..pg.clone()
1149 }),
1150 Some((1, 5))
1151 );
1152 }
1153
1154 #[test]
1155 fn diff_rows_reports_every_kind_in_key_order() {
1156 let key = vec!["id".to_string()];
1157 let source = vec![
1158 json!({"id": 1, "v": "a"}),
1159 json!({"id": 2, "v": "b"}),
1160 json!({"id": 3, "v": "c"}),
1161 json!({"id": 5, "v": "e"}),
1162 json!({"id": 5, "v": "e2"}),
1163 ];
1164 let dest = vec![
1165 json!({"id": 1, "v": "a", "_faucet_run_id": "r"}),
1166 json!({"id": 2, "v": "B"}),
1167 json!({"id": 4, "v": "d"}),
1168 json!({"id": 6, "v": "f"}),
1169 json!({"id": 6, "v": "f"}),
1170 ];
1171 let ex = vec!["_faucet_*".to_string()];
1172 let diffs = diff_rows(&source, &dest, &key, None, &ex, &norm());
1173 let kinds: Vec<String> = diffs
1174 .iter()
1175 .map(|d| format!("{}:{:?}", d.key["id"], d.kind))
1176 .collect();
1177 assert_eq!(diffs.len(), 5, "{kinds:?}");
1178 assert_eq!(
1179 diffs[0].kind,
1180 DifferenceKind::Changed {
1181 columns: vec!["v".into()]
1182 }
1183 );
1184 assert_eq!(diffs[0].key, json!({"id": 2}));
1185 assert_eq!(diffs[1].kind, DifferenceKind::MissingInDest);
1186 assert_eq!(diffs[1].key, json!({"id": 3}));
1187 assert_eq!(diffs[2].kind, DifferenceKind::ExtraInDest);
1188 assert_eq!(diffs[2].key, json!({"id": 4}));
1189 assert_eq!(
1190 diffs[3].kind,
1191 DifferenceKind::Duplicate {
1192 side: Side::Source,
1193 count: 2
1194 }
1195 );
1196 assert_eq!(
1197 diffs[4].kind,
1198 DifferenceKind::Duplicate {
1199 side: Side::Destination,
1200 count: 2
1201 }
1202 );
1203 assert!(diffs[0].needs_upsert() && diffs[1].needs_upsert());
1204 assert!(diffs[2].needs_delete() && !diffs[2].needs_upsert());
1205 assert!(!diffs[3].needs_upsert() && !diffs[3].needs_delete());
1206 assert!(diff_rows(&source[..1], &dest[..1], &key, None, &ex, &norm()).is_empty());
1208 let three = vec![json!({"id": 9}), json!({"id": 9}), json!({"id": 9})];
1211 let d = diff_rows(
1212 &three,
1213 &[json!({"id": 9}), json!({"id": 9})],
1214 &key,
1215 None,
1216 &[],
1217 &norm(),
1218 );
1219 assert_eq!(d.len(), 1, "a duplicated key is reported once: {d:?}");
1220 assert_eq!(
1221 d[0].kind,
1222 DifferenceKind::Duplicate {
1223 side: Side::Source,
1224 count: 3
1225 }
1226 );
1227 }
1228
1229 #[test]
1230 fn diff_rows_with_explicit_columns_ignores_the_rest() {
1231 let key = vec!["id".to_string()];
1232 let cols = vec!["amount".to_string()];
1233 let s = vec![json!({"id": 1, "amount": 10, "note": "x"})];
1234 let d = vec![json!({"id": 1, "amount": 10, "note": "y"})];
1235 assert!(diff_rows(&s, &d, &key, Some(&cols), &[], &norm()).is_empty());
1236 let d2 = vec![json!({"id": 1, "amount": 11})];
1237 let diffs = diff_rows(&s, &d2, &key, Some(&cols), &[], &norm());
1238 assert_eq!(
1239 diffs[0].kind,
1240 DifferenceKind::Changed {
1241 columns: vec!["amount".into()]
1242 }
1243 );
1244 }
1245
1246 #[test]
1247 fn range_width_display_and_number_canonical_forms() {
1248 let inverted = KeyRange {
1249 lo: Some(5),
1250 hi: Some(5),
1251 };
1252 assert_eq!(inverted.width(), Some(0));
1253 assert_eq!(format!("{inverted}"), "[5, 5)");
1254 assert_eq!(format!("{}", KeyRange::ALL), "all");
1255 let norm = Normalizer::default();
1256 assert_eq!(
1257 norm.canonical(Some(&json!(u64::MAX))),
1258 format!("n{}", u64::MAX)
1259 );
1260 assert_eq!(norm.canonical(Some(&json!(-3))), "n-3");
1261 assert_eq!(norm.canonical(Some(&json!(2.0))), "n2");
1262 }
1263
1264 #[test]
1265 fn timestamps_parse_every_supported_spelling_or_none() {
1266 assert!(parse_timestamp_micros("2026-01-01T00:00:00Z").is_some());
1267 assert!(parse_timestamp_micros("2026-01-01 00:00:00.5").is_some());
1268 assert!(parse_timestamp_micros("2026-01-01 00:00:00+02:00").is_some());
1269 assert!(parse_timestamp_micros("2026-01-01").is_some());
1270 assert_eq!(parse_timestamp_micros("2026-13-99"), None);
1271 assert_eq!(parse_timestamp_micros("not a date"), None);
1272 assert_eq!(parse_timestamp_micros("20260101"), None);
1273 }
1274
1275 #[test]
1276 fn bounds_union_needs_both_ends() {
1277 let a = ContentDigest {
1278 key_min: Some(1),
1279 ..Default::default()
1280 };
1281 let b = ContentDigest::default();
1282 assert_eq!(a.bounds_union(&b), None, "no max anywhere");
1283 let sd = ServerDigest {
1284 algorithm: "x".into(),
1285 rows: 0,
1286 digest: "0".into(),
1287 key_min: None,
1288 key_max: Some(9),
1289 };
1290 assert_eq!(sd.bounds_union(&sd), None, "no min anywhere");
1291 let other = ServerDigest {
1292 key_min: Some(2),
1293 ..sd.clone()
1294 };
1295 assert_eq!(sd.bounds_union(&other), Some((2, 9)));
1296 }
1297
1298 #[test]
1299 fn duplicate_destination_keys_are_reported_once_on_that_side() {
1300 let source = vec![json!({"id": 1, "v": "a"})];
1301 let dest = vec![json!({"id": 1, "v": "a"}), json!({"id": 1, "v": "b"})];
1302 let d = diff_rows(
1303 &source,
1304 &dest,
1305 &["id".into()],
1306 None,
1307 &[],
1308 &Normalizer::default(),
1309 );
1310 assert_eq!(d.len(), 1, "{d:?}");
1311 assert_eq!(
1312 d[0].kind,
1313 DifferenceKind::Duplicate {
1314 side: Side::Destination,
1315 count: 2
1316 }
1317 );
1318 assert!(!d[0].needs_upsert() && !d[0].needs_delete());
1319 }
1320
1321 #[test]
1322 fn report_healed_only_when_every_difference_was_repaired() {
1323 let mut r = VerifyReport::default();
1324 assert!(!r.healed(), "nothing to heal");
1325 r.differences.push(Difference {
1326 key: json!({"id": 1}),
1327 kind: DifferenceKind::MissingInDest,
1328 });
1329 r.differences.push(Difference {
1330 key: json!({"id": 2}),
1331 kind: DifferenceKind::ExtraInDest,
1332 });
1333 assert!(!r.healed(), "no repair ran");
1334 r.repaired_upserts = Some(1);
1335 r.repaired_deletes = Some(0);
1336 assert!(!r.healed(), "the extra row was not deleted");
1337 r.repaired_deletes = Some(1);
1338 assert!(r.healed());
1339 r.truncated = true;
1340 assert!(!r.healed(), "a truncated report cannot claim to be healed");
1341 r.truncated = false;
1344 let listed = r.differences.len() as u64;
1345 r.differences_found = Some(listed + 3);
1346 assert_eq!(r.differing_keys(), listed + 3);
1347 r.repaired_upserts = Some(listed + 2);
1348 r.repaired_deletes = Some(0);
1349 assert!(!r.healed(), "two found differences were not repaired");
1350 r.repaired_deletes = Some(1);
1351 assert!(r.healed());
1352 }
1353
1354 #[test]
1355 fn report_tallies_and_equality() {
1356 let mut r = VerifyReport::default();
1357 assert!(r.equal());
1358 r.differences.push(Difference {
1359 key: json!({"id": 1}),
1360 kind: DifferenceKind::MissingInDest,
1361 });
1362 r.differences.push(Difference {
1363 key: json!({"id": 2}),
1364 kind: DifferenceKind::ExtraInDest,
1365 });
1366 r.differences.push(Difference {
1367 key: json!({"id": 3}),
1368 kind: DifferenceKind::Changed { columns: vec![] },
1369 });
1370 r.differences.push(Difference {
1371 key: json!({"id": 4}),
1372 kind: DifferenceKind::Duplicate {
1373 side: Side::Source,
1374 count: 2,
1375 },
1376 });
1377 assert_eq!(r.tally(), (1, 1, 1, 1));
1378 assert!(!r.equal());
1379 let truncated = VerifyReport {
1380 truncated: true,
1381 ..Default::default()
1382 };
1383 assert!(!truncated.equal(), "a truncated scan proves nothing");
1384 let json = serde_json::to_value(&r).unwrap();
1385 assert_eq!(json["differences"][0]["kind"], "missing_in_dest");
1386 assert_eq!(json["differences"][2]["kind"], "changed");
1387 }
1388}