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.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
280pub fn parse_timestamp_micros(s: &str) -> Option<i64> {
284 let s = s.trim();
285 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
307pub(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
321pub 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
359pub 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
367pub 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
382pub 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#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
396pub struct ContentDigest {
397 pub rows: u64,
398 pub sum: u128,
400 pub xor: u128,
402 pub key_min: Option<i64>,
404 pub key_max: Option<i64>,
405}
406
407impl ContentDigest {
408 pub fn same(&self, other: &ContentDigest) -> bool {
410 self.rows == other.rows && self.sum == other.sum && self.xor == other.xor
411 }
412
413 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#[derive(Debug, Default)]
429pub struct DigestAccumulator {
430 digest: ContentDigest,
431}
432
433impl DigestAccumulator {
434 pub fn new() -> Self {
435 Self::default()
436 }
437
438 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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
460pub struct ServerDigest {
461 pub algorithm: String,
462 pub rows: u64,
463 pub digest: String,
466 pub key_min: Option<i64>,
467 pub key_max: Option<i64>,
468}
469
470impl ServerDigest {
471 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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
496#[serde(tag = "kind", rename_all = "snake_case")]
497pub enum DifferenceKind {
498 MissingInDest,
500 ExtraInDest,
502 Changed { columns: Vec<String> },
504 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#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
517pub struct Difference {
518 pub key: Value,
520 #[serde(flatten)]
521 pub kind: DifferenceKind,
522}
523
524impl Difference {
525 pub fn needs_upsert(&self) -> bool {
527 matches!(
528 self.kind,
529 DifferenceKind::MissingInDest | DifferenceKind::Changed { .. }
530 )
531 }
532
533 pub fn needs_delete(&self) -> bool {
535 matches!(self.kind, DifferenceKind::ExtraInDest)
536 }
537}
538
539fn 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
584pub 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 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#[derive(Debug, Clone, Default, Serialize, Deserialize)]
682pub struct VerifyReport {
683 pub ranges_compared: u64,
685 pub ranges_differing: u64,
687 pub rows_fetched_source: u64,
689 pub rows_fetched_dest: u64,
691 pub server_digests: bool,
694 pub differences: Vec<Difference>,
695 pub truncated: bool,
697 #[serde(skip_serializing_if = "Option::is_none")]
699 pub repaired_upserts: Option<u64>,
700 #[serde(skip_serializing_if = "Option::is_none")]
702 pub repaired_deletes: Option<u64>,
703}
704
705impl VerifyReport {
706 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 pub fn equal(&self) -> bool {
725 self.differences.is_empty() && !self.truncated
726 }
727
728 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 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 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 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 assert!(diff_rows(&source[..1], &dest[..1], &key, None, &ex, &norm()).is_empty());
1105 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}