Skip to main content

zenkey_fleet/model/
timeline.rs

1//! The fleet timeline (#216): **deliberately no edges** — a merged ordering
2//! shows *when* things were seen on which clock, never that one caused
3//! another. A line drawn between two lanes would claim a causality no
4//! observer on this bus can establish, so there is no line; there is a
5//! position on a named axis, and the axis says what a position means.
6//!
7//! Every pane an explorer had was one key or one flat scrollback. The
8//! question an operator asks is "what happened, across the fleet, in that
9//! ten seconds?" — and answering it honestly means three things at once:
10//! one ordering, stated per axis; lanes per origin and producer, so the
11//! fleet's parallelism is visible; and the three provenances of a position
12//! (arrival, HLC, sequence number) kept apart, because they are three
13//! different measurements with three different claims.
14//!
15//! # The two axes, and why they never mix
16//!
17//! **Arrival** is this observer's monotonic clock, µs since the window
18//! epoch. Every sample has one, it is total, and it says nothing about when
19//! anything was produced — only when it was seen here.
20//!
21//! **HLC** is the sample's `uhlc` timestamp, whose id names the node that
22//! stamped it (RFC 09 §5.1 **O7** — not necessarily the publisher). What an
23//! HLC order can claim rests on a fact about zenoh 1.10, verified in its
24//! source: `treat_timestamp!` in `net/routing/dispatcher/pubsub.rs` updates
25//! a node's HLC on receive **only when the node has an HLC**, and
26//! `net/runtime/mod.rs` builds one only where `timestamping.enabled` is true
27//! for the node's whatami — routers default **true**, peers and clients
28//! **false**. The explorer session sets no timestamping, so an observer
29//! holds no HLC at all. Three consequences, each of which is a field here:
30//!
31//! * values from **one** stamper are causally ordered — that node's HLC is
32//!   monotonic and was updated by everything it forwarded
33//!   ([`HlcClaim::HappensBefore`]);
34//! * values from **different** stampers compare as skewed wall clocks
35//!   unless they passed through a common timestamping node, which this
36//!   observer cannot see ([`HlcClaim::SkewedWallClock`]);
37//! * an observer can never place "now" on the HLC axis, so a break — a
38//!   drop this observer suffered — has an arrival position and **no HLC
39//!   position**. [`HlcOrdering`] carries no breaks; the report states the
40//!   total and points at `--order arrival` for where they fell.
41//!
42//! `drop_future_timestamp = false` also lets a router re-stamp a sample
43//! whose HLC runs ahead; O7's `Foreign` provenance covers it, and the
44//! per-lane provenance counts keep that population visible.
45//!
46//! An **unstamped** sample has no HLC position, and that is enforced by the
47//! type system rather than by a warning: [`Placed<HlcAxis>`] has exactly one
48//! constructor, [`Placed::<HlcAxis>::new`], which returns
49//! `Err(`[`Unstamped`]`)` for a row without an HLC — so an unstamped row
50//! cannot be *selected onto* the HLC axis at all. It lives in
51//! [`LaneId::Unstamped`] on the arrival axis, and the HLC report counts the
52//! exclusion ([`HlcOrdering::unstamped_excluded`]).
53//!
54//! # The third provenance
55//!
56//! A per-`SampleSource` sequence-number lane would be the one ordering a
57//! *publisher* vouches for. zenoh 1.9 and 1.10 deliver no `SourceInfo` to a
58//! subscriber (`tests/stamper.rs` pins it), so [`SnLane::Unavailable`] is a
59//! structural state with a fixed reason, never an empty vector: an empty
60//! lane would read as "nobody skipped a number".
61//!
62//! # Live and replay are one projection
63//!
64//! [`TimelineRow::from_view`] and [`Ingested::from_zrec`] build the same row
65//! from a live [`SampleView`] and from a `.zrec` line, and the test in
66//! `tests/timeline_identity.rs` asserts the two reports are equal. That is
67//! what this layer is for (nothing here takes a session), and it is what
68//! makes "the same ten seconds, from the file" a claim rather than a hope.
69
70use std::collections::{BTreeMap, BTreeSet};
71use std::marker::PhantomData;
72use std::str::FromStr;
73use std::time::Instant;
74
75use crate::bus::monitor::{SampleView, StampProvenance};
76use crate::model::facts::{KeyFacts, KeyShape};
77use crate::report::{
78    ARRIVAL_CLOCK, AxisLabel, BreakKind, HlcClaim, LaneId, LaneSummary, OrderLabel, Provenance,
79    ProvenanceCounts, RowKind, SN_UNAVAILABLE_REASON, SnLaneReport, TimelineEntry, TimelineReport,
80    TimelineSource,
81};
82use crate::tape::record::ZrecItem;
83
84/// A sample's HLC, decomposed: the value, who stamped it, and how far that
85/// stamper can be attributed to the publisher (#213).
86#[derive(Debug, Clone, PartialEq, Eq)]
87pub struct HlcStamp {
88    pub ntp64: u64,
89    /// The stamping node's id, as `uhlc` prints it — the `.zrec` spelling.
90    pub stamper: String,
91    pub provenance: Provenance,
92}
93
94impl HlcStamp {
95    /// `<ntp64>/<stamper>` — what `uhlc::Timestamp: Display` prints and
96    /// `FromStr` reads, so it round-trips through a `.zrec`.
97    pub fn to_wire(&self) -> String {
98        format!("{}/{}", self.ntp64, self.stamper)
99    }
100}
101
102/// One sample, reduced to what a timeline needs: where it sits on each
103/// axis, which lane it belongs to, and what it was. No payload — the
104/// timeline is about *when*, and a pane that wants the bytes has the
105/// retained window.
106#[derive(Debug, Clone, PartialEq, Eq)]
107pub struct TimelineRow {
108    pub key: String,
109    pub lane: LaneId,
110    /// Arrival, µs since the window epoch (the observer's monotonic clock).
111    pub t_us: u64,
112    /// `None` exactly when the lane is [`LaneId::Unstamped`].
113    pub hlc: Option<HlcStamp>,
114    /// The publishing entity, `zid:eid`, when `SourceInfo` rode the sample.
115    pub source: Option<String>,
116    /// That entity's sequence number for this sample.
117    pub sn: Option<u32>,
118    pub kind: RowKind,
119    pub payload_bytes: usize,
120}
121
122impl TimelineRow {
123    /// A row from a live sample, `t_us` measured from `epoch` — the instant
124    /// the caller's watches were all declared. A sample received before the
125    /// epoch saturates to 0, as `ZrecWriter` does.
126    pub fn from_view(view: &SampleView, epoch: Instant, base: &str) -> TimelineRow {
127        let t_us = u64::try_from(view.received.saturating_duration_since(epoch).as_micros())
128            .unwrap_or(u64::MAX);
129        let hlc = view.timestamp.as_ref().map(|t| HlcStamp {
130            ntp64: t.get_time().as_u64(),
131            stamper: t.get_id().to_string(),
132            provenance: match view.stamped_by {
133                Some(StampProvenance::SelfStamped) => Provenance::SelfStamped,
134                Some(StampProvenance::Foreign { .. }) => Provenance::Foreign,
135                Some(StampProvenance::Unattributable { .. }) | None => Provenance::Unattributable,
136            },
137        });
138        TimelineRow::assemble(
139            view.key.clone(),
140            t_us,
141            hlc,
142            view.source.map(|s| format!("{}:{}", s.zid, s.eid)),
143            view.source.map(|s| s.sn),
144            if view.kind == zenoh::sample::SampleKind::Delete {
145                RowKind::Delete
146            } else {
147                RowKind::Put
148            },
149            view.payload.len(),
150            base,
151        )
152    }
153
154    /// The one place a lane is decided: unstamped first, then the key.
155    #[allow(clippy::too_many_arguments)]
156    fn assemble(
157        key: String,
158        t_us: u64,
159        hlc: Option<HlcStamp>,
160        source: Option<String>,
161        sn: Option<u32>,
162        kind: RowKind,
163        payload_bytes: usize,
164        base: &str,
165    ) -> TimelineRow {
166        let lane = match &hlc {
167            None => LaneId::Unstamped,
168            Some(_) => match KeyFacts::project(base, &key).shape {
169                KeyShape::V1(f) => LaneId::Origin {
170                    origin: f.origin,
171                    producer: f.producer,
172                },
173                KeyShape::NotUnderBase | KeyShape::Unparsed { .. } => LaneId::Foreign,
174            },
175        };
176        TimelineRow {
177            key,
178            lane,
179            t_us,
180            hlc,
181            source,
182            sn,
183            kind,
184            payload_bytes,
185        }
186    }
187}
188
189/// A break in the sequence: what the observer did not see, or merged.
190#[derive(Debug, Clone, Copy, PartialEq, Eq)]
191pub enum Break {
192    /// Samples missed while behind (`StreamItem::Dropped`, or a `.zrec`
193    /// drop record) — O6.
194    Dropped(u64),
195    /// Distinct samples a consumer merged into fewer (a GUI tick).
196    Coalesced(u64),
197}
198
199impl Break {
200    pub fn n(self) -> u64 {
201        match self {
202            Break::Dropped(n) | Break::Coalesced(n) => n,
203        }
204    }
205
206    fn kind(self) -> BreakKind {
207        match self {
208            Break::Dropped(_) => BreakKind::Dropped,
209            Break::Coalesced(_) => BreakKind::Coalesced,
210        }
211    }
212}
213
214/// A break at its arrival position: after the first `after` rows, in the
215/// order they were ingested. An index rather than a time, because a drop
216/// has no instant of its own — the observer learns of it *between* two
217/// samples — and an index is what a live drain and a `.zrec` reader can
218/// agree on exactly.
219#[derive(Debug, Clone, PartialEq, Eq)]
220pub struct PlacedBreak {
221    pub after: usize,
222    /// The lane the break belongs to, when a consumer coalesced one lane.
223    /// A monitor drop is lane-less: the broadcast does not know whose
224    /// samples it lost.
225    pub lane: Option<LaneId>,
226    pub kind: Break,
227}
228
229/// What one `.zrec` line turns into.
230#[derive(Debug, Clone, PartialEq, Eq)]
231pub enum Ingested {
232    Row(TimelineRow),
233    Break(Break),
234    /// A version-2 preamble row (RFC 13 §4.1): state at capture start,
235    /// not an arrival — it has no place on either axis, and a reader
236    /// counts it apart from observed rows (O6). The key, so a consumer can
237    /// say which.
238    Preamble {
239        key: String,
240    },
241    /// A version-2 trigger record: what fired, and to what. A marker the
242    /// consumer may place where it fell; never a row.
243    Trigger {
244        rule: String,
245        to: crate::report::CondState,
246    },
247}
248
249impl Ingested {
250    /// A `.zrec` line as the live path would have seen it (RFC 09 §5.2):
251    /// `t` verbatim as the arrival offset, `timestamp` parsed back through
252    /// `uhlc::Timestamp: FromStr`, and the stamper judged against the
253    /// row's `source` exactly as [`StampProvenance`] judges a live sample
254    /// — `Unattributable` unless the row carried one, so live and replay
255    /// classify identically.
256    ///
257    /// A row without `t` (a hand-piped ndjson row rather than a capture)
258    /// sits at 0: it has no pacing, and inventing one would be an ordering
259    /// claim. A `timestamp` that does not parse reads as unstamped — the
260    /// row is kept (O1), and the arrival axis still holds it.
261    pub fn from_zrec(item: &ZrecItem, base: &str) -> Ingested {
262        match item {
263            ZrecItem::Dropped(n) => Ingested::Break(Break::Dropped(*n)),
264            ZrecItem::Preamble { row, .. } => Ingested::Preamble {
265                key: row.key.clone(),
266            },
267            ZrecItem::Trigger(t) => Ingested::Trigger {
268                rule: t.rule.clone(),
269                to: t.to,
270            },
271            ZrecItem::Sample {
272                row,
273                t_us,
274                timestamp,
275                source,
276            } => {
277                let (entity, sn) = match source.as_deref() {
278                    Some(s) => match s.rsplit_once('#') {
279                        Some((entity, sn)) => (Some(entity.to_string()), sn.parse::<u32>().ok()),
280                        None => (Some(s.to_string()), None),
281                    },
282                    None => (None, None),
283                };
284                let hlc = timestamp
285                    .as_deref()
286                    .and_then(|s| zenoh::time::Timestamp::from_str(s).ok())
287                    .map(|t| HlcStamp {
288                        ntp64: t.get_time().as_u64(),
289                        stamper: t.get_id().to_string(),
290                        provenance: provenance_of(t.get_id(), entity.as_deref()),
291                    });
292                Ingested::Row(TimelineRow::assemble(
293                    row.key.clone(),
294                    t_us.unwrap_or(0),
295                    hlc,
296                    entity,
297                    sn,
298                    if row.delete {
299                        RowKind::Delete
300                    } else {
301                        RowKind::Put
302                    },
303                    row.payload.len(),
304                    base,
305                ))
306            }
307        }
308    }
309}
310
311/// [`StampProvenance::of`]'s judgement over the `.zrec` spelling of a
312/// source: the entity's zid against the stamper's id, exact, and
313/// `Unattributable` when there is nothing to compare against (O4).
314fn provenance_of(stamper: &zenoh::time::TimestampId, entity: Option<&str>) -> Provenance {
315    let Some(entity) = entity else {
316        return Provenance::Unattributable;
317    };
318    let zid = entity.split(':').next().unwrap_or(entity);
319    match zenoh::config::ZenohId::from_str(zid) {
320        Ok(zid) if zenoh::time::TimestampId::from(zid) == *stamper => Provenance::SelfStamped,
321        Ok(_) => Provenance::Foreign,
322        Err(_) => Provenance::Unattributable,
323    }
324}
325
326/// Which axis to order on.
327#[derive(Debug, Clone, Copy, PartialEq, Eq)]
328pub enum Order {
329    Arrival,
330    Hlc,
331}
332
333impl Order {
334    pub fn label(self) -> OrderLabel {
335        match self {
336            Order::Arrival => OrderLabel::Arrival,
337            Order::Hlc => OrderLabel::Hlc,
338        }
339    }
340}
341
342/// The arrival axis: every row has a position on it.
343#[derive(Debug, Clone, Copy, PartialEq, Eq)]
344pub struct ArrivalAxis;
345
346/// The HLC axis: only a stamped row has a position on it.
347#[derive(Debug, Clone, Copy, PartialEq, Eq)]
348pub struct HlcAxis;
349
350/// The refusal: this row carries no HLC and has no place on that axis.
351#[derive(Debug, Clone, Copy, PartialEq, Eq)]
352pub struct Unstamped;
353
354/// A row placed on one axis. The axis is a type parameter so that a
355/// `Vec<Placed<HlcAxis>>` *cannot* hold an unstamped row: the only way to
356/// make one is [`Placed::<HlcAxis>::new`], and it refuses.
357///
358/// The arrival constructor is total:
359///
360/// ```
361/// use zenkey_fleet::model::timeline::{ArrivalAxis, Placed, TimelineRow};
362/// use zenkey_fleet::report::{LaneId, RowKind};
363/// let row = TimelineRow {
364///     key: "plain/key".into(), lane: LaneId::Unstamped, t_us: 1, hlc: None,
365///     source: None, sn: None, kind: RowKind::Put, payload_bytes: 0,
366/// };
367/// let placed: Placed<ArrivalAxis> = Placed::<ArrivalAxis>::new(row);
368/// assert_eq!(placed.row().t_us, 1);
369/// ```
370///
371/// The HLC constructor is not — it returns a `Result`, and there is no
372/// other constructor, so a function that promises a `Placed<HlcAxis>` for
373/// an arbitrary row does not compile:
374///
375/// ```compile_fail
376/// use zenkey_fleet::model::timeline::{HlcAxis, Placed, TimelineRow};
377/// fn onto_the_hlc_axis(row: TimelineRow) -> Placed<HlcAxis> {
378///     Placed::<HlcAxis>::new(row)
379/// }
380/// ```
381#[derive(Debug, Clone, PartialEq, Eq)]
382pub struct Placed<A> {
383    pos: usize,
384    seq: usize,
385    row: TimelineRow,
386    _axis: PhantomData<A>,
387}
388
389impl<A> Placed<A> {
390    /// Position in the merged ordering on this axis, 0-based. Assigned by
391    /// the ordering that ranked it; 0 until then.
392    pub fn pos(&self) -> usize {
393        self.pos
394    }
395
396    pub fn row(&self) -> &TimelineRow {
397        &self.row
398    }
399
400    pub fn into_row(self) -> TimelineRow {
401        self.row
402    }
403}
404
405impl Placed<ArrivalAxis> {
406    /// Every row arrives; every row is placeable here.
407    pub fn new(row: TimelineRow) -> Placed<ArrivalAxis> {
408        Placed {
409            pos: 0,
410            seq: 0,
411            row,
412            _axis: PhantomData,
413        }
414    }
415}
416
417impl Placed<HlcAxis> {
418    /// The **only** constructor: a row without an HLC has no position on
419    /// this axis and is refused, never defaulted to its arrival time.
420    ///
421    /// Belt and braces on the lane too — a row assembled by hand into
422    /// [`LaneId::Unstamped`] is refused whatever its `hlc` field says, so
423    /// the lane's invariant ("exists only on the arrival axis") cannot be
424    /// broken by a struct literal.
425    pub fn new(row: TimelineRow) -> Result<Placed<HlcAxis>, Unstamped> {
426        if row.hlc.is_none() || row.lane == LaneId::Unstamped {
427            return Err(Unstamped);
428        }
429        Ok(Placed {
430            pos: 0,
431            seq: 0,
432            row,
433            _axis: PhantomData,
434        })
435    }
436
437    fn stamp(&self) -> &HlcStamp {
438        self.row
439            .hlc
440            .as_ref()
441            .expect("a Placed<HlcAxis> is stamped by construction")
442    }
443}
444
445/// One slot of a merged ordering: which lane, and which entry in it — or a
446/// break, which belongs to the sequence and to no lane's vector.
447#[derive(Debug, Clone, PartialEq, Eq)]
448pub enum Slot {
449    Sample { lane: LaneId, index: usize },
450    Break(usize),
451}
452
453/// The window ordered by arrival: every row placed, breaks at the position
454/// the observer learned of them.
455#[derive(Debug, Clone, PartialEq, Eq)]
456pub struct ArrivalOrdering {
457    pub lanes: BTreeMap<LaneId, Vec<Placed<ArrivalAxis>>>,
458    pub breaks: Vec<PlacedBreak>,
459    /// The merged sequence, `pos` order.
460    pub merged: Vec<Slot>,
461}
462
463/// The window ordered by HLC: stamped rows placed, the rest counted, and
464/// the claim the axis can make. **No breaks**: a drop has no HLC position.
465#[derive(Debug, Clone, PartialEq, Eq)]
466pub struct HlcOrdering {
467    pub lanes: BTreeMap<LaneId, Vec<Placed<HlcAxis>>>,
468    /// Rows refused by [`Placed::<HlcAxis>::new`] — reported, so an HLC
469    /// listing that is shorter than the arrival one says why.
470    pub unstamped_excluded: usize,
471    pub claim: HlcClaim,
472    pub merged: Vec<Slot>,
473}
474
475/// The sequence-number provenance, present or structurally not.
476#[derive(Debug, Clone, PartialEq, Eq)]
477pub enum SnLane {
478    /// No row carried a sequence number. The reason is fixed because the
479    /// cause is: this zenoh delivers none.
480    Unavailable { reason: &'static str },
481    /// Some rows did — per publishing entity, `(seq, sn)` pairs in arrival
482    /// order, which is what a gap or a reset would be read from.
483    Present {
484        sources: BTreeMap<String, Vec<(usize, u32)>>,
485    },
486}
487
488impl SnLane {
489    fn report(&self) -> SnLaneReport {
490        match self {
491            SnLane::Unavailable { reason } => SnLaneReport::Unavailable { reason },
492            SnLane::Present { sources } => SnLaneReport::Present {
493                sources: sources.len(),
494                samples: sources.values().map(Vec::len).sum(),
495            },
496        }
497    }
498}
499
500/// The sequence-number lane over a window.
501pub fn sn_lane(rows: &[TimelineRow]) -> SnLane {
502    let mut sources: BTreeMap<String, Vec<(usize, u32)>> = BTreeMap::new();
503    for (seq, row) in rows.iter().enumerate() {
504        if let (Some(source), Some(sn)) = (&row.source, row.sn) {
505            sources.entry(source.clone()).or_default().push((seq, sn));
506        }
507    }
508    if sources.is_empty() {
509        SnLane::Unavailable {
510            reason: SN_UNAVAILABLE_REASON,
511        }
512    } else {
513        SnLane::Present { sources }
514    }
515}
516
517/// Order a window by arrival. Rows are ranked by `(t_us, ingest order)`,
518/// so two samples that saturated to the same offset keep the order they
519/// were seen in; a break with `after = k` is emitted before the row that
520/// was ingested `k`-th.
521pub fn arrival(rows: &[TimelineRow], breaks: &[PlacedBreak]) -> ArrivalOrdering {
522    let mut placed: Vec<Placed<ArrivalAxis>> = rows
523        .iter()
524        .cloned()
525        .enumerate()
526        .map(|(seq, row)| Placed {
527            seq,
528            ..Placed::<ArrivalAxis>::new(row)
529        })
530        .collect();
531    placed.sort_by_key(|p| (p.row.t_us, p.seq));
532
533    let mut pending: Vec<(usize, &PlacedBreak)> = breaks.iter().enumerate().collect();
534    pending.sort_by_key(|(i, b)| (b.after, *i));
535    let mut pending = pending.into_iter().peekable();
536
537    let mut lanes: BTreeMap<LaneId, Vec<Placed<ArrivalAxis>>> = BTreeMap::new();
538    let mut merged = Vec::with_capacity(placed.len() + breaks.len());
539    let mut pos = 0usize;
540    for mut p in placed {
541        while let Some((i, b)) = pending.peek()
542            && b.after <= p.seq
543        {
544            merged.push(Slot::Break(*i));
545            pos += 1;
546            pending.next();
547        }
548        p.pos = pos;
549        pos += 1;
550        let lane = p.row.lane.clone();
551        let entries = lanes.entry(lane.clone()).or_default();
552        merged.push(Slot::Sample {
553            lane,
554            index: entries.len(),
555        });
556        entries.push(p);
557    }
558    for (i, _) in pending {
559        merged.push(Slot::Break(i));
560    }
561    ArrivalOrdering {
562        lanes,
563        breaks: breaks.to_vec(),
564        merged,
565    }
566}
567
568/// Order a window by HLC. Rows are ranked by `(ntp64, stamper, ingest
569/// order)`: the stamper tie-break keeps the listing deterministic across
570/// two clocks that happen to agree, and the ingest tie-break keeps it so
571/// for one clock that stamped twice at once.
572pub fn hlc(rows: &[TimelineRow]) -> HlcOrdering {
573    let mut unstamped_excluded = 0usize;
574    let mut placed: Vec<Placed<HlcAxis>> = Vec::with_capacity(rows.len());
575    for (seq, row) in rows.iter().cloned().enumerate() {
576        match Placed::<HlcAxis>::new(row) {
577            Ok(p) => placed.push(Placed { seq, ..p }),
578            Err(Unstamped) => unstamped_excluded += 1,
579        }
580    }
581    placed.sort_by(|a, b| {
582        let (sa, sb) = (a.stamp(), b.stamp());
583        (sa.ntp64, &sa.stamper, a.seq).cmp(&(sb.ntp64, &sb.stamper, b.seq))
584    });
585
586    let stampers: BTreeSet<String> = placed.iter().map(|p| p.stamp().stamper.clone()).collect();
587    let claim = match stampers.len() {
588        0 => HlcClaim::NoStampedSamples,
589        1 => HlcClaim::HappensBefore {
590            stamper: stampers.into_iter().next().expect("one stamper"),
591        },
592        _ => HlcClaim::SkewedWallClock { stampers },
593    };
594
595    let mut lanes: BTreeMap<LaneId, Vec<Placed<HlcAxis>>> = BTreeMap::new();
596    let mut merged = Vec::with_capacity(placed.len());
597    for (pos, mut p) in placed.into_iter().enumerate() {
598        p.pos = pos;
599        let lane = p.row.lane.clone();
600        let entries = lanes.entry(lane.clone()).or_default();
601        merged.push(Slot::Sample {
602            lane,
603            index: entries.len(),
604        });
605        entries.push(p);
606    }
607    HlcOrdering {
608        lanes,
609        unstamped_excluded,
610        claim,
611        merged,
612    }
613}
614
615/// A window and what was asked to get it — the input to [`timeline`].
616#[derive(Debug, Clone, PartialEq)]
617pub struct Window {
618    pub rows: Vec<TimelineRow>,
619    pub breaks: Vec<PlacedBreak>,
620    /// The selectors watched (O5).
621    pub scopes: Vec<String>,
622    pub window_s: Option<f64>,
623    pub source: TimelineSource,
624    /// Keys the bounded statistics table retired during the window (O6).
625    pub keys_evicted: u64,
626}
627
628/// The report for one window on one axis.
629pub fn timeline(window: &Window, order: Order) -> TimelineReport {
630    let order_by = order.label();
631    let sn = sn_lane(&window.rows).report();
632    let dropped = window
633        .breaks
634        .iter()
635        .filter_map(|b| match b.kind {
636            Break::Dropped(n) => Some(n),
637            Break::Coalesced(_) => None,
638        })
639        .sum();
640    let coalesced = window
641        .breaks
642        .iter()
643        .filter_map(|b| match b.kind {
644            Break::Coalesced(n) => Some(n),
645            Break::Dropped(_) => None,
646        })
647        .sum();
648    let (axis, lanes, unstamped_excluded, rows) = match order {
649        Order::Arrival => {
650            let o = arrival(&window.rows, &window.breaks);
651            let lanes = summarise(&o.lanes, |p| &p.row);
652            let rows = o
653                .merged
654                .iter()
655                .enumerate()
656                .map(|(pos, slot)| match slot {
657                    Slot::Sample { lane, index } => {
658                        entry(order_by, &o.lanes[lane][*index].row, pos)
659                    }
660                    Slot::Break(i) => {
661                        let b = &o.breaks[*i];
662                        TimelineEntry::Break {
663                            order_by,
664                            pos,
665                            lane: b.lane.clone(),
666                            kind: b.kind.kind(),
667                            n: b.kind.n(),
668                        }
669                    }
670                })
671                .collect();
672            (
673                AxisLabel::Arrival {
674                    clock: ARRIVAL_CLOCK,
675                },
676                lanes,
677                0,
678                rows,
679            )
680        }
681        Order::Hlc => {
682            let o = hlc(&window.rows);
683            let lanes = summarise(&o.lanes, |p| &p.row);
684            let rows = o
685                .merged
686                .iter()
687                .enumerate()
688                .map(|(pos, slot)| match slot {
689                    Slot::Sample { lane, index } => {
690                        entry(order_by, &o.lanes[lane][*index].row, pos)
691                    }
692                    Slot::Break(_) => unreachable!("an HLC ordering places no breaks"),
693                })
694                .collect();
695            (
696                AxisLabel::Hlc { claim: o.claim },
697                lanes,
698                o.unstamped_excluded,
699                rows,
700            )
701        }
702    };
703    TimelineReport {
704        order_by,
705        axis,
706        scopes: window.scopes.clone(),
707        window_s: window.window_s,
708        source: window.source.clone(),
709        lanes,
710        sn_lane: sn,
711        unstamped_excluded,
712        dropped,
713        coalesced,
714        keys_evicted: window.keys_evicted,
715        rows,
716    }
717}
718
719fn entry(order_by: OrderLabel, row: &TimelineRow, pos: usize) -> TimelineEntry {
720    TimelineEntry::Sample {
721        order_by,
722        pos,
723        lane: row.lane.clone(),
724        key: row.key.clone(),
725        t_us: row.t_us,
726        hlc: row.hlc.as_ref().map(HlcStamp::to_wire),
727        stamped_by: row.hlc.as_ref().map(|h| h.stamper.clone()),
728        provenance: row.hlc.as_ref().map(|h| h.provenance),
729        kind: row.kind,
730    }
731}
732
733fn summarise<A>(
734    lanes: &BTreeMap<LaneId, Vec<Placed<A>>>,
735    row: impl Fn(&Placed<A>) -> &TimelineRow,
736) -> Vec<LaneSummary> {
737    lanes
738        .iter()
739        .map(|(lane, placed)| {
740            let mut provenance = ProvenanceCounts::default();
741            let mut stampers = BTreeSet::new();
742            let (mut first, mut last) = (u64::MAX, 0u64);
743            for p in placed {
744                let r = row(p);
745                first = first.min(r.t_us);
746                last = last.max(r.t_us);
747                if let Some(h) = &r.hlc {
748                    stampers.insert(h.stamper.clone());
749                    match h.provenance {
750                        Provenance::SelfStamped => provenance.self_stamped += 1,
751                        Provenance::Foreign => provenance.foreign += 1,
752                        Provenance::Unattributable => provenance.unattributable += 1,
753                    }
754                }
755            }
756            LaneSummary {
757                lane: lane.clone(),
758                samples: placed.len(),
759                first_t_us: if placed.is_empty() { 0 } else { first },
760                last_t_us: last,
761                stampers,
762                provenance,
763            }
764        })
765        .collect()
766}
767
768#[cfg(test)]
769mod tests {
770    use super::*;
771
772    const BASE: &str = "acme";
773
774    fn stamped(key: &str, t_us: u64, ntp64: u64, stamper: &str) -> TimelineRow {
775        TimelineRow::assemble(
776            key.into(),
777            t_us,
778            Some(HlcStamp {
779                ntp64,
780                stamper: stamper.into(),
781                provenance: Provenance::Unattributable,
782            }),
783            None,
784            None,
785            RowKind::Put,
786            3,
787            BASE,
788        )
789    }
790
791    fn unstamped(key: &str, t_us: u64) -> TimelineRow {
792        TimelineRow::assemble(key.into(), t_us, None, None, None, RowKind::Put, 3, BASE)
793    }
794
795    fn keys(report: &TimelineReport) -> Vec<(&str, usize)> {
796        report
797            .rows
798            .iter()
799            .filter_map(|e| match e {
800                TimelineEntry::Sample { key, pos, .. } => Some((key.as_str(), *pos)),
801                TimelineEntry::Break { .. } => None,
802            })
803            .collect()
804    }
805
806    fn window(rows: Vec<TimelineRow>, breaks: Vec<PlacedBreak>) -> Window {
807        Window {
808            rows,
809            breaks,
810            scopes: vec!["acme/v1/**".into()],
811            window_s: Some(1.0),
812            source: TimelineSource::Live,
813            keys_evicted: 0,
814        }
815    }
816
817    const A: &str = "acme/v1/h-3fa9c2d41b7e/telemetry/sysinfo/a";
818    const B: &str = "acme/v1/h-3fa9c2d41b7e/telemetry/sysinfo/b";
819
820    /// The acceptance criterion: a synthetic reorder — A seen first, B
821    /// stamped first — is visible as hlc-order ≠ arrival-order.
822    #[test]
823    fn a_reorder_is_visible_as_two_different_orders() {
824        let w = window(
825            vec![stamped(A, 10, 200, "33"), stamped(B, 20, 100, "33")],
826            vec![],
827        );
828        let by_arrival = timeline(&w, Order::Arrival);
829        let by_hlc = timeline(&w, Order::Hlc);
830        assert_eq!(keys(&by_arrival), [(A, 0), (B, 1)]);
831        assert_eq!(keys(&by_hlc), [(B, 0), (A, 1)]);
832        assert_eq!(
833            by_hlc.axis,
834            AxisLabel::Hlc {
835                claim: HlcClaim::HappensBefore {
836                    stamper: "33".into()
837                }
838            }
839        );
840        assert_eq!(
841            by_arrival.axis,
842            AxisLabel::Arrival {
843                clock: ARRIVAL_CLOCK
844            }
845        );
846        // Both rows share one lane; the lane says who stamped it.
847        assert_eq!(by_hlc.lanes.len(), 1);
848        assert_eq!(by_hlc.lanes[0].stampers.len(), 1);
849        assert_eq!(by_hlc.lanes[0].provenance.unattributable, 2);
850    }
851
852    /// Two stampers: the order is a comparison of wall clocks and says so.
853    #[test]
854    fn mixed_stampers_make_the_hlc_axis_a_skewed_wall_clock() {
855        let w = window(
856            vec![stamped(A, 10, 200, "33"), stamped(B, 20, 100, "44")],
857            vec![],
858        );
859        let by_hlc = timeline(&w, Order::Hlc);
860        assert_eq!(
861            by_hlc.axis,
862            AxisLabel::Hlc {
863                claim: HlcClaim::SkewedWallClock {
864                    stampers: ["33".to_string(), "44".to_string()].into_iter().collect()
865                }
866            }
867        );
868    }
869
870    /// An unstamped row: its own lane on arrival, refused on HLC, counted.
871    #[test]
872    fn an_unstamped_row_has_an_arrival_lane_and_no_hlc_position() {
873        let plain = unstamped("plain/key", 5);
874        assert_eq!(plain.lane, LaneId::Unstamped);
875        assert_eq!(Placed::<HlcAxis>::new(plain.clone()), Err(Unstamped));
876
877        let w = window(vec![stamped(A, 10, 200, "33"), plain], vec![]);
878        let by_arrival = timeline(&w, Order::Arrival);
879        assert_eq!(by_arrival.unstamped_excluded, 0);
880        assert!(
881            by_arrival
882                .lanes
883                .iter()
884                .any(|l| l.lane == LaneId::Unstamped && l.samples == 1)
885        );
886        let by_hlc = timeline(&w, Order::Hlc);
887        assert_eq!(by_hlc.unstamped_excluded, 1);
888        assert_eq!(keys(&by_hlc), [(A, 0)]);
889        assert!(by_hlc.lanes.iter().all(|l| l.lane != LaneId::Unstamped));
890    }
891
892    /// A v1 key from an origin, unstamped, still goes to the unstamped
893    /// lane — the lane is about the axis it can live on, not the key.
894    #[test]
895    fn an_unstamped_v1_key_is_in_the_unstamped_lane_not_its_origins() {
896        let row = unstamped(A, 1);
897        assert_eq!(row.lane, LaneId::Unstamped);
898        // And a hand-built row cannot smuggle itself onto the HLC axis by
899        // lying about its lane.
900        let lying = TimelineRow {
901            lane: LaneId::Unstamped,
902            ..stamped(A, 1, 1, "33")
903        };
904        assert_eq!(Placed::<HlcAxis>::new(lying), Err(Unstamped));
905    }
906
907    /// Nothing under the base is a lane of its own, never a dropped row.
908    #[test]
909    fn a_key_outside_the_base_is_foreign_and_kept() {
910        let row = stamped("other/v1/h-3fa9c2d41b7e/state/x/y", 1, 1, "33");
911        assert_eq!(row.lane, LaneId::Foreign);
912        let w = window(vec![row], vec![]);
913        assert_eq!(timeline(&w, Order::Hlc).rows.len(), 1);
914    }
915
916    #[test]
917    fn the_sn_lane_is_unavailable_never_empty_until_a_row_carries_one() {
918        assert_eq!(
919            sn_lane(&[unstamped(A, 1)]),
920            SnLane::Unavailable {
921                reason: SN_UNAVAILABLE_REASON
922            }
923        );
924        let with_sn = TimelineRow {
925            source: Some("zid:7".into()),
926            sn: Some(41),
927            ..unstamped(A, 1)
928        };
929        assert!(matches!(sn_lane(&[with_sn]), SnLane::Present { .. }));
930    }
931
932    /// Breaks keep their arrival position on the arrival axis and have
933    /// none on the HLC axis — the total still rides the envelope.
934    #[test]
935    fn breaks_keep_position_on_arrival_and_are_totalled_on_hlc() {
936        let w = window(
937            vec![stamped(A, 10, 100, "33"), stamped(B, 20, 200, "33")],
938            vec![PlacedBreak {
939                after: 1,
940                lane: None,
941                kind: Break::Dropped(7),
942            }],
943        );
944        let by_arrival = timeline(&w, Order::Arrival);
945        let shape: Vec<&str> = by_arrival
946            .rows
947            .iter()
948            .map(|e| match e {
949                TimelineEntry::Sample { key, .. } => key.as_str(),
950                TimelineEntry::Break { .. } => "<break>",
951            })
952            .collect();
953        assert_eq!(shape, [A, "<break>", B]);
954        assert!(matches!(
955            by_arrival.rows[1],
956            TimelineEntry::Break {
957                pos: 1,
958                kind: BreakKind::Dropped,
959                n: 7,
960                ..
961            }
962        ));
963        assert_eq!(by_arrival.dropped, 7);
964        let by_hlc = timeline(&w, Order::Hlc);
965        assert_eq!(by_hlc.dropped, 7);
966        assert!(
967            by_hlc
968                .rows
969                .iter()
970                .all(|e| matches!(e, TimelineEntry::Sample { .. }))
971        );
972    }
973
974    #[test]
975    fn a_zrec_drop_record_is_a_break_and_a_row_reads_its_stamp_back() {
976        assert_eq!(
977            Ingested::from_zrec(&ZrecItem::Dropped(3), BASE),
978            Ingested::Break(Break::Dropped(3))
979        );
980        let item = ZrecItem::Sample {
981            row: crate::tape::ingest::IngestRow {
982                key: A.into(),
983                payload: vec![1, 2, 3],
984                encoding: None,
985                qos: None,
986                delete: false,
987                attachment: None,
988            },
989            t_us: Some(42),
990            timestamp: Some("100/33".into()),
991            source: None,
992        };
993        let Ingested::Row(row) = Ingested::from_zrec(&item, BASE) else {
994            panic!("a sample line is a row");
995        };
996        assert_eq!(row.t_us, 42);
997        assert_eq!(
998            row.hlc,
999            Some(HlcStamp {
1000                ntp64: 100,
1001                stamper: "33".into(),
1002                provenance: Provenance::Unattributable
1003            })
1004        );
1005        assert_eq!(row.payload_bytes, 3);
1006        assert_eq!(
1007            row.lane,
1008            LaneId::Origin {
1009                origin: "h-3fa9c2d41b7e".into(),
1010                producer: Some("sysinfo".into())
1011            }
1012        );
1013    }
1014}