1use 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#[derive(Debug, Clone, PartialEq, Eq)]
87pub struct HlcStamp {
88 pub ntp64: u64,
89 pub stamper: String,
91 pub provenance: Provenance,
92}
93
94impl HlcStamp {
95 pub fn to_wire(&self) -> String {
98 format!("{}/{}", self.ntp64, self.stamper)
99 }
100}
101
102#[derive(Debug, Clone, PartialEq, Eq)]
107pub struct TimelineRow {
108 pub key: String,
109 pub lane: LaneId,
110 pub t_us: u64,
112 pub hlc: Option<HlcStamp>,
114 pub source: Option<String>,
116 pub sn: Option<u32>,
118 pub kind: RowKind,
119 pub payload_bytes: usize,
120}
121
122impl TimelineRow {
123 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 #[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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
191pub enum Break {
192 Dropped(u64),
195 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#[derive(Debug, Clone, PartialEq, Eq)]
220pub struct PlacedBreak {
221 pub after: usize,
222 pub lane: Option<LaneId>,
226 pub kind: Break,
227}
228
229#[derive(Debug, Clone, PartialEq, Eq)]
231pub enum Ingested {
232 Row(TimelineRow),
233 Break(Break),
234 Preamble {
239 key: String,
240 },
241 Trigger {
244 rule: String,
245 to: crate::report::CondState,
246 },
247}
248
249impl Ingested {
250 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
311fn 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#[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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
344pub struct ArrivalAxis;
345
346#[derive(Debug, Clone, Copy, PartialEq, Eq)]
348pub struct HlcAxis;
349
350#[derive(Debug, Clone, Copy, PartialEq, Eq)]
352pub struct Unstamped;
353
354#[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 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 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 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#[derive(Debug, Clone, PartialEq, Eq)]
448pub enum Slot {
449 Sample { lane: LaneId, index: usize },
450 Break(usize),
451}
452
453#[derive(Debug, Clone, PartialEq, Eq)]
456pub struct ArrivalOrdering {
457 pub lanes: BTreeMap<LaneId, Vec<Placed<ArrivalAxis>>>,
458 pub breaks: Vec<PlacedBreak>,
459 pub merged: Vec<Slot>,
461}
462
463#[derive(Debug, Clone, PartialEq, Eq)]
466pub struct HlcOrdering {
467 pub lanes: BTreeMap<LaneId, Vec<Placed<HlcAxis>>>,
468 pub unstamped_excluded: usize,
471 pub claim: HlcClaim,
472 pub merged: Vec<Slot>,
473}
474
475#[derive(Debug, Clone, PartialEq, Eq)]
477pub enum SnLane {
478 Unavailable { reason: &'static str },
481 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
500pub 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
517pub 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
568pub 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#[derive(Debug, Clone, PartialEq)]
617pub struct Window {
618 pub rows: Vec<TimelineRow>,
619 pub breaks: Vec<PlacedBreak>,
620 pub scopes: Vec<String>,
622 pub window_s: Option<f64>,
623 pub source: TimelineSource,
624 pub keys_evicted: u64,
626}
627
628pub 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 #[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 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 #[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 #[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 #[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 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 #[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 #[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}