Skip to main content

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}