zenkey_fleet/model/stats.rs
1//! Windowed per-key statistics (issues #13/#15): message/byte counters and
2//! an exponentially-weighted rate, keyed by wire key. Backs `zenctl rate`
3//! (`--bytes` and all), `echo --rate`, and zengui's tree badges.
4//!
5//! Perf posture (report §14): lookups borrow (`&str` against the `String`
6//! keys — no per-sample allocation on the hot hit path); one allocation per
7//! *new* key is the floor.
8
9use std::collections::VecDeque;
10use std::sync::Arc;
11use std::time::{Duration, Instant};
12
13use crate::model::bounded::{BoundedLru, DEFAULT_MAX_KEYS};
14use crate::model::tree::{TreeRow, TreeRows};
15use crate::report::{LatencyReport, LatencySummary};
16
17/// How many per-key latency observations the summary window keeps. Bounded
18/// like everything else an hours-long observer accumulates (O6).
19const LAT_WINDOW: usize = 256;
20
21/// Which clock stamped a latency observation — the storage form of
22/// [`crate::bus::monitor::StampProvenance`], without the stamper's identity.
23#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
24pub enum StampClass {
25 SelfStamped,
26 Foreign,
27 Unattributable,
28}
29
30/// How many distinct stampers one key retains. A key sees its publisher and,
31/// at most, the routers on its path; more than this is a fleet-shaped
32/// question for `doctor`, not a per-key one.
33const MAX_STAMPERS: usize = 4;
34
35/// The largest wrapping SN advance still read as forward progress (loss),
36/// half the `u32` space. Beyond it the shorter way round is *backwards* —
37/// a duplicate burst or a restarted publisher — which is counted as a
38/// reset ([`KeyStats::sn_resets`]), never as a few billion lost samples.
39/// The heuristic is the standard serial-number-arithmetic split (the RFC
40/// 1982 shape): no real publisher legitimately skips 2^31 samples between
41/// two arrivals.
42const SN_RESET_WINDOW: u32 = u32::MAX / 2;
43
44/// One key's running statistics.
45#[derive(Debug, Clone)]
46pub struct KeyStats {
47 pub count: u64,
48 pub bytes: u64,
49 /// EWMA of the instantaneous rate (Hz), time-decayed.
50 pub rate_hz: f64,
51 pub last_seen: Instant,
52 /// Consecutive source-sequence-number gap count, when publishers attach
53 /// SourceInfo (unstable API) — loss visibility, `--loss`. Wrap-safe:
54 /// the SN is a `u32`, so `u32::MAX → 0` is the next sample, not a
55 /// 4-billion-sample gap (see the private `SN_RESET_WINDOW`).
56 pub sn_gaps: u64,
57 /// Times the sequence numbering restarted — a backwards or absurd jump
58 /// (beyond the private `SN_RESET_WINDOW`), which is a publisher restart, not
59 /// loss. Its own number: folding a restart into `sn_gaps` would invent
60 /// millions of "lost" samples nobody sent (RFC 09 §5.1 O6 — the kinds
61 /// are never folded).
62 pub sn_resets: u64,
63 /// Samples that carried **no** HLC timestamp — an observation of their
64 /// own, counted separately: an unstamped sample has no latency, which
65 /// is not the same as zero latency (#119).
66 pub unstamped: u64,
67 last_sn: Option<u32>,
68 /// Bounded window of observed skewed latencies, µs, each tagged with who
69 /// stamped it — the tag is what keeps the three populations apart (#213).
70 lat: VecDeque<(i64, StampClass)>,
71 /// Distinct third-party stampers seen, bounded by [`MAX_STAMPERS`].
72 stampers: std::collections::BTreeSet<zenoh::time::TimestampId>,
73 /// Stampers the bound refused (O6).
74 stampers_dropped: u64,
75}
76
77impl KeyStats {
78 /// The window's distributions, split by who stamped them (#213).
79 ///
80 /// `None` before any stamped sample — which is not zero latency, and is
81 /// why [`KeyStats::unstamped`] is counted beside this rather than folded
82 /// into it.
83 pub fn latency(&self) -> Option<LatencyReport> {
84 if self.lat.is_empty() {
85 return None;
86 }
87 let of = |class: StampClass| {
88 summarise(
89 self.lat
90 .iter()
91 .filter(|(_, c)| *c == class)
92 .map(|(us, _)| *us),
93 )
94 };
95 Some(LatencyReport {
96 self_stamped: of(StampClass::SelfStamped),
97 foreign: of(StampClass::Foreign),
98 unattributable: of(StampClass::Unattributable),
99 stampers: self.stampers.iter().map(|id| id.to_string()).collect(),
100 stampers_dropped: self.stampers_dropped,
101 })
102 }
103}
104
105/// Retain a third-party stamper, or count the bound that refused it (O6).
106fn note_stamper(
107 set: &mut std::collections::BTreeSet<zenoh::time::TimestampId>,
108 dropped: &mut u64,
109 stamper: Option<zenoh::time::TimestampId>,
110) {
111 let Some(id) = stamper else { return };
112
113 if set.contains(&id) {
114 return;
115 }
116 if set.len() >= MAX_STAMPERS {
117 *dropped += 1;
118 return;
119 }
120 set.insert(id);
121}
122
123/// The distribution of one population, or `None` when it is empty.
124fn summarise(values: impl Iterator<Item = i64>) -> Option<LatencySummary> {
125 let mut sorted: Vec<i64> = values.collect();
126 if sorted.is_empty() {
127 return None;
128 }
129 sorted.sort_unstable();
130 let at = |q: f64| sorted[((sorted.len() - 1) as f64 * q) as usize];
131 Some(LatencySummary {
132 min_us: sorted[0],
133 median_us: at(0.5),
134 p95_us: at(0.95),
135 max_us: *sorted.last().expect("non-empty"),
136 samples: sorted.len(),
137 })
138}
139
140/// The table. Feed it samples; read it per key or in aggregate.
141///
142/// **Bounded.** A CLI runs for `--for` seconds and exits, so an unbounded
143/// map was fine; a GUI left open overnight on a bus carrying content-addressed
144/// or per-request keys would grow one entry per key forever. The table
145/// therefore keeps at most [`DEFAULT_MAX_KEYS`] entries, evicting the
146/// least-recently-seen first — the keys that stopped publishing are the ones a
147/// live view has least use for — and **counts every eviction**, so a shrinking
148/// key set is never mistaken for a quiet bus (RFC 09 §5.1).
149///
150/// The keys are `Arc<str>` rather than `String` so that
151/// [`rows`](Self::rows) — the copy the ingest lock is held for (#330) — is a
152/// refcount bump per key and not a per-key allocation. Lookups still borrow:
153/// `Arc<str>: Borrow<str>`, so `get(&str)` allocates nothing on the hot hit
154/// path.
155#[derive(Debug)]
156pub struct StatsTable {
157 keys: BoundedLru<Arc<str>, KeyStats>,
158 evicted: u64,
159 unwatched: u64,
160}
161
162impl Default for StatsTable {
163 fn default() -> Self {
164 StatsTable::with_capacity(DEFAULT_MAX_KEYS)
165 }
166}
167
168/// EWMA time constant (~2 s: responsive enough for a UI badge, smooth
169/// enough not to flicker); samples older than ~tau contribute e^-1.
170const TAU: Duration = Duration::from_secs(2);
171
172impl StatsTable {
173 pub fn new() -> Self {
174 Self::default()
175 }
176
177 /// A table bounded at `max_keys` entries.
178 pub fn with_capacity(max_keys: usize) -> Self {
179 StatsTable {
180 keys: BoundedLru::with_capacity(max_keys),
181 evicted: 0,
182 unwatched: 0,
183 }
184 }
185
186 /// Keys dropped to stay within the bound.
187 ///
188 /// Non-zero means the view is partial: some keys that carried traffic are
189 /// no longer represented in [`len`](Self::len), [`totals`](Self::totals) or
190 /// any tree built from this table.
191 pub fn evicted(&self) -> u64 {
192 self.evicted
193 }
194
195 /// The bound in force.
196 pub fn max_keys(&self) -> usize {
197 self.keys.max_keys()
198 }
199
200 /// Keys retired because no active watch covers them any more
201 /// ([`retire_unwatched`](Self::retire_unwatched)).
202 ///
203 /// The third O6 category, deliberately distinct from
204 /// [`evicted`](Self::evicted) ("chose to forget under the bound") and the
205 /// broadcast's dropped ("could not keep up"): this one is "stopped
206 /// looking, by request" — and a key set that shrinks because the user
207 /// unwatched a subtree must say so, or it reads as a quieting bus.
208 pub fn unwatched(&self) -> u64 {
209 self.unwatched
210 }
211
212 /// Retire every key that `gone` covers and no selector in `kept` still
213 /// covers, counting them under [`unwatched`](Self::unwatched). Returns
214 /// how many were retired. Selectors that fail to parse as key
215 /// expressions cover nothing (`gone`) / keep nothing (`kept`).
216 pub fn retire_unwatched(&mut self, gone: &str, kept: &[String]) -> usize {
217 use zenoh::key_expr::keyexpr;
218 // Borrowed throughout: `keyexpr::new(&str)` validates without
219 // allocating, where `KeyExpr::new(String)` builds an `OwnedKeyExpr`.
220 // The old form cloned the key *and* built an owned expr for every key
221 // in the table on every unwatch — 100k allocations at the default
222 // bound (`docs/zero-copy.md`).
223 let Ok(gone) = keyexpr::new(gone) else {
224 return 0;
225 };
226 let kept: Vec<&keyexpr> = kept
227 .iter()
228 .filter_map(|k| keyexpr::new(k.as_str()).ok())
229 .collect();
230 let doomed: Vec<Arc<str>> = self
231 .keys
232 .keys()
233 .filter(|key| match keyexpr::new(&***key) {
234 Ok(ke) => gone.intersects(ke) && !kept.iter().any(|k| k.intersects(ke)),
235 Err(_) => false,
236 })
237 .cloned()
238 .collect();
239 for key in &doomed {
240 self.keys.remove(&**key);
241 }
242 self.unwatched += doomed.len() as u64;
243 doomed.len()
244 }
245
246 /// Record one sample. `now` is injected for deterministic tests;
247 /// `latency` is the pre-computed skewed latency (#119) with the class of
248 /// clock that produced it (#213) — `None` for an unstamped sample, which
249 /// is counted, not defaulted. `stamper` names a third-party stamping node
250 /// when there was one.
251 pub fn record(
252 &mut self,
253 key: &str,
254 payload_len: usize,
255 sn: Option<u32>,
256 now: Instant,
257 latency: Option<(i64, StampClass)>,
258 stamper: Option<zenoh::time::TimestampId>,
259 ) {
260 if let Some(s) = self.keys.get_mut(key) {
261 let dt = now.saturating_duration_since(s.last_seen).as_secs_f64();
262 if dt > 0.0 {
263 let alpha = 1.0 - (-dt / TAU.as_secs_f64()).exp();
264 let instant_rate = 1.0 / dt;
265 s.rate_hz += alpha * (instant_rate - s.rate_hz);
266 }
267 s.count += 1;
268 s.bytes += payload_len as u64;
269 s.last_seen = now;
270 if let (Some(prev), Some(cur)) = (s.last_sn, sn) {
271 // Wrapping arithmetic (deep-review D6): `cur > prev + 1`
272 // overflowed in debug at `prev == u32::MAX` and read the
273 // wrap `u32::MAX → 0` as a ~2^32 gap in release. The
274 // wrapping delta makes the wrap a plain `1` (no gap);
275 // `0` is a duplicate (neither loss nor reset); anything
276 // past [`SN_RESET_WINDOW`] went backwards — a restart,
277 // counted as a reset, not loss.
278 let delta = cur.wrapping_sub(prev);
279 if delta > SN_RESET_WINDOW {
280 s.sn_resets += 1;
281 } else if delta > 1 {
282 s.sn_gaps += u64::from(delta - 1);
283 }
284 }
285 s.last_sn = sn;
286 match latency {
287 Some(observed) => {
288 if s.lat.len() >= LAT_WINDOW {
289 s.lat.pop_front();
290 }
291 s.lat.push_back(observed);
292 }
293 None => s.unstamped += 1,
294 }
295 note_stamper(&mut s.stampers, &mut s.stampers_dropped, stamper);
296 } else {
297 // Recency is the injected `last_seen`, not arrival order: `now` is
298 // the test seam, and eviction must follow the timeline it states.
299 self.evicted += self.keys.admit(|s| s.last_seen) as u64;
300 self.keys.insert(
301 Arc::from(key),
302 KeyStats {
303 count: 1,
304 bytes: payload_len as u64,
305 rate_hz: 0.0,
306 last_seen: now,
307 sn_gaps: 0,
308 sn_resets: 0,
309 unstamped: u64::from(latency.is_none()),
310 last_sn: sn,
311 lat: latency.into_iter().collect(),
312 stampers: {
313 let mut set = std::collections::BTreeSet::new();
314 let mut dropped = 0;
315 note_stamper(&mut set, &mut dropped, stamper);
316 set
317 },
318 stampers_dropped: 0,
319 },
320 );
321 }
322 }
323
324 pub fn get(&self, key: &str) -> Option<&KeyStats> {
325 self.keys.get(key)
326 }
327
328 pub fn iter(&self) -> impl Iterator<Item = (&str, &KeyStats)> {
329 self.keys.iter().map(|(k, v)| (&**k, v))
330 }
331
332 /// The compact rows a [`KeyTreeSnapshot`](crate::KeyTreeSnapshot) is
333 /// folded from, plus the table's own O6 counters (#330).
334 ///
335 /// This is the **whole** of what the tree needs, and it is deliberately a
336 /// copy: [`MonitorCore::tick`](crate::MonitorCore::tick) holds the ingest
337 /// mutex for exactly this call and folds afterwards, so the network
338 /// callback thread waits on an O(keys) walk of `Copy` fields and one
339 /// refcount bump per key — never on the O(keys × chunks) `BTreeMap`
340 /// descent with a `String` allocation per new node that the fold is.
341 /// Before the split, four ticks a second each held the lock for the whole
342 /// rebuild, and `Monitor::watch`'s promise that a slow UI cannot push
343 /// back into the network layer was false for as long as each one took.
344 pub fn rows(&self) -> TreeRows {
345 TreeRows {
346 rows: self
347 .keys
348 .iter()
349 .map(|(key, s)| TreeRow {
350 key: Arc::clone(key),
351 count: s.count,
352 bytes: s.bytes,
353 rate_hz: s.rate_hz,
354 last_seen: s.last_seen,
355 })
356 .collect(),
357 keys: self.keys.len(),
358 evicted: self.evicted,
359 unwatched: self.unwatched,
360 }
361 }
362
363 pub fn len(&self) -> usize {
364 self.keys.len()
365 }
366
367 pub fn is_empty(&self) -> bool {
368 self.keys.is_empty()
369 }
370
371 /// Aggregate totals: (samples, bytes, summed EWMA rate).
372 pub fn totals(&self) -> (u64, u64, f64) {
373 self.keys.values().fold((0, 0, 0.0), |(c, b, r), s| {
374 (c + s.count, b + s.bytes, r + s.rate_hz)
375 })
376 }
377}
378
379#[cfg(test)]
380mod tests {
381 use super::*;
382
383 /// An unbounded table is a leak for any observer that runs for hours on a
384 /// bus with content-addressed or per-request keys.
385 #[test]
386 fn the_table_is_bounded() {
387 let mut t = StatsTable::with_capacity(100);
388 let now = Instant::now();
389 for i in 0..1000 {
390 t.record(&format!("demo/k{i}"), 4, None, now, None, None);
391 }
392 assert!(t.len() <= 100, "len {} exceeds the bound", t.len());
393 assert!(t.evicted() > 0);
394 // Nothing vanishes silently: every key seen is either present or counted.
395 assert_eq!(t.len() as u64 + t.evicted(), 1000);
396 }
397
398 /// Eviction is least-recently-seen: a key still publishing must outlive a
399 /// key that went quiet, or a live view would drop exactly what it is for.
400 #[test]
401 fn eviction_drops_the_least_recently_seen() {
402 let mut t = StatsTable::with_capacity(10);
403 let t0 = Instant::now();
404
405 // Ten keys, oldest first.
406 for i in 0..10 {
407 t.record(
408 &format!("old/k{i}"),
409 4,
410 None,
411 t0 + Duration::from_millis(i),
412 None,
413 None,
414 );
415 }
416 // One of them keeps publishing, much later.
417 let fresh = t0 + Duration::from_secs(60);
418 t.record("old/k0", 4, None, fresh, None, None);
419
420 // Now push new keys in, forcing eviction. Each is strictly newer than
421 // `old/k0`'s refresh, so there is no tie for "oldest" to break.
422 for i in 1..=5 {
423 t.record(
424 &format!("new/k{i}"),
425 4,
426 None,
427 fresh + Duration::from_millis(i),
428 None,
429 None,
430 );
431 }
432
433 assert!(
434 t.get("old/k0").is_some(),
435 "a key that is still publishing must survive"
436 );
437 assert!(
438 t.get("old/k1").is_none(),
439 "a key that went quiet should have been evicted first"
440 );
441 }
442
443 /// #119: the latency window summarises stamped samples and counts
444 /// unstamped ones separately — no latency is not zero latency, and a
445 /// negative value is the skew evidence, kept.
446 #[test]
447 fn latency_is_summarised_and_unstamped_is_counted_not_defaulted() {
448 let mut t = StatsTable::new();
449 let now = Instant::now();
450 for us in [1000, -200, 5000, 3000] {
451 t.record("k", 4, None, now, Some((us, StampClass::SelfStamped)), None);
452 }
453 t.record("k", 4, None, now, None, None);
454 let s = t.get("k").unwrap();
455 assert_eq!(s.unstamped, 1);
456 let lat = s.latency().unwrap();
457 let own = lat
458 .self_stamped
459 .expect("the publisher stamped these itself");
460 assert_eq!(own.min_us, -200, "negative skew is shown, not clamped");
461 assert_eq!(own.max_us, 5000);
462 assert_eq!(own.samples, 4);
463 assert!(own.median_us >= -200 && own.median_us <= 5000);
464 assert!(lat.foreign.is_none(), "nothing else stamped anything");
465 assert!(lat.stampers.is_empty(), "no third party to name");
466
467 // Never stamped: no summary, rather than an invented zero.
468 t.record("quiet", 4, None, now, None, None);
469 assert!(t.get("quiet").unwrap().latency().is_none());
470 assert_eq!(t.get("quiet").unwrap().unstamped, 1);
471 }
472
473 /// Updating a known key must never evict — the bound is on distinct keys,
474 /// not on samples.
475 #[test]
476 fn repeated_keys_never_trigger_eviction() {
477 let mut t = StatsTable::with_capacity(4);
478 let t0 = Instant::now();
479 for i in 0..1000 {
480 t.record(
481 "demo/one",
482 4,
483 None,
484 t0 + Duration::from_millis(i),
485 None,
486 None,
487 );
488 }
489 assert_eq!(t.len(), 1);
490 assert_eq!(t.evicted(), 0);
491 assert_eq!(t.get("demo/one").unwrap().count, 1000);
492 }
493
494 /// A degenerate bound must not panic or spin.
495 #[test]
496 fn a_capacity_of_one_still_works() {
497 let mut t = StatsTable::with_capacity(1);
498 let now = Instant::now();
499 t.record("a", 1, None, now, None, None);
500 t.record("b", 1, None, now, None, None);
501 assert_eq!(t.len(), 1);
502 assert_eq!(t.evicted(), 1);
503 // Zero is clamped rather than accepted.
504 assert_eq!(StatsTable::with_capacity(0).max_keys(), 1);
505 }
506
507 #[test]
508 fn rates_converge_and_gaps_count() {
509 let mut t = StatsTable::new();
510 let t0 = Instant::now();
511 // 10 Hz for 100 samples: the EWMA converges near 10.
512 for i in 0..100u32 {
513 t.record(
514 "v1/h-a/telemetry/x/m",
515 8,
516 Some(i),
517 t0 + Duration::from_millis(100 * u64::from(i)),
518 None,
519 None,
520 );
521 }
522 let s = t.get("v1/h-a/telemetry/x/m").unwrap();
523 assert_eq!(s.count, 100);
524 assert_eq!(s.bytes, 800);
525 assert!((s.rate_hz - 10.0).abs() < 1.0, "rate {}", s.rate_hz);
526 assert_eq!(s.sn_gaps, 0);
527
528 // A sequence jump records the gap.
529 t.record(
530 "v1/h-a/telemetry/x/m",
531 8,
532 Some(105),
533 t0 + Duration::from_millis(10_100),
534 None,
535 None,
536 );
537 assert_eq!(t.get("v1/h-a/telemetry/x/m").unwrap().sn_gaps, 5);
538 }
539
540 #[test]
541 fn totals_aggregate() {
542 let mut t = StatsTable::new();
543 let now = Instant::now();
544 t.record("a", 10, None, now, None, None);
545 t.record("b", 20, None, now, None, None);
546 let (count, bytes, _) = t.totals();
547 assert_eq!((count, bytes), (2, 30));
548 assert_eq!(t.len(), 2);
549 }
550
551 /// Unwatch retirement: covered-by-gone and not-by-kept keys leave the
552 /// table, counted separately from bound eviction (O6's third category).
553 #[test]
554 fn retire_unwatched_respects_remaining_coverage() {
555 let mut t = StatsTable::new();
556 let now = Instant::now();
557 t.record("v1/h-a/telemetry/x/m1", 4, None, now, None, None);
558 t.record("v1/h-a/state/x/health", 4, None, now, None, None);
559 t.record("v1/h-b/telemetry/y/m2", 4, None, now, None, None);
560
561 // Release the telemetry watch, but keep watching h-a entirely.
562 let retired = t.retire_unwatched("v1/*/telemetry/**", &["v1/h-a/**".to_string()]);
563 assert_eq!(retired, 1, "only h-b's telemetry loses coverage");
564 assert!(
565 t.get("v1/h-a/telemetry/x/m1").is_some(),
566 "still covered by kept"
567 );
568 assert!(t.get("v1/h-b/telemetry/y/m2").is_none());
569 assert_eq!(t.unwatched(), 1);
570
571 // Release the rest: everything goes, and the ledger adds up.
572 let retired = t.retire_unwatched("**", &[]);
573 assert_eq!(retired, 2);
574 assert_eq!(t.len(), 0);
575 assert_eq!(t.unwatched(), 3);
576 }
577
578 /// A selector that is not a valid keyexpr covers nothing — no panic, no
579 /// accidental mass retirement.
580 #[test]
581 fn retire_unwatched_tolerates_bad_selectors() {
582 let mut t = StatsTable::new();
583 t.record("a/b", 1, None, Instant::now(), None, None);
584 assert_eq!(t.retire_unwatched("", &[]), 0);
585 assert_eq!(t.len(), 1);
586 }
587
588 /// #213: a publisher-stamped sample and a router-stamped one measure from
589 /// different clocks. Averaging them yields a number that describes
590 /// neither, so the summary keeps them apart — and names the third party.
591 #[test]
592 fn two_stampers_are_never_folded_into_one_median() {
593 let mut t = StatsTable::new();
594 let now = Instant::now();
595 let router = zenoh::time::TimestampId::rand();
596
597 // The publisher stamps its own: tight, sub-millisecond.
598 for us in [100, 120, 140, 160] {
599 t.record("k", 4, None, now, Some((us, StampClass::SelfStamped)), None);
600 }
601 // A router stamps the rest, much further from us.
602 for us in [9000, 9500, 10_000] {
603 t.record(
604 "k",
605 4,
606 None,
607 now,
608 Some((us, StampClass::Foreign)),
609 Some(router),
610 );
611 }
612
613 let lat = t
614 .get("k")
615 .unwrap()
616 .latency()
617 .expect("something was stamped");
618 let own = lat.self_stamped.expect("the publisher-stamped population");
619 let far = lat.foreign.expect("the router-stamped population");
620 assert_eq!(own.samples, 4);
621 assert_eq!(far.samples, 3);
622 assert_eq!(own.max_us, 160);
623 assert_eq!(far.min_us, 9000);
624 assert!(
625 own.median_us < far.median_us,
626 "two populations, two medians: {} vs {}",
627 own.median_us,
628 far.median_us
629 );
630 assert_eq!(
631 lat.stampers,
632 vec![router.to_string()],
633 "the third-party stamper is named, not averaged away"
634 );
635 assert!(lat.unattributable.is_none());
636
637 // A sample with no SourceInfo is *unknown*, never "foreign" (O4).
638 let orphan = zenoh::time::TimestampId::rand();
639 t.record(
640 "u",
641 4,
642 None,
643 now,
644 Some((7, StampClass::Unattributable)),
645 Some(orphan),
646 );
647 let u = t.get("u").unwrap().latency().unwrap();
648 assert!(u.unattributable.is_some());
649 assert!(u.foreign.is_none(), "unknown is not foreign");
650 }
651
652 /// The stamper set is bounded like everything else an hours-long observer
653 /// accumulates, and it reports what the bound cost (O6).
654 #[test]
655 fn the_stamper_set_is_bounded_and_says_what_it_dropped() {
656 let mut t = StatsTable::new();
657 let now = Instant::now();
658 for _ in 0..(MAX_STAMPERS + 3) {
659 t.record(
660 "k",
661 4,
662 None,
663 now,
664 Some((10, StampClass::Foreign)),
665 Some(zenoh::time::TimestampId::rand()),
666 );
667 }
668 let lat = t.get("k").unwrap().latency().unwrap();
669 assert_eq!(lat.stampers.len(), MAX_STAMPERS);
670 assert_eq!(lat.stampers_dropped, 3, "the bound reports its cost");
671 }
672
673 /// The caveat travels with the numbers and names which clock produced
674 /// them — the mislabel #213 exists to fix.
675 #[test]
676 fn the_caveat_names_the_clock_it_measured_from() {
677 let self_only = LatencyReport {
678 self_stamped: Some(LatencySummary {
679 min_us: 1,
680 median_us: 2,
681 p95_us: 3,
682 max_us: 4,
683 samples: 4,
684 }),
685 ..LatencyReport::default()
686 };
687 assert!(
688 self_only.caveat().contains("the publisher's own HLC"),
689 "{}",
690 self_only.caveat()
691 );
692
693 let router_only = LatencyReport {
694 foreign: self_only.self_stamped,
695 stampers: vec!["abcd".into()],
696 ..LatencyReport::default()
697 };
698 let note = router_only.caveat();
699 assert!(
700 note.contains("stamped in transit, not the publisher's"),
701 "{note}"
702 );
703 assert!(note.contains("abcd"), "the stamper is named: {note}");
704
705 let both = LatencyReport {
706 self_stamped: self_only.self_stamped,
707 foreign: self_only.self_stamped,
708 ..LatencyReport::default()
709 };
710 assert!(both.caveat().contains("kept apart"), "{}", both.caveat());
711 }
712}