zenkey_fleet/report/snapshot.rs
1//! The snapshot plane (RFC 13 §4.4, v1.34; #219): the `.zsnap` header, the
2//! row a snapshot is made of, what taking one reports, and what comparing
3//! two says.
4//!
5//! A snapshot is a fan-in GET kept on disk — the sibling of a `.zrec`
6//! capture — and the one obligation a capture does not carry is stated in
7//! the header and repeated by every renderer: **it was collected *over* a
8//! span, never *at* an instant** ([`ZsnapHeader::collection_span_s`]).
9//!
10//! Like [`ZrecHeader`](super::ZrecHeader), every shape here is read as well
11//! as written — a `.zsnap` outlives the build that wrote it, and a diff reads
12//! two of them back — so the whole module derives `Deserialize`. The row
13//! spellings mirror [`SampleRow`](super::SampleRow)'s where the two carry the
14//! same fact (`key`, `delete`, `bytes`, `encoding`, `timestamp`, `source`),
15//! so a reader of one dialect reads the other.
16
17use serde::{Deserialize, Serialize};
18
19use super::asked::{Asked, u64_is_zero};
20use super::diff::{ByteDiff, ValueDiff};
21use super::judgement::Judgement;
22
23/// The first line of a `.zsnap` file (RFC 13 §4.4): what was asked, under
24/// which base, when, and — the fact a capture does not need — over what
25/// span. The `base` is the operator's *stated* deployment base at collection
26/// time; recorded keys are full wire keys and are never re-derived from it
27/// (O3).
28#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
29pub struct ZsnapHeader {
30 /// Format version ([`ZSNAP_VERSION`](crate::tape::snapshot::ZSNAP_VERSION)).
31 pub zsnap: u32,
32 /// The full wire selectors the snapshot GET, one fan-in each. A `**`
33 /// never crosses an `@`-chunk, so a `**` snapshot excludes the verbatim
34 /// planes by construction (O5).
35 pub selectors: Vec<String>,
36 /// The deployment base the operator resolved at collection time (may be
37 /// empty: the base-less bus-root deployment).
38 pub base: String,
39 /// Collection start, RFC 3339 wall clock.
40 pub collected_at: String,
41 /// How long the collection took, first GET issued to last reply drained,
42 /// the roster ask included. **The number every rendering states**: a
43 /// fan-in GET is collected over this, not at [`collected_at`](Self::collected_at).
44 pub collection_span_s: f64,
45 /// GETs issued — one per selector, whether or not it answered.
46 pub asked: u64,
47 /// Value replies received across every GET, *before* last-writer-wins
48 /// folded them per key — so `answered - superseded` is the row count.
49 pub answered: u64,
50 /// Replies that arrived and were not kept past the observer's bound
51 /// (O6). Absent at zero.
52 #[serde(default, skip_serializing_if = "u64_is_zero")]
53 pub elided: u64,
54 /// Error replies (RFC 05 §3 envelopes) — a refusal is not a value and
55 /// not silence. Absent at zero.
56 #[serde(default, skip_serializing_if = "u64_is_zero")]
57 pub errors: u64,
58 /// Answers that lost last-writer-wins to a newer reply on the same key
59 /// (RFC 04 §1.2). Absent at zero.
60 #[serde(default, skip_serializing_if = "u64_is_zero")]
61 pub superseded: u64,
62 /// How many origins the liveliness roster reported, when it was asked.
63 /// Absent when it was not (O4) — and then every row's holder is
64 /// [`Holder::Unattributed`].
65 #[serde(default, skip_serializing_if = "Asked::is_not_asked")]
66 pub roster: Asked<usize>,
67}
68
69/// One key, as the snapshot could establish it (RFC 13 §4.4).
70#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
71pub struct SnapshotRow {
72 /// Full wire key, verbatim — conformant or not (O1).
73 pub key: String,
74 /// A tombstone answered: authoritative retirement, never an empty put
75 /// (RFC 04 §1.2). Always written.
76 pub delete: bool,
77 /// Base64 of the exact wire payload — lossless. Absent on a delete row:
78 /// the tombstone is the whole fact.
79 #[serde(default, skip_serializing_if = "Option::is_none")]
80 pub bytes: Option<String>,
81 /// The reply's declared encoding, verbatim, when it carried one.
82 #[serde(default, skip_serializing_if = "Option::is_none")]
83 pub encoding: Option<String>,
84 /// The value's HLC, when one rode it — provenance, whose clock
85 /// [`stamper`](Self::stamper) says.
86 #[serde(default, skip_serializing_if = "Option::is_none")]
87 pub timestamp: Option<String>,
88 /// O7's classification of the HLC: whose clock stamped it. `None`
89 /// exactly when [`timestamp`](Self::timestamp) is.
90 #[serde(default, skip_serializing_if = "Option::is_none")]
91 pub stamper: Option<StamperWire>,
92 /// The publishing entity, `zid:eid#sn`, when `SourceInfo` rode the reply
93 /// — the [`SampleRow`](super::SampleRow) spelling. Usually absent.
94 #[serde(default, skip_serializing_if = "Option::is_none")]
95 pub source: Option<String>,
96 /// The zenoh id of the session that *answered*, where the reply named
97 /// its replier (RFC 13 §4.4 `source_zid`). Not the same fact as
98 /// [`source`](Self::source): a storage answers for a publisher it is not.
99 #[serde(default, skip_serializing_if = "Option::is_none")]
100 pub source_zid: Option<String>,
101 /// O2's rung, including "registry not loaded" — which is not
102 /// "unregistered" (O4).
103 pub registration: RegistrationWire,
104 /// The three-valued payload verdict, never a boolean (#159).
105 pub verdict: VerdictWire,
106 /// Who holds this value: evidence, not inference.
107 pub holder: Holder,
108}
109
110impl SnapshotRow {
111 /// The exact wire payload, decoded from `bytes`. `None` on a delete row
112 /// (the tombstone is the whole fact), on a row that carries no `bytes`,
113 /// or on base64 that does not decode — a file a hand edited.
114 pub fn payload(&self) -> Option<Vec<u8>> {
115 use base64::Engine as _;
116 if self.delete {
117 return None;
118 }
119 base64::engine::general_purpose::STANDARD
120 .decode(self.bytes.as_deref()?)
121 .ok()
122 }
123}
124
125/// Who stamped a value's HLC (RFC 09 §5.1 O7), on the wire — the
126/// [`StampProvenance`](crate::StampProvenance) vocabulary, tagged `kind`.
127#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
128#[serde(tag = "kind", rename_all = "snake_case")]
129pub enum StamperWire {
130 /// The publishing session stamped it: the publisher's own clock.
131 SelfStamped,
132 /// Another node stamped it (commonly a router); `id` is that node's.
133 Foreign { id: String },
134 /// Stamped, and nothing to compare the stamper against — unknown, not
135 /// foreign (O4).
136 Unattributable { id: String },
137}
138
139/// Where the registry ladder stopped for a key (RFC 09 §5.1 O2) — the
140/// [`TopicVerdict`](super::TopicVerdict) vocabulary, spelled identically so
141/// a script that reads `topic info` reads a snapshot row.
142#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
143#[serde(rename_all = "snake_case")]
144pub enum RegistrationWire {
145 /// Parses, refines, declared.
146 Registered,
147 /// Parses as a v1 data key; the producer's slice does not declare it.
148 Unregistered,
149 /// Parses; no loaded slice covers this producer (or service origin).
150 NoSliceForProducer,
151 /// Parses, but onto a verbatim plane — no `[[subject]]` surface exists
152 /// (RFC 03 §1.4).
153 NotADataClass,
154 /// A legal Zenoh key that is not this convention's (O1: a fact).
155 NotV1,
156 /// Sits under a different deployment base than the one stated.
157 NotUnderBase,
158 /// Parses as a data key, and no registry was loaded — "not asked" is
159 /// not "answered no" (O4).
160 RegistryNotLoaded,
161}
162
163/// The RFC 08 §7 conformance verdict on the wire, tagged `state` — the
164/// [`Verdict`](zenkey::schema::validate::Verdict) three-state, never
165/// collapsed to a boolean.
166#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
167#[serde(tag = "state", rename_all = "snake_case")]
168pub enum VerdictWire {
169 /// Checked and conformant.
170 Valid,
171 /// Checked and non-conformant, each violation one sentence.
172 Invalid { violations: Vec<String> },
173 /// Not checked, and why — a token (`no_schema`, `no_registry`,
174 /// `feature_off`, `kind_unsupported`, `undecodable`, `bad_schema`, or
175 /// `tombstone` for a delete row, which carries nothing to check).
176 NotValidated { reason: String },
177}
178
179/// Who holds a value (RFC 13 §4.4): **evidence, not inference**. Tagged
180/// `kind`.
181#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
182#[serde(tag = "kind", rename_all = "snake_case")]
183pub enum Holder {
184 /// The key's origin held an `alive` token during the collection —
185 /// alive *at collection*, not fresh (freshness stays RFC 04 §4's).
186 Live {
187 origin: String,
188 answered_by: AnsweredBy,
189 },
190 /// A value answered and no token was held: a storage remembers it,
191 /// nobody is saying it now.
192 StorageOnly { origin: String },
193 /// The roster was not asked, or the key names no origin (O1).
194 Unattributed { reason: String },
195}
196
197/// Whether the replier was the stamping entity (RFC 13 §4.4).
198#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
199#[serde(rename_all = "snake_case")]
200pub enum AnsweredBy {
201 /// The reply came from the session whose clock stamped the value.
202 Stamper,
203 /// Both identities are known and they differ — a storage or a cache
204 /// answered for the publisher.
205 Other,
206 /// One side or the other is unknown: no replier id, or no stamp.
207 Unknown,
208}
209
210/// A whole `.zsnap` in memory: the header and every row, in key order.
211#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
212pub struct Snapshot {
213 pub header: ZsnapHeader,
214 pub rows: Vec<SnapshotRow>,
215}
216
217/// What taking a snapshot did — the report `zenctl snapshot` renders.
218#[derive(Debug, Clone, PartialEq, Serialize)]
219pub struct SnapshotReport {
220 /// The header as written: a snapshot names its question (O4) and its
221 /// span.
222 pub header: ZsnapHeader,
223 /// Where the file went, when it went to one.
224 #[serde(skip_serializing_if = "Option::is_none")]
225 pub out: Option<String>,
226 /// Rows whose holder is [`Holder::Live`].
227 pub live: u64,
228 /// Rows whose holder is [`Holder::StorageOnly`].
229 pub storage_only: u64,
230 /// Rows whose holder is [`Holder::Unattributed`].
231 pub unattributed: u64,
232 /// Selectors whose GET could not be issued at all — asked, and the
233 /// question never reached the bus. Counted in `asked`, absent from
234 /// `answered`, and named here so the file says which of its selectors
235 /// it does not cover (O5).
236 #[serde(skip_serializing_if = "Vec::is_empty")]
237 pub incomplete: Vec<String>,
238}
239
240/// What two snapshots disagree about (RFC 13 §4.4): both spans stated,
241/// the facets kept apart, and an origin that could not be paired listed
242/// rather than dropped.
243#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
244pub struct SnapshotDiff {
245 /// The earlier side's header — its span is half of what a diff MUST
246 /// state.
247 pub a: ZsnapHeader,
248 /// The later side's header.
249 pub b: ZsnapHeader,
250 /// Keys in `b` and not in `a`.
251 pub added: Vec<String>,
252 /// Keys in `a` and not in `b`.
253 pub removed: Vec<String>,
254 /// Keys in both whose value, verdict, registration, holder or stamp
255 /// moved — each facet reported on its own.
256 pub changed: Vec<KeyChange>,
257 /// Keys in both that are identical on every facet.
258 pub unchanged: u64,
259 /// Differing keys past the listing bound — counted, never dropped (O6).
260 /// Absent at zero.
261 #[serde(default, skip_serializing_if = "u64_is_zero")]
262 pub truncated: u64,
263 /// The origin alignment across deployments, when one was asked for
264 /// (#220). Absent when not.
265 #[serde(default, skip_serializing_if = "Asked::is_not_asked")]
266 pub origin_map: Asked<Vec<OriginPair>>,
267 /// Origins the alignment could not pair — listed, never dropped
268 /// (RFC 13 §4.4). Absent when empty.
269 #[serde(default, skip_serializing_if = "Vec::is_empty")]
270 pub unmapped: Vec<Unmapped>,
271 /// The per-subject roll-up, when asked for. Absent when not — and
272 /// absent on a [`refused`](Self::refused) diff, where the alignment was
273 /// asked and the comparison declined.
274 #[serde(default, skip_serializing_if = "Asked::is_not_asked")]
275 pub by_subject: Asked<Vec<SubjectDelta>>,
276}
277
278impl SnapshotDiff {
279 /// The alignment was asked and left origins unpaired, so the
280 /// comparison was **not made** (#220): the pairs and the unpaired ride
281 /// the report, nothing is listed as added, removed or changed, and the
282 /// judgement is the reserved non-verdict — a diff that compared around
283 /// an origin it could not place would be confident nonsense.
284 pub fn refused(&self) -> bool {
285 self.origin_map.is_asked() && !self.unmapped.is_empty()
286 }
287
288 /// Whether anything at all differs — added, removed, changed, or
289 /// differences past the bound.
290 pub fn differs(&self) -> bool {
291 !self.added.is_empty()
292 || !self.removed.is_empty()
293 || !self.changed.is_empty()
294 || self.truncated > 0
295 }
296
297 /// The RFC 13 §1.2 projection: a difference is the finding
298 /// (`Established`, exit 1), identity the clean answer. A diff over two
299 /// parsed files is always asked; the one unestablished pole is a
300 /// [`refused`](Self::refused) alignment, where the observation cannot
301 /// carry the claim (`Unobservable`, exit 2).
302 pub fn to_judgement(&self) -> Judgement {
303 if self.refused() {
304 Judgement::Unobservable {
305 reason: format!(
306 "{} origin(s) could not be paired; the comparison was not made",
307 self.unmapped.len()
308 ),
309 }
310 } else if self.differs() {
311 Judgement::Established
312 } else {
313 Judgement::NotEstablished {
314 reason: "the two snapshots are identical on every facet".into(),
315 }
316 }
317 }
318}
319
320/// One key present in both snapshots that moved on at least one facet.
321/// A facet pair is present only when that facet differs; `timestamp` is
322/// always carried because the two stamps are what date the change.
323#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
324pub struct KeyChange {
325 pub key: String,
326 /// The structural diff, when both sides carried a structural value.
327 #[serde(default, skip_serializing_if = "Option::is_none")]
328 pub value: Option<ValueDiff>,
329 /// The byte comparison, when at least one side did not.
330 #[serde(default, skip_serializing_if = "Option::is_none")]
331 pub bytes: Option<ByteDiff>,
332 /// `(a, b)` when the verdict moved.
333 #[serde(default, skip_serializing_if = "Option::is_none")]
334 pub verdict: Option<(VerdictWire, VerdictWire)>,
335 /// `(a, b)` when the registration rung moved.
336 #[serde(default, skip_serializing_if = "Option::is_none")]
337 pub registration: Option<(RegistrationWire, RegistrationWire)>,
338 /// `(a, b)` when the holder moved.
339 #[serde(default, skip_serializing_if = "Option::is_none")]
340 pub holder: Option<(Holder, Holder)>,
341 /// `(a, b)` stamps, each absent where the value carried none.
342 pub timestamp: (Option<String>, Option<String>),
343}
344
345/// One origin in `a` aligned with one in `b`, and on what evidence.
346#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
347pub struct OriginPair {
348 pub a: String,
349 pub b: String,
350 pub evidence: MapEvidence,
351}
352
353/// Why two origins were paired. Tagged `kind`.
354#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
355#[serde(tag = "kind", rename_all = "snake_case")]
356pub enum MapEvidence {
357 /// The operator said so (`--map a=b`).
358 Explicit,
359 /// The `source` label both sides' identity-bridge documents carry
360 /// (RFC 06 §6.2: `state/<producer>/health` or `…/sensor`, `host_id`
361 /// beside `source`) agreed, verified on both sides and claimed by no
362 /// other origin on either. `source` is the label itself.
363 Label { source: String },
364 /// The two origins serve the same producer set and nothing else does.
365 ProducerSet,
366}
367
368/// An origin the alignment could not pair — listed, never dropped.
369#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
370pub struct Unmapped {
371 pub origin: String,
372 pub side: Side,
373 pub reason: String,
374}
375
376/// Which snapshot an unpaired origin belongs to.
377#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
378#[serde(rename_all = "snake_case")]
379pub enum Side {
380 A,
381 B,
382}
383
384/// The per-subject roll-up of a diff: one row per subject path across every
385/// origin, so a fleet-wide drift reads as one line rather than N.
386#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
387pub struct SubjectDelta {
388 pub subject: String,
389 /// Keys compared under this subject (present on both sides).
390 pub compared: u64,
391 /// Of those, how many differ.
392 pub differing: u64,
393 pub only_in_a: u64,
394 pub only_in_b: u64,
395 /// One differing key, when there is one, for the human render.
396 #[serde(default, skip_serializing_if = "Option::is_none")]
397 pub example: Option<KeyChange>,
398}
399
400#[cfg(test)]
401mod tests {
402 use super::*;
403 use serde_json::json;
404
405 fn header() -> ZsnapHeader {
406 ZsnapHeader {
407 zsnap: 1,
408 selectors: vec!["acme/v1/**".into()],
409 base: "acme".into(),
410 collected_at: "2026-09-06T00:00:00Z".into(),
411 collection_span_s: 1.25,
412 asked: 1,
413 answered: 2,
414 elided: 0,
415 errors: 0,
416 superseded: 0,
417 roster: Asked::NotAsked,
418 }
419 }
420
421 /// The header's O4/O6 spellings: a counter at zero is absent, a roster
422 /// not asked is absent — and both read back to the value that wrote
423 /// them.
424 #[test]
425 fn a_header_omits_zero_counters_and_an_unasked_roster() {
426 let h = header();
427 let v = serde_json::to_value(&h).unwrap();
428 assert_eq!(
429 v,
430 json!({
431 "zsnap": 1,
432 "selectors": ["acme/v1/**"],
433 "base": "acme",
434 "collected_at": "2026-09-06T00:00:00Z",
435 "collection_span_s": 1.25,
436 "asked": 1,
437 "answered": 2,
438 })
439 );
440 let back: ZsnapHeader = serde_json::from_value(v).unwrap();
441 assert_eq!(back, h);
442
443 let asked = ZsnapHeader {
444 elided: 3,
445 superseded: 1,
446 roster: Asked::Asked(2),
447 ..header()
448 };
449 let v = serde_json::to_value(&asked).unwrap();
450 assert_eq!(v["elided"], 3);
451 assert_eq!(v["superseded"], 1);
452 assert_eq!(v["roster"], 2);
453 assert!(v.get("errors").is_none(), "still zero, still absent");
454 let back: ZsnapHeader = serde_json::from_value(v).unwrap();
455 assert_eq!(back, asked);
456 }
457
458 /// The tag spellings of the four tagged vocabularies, pinned once.
459 #[test]
460 fn the_tagged_vocabularies_spell_snake_case_kinds() {
461 assert_eq!(
462 serde_json::to_value(StamperWire::Foreign { id: "ab12".into() }).unwrap(),
463 json!({"kind": "foreign", "id": "ab12"})
464 );
465 assert_eq!(
466 serde_json::to_value(StamperWire::SelfStamped).unwrap(),
467 json!({"kind": "self_stamped"})
468 );
469 assert_eq!(
470 serde_json::to_value(VerdictWire::NotValidated {
471 reason: "no_registry".into()
472 })
473 .unwrap(),
474 json!({"state": "not_validated", "reason": "no_registry"})
475 );
476 assert_eq!(
477 serde_json::to_value(VerdictWire::Invalid {
478 violations: vec!["/status: not one of …".into()]
479 })
480 .unwrap(),
481 json!({"state": "invalid", "violations": ["/status: not one of …"]})
482 );
483 assert_eq!(
484 serde_json::to_value(Holder::Live {
485 origin: "h-3fa9c2d41b7e".into(),
486 answered_by: AnsweredBy::Stamper,
487 })
488 .unwrap(),
489 json!({"kind": "live", "origin": "h-3fa9c2d41b7e", "answered_by": "stamper"})
490 );
491 assert_eq!(
492 serde_json::to_value(Holder::Unattributed {
493 reason: "roster not asked".into()
494 })
495 .unwrap(),
496 json!({"kind": "unattributed", "reason": "roster not asked"})
497 );
498 assert_eq!(
499 serde_json::to_value(RegistrationWire::RegistryNotLoaded).unwrap(),
500 json!("registry_not_loaded")
501 );
502 assert_eq!(
503 serde_json::to_value(MapEvidence::Label {
504 source: "pve".into()
505 })
506 .unwrap(),
507 json!({"kind": "label", "source": "pve"})
508 );
509 assert_eq!(serde_json::to_value(Side::A).unwrap(), json!("a"));
510 }
511
512 /// A row omits what it does not hold and never nulls it (O4); a delete
513 /// row has no `bytes` at all.
514 #[test]
515 fn a_delete_row_carries_no_payload_and_no_null() {
516 let row = SnapshotRow {
517 key: "acme/v1/h-3fa9c2d41b7e/state/sysinfo/health".into(),
518 delete: true,
519 bytes: None,
520 encoding: None,
521 timestamp: None,
522 stamper: None,
523 source: None,
524 source_zid: None,
525 registration: RegistrationWire::Registered,
526 verdict: VerdictWire::NotValidated {
527 reason: "tombstone".into(),
528 },
529 holder: Holder::StorageOnly {
530 origin: "h-3fa9c2d41b7e".into(),
531 },
532 };
533 assert_eq!(
534 serde_json::to_value(&row).unwrap(),
535 json!({
536 "key": "acme/v1/h-3fa9c2d41b7e/state/sysinfo/health",
537 "delete": true,
538 "registration": "registered",
539 "verdict": {"state": "not_validated", "reason": "tombstone"},
540 "holder": {"kind": "storage_only", "origin": "h-3fa9c2d41b7e"},
541 })
542 );
543 }
544
545 /// A diff with nothing asked beyond the two files carries neither the
546 /// origin map nor the subject roll-up, and a zero truncation is absent.
547 #[test]
548 fn an_unasked_alignment_is_absent_from_a_diff() {
549 let d = SnapshotDiff {
550 a: header(),
551 b: header(),
552 added: vec![],
553 removed: vec![],
554 changed: vec![],
555 unchanged: 4,
556 truncated: 0,
557 origin_map: Asked::NotAsked,
558 unmapped: vec![],
559 by_subject: Asked::NotAsked,
560 };
561 let v = serde_json::to_value(&d).unwrap();
562 for absent in ["truncated", "origin_map", "unmapped", "by_subject"] {
563 assert!(v.get(absent).is_none(), "{absent} should be absent: {v}");
564 }
565 assert_eq!(v["unchanged"], 4);
566 assert!(!d.differs());
567 assert_eq!(crate::report::judgement_exit_code(&d.to_judgement()), 0);
568 }
569}