zenkey_fleet/report/tape.rs
1//! The tape plane (RFC 09 §5.2): the `.zrec` header, the row dialect a
2//! capture is made of, and what a capture or a replay reports afterwards.
3//!
4//! [`ZrecHeader`] is the one shape in this module that is read as well as
5//! written — a `.zrec` on disk outlives the process that wrote it, so the
6//! header is a contract in both directions and is the only report shape
7//! deriving `Deserialize`.
8
9use serde::{Deserialize, Serialize};
10
11/// The first line of a `.zrec` file: what was asked, under which base, and
12/// when (RFC 09 §5.1 O4 — a capture names its question). The `base` is the
13/// operator's *stated* deployment base at capture time; recorded keys are
14/// full wire keys and are never re-derived from it (O3).
15#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
16pub struct ZrecHeader {
17 /// Format version ([`ZREC_VERSION`](crate::tape::record::ZREC_VERSION)).
18 pub zrec: u32,
19 /// The full wire selectors the capture watched. A wildcard selector
20 /// never crosses an `@`-chunk, so a `**` capture excludes the verbatim
21 /// planes by construction (O5) — the reader states that rather than
22 /// letting the file claim "everything".
23 pub selectors: Vec<String>,
24 /// The deployment base the operator resolved at capture time
25 /// (may be empty: the base-less bus-root deployment).
26 pub base: String,
27 /// Capture start, RFC 3339 wall clock — provenance, not a pacing clock
28 /// (pacing rides each row's `t`).
29 pub captured_at: String,
30}
31
32/// What a capture did — the shared report shape both frontends render.
33#[derive(Debug, Clone, Serialize)]
34pub struct RecordReport {
35 /// The header as written: a capture names its question (O4).
36 pub header: ZrecHeader,
37 /// Where the capture went, when it went to a file.
38 #[serde(skip_serializing_if = "Option::is_none")]
39 pub out: Option<String>,
40 /// Samples written.
41 pub samples: u64,
42 /// Samples the capture missed while behind — stored in the file as
43 /// interleaved drop records *and* totalled here (O6).
44 pub dropped: u64,
45 /// Wall-clock capture length.
46 pub duration_ms: u64,
47}
48
49/// What a replay did — the shared report shape both frontends render.
50#[derive(Debug, Clone, Serialize)]
51pub struct ReplayReport {
52 /// The capture header, echoed: a replay names what it replayed.
53 pub header: ZrecHeader,
54 pub dry_run: bool,
55 pub speed: f64,
56 /// Rows published (dry run: rows that would have been).
57 pub published: u64,
58 /// Tombstones sent (dry run: would have been).
59 pub tombstones: u64,
60 /// Rows that could not be parsed — counted, never skipped.
61 pub malformed: u64,
62 /// Delete rows the retire gate refused.
63 pub refused: u64,
64 /// Samples the *capture* missed (summed from the file's drop records):
65 /// this replay is a partial view and says so (O6).
66 pub capture_dropped: u64,
67 /// The first few malformed/refused reasons, for the human render.
68 #[serde(skip_serializing_if = "Vec::is_empty")]
69 pub first_errors: Vec<String>,
70}
71
72/// One sample, as the explorers write it (#235).
73///
74/// The write side of this module's dialect: `.zrec` rows (RFC 09 §5.2),
75/// `zenctl echo --format ndjson`, and zengui's echo export are all
76/// this struct, so [`parse_row`](crate::tape::ingest::parse_row) reads back what any of them wrote. Every
77/// optional field is `skip_serializing_if`: a writer that does not hold a
78/// fact omits it rather than nulling it, because a `null` here would claim
79/// a question was asked and answered negatively (RFC 09 §5.1 O4). That is
80/// also what keeps the three writers' rows a *subset* relationship rather
81/// than three shapes — a `.zrec` row carries no `origin`, an echo row
82/// carries no pacing offset, and neither is lying about the other.
83#[derive(Debug, Clone, Default, PartialEq, Serialize)]
84pub struct SampleRow {
85 /// Full wire key, as received — explorers run un-namespaced (RFC 09 §5).
86 pub key: String,
87 /// The convention-parsed origin chunk, when the key parses at all.
88 ///
89 /// Absent means the key did not parse under the observer's base, which
90 /// is a fact about the key and not a claim about the fleet (O1/O3).
91 #[serde(skip_serializing_if = "Option::is_none")]
92 pub origin: Option<String>,
93 /// The parsed subject tail, joined — absent on the same terms as
94 /// [`SampleRow::origin`].
95 #[serde(skip_serializing_if = "Option::is_none")]
96 pub subject: Option<String>,
97 /// Microseconds since the capture epoch: the **observer's arrival
98 /// clock**, and the only thing replay paces by (RFC 09 §5.2). A live
99 /// stream has no epoch to be relative to, so it omits this.
100 #[serde(skip_serializing_if = "Option::is_none")]
101 pub t: Option<u64>,
102 /// The registry-declared type name, when a decode was asked for and
103 /// resolved one. `--no-decode` never asks, so it omits this rather than
104 /// nulling it (O4).
105 #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
106 pub type_name: Option<String>,
107 /// Whether [`SampleRow::value`] is a schema decode rather than a
108 /// structural rendering. Absent when nothing was decoded.
109 #[serde(skip_serializing_if = "Option::is_none")]
110 pub typed: Option<bool>,
111 /// The sample's declared encoding, verbatim (RFC 08 §7: sample beats
112 /// registry beats sniff).
113 #[serde(skip_serializing_if = "Option::is_none")]
114 pub encoding: Option<String>,
115 /// The sample's HLC, when one rode it. Whose clock it is depends on who
116 /// stamped it (RFC 09 §5.1 O7); this field says only that it exists.
117 #[serde(skip_serializing_if = "Option::is_none")]
118 pub timestamp: Option<String>,
119 /// The RFC 04 §3 QoS profile **name**, and only when the wire's actual
120 /// axes match one.
121 ///
122 /// This is the field [`parse_row`](crate::tape::ingest::parse_row) resolves through
123 /// `zenkey::qos::QosProfile::from_name`, so it must never carry
124 /// anything else — axes matching no profile are not approximated, they
125 /// ride [`SampleRow::qos_axes`] instead. Writing the axes here is
126 /// exactly the bug in #235.
127 #[serde(skip_serializing_if = "Option::is_none")]
128 pub qos: Option<String>,
129 /// The wire's actual QoS axes as one token,
130 /// `priority/congestion/reliability[+express]` (#120).
131 ///
132 /// A fact worth carrying and *not* a profile name: a fleet is free to
133 /// publish axes no profile declares, and the declared-vs-observed
134 /// comparison is the point.
135 #[serde(skip_serializing_if = "Option::is_none")]
136 pub qos_axes: Option<String>,
137 /// A tombstone: authoritative retirement, never an empty put
138 /// (RFC 04 §1.2). Always written — every sample is one or the other,
139 /// and that is a fact rather than an unanswered question.
140 pub delete: bool,
141 /// Base64 of the exact wire payload — lossless and round-trippable,
142 /// which is why [`parse_row`](crate::tape::ingest::parse_row) prefers it over [`SampleRow::value`].
143 #[serde(skip_serializing_if = "Option::is_none")]
144 pub bytes: Option<String>,
145 /// The payload as a *rendering*: a schema decode when one was asked
146 /// for and succeeded, else the structural degradation. It does not
147 /// round-trip a binary payload, and RFC 09 §5.2 says so.
148 #[serde(skip_serializing_if = "Option::is_none")]
149 pub value: Option<serde_json::Value>,
150 /// True payload size, whatever the rendering above shows.
151 ///
152 /// Deliberately not spelled `bytes`: that key has meant the base64
153 /// payload since RFC 09 §5.2, and a byte count under it makes the row
154 /// unreadable rather than merely lossy (#235).
155 #[serde(skip_serializing_if = "Option::is_none")]
156 pub payload_bytes: Option<usize>,
157 /// The publishing entity, `zid:eid#sn`, when `SourceInfo` rode the
158 /// sample. Usually absent: zenoh 1.9 and 1.10 deliver none to a
159 /// subscriber — 1.10 even dropped setting it through the advanced API
160 /// (eclipse-zenoh/zenoh#2563) — RFC 09 §5.1 O7's practical note.
161 #[serde(skip_serializing_if = "Option::is_none")]
162 pub source: Option<String>,
163 /// The attachment as a rendering (#117), on the same terms as
164 /// [`SampleRow::value`].
165 #[serde(skip_serializing_if = "Option::is_none")]
166 pub attachment: Option<serde_json::Value>,
167 /// Base64 of the exact attachment bytes; wins over the rendering.
168 #[serde(skip_serializing_if = "Option::is_none")]
169 pub attachment_b64: Option<String>,
170 /// True attachment size, whatever the rendering above shows.
171 #[serde(skip_serializing_if = "Option::is_none")]
172 pub attachment_bytes: Option<usize>,
173 /// The RFC 08 §7 validation verdict, present only when the pipeline
174 /// was asked — and then always, so "valid" and "not checked" cannot be
175 /// confused by a shared absence (#159).
176 #[serde(skip_serializing_if = "Option::is_none")]
177 pub verdict: Option<String>,
178 /// The failed constraints, when the verdict is `invalid`.
179 #[serde(skip_serializing_if = "Option::is_none")]
180 pub violations: Option<Vec<String>>,
181 /// Why a decode that was asked for did not happen.
182 #[serde(skip_serializing_if = "Option::is_none")]
183 pub decode_error: Option<String>,
184}
185
186/// The wire's QoS axes as one stable token: `priority/congestion/reliability`,
187/// `+express` when set — lowercase, cut/awk-friendly, never `Debug`.
188///
189/// **The engine's word, not a frontend's.** This is the spelling
190/// [`SampleRow::qos_axes`] carries, which
191/// [`parse_row`](crate::tape::ingest::parse_row) reads back for
192/// `pub --from ndjson` and `.zrec` replay (RFC 09 §5.2) — so it is a
193/// round-trip contract, not a rendering choice. Both frontends used to spell
194/// it independently, fifteen literals each, agreeing by a comment that said
195/// they agreed (#353). The precedent is `LatencyReport::caveat`, worded here
196/// in #213 for exactly this reason: two frontends must not be able to
197/// describe one fact differently.
198pub fn qos_axes_token(
199 priority: zenoh::qos::Priority,
200 congestion_control: zenoh::qos::CongestionControl,
201 reliability: zenoh::qos::Reliability,
202 express: bool,
203) -> String {
204 use zenoh::qos::{CongestionControl as Cc, Priority as P, Reliability as R};
205 let p = match priority {
206 P::RealTime => "real_time",
207 P::InteractiveHigh => "interactive_high",
208 P::InteractiveLow => "interactive_low",
209 P::DataHigh => "data_high",
210 P::Data => "data",
211 P::DataLow => "data_low",
212 P::Background => "background",
213 };
214 let c = match congestion_control {
215 Cc::Drop => "drop",
216 Cc::Block => "block",
217 // `CongestionControl` is `#[non_exhaustive]` upstream: a variant this
218 // build has never heard of renders as `other` rather than as one of
219 // the two it knows.
220 _ => "other",
221 };
222 let r = match reliability {
223 R::BestEffort => "best_effort",
224 R::Reliable => "reliable",
225 };
226 format!("{p}/{c}/{r}{}", if express { "+express" } else { "" })
227}
228
229#[cfg(test)]
230mod qos_axes_tests {
231 use super::*;
232 use zenoh::qos::{CongestionControl as Cc, Priority as P, Reliability as R};
233
234 /// The token is a round-trip contract, not a rendering: `.zrec` replay
235 /// and `pub --from ndjson` read it back (RFC 09 §5.2). Pinned here rather
236 /// than in either frontend, because it is neither frontend's (#353).
237 #[test]
238 fn the_axes_token_is_stable() {
239 assert_eq!(
240 qos_axes_token(P::Data, Cc::Drop, R::BestEffort, false),
241 "data/drop/best_effort"
242 );
243 assert_eq!(
244 qos_axes_token(P::RealTime, Cc::Block, R::Reliable, true),
245 "real_time/block/reliable+express"
246 );
247 assert_eq!(
248 qos_axes_token(P::InteractiveHigh, Cc::Block, R::Reliable, false),
249 "interactive_high/block/reliable"
250 );
251 assert_eq!(
252 qos_axes_token(P::Background, Cc::Drop, R::BestEffort, true),
253 "background/drop/best_effort+express"
254 );
255 }
256
257 /// Every declared profile renders a token, and the five are distinct —
258 /// which is what makes declared-vs-observed a comparison at all (#120).
259 #[test]
260 fn every_qos_profile_has_a_distinct_axes_token() {
261 let tokens: Vec<String> = zenkey::QosProfile::ALL
262 .iter()
263 .map(|p| {
264 qos_axes_token(
265 p.priority(),
266 p.congestion_control(),
267 p.reliability(),
268 p.express(),
269 )
270 })
271 .collect();
272 let mut unique = tokens.clone();
273 unique.sort();
274 unique.dedup();
275 assert_eq!(unique.len(), tokens.len(), "{tokens:?}");
276 }
277}