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