clankers_data/
replay.rs

1use std::path::Path;
2use std::time::{Duration, Instant};
3
4use clankers_core::{LatencyStats, RobotResult};
5
6use crate::inspect::{McapLog, McapRecord};
7
8/// Replay engine that feeds MCAP messages in timestamp order.
9pub struct Replay {
10    messages: Vec<McapRecord>,
11    path: String,
12}
13
14#[derive(Debug, Clone, Default)]
15pub struct ReplaySummary {
16    pub input_messages: u64,
17    pub output_messages: u64,
18    pub dropped_messages: u64,
19    pub deadline_misses: u64,
20}
21
22#[derive(Debug, Clone)]
23pub struct ReplayResult {
24    pub summary: ReplaySummary,
25    pub latency: LatencyStats,
26    pub path: String,
27}
28
29impl Replay {
30    pub fn from_mcap(path: impl AsRef<Path>) -> RobotResult<Self> {
31        let path = path.as_ref();
32        let log = McapLog::open(path)?;
33        let messages = log.messages()?;
34        Ok(Self {
35            path: path.display().to_string(),
36            messages,
37        })
38    }
39
40    pub fn messages(&self) -> &[McapRecord] {
41        &self.messages
42    }
43
44    pub fn topics(&self) -> Vec<String> {
45        let mut topics: Vec<String> = self
46            .messages
47            .iter()
48            .map(|m| m.topic.clone())
49            .collect::<std::collections::HashSet<_>>()
50            .into_iter()
51            .collect();
52        topics.sort();
53        topics
54    }
55
56    /// Replay messages through a callback, measuring per-message latency.
57    pub async fn run<F, Fut>(&self, mut handler: F) -> RobotResult<ReplayResult>
58    where
59        F: FnMut(McapRecord) -> Fut,
60        Fut: std::future::Future<Output = RobotResult<()>>,
61    {
62        let mut latency = LatencyStats::new();
63        let mut dropped = 0u64;
64        let mut deadline_misses = 0u64;
65        let deadline = Duration::from_millis(20);
66
67        for msg in &self.messages {
68            let start = Instant::now();
69            if let Err(e) = handler(msg.clone()).await {
70                tracing::warn!(error = %e, topic = %msg.topic, "replay handler error");
71                dropped += 1;
72            }
73            let elapsed = start.elapsed();
74            latency.record(elapsed);
75            if elapsed > deadline {
76                deadline_misses += 1;
77            }
78        }
79
80        let input_count = self.messages.len() as u64;
81        Ok(ReplayResult {
82            summary: ReplaySummary {
83                input_messages: input_count,
84                output_messages: input_count.saturating_sub(dropped),
85                dropped_messages: dropped,
86                deadline_misses,
87            },
88            latency,
89            path: self.path.clone(),
90        })
91    }
92
93    pub fn format_summary(result: &ReplayResult) -> String {
94        format!(
95            "Replay summary:\n  file: {}\n  input messages: {}\n  output messages: {}\n  dropped messages: {}\n  deadline misses: {}\n\n{}",
96            result.path,
97            result.summary.input_messages,
98            result.summary.output_messages,
99            result.summary.dropped_messages,
100            result.summary.deadline_misses,
101            result.latency.format_report(),
102        )
103    }
104}
clankers_data/replay.rs

clankers_data/
replay.rs
