katra_trace/
reader.rs

1//! Trace reader: iterates records from a trace file with validation.
2
3use std::fs::File;
4use std::io::{BufReader, Read};
5use std::path::Path;
6
7use katra_core::{KatraError, Result};
8
9use crate::header::{FORMAT_VERSION, HostInfo, SCHEMA_HASH, TRACE_MAGIC, TraceHeader};
10use crate::record::TraceRecord;
11
12/// Errors surfaced while opening or reading a trace.
13#[derive(Debug)]
14pub struct TraceReaderError {
15    /// Human-readable description.
16    pub message: String,
17}
18
19/// Reads records from a trace file.
20pub struct TraceReader {
21    inner: BufReader<File>,
22    header: Option<TraceHeader>,
23    count: u64,
24}
25
26impl TraceReader {
27    /// Open and validate the header of a trace file.
28    pub fn open(path: &Path) -> Result<Self> {
29        let file = File::open(path).map_err(KatraError::from)?;
30        let mut r = TraceReader { inner: BufReader::new(file), header: None, count: 0 };
31        r.read_header()?;
32        Ok(r)
33    }
34
35    fn read_exact(&mut self, buf: &mut [u8]) -> Result<()> {
36        self.inner.read_exact(buf).map_err(|e| KatraError::Corrupt(format!("truncated trace: {e}")))
37    }
38
39    fn read_header(&mut self) -> Result<()> {
40        let mut magic = [0u8; 8];
41        self.read_exact(&mut magic)?;
42        if magic != TRACE_MAGIC {
43            return Err(KatraError::Corrupt(format!(
44                "bad magic: {magic:02x?} (expected {TRACE_MAGIC:02x?})"
45            )));
46        }
47        let mut version_bytes = [0u8; 4];
48        self.read_exact(&mut version_bytes)?;
49        let version = u32::from_le_bytes(version_bytes);
50        if version != FORMAT_VERSION {
51            return Err(KatraError::Protocol(format!(
52                "trace format version {version} != supported {FORMAT_VERSION}"
53            )));
54        }
55        let mut len_bytes = [0u8; 4];
56        self.read_exact(&mut len_bytes)?;
57        let len = u32::from_le_bytes(len_bytes) as usize;
58        let mut buf = vec![0u8; len];
59        self.read_exact(&mut buf)?;
60        let header: TraceHeader = bincode::deserialize(&buf)
61            .map_err(|e| KatraError::Corrupt(format!("bad header: {e}")))?;
62        if header.schema_hash != SCHEMA_HASH {
63            return Err(KatraError::Protocol(format!(
64                "schema hash {} != expected {}; trace was produced by an incompatible build",
65                header.schema_hash, SCHEMA_HASH
66            )));
67        }
68        self.header = Some(header);
69        Ok(())
70    }
71
72    /// The validated header.
73    pub fn header(&self) -> Result<&TraceHeader> {
74        self.header.as_ref().ok_or_else(|| KatraError::Protocol("header not read".into()))
75    }
76
77    /// Read the next record, or `None` at end of stream.
78    pub fn next_record(&mut self) -> Result<Option<TraceRecord>> {
79        let mut len_bytes = [0u8; 4];
80        match self.inner.read_exact(&mut len_bytes) {
81            Ok(()) => {}
82            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(None),
83            Err(e) => return Err(KatraError::from(e)),
84        }
85        let len = u32::from_le_bytes(len_bytes) as usize;
86        let mut buf = vec![0u8; len];
87        self.read_exact(&mut buf)?;
88        let record: TraceRecord = bincode::deserialize(&buf)
89            .map_err(|e| KatraError::Corrupt(format!("bad record: {e}")))?;
90        self.count += 1;
91        Ok(Some(record))
92    }
93
94    /// Number of records read so far.
95    pub fn count(&self) -> u64 {
96        self.count
97    }
98}
99
100/// Collect basic host information for the trace header, from `/proc`
101/// (Linux) or defaults elsewhere. Best-effort; never fails.
102pub fn host_info() -> HostInfo {
103    HostInfo {
104        os: Some(std::env::consts::OS.to_string()),
105        kernel: read_first_line("/proc/sys/kernel/osrelease"),
106        cpu_model: read_proc_cpuinfo_model(),
107        cpu_cores: std::thread::available_parallelism().ok().map(|n| n.get() as u32),
108        ram_bytes: read_proc_meminfo_total(),
109    }
110}
111
112fn read_first_line(path: &str) -> Option<String> {
113    std::fs::read_to_string(path)
114        .ok()
115        .and_then(|s| s.lines().next().map(|l| l.trim().to_string()))
116        .filter(|s| !s.is_empty())
117}
118
119fn read_proc_cpuinfo_model() -> Option<String> {
120    let text = std::fs::read_to_string("/proc/cpuinfo").ok()?;
121    for line in text.lines() {
122        if let Some((key, value)) = line.split_once(':') {
123            if key.trim() == "model name" {
124                let v = value.trim().to_string();
125                if !v.is_empty() {
126                    return Some(v);
127                }
128            }
129        }
130    }
131    None
132}
133
134fn read_proc_meminfo_total() -> Option<u64> {
135    let text = std::fs::read_to_string("/proc/meminfo").ok()?;
136    for line in text.lines() {
137        if let Some(rest) = line.strip_prefix("MemTotal:") {
138            let kb: u64 = rest.trim().trim_end_matches("kB").trim().parse().ok()?;
139            return Some(kb * 1024);
140        }
141    }
142    None
143}
katra_trace/reader.rs

katra_trace/
reader.rs
