digdigdig3_station/storage/
event_log.rs

1//! EventLog — append-only binary record store.
2//!
3//! Format compatible with mli StorageRoot. Each record:
4//!   [i64 ts_ms LE][u32 payload_len LE][payload_bytes]
5//!
6//! File layout: `{root}/{symbol}/{stream_kind}.bin`
7//!
8//! Append-only. Truncated tail records are silently skipped on read.
9
10use std::fs::{create_dir_all, File, OpenOptions};
11use std::io::{BufReader, Read, Write};
12use std::path::PathBuf;
13
14use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt};
15
16/// Append-only binary event log, one file per `(symbol, stream_kind)`.
17pub struct EventLog {
18    root: PathBuf,
19}
20
21/// A single record to append.
22pub struct EventRecord<'a> {
23    /// Exchange timestamp in milliseconds (UTC).
24    pub ts_ms: i64,
25    /// Raw payload bytes (caller serialises to JSON before passing in).
26    pub payload: &'a [u8],
27}
28
29impl EventLog {
30    /// Create an `EventLog` rooted at `root`. Creates the directory if absent.
31    pub fn new(root: impl Into<PathBuf>) -> std::io::Result<Self> {
32        let root = root.into();
33        create_dir_all(&root)?;
34        Ok(Self { root })
35    }
36
37    /// Append one record to `{symbol}/{stream_kind}.bin`.
38    ///
39    /// Creates subdirectory and file on first call.
40    pub fn append(
41        &self,
42        symbol: &str,
43        stream_kind: &str,
44        record: &EventRecord<'_>,
45    ) -> std::io::Result<()> {
46        let dir = self.root.join(symbol);
47        create_dir_all(&dir)?;
48        let path = dir.join(format!("{stream_kind}.bin"));
49
50        let mut f = OpenOptions::new()
51            .create(true)
52            .append(true)
53            .open(&path)?;
54
55        let mut buf = Vec::with_capacity(8 + 4 + record.payload.len());
56        buf.write_i64::<LittleEndian>(record.ts_ms)?;
57        buf.write_u32::<LittleEndian>(record.payload.len() as u32)?;
58        buf.write_all(record.payload)?;
59        f.write_all(&buf)?;
60        Ok(())
61    }
62
63    /// Read all records from `{symbol}/{stream_kind}.bin`.
64    ///
65    /// Returns `(ts_ms, payload_bytes)` pairs. Truncated tail records are
66    /// silently skipped (truncation-safe).
67    pub fn read_all(
68        &self,
69        symbol: &str,
70        stream_kind: &str,
71    ) -> std::io::Result<Vec<(i64, Vec<u8>)>> {
72        let path = self.root.join(symbol).join(format!("{stream_kind}.bin"));
73        if !path.exists() {
74            return Ok(vec![]);
75        }
76        let mut reader = BufReader::new(File::open(&path)?);
77        let mut out = Vec::new();
78        loop {
79            let ts = match reader.read_i64::<LittleEndian>() {
80                Ok(v) => v,
81                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
82                Err(e) => return Err(e),
83            };
84            let len = match reader.read_u32::<LittleEndian>() {
85                Ok(v) => v as usize,
86                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
87                Err(e) => return Err(e),
88            };
89            let mut payload = vec![0u8; len];
90            match reader.read_exact(&mut payload) {
91                Ok(()) => out.push((ts, payload)),
92                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
93                Err(e) => return Err(e),
94            }
95        }
96        Ok(out)
97    }
98
99    /// Read records with `ts_ms` in `[from_ms, to_ms]` (inclusive).
100    pub fn read_range(
101        &self,
102        symbol: &str,
103        stream_kind: &str,
104        from_ms: i64,
105        to_ms: i64,
106    ) -> std::io::Result<Vec<(i64, Vec<u8>)>> {
107        let all = self.read_all(symbol, stream_kind)?;
108        Ok(all
109            .into_iter()
110            .filter(|(ts, _)| *ts >= from_ms && *ts <= to_ms)
111            .collect())
112    }
113
114    /// Lazy iterator over records — avoids loading the entire file into memory.
115    ///
116    /// Returns an iterator that yields `Ok((ts_ms, payload))` or `Err` on I/O
117    /// failure. Truncated tail records produce `None` (iterator stops cleanly).
118    pub fn iter(
119        &self,
120        symbol: &str,
121        stream_kind: &str,
122    ) -> std::io::Result<EventLogIter> {
123        let path = self.root.join(symbol).join(format!("{stream_kind}.bin"));
124        let reader = if path.exists() {
125            Some(BufReader::new(File::open(&path)?))
126        } else {
127            None
128        };
129        Ok(EventLogIter { reader })
130    }
131}
132
133/// Lazy iterator returned by [`EventLog::iter`].
134pub struct EventLogIter {
135    reader: Option<BufReader<File>>,
136}
137
138impl Iterator for EventLogIter {
139    type Item = std::io::Result<(i64, Vec<u8>)>;
140
141    fn next(&mut self) -> Option<Self::Item> {
142        let r = self.reader.as_mut()?;
143
144        let ts = match r.read_i64::<LittleEndian>() {
145            Ok(v) => v,
146            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return None,
147            Err(e) => return Some(Err(e)),
148        };
149        let len = match r.read_u32::<LittleEndian>() {
150            Ok(v) => v as usize,
151            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return None,
152            Err(e) => return Some(Err(e)),
153        };
154        let mut payload = vec![0u8; len];
155        match r.read_exact(&mut payload) {
156            Ok(()) => Some(Ok((ts, payload))),
157            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => None,
158            Err(e) => Some(Err(e)),
159        }
160    }
161}
digdigdig3_station/storage/event_log.rs

digdigdig3_station/storage/
event_log.rs
