1use std::collections::BTreeMap;
24
25use crate::model::diff::{byte_diff, diff as value_diff};
26use crate::model::facts::{KeyFacts, KeyShape, OriginKind};
27use crate::model::origin_map::MapPlan;
28use crate::report::{
29 Asked, Holder, KeyChange, Snapshot, SnapshotDiff, SnapshotRow, SubjectDelta, ZsnapHeader,
30};
31
32#[derive(Debug, Clone, Copy)]
35pub struct DiffOpts {
36 pub max_changes: usize,
39 pub max_keys: usize,
42 pub stamps_alone: bool,
48}
49
50impl Default for DiffOpts {
51 fn default() -> Self {
52 DiffOpts {
53 max_changes: 20,
54 max_keys: crate::model::bounded::DEFAULT_MAX_KEYS,
55 stamps_alone: true,
56 }
57 }
58}
59
60enum Outcome {
62 Added,
63 Removed,
64 Changed(Box<KeyChange>),
65 Unchanged,
66}
67
68pub fn diff_snapshots(a: &Snapshot, b: &Snapshot, opts: DiffOpts) -> SnapshotDiff {
70 bound(&a.header, &b.header, compare(a, b, opts), opts.max_keys)
71}
72
73pub fn diff_normalized(a: &Snapshot, b: &Snapshot, plan: &MapPlan, opts: DiffOpts) -> SnapshotDiff {
88 let mut out = SnapshotDiff {
89 a: a.header.clone(),
90 b: b.header.clone(),
91 added: Vec::new(),
92 removed: Vec::new(),
93 changed: Vec::new(),
94 unchanged: 0,
95 truncated: 0,
96 origin_map: Asked::Asked(plan.pairs.clone()),
97 unmapped: plan.unmapped.clone(),
98 by_subject: Asked::NotAsked,
99 };
100 if !plan.is_complete() {
101 return out;
102 }
103 let opts = DiffOpts {
104 stamps_alone: false,
105 ..opts
106 };
107 let b_to_a = plan.b_to_a();
108 let a_norm = Snapshot {
109 header: a.header.clone(),
110 rows: a
111 .rows
112 .iter()
113 .map(|r| canonical_bridge(r, &a.header.base, None))
114 .collect(),
115 };
116 let b_norm = Snapshot {
117 header: b.header.clone(),
118 rows: b
119 .rows
120 .iter()
121 .map(|r| rewrite_row(r, &b.header.base, &a.header.base, &b_to_a))
122 .collect(),
123 };
124 let outcomes = compare(&a_norm, &b_norm, opts);
125
126 let mut subjects: BTreeMap<String, SubjectDelta> = BTreeMap::new();
129 for (key, outcome) in &outcomes {
130 let subject = subject_of(&a.header.base, key);
131 let s = subjects
132 .entry(subject.clone())
133 .or_insert_with(|| SubjectDelta {
134 subject,
135 compared: 0,
136 differing: 0,
137 only_in_a: 0,
138 only_in_b: 0,
139 example: None,
140 });
141 match outcome {
142 Outcome::Added => s.only_in_b += 1,
143 Outcome::Removed => s.only_in_a += 1,
144 Outcome::Unchanged => s.compared += 1,
145 Outcome::Changed(c) => {
146 s.compared += 1;
147 s.differing += 1;
148 if s.example.is_none() {
149 s.example = Some((**c).clone());
150 }
151 }
152 }
153 }
154 let bounded = bound(&a.header, &b.header, outcomes, opts.max_keys);
155 out.added = bounded.added;
156 out.removed = bounded.removed;
157 out.changed = bounded.changed;
158 out.unchanged = bounded.unchanged;
159 out.truncated = bounded.truncated;
160 out.by_subject = Asked::Asked(subjects.into_values().collect());
161 out
162}
163
164fn subject_of(base: &str, key: &str) -> String {
169 let facts = KeyFacts::project(base, key);
170 let (KeyShape::V1(f), Some(relative)) = (&facts.shape, zenkey::grammar::strip_base(base, key))
171 else {
172 return key.to_string();
173 };
174 let mut chunks = relative.split('/');
175 chunks.next(); if f.origin_kind == OriginKind::Host {
177 chunks.next(); }
179 chunks.collect::<Vec<_>>().join("/")
180}
181
182fn rewrite_row(
186 row: &SnapshotRow,
187 from_base: &str,
188 to_base: &str,
189 b_to_a: &BTreeMap<&str, &str>,
190) -> SnapshotRow {
191 let facts = KeyFacts::project(from_base, &row.key);
192 let (KeyShape::V1(f), Some(relative)) = (
193 &facts.shape,
194 zenkey::grammar::strip_base(from_base, &row.key),
195 ) else {
196 return row.clone();
199 };
200 let mapped = if f.origin_kind == OriginKind::Host {
201 b_to_a.get(f.origin.as_str()).copied()
202 } else {
203 None
204 };
205 let mut chunks: Vec<&str> = relative.split('/').collect();
206 if let Some(to) = mapped
207 && chunks.len() > 1
208 {
209 chunks[1] = to;
210 }
211 let mut out = row.clone();
212 out.key = zenkey::grammar::with_base(to_base, chunks.join("/"));
213 if let Some(to) = mapped {
214 out.holder = match &row.holder {
215 Holder::Live {
216 origin,
217 answered_by,
218 } if origin == &f.origin => Holder::Live {
219 origin: to.to_string(),
220 answered_by: *answered_by,
221 },
222 Holder::StorageOnly { origin } if origin == &f.origin => Holder::StorageOnly {
223 origin: to.to_string(),
224 },
225 other => other.clone(),
226 };
227 }
228 canonical_bridge(&out, to_base, mapped.map(|to| (f.origin.as_str(), to)))
229}
230
231fn canonical_bridge(row: &SnapshotRow, base: &str, rename: Option<(&str, &str)>) -> SnapshotRow {
238 use base64::Engine as _;
239 let facts = KeyFacts::project(base, &row.key);
240 let KeyShape::V1(f) = &facts.shape else {
241 return row.clone();
242 };
243 let is_bridge = f.origin_kind == OriginKind::Host
244 && f.class == "state"
245 && matches!(f.subject.as_slice(), [s] if s == "health" || s == "sensor");
246 if !is_bridge || row.delete {
247 return row.clone();
248 }
249 let Some(serde_json::Value::Object(mut doc)) = structural_of(row) else {
250 return row.clone();
251 };
252 let own = rename.map(|(from, _)| from).unwrap_or(f.origin.as_str());
253 if doc.get("host_id").and_then(|v| v.as_str()) != Some(own) {
254 return row.clone();
255 }
256 if let Some((_, to)) = rename {
257 doc.insert("host_id".into(), serde_json::Value::String(to.to_string()));
258 }
259 let mut out = row.clone();
260 let bytes = serde_json::to_vec(&serde_json::Value::Object(doc)).unwrap_or_default();
261 out.bytes = Some(base64::engine::general_purpose::STANDARD.encode(bytes));
262 out
263}
264
265fn compare(a: &Snapshot, b: &Snapshot, opts: DiffOpts) -> Vec<(String, Outcome)> {
268 fn by_key(s: &Snapshot) -> BTreeMap<&str, &SnapshotRow> {
269 s.rows.iter().map(|r| (r.key.as_str(), r)).collect()
270 }
271 let (ra, rb) = (by_key(a), by_key(b));
272 let mut out = Vec::with_capacity(ra.len() + rb.len());
273 for (key, row_a) in &ra {
274 let outcome = match rb.get(key) {
275 None => Outcome::Removed,
276 Some(row_b) => match key_change(row_a, row_b, opts) {
277 None => Outcome::Unchanged,
278 Some(change) => Outcome::Changed(Box::new(change)),
279 },
280 };
281 out.push(((*key).to_string(), outcome));
282 }
283 for key in rb.keys() {
284 if !ra.contains_key(key) {
285 out.push(((*key).to_string(), Outcome::Added));
286 }
287 }
288 out
289}
290
291fn bound(
295 a: &ZsnapHeader,
296 b: &ZsnapHeader,
297 outcomes: Vec<(String, Outcome)>,
298 max_keys: usize,
299) -> SnapshotDiff {
300 let mut out = SnapshotDiff {
301 a: a.clone(),
302 b: b.clone(),
303 added: Vec::new(),
304 removed: Vec::new(),
305 changed: Vec::new(),
306 unchanged: 0,
307 truncated: 0,
308 origin_map: Asked::NotAsked,
309 unmapped: Vec::new(),
310 by_subject: Asked::NotAsked,
311 };
312 let mut listed = 0usize;
313 for (key, outcome) in outcomes {
314 if matches!(outcome, Outcome::Unchanged) {
315 out.unchanged += 1;
316 continue;
317 }
318 if listed >= max_keys {
319 out.truncated += 1;
320 continue;
321 }
322 listed += 1;
323 match outcome {
324 Outcome::Added => out.added.push(key),
325 Outcome::Removed => out.removed.push(key),
326 Outcome::Changed(c) => out.changed.push(*c),
327 Outcome::Unchanged => unreachable!("counted above"),
328 }
329 }
330 out
331}
332
333pub(crate) fn structural_of(row: &SnapshotRow) -> Option<serde_json::Value> {
340 let bytes = payload(row)?;
341 #[cfg(feature = "decode")]
342 {
343 crate::model::decode::structural_value(&bytes)
344 }
345 #[cfg(not(feature = "decode"))]
346 {
347 serde_json::from_slice(&bytes).ok()
348 }
349}
350
351fn payload(row: &SnapshotRow) -> Option<Vec<u8>> {
352 row.payload()
353}
354
355fn key_change(a: &SnapshotRow, b: &SnapshotRow, opts: DiffOpts) -> Option<KeyChange> {
357 let mut change = KeyChange {
358 key: a.key.clone(),
359 value: None,
360 bytes: None,
361 verdict: None,
362 registration: None,
363 holder: None,
364 timestamp: (a.timestamp.clone(), b.timestamp.clone()),
365 };
366 let mut moved = false;
367
368 if a.delete != b.delete || a.bytes != b.bytes {
372 moved = true;
373 match (structural_of(a), structural_of(b)) {
374 (Some(va), Some(vb)) => {
375 let d = value_diff(&va, &vb, opts.max_changes);
376 if d.is_empty() {
377 change.bytes = Some(byte_diff(
380 &payload(a).unwrap_or_default(),
381 &payload(b).unwrap_or_default(),
382 ));
383 } else {
384 change.value = Some(d);
385 }
386 }
387 _ => {
388 change.bytes = Some(byte_diff(
389 &payload(a).unwrap_or_default(),
390 &payload(b).unwrap_or_default(),
391 ));
392 }
393 }
394 }
395 if a.verdict != b.verdict {
396 moved = true;
397 change.verdict = Some((a.verdict.clone(), b.verdict.clone()));
398 }
399 if a.registration != b.registration {
400 moved = true;
401 change.registration = Some((a.registration, b.registration));
402 }
403 if a.holder != b.holder {
404 moved = true;
405 change.holder = Some((a.holder.clone(), b.holder.clone()));
406 }
407 if opts.stamps_alone && a.timestamp != b.timestamp {
413 moved = true;
414 }
415 moved.then_some(change)
416}
417
418#[cfg(test)]
419mod tests {
420 use super::*;
421 use crate::report::{AnsweredBy, Holder, RegistrationWire, VerdictWire, ZsnapHeader};
422
423 fn header() -> ZsnapHeader {
424 ZsnapHeader {
425 zsnap: 1,
426 selectors: vec!["v1/**".into()],
427 base: String::new(),
428 collected_at: "2026-09-06T00:00:00Z".into(),
429 collection_span_s: 0.5,
430 asked: 1,
431 answered: 0,
432 elided: 0,
433 errors: 0,
434 superseded: 0,
435 roster: Asked::NotAsked,
436 }
437 }
438
439 fn row(key: &str, body: &[u8]) -> SnapshotRow {
440 use base64::Engine as _;
441 SnapshotRow {
442 key: key.into(),
443 delete: false,
444 bytes: Some(base64::engine::general_purpose::STANDARD.encode(body)),
445 encoding: Some("application/json".into()),
446 timestamp: None,
447 stamper: None,
448 source: None,
449 source_zid: None,
450 registration: RegistrationWire::RegistryNotLoaded,
451 verdict: VerdictWire::NotValidated {
452 reason: "no_registry".into(),
453 },
454 holder: Holder::Unattributed {
455 reason: "roster not asked".into(),
456 },
457 }
458 }
459
460 fn snap(rows: Vec<SnapshotRow>) -> Snapshot {
461 Snapshot {
462 header: header(),
463 rows,
464 }
465 }
466
467 #[test]
468 fn a_self_diff_is_empty() {
469 let s = snap(vec![
470 row("v1/h-aaaaaaaaaaaa/state/p/a", b"{\"x\":1}"),
471 row("v1/h-aaaaaaaaaaaa/state/p/b", b"text"),
472 ]);
473 let d = diff_snapshots(&s, &s, DiffOpts::default());
474 assert!(!d.differs());
475 assert_eq!(d.unchanged, 2);
476 assert!(d.changed.is_empty() && d.added.is_empty() && d.removed.is_empty());
477 }
478
479 #[test]
482 fn a_non_json_payload_falls_back_to_bytes() {
483 let a = snap(vec![
484 row("k/json", b"{\"x\":1}"),
485 row("k/text", b"hello world"),
486 ]);
487 let b = snap(vec![
488 row("k/json", b"{\"x\":2}"),
489 row("k/text", b"hello there"),
490 ]);
491 let d = diff_snapshots(&a, &b, DiffOpts::default());
492 assert_eq!(d.changed.len(), 2);
493 let json = d.changed.iter().find(|c| c.key == "k/json").unwrap();
494 assert!(json.value.is_some() && json.bytes.is_none());
495 assert_eq!(json.value.as_ref().unwrap().changes[0].path(), "x");
496 let text = d.changed.iter().find(|c| c.key == "k/text").unwrap();
497 assert!(text.bytes.is_some() && text.value.is_none());
498 assert_eq!(text.bytes.unwrap().common_prefix, 6);
499 }
500
501 #[test]
504 fn a_facet_pair_rides_only_when_that_facet_moved() {
505 let mut live = row("k", b"{}");
506 live.holder = Holder::Live {
507 origin: "h-aaaaaaaaaaaa".into(),
508 answered_by: AnsweredBy::Stamper,
509 };
510 let a = snap(vec![row("k", b"{}")]);
511 let b = snap(vec![live]);
512 let d = diff_snapshots(&a, &b, DiffOpts::default());
513 let c = &d.changed[0];
514 assert!(c.holder.is_some());
515 assert!(c.value.is_none() && c.bytes.is_none());
516 assert!(c.verdict.is_none() && c.registration.is_none());
517 }
518
519 #[test]
520 fn added_and_removed_keys_are_listed_by_side() {
521 let a = snap(vec![row("only/a", b"1"), row("both", b"1")]);
522 let b = snap(vec![row("only/b", b"1"), row("both", b"1")]);
523 let d = diff_snapshots(&a, &b, DiffOpts::default());
524 assert_eq!(d.added, ["only/b"]);
525 assert_eq!(d.removed, ["only/a"]);
526 assert_eq!(d.unchanged, 1);
527 assert!(d.differs());
528 }
529
530 #[test]
533 fn differing_keys_past_the_bound_are_counted() {
534 let a = snap((0..5).map(|i| row(&format!("k/{i}"), b"1")).collect());
535 let b = snap((3..8).map(|i| row(&format!("k/{i}"), b"2")).collect());
536 let d = diff_snapshots(
537 &a,
538 &b,
539 DiffOpts {
540 max_keys: 3,
541 ..DiffOpts::default()
542 },
543 );
544 let listed = d.added.len() + d.removed.len() + d.changed.len();
545 assert_eq!(listed, 3);
546 assert_eq!(
547 d.truncated, 5,
548 "3 removed + 2 changed + 3 added = 8, 3 listed"
549 );
550 assert!(d.differs());
551 }
552
553 #[test]
556 fn a_tombstone_against_a_value_is_a_byte_change() {
557 let mut gone = row("k", b"{}");
558 gone.delete = true;
559 gone.bytes = None;
560 let d = diff_snapshots(
561 &snap(vec![row("k", b"{}")]),
562 &snap(vec![gone]),
563 DiffOpts::default(),
564 );
565 let c = &d.changed[0];
566 assert!(c.bytes.is_some());
567 assert_eq!(c.bytes.unwrap().new_len, 0);
568 }
569
570 use crate::model::origin_map::tests::{host, snap as fleet};
573 use crate::model::origin_map::{origin_profiles, plan_map};
574 use crate::report::MapEvidence;
575
576 const A1: &str = "h-aaaaaaaaaaa1";
577 const A2: &str = "h-aaaaaaaaaaa2";
578 const B1: &str = "h-bbbbbbbbbbb1";
579 const B2: &str = "h-bbbbbbbbbbb2";
580
581 fn renamed(base_b: &str) -> (Snapshot, Snapshot) {
582 let a = fleet(
583 "prod",
584 [
585 host("prod", A1, "web", &[]),
586 host("prod", A2, "db", &["logs"]),
587 ]
588 .concat(),
589 );
590 let mut b = fleet(
591 base_b,
592 [
593 host(base_b, B1, "web", &[]),
594 host(base_b, B2, "db", &["logs"]),
595 ]
596 .concat(),
597 );
598 for r in &mut b.rows {
600 r.timestamp = Some("7f3b2a1c00000009/ef56".into());
601 }
602 (a, b)
603 }
604
605 fn plan(a: &Snapshot, b: &Snapshot) -> MapPlan {
606 plan_map(&origin_profiles(a), &origin_profiles(b), &[]).unwrap()
607 }
608
609 #[test]
614 fn a_renamed_fleet_diffs_to_zero_once_aligned() {
615 let (a, b) = renamed("stg");
616 let plain = diff_snapshots(&a, &b, DiffOpts::default());
617 assert!(plain.differs(), "verbatim, nothing lines up");
618 assert!(plain.origin_map.is_not_asked() && plain.by_subject.is_not_asked());
619
620 let d = diff_normalized(&a, &b, &plan(&a, &b), DiffOpts::default());
621 assert!(!d.differs(), "{d:?}");
622 assert_eq!(d.unchanged, 5);
623 assert!(d.unmapped.is_empty());
624 let pairs = d.origin_map.as_option().unwrap();
625 assert_eq!(pairs.len(), 2);
626 assert!(matches!(pairs[0].evidence, MapEvidence::Label { .. }));
627 let subjects = d.by_subject.as_option().unwrap();
628 assert!(
629 subjects
630 .iter()
631 .all(|s| s.differing == 0 && s.only_in_a == 0 && s.only_in_b == 0)
632 );
633 assert_eq!(
634 subjects
635 .iter()
636 .map(|s| s.subject.as_str())
637 .collect::<Vec<_>>(),
638 [
639 "state/logs/rotated",
640 "state/sysinfo/health",
641 "telemetry/sysinfo/disk/root/used"
642 ]
643 );
644 assert_eq!(subjects[1].compared, 2, "one health document per origin");
645 assert_eq!(crate::report::judgement_exit_code(&d.to_judgement()), 0);
646 }
647
648 #[test]
651 fn a_changed_value_on_one_host_reads_as_one_of_n_on_its_subject() {
652 let (a, mut b) = renamed("prod");
653 let disk = b
654 .rows
655 .iter_mut()
656 .find(|r| r.key == format!("prod/v1/{B2}/telemetry/sysinfo/disk/root/used"))
657 .unwrap();
658 disk.bytes = Some({
659 use base64::Engine as _;
660 base64::engine::general_purpose::STANDARD.encode(r#"{"value":97.0}"#)
661 });
662 let d = diff_normalized(&a, &b, &plan(&a, &b), DiffOpts::default());
663 assert!(d.differs());
664 assert_eq!(d.changed.len(), 1);
665 assert_eq!(
666 d.changed[0].key,
667 format!("prod/v1/{A2}/telemetry/sysinfo/disk/root/used"),
668 "the changed key is spelled in a's origin"
669 );
670 let subjects = d.by_subject.as_option().unwrap();
671 let disk = subjects
672 .iter()
673 .find(|s| s.subject == "telemetry/sysinfo/disk/root/used")
674 .unwrap();
675 assert_eq!((disk.compared, disk.differing), (2, 1));
676 assert!(disk.example.as_ref().unwrap().value.is_some());
677 assert_eq!(crate::report::judgement_exit_code(&d.to_judgement()), 1);
678 }
679
680 #[test]
683 fn a_key_only_one_side_holds_rolls_up_as_only_in() {
684 let (a, mut b) = renamed("prod");
685 b.rows.retain(|r| !r.key.ends_with("/logs/rotated"));
686 let d = diff_normalized(&a, &b, &plan(&a, &b), DiffOpts::default());
687 assert_eq!(d.removed, [format!("prod/v1/{A2}/state/logs/rotated")]);
688 let logs = &d.by_subject.as_option().unwrap()[0];
689 assert_eq!(logs.subject, "state/logs/rotated");
690 assert_eq!((logs.compared, logs.only_in_a, logs.only_in_b), (0, 1, 0));
691 assert!(
692 d.changed.iter().all(|c| c.holder.is_none()),
693 "a renamed holder is not a moved holder: {:?}",
694 d.changed
695 );
696 }
697
698 #[test]
702 fn an_incomplete_plan_is_refused_not_compared_around() {
703 let (a, mut b) = renamed("prod");
704 b.rows
705 .extend(host("prod", "h-bbbbbbbbbbb3", "db", &["logs"]));
706 let plan = plan(&a, &b);
707 assert_eq!(
708 plan.unmapped.len(),
709 3,
710 "db is claimed twice in b, so a's db is unpaired too"
711 );
712 let d = diff_normalized(&a, &b, &plan, DiffOpts::default());
713 assert!(d.refused());
714 assert_eq!(d.unmapped.len(), plan.unmapped.len(), "never dropped");
715 assert_eq!(
716 d.origin_map.as_option().unwrap().len(),
717 1,
718 "web still paired"
719 );
720 assert!(d.by_subject.is_not_asked());
721 assert!(d.added.is_empty() && d.removed.is_empty() && d.changed.is_empty());
722 assert_eq!(d.unchanged, 0);
723 assert!(!d.differs());
724 assert_eq!(crate::report::judgement_exit_code(&d.to_judgement()), 2);
725 }
726
727 #[test]
730 fn a_foreign_host_id_is_not_rewritten() {
731 let (a, mut b) = renamed("prod");
732 let health = b
733 .rows
734 .iter_mut()
735 .find(|r| r.key == format!("prod/v1/{B1}/state/sysinfo/health"))
736 .unwrap();
737 health.bytes = Some({
738 use base64::Engine as _;
739 base64::engine::general_purpose::STANDARD
740 .encode(r#"{"host_id":"h-000000000000","source":"web","status":"ok"}"#)
741 });
742 let plan = plan_map(
744 &origin_profiles(&a),
745 &origin_profiles(&b),
746 &[(
747 zenkey::origin::HostId::parse(A1).unwrap(),
748 zenkey::origin::HostId::parse(B1).unwrap(),
749 )],
750 )
751 .unwrap();
752 let d = diff_normalized(&a, &b, &plan, DiffOpts::default());
753 let c = d
754 .changed
755 .iter()
756 .find(|c| c.key.ends_with("/health"))
757 .unwrap();
758 let v = c.value.as_ref().unwrap();
759 assert_eq!(v.changes[0].path(), "host_id");
760 }
761
762 #[test]
763 fn a_subject_keeps_a_service_origin_and_a_foreign_key_verbatim() {
764 assert_eq!(
765 subject_of("acme", "acme/v1/h-aaaaaaaaaaa1/state/sysinfo/health"),
766 "state/sysinfo/health"
767 );
768 assert_eq!(
769 subject_of("acme", "acme/v1/@catalog/state/entity/x"),
770 "@catalog/state/entity/x"
771 );
772 assert_eq!(subject_of("acme", "acme/plain/leak"), "acme/plain/leak");
773 assert_eq!(
774 subject_of("", "v1/h-aaaaaaaaaaa1/telemetry/sysinfo-2/cpu"),
775 "telemetry/sysinfo-2/cpu"
776 );
777 }
778}