brokapi_runtime/
record.rs

1//! Writing the event stream to a file and playing it back.
2//!
3//! A research program needs the same events twice: once live, to record, and later from a file, as
4//! many times as it takes to debug a strategy. This module does the file part with no dependencies
5//! beyond JSON. The format is one JSON line per [`Envelope`], after a header line that names the
6//! format and the schema of the envelopes:
7//!
8//! ```text
9//! {"brokapi_log":1,"envelope_schema":1}
10//! {"schema":1,"source":"bcs-md","sub":1,"ts":{...},"payload":{"OrderBook":{...}}}
11//! ```
12//!
13//! | Need | Item |
14//! |---|---|
15//! | Append envelopes to a file | [`EnvelopeWriter`] |
16//! | Read them back in order | [`EnvelopeReader`] |
17//! | Feed them to a channel, as fast as possible or at the recorded pace | [`replay`], [`ReplaySpeed`] |
18//!
19//! What the file keeps and what it does not:
20//! - Prices, quantities and money are written as decimal strings, so they come back exactly.
21//! - The monotonic clock [`Timestamps::rx_mono`](brokapi_core::Timestamps::rx_mono) is not stored. A
22//!   replayed envelope carries the time of reading in it. Use `rx_ts` and `broker_ts` to order and
23//!   measure replayed data, never `rx_mono`.
24//! - A line that cannot be read stops the read with an error that names the line. A reader never
25//!   skips a bad line silently.
26//! - An envelope with a `schema` newer than this build understands is refused
27//!   ([`LogError::Schema`]), as the rules of [`ENVELOPE_SCHEMA`] ask.
28//!
29//! This is a log of a stream, not a catalog: there is no index by instrument or time, so filter
30//! while you read. Compare NautilusTrader, whose data catalog keeps typed data in Parquet files and
31//! queries it by instrument and time: a catalog is the right tool for years of data, this log is
32//! the smallest thing that records a live run and plays it back.
33//!
34//! # Examples
35//!
36//! ```
37//! use std::sync::Arc;
38//! use std::time::Instant;
39//! use brokapi_core::{Envelope, Event, Timestamps, UnixNanos};
40//! use brokapi_runtime::record::{EnvelopeReader, EnvelopeWriter};
41//!
42//! let ts = Timestamps::new(UnixNanos::from_nanos(7), Instant::now());
43//! let env = Envelope::new(Arc::from("bcs-md"), ts, Event::Balances(vec![]));
44//!
45//! let mut writer = EnvelopeWriter::new(Vec::new())?;
46//! writer.write(&env)?;
47//! let bytes = writer.finish()?;
48//!
49//! let back: Vec<Envelope> = EnvelopeReader::new(bytes.as_slice())?.collect::<Result<_, _>>()?;
50//! assert_eq!(back, vec![env]);
51//! # Ok::<(), Box<dyn std::error::Error>>(())
52//! ```
53
54use std::fmt;
55use std::io::{self, BufRead, Write};
56use std::time::Duration;
57
58use brokapi_core::{ENVELOPE_SCHEMA, Envelope};
59use serde::{Deserialize, Serialize};
60use tokio::sync::mpsc;
61
62/// The version of the file format. It is in the header line of every file.
63pub const LOG_FORMAT: u32 = 1;
64
65#[derive(Serialize, Deserialize)]
66struct Header {
67    brokapi_log: u32,
68    envelope_schema: u16,
69}
70
71/// Why a log could not be written or read.
72///
73/// # Examples
74///
75/// ```
76/// use brokapi_runtime::record::{EnvelopeReader, LogError};
77///
78/// let err = EnvelopeReader::new(&b"not a log\n"[..]).err().expect("refused");
79/// assert!(matches!(err, LogError::Format { line: 1, .. }));
80/// ```
81#[derive(Debug)]
82pub enum LogError {
83    /// The file or the stream failed. Check the disk and the path.
84    Io(io::Error),
85    /// A line is not what the format says. The file is damaged or is not a brokapi log. `line` counts
86    /// from 1, the header is line 1.
87    Format {
88        /// The number of the bad line.
89        line: u64,
90        /// What is wrong.
91        message: String,
92    },
93    /// The file was written by a newer version. Update the library to read it.
94    Schema {
95        /// What the file declares.
96        found: u32,
97        /// The newest this build reads.
98        supported: u32,
99    },
100}
101
102impl fmt::Display for LogError {
103    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
104        match self {
105            Self::Io(e) => write!(f, "log I/O error: {e}"),
106            Self::Format { line, message } => write!(f, "log line {line}: {message}"),
107            Self::Schema { found, supported } => write!(
108                f,
109                "log has format or schema {found}, this build reads up to {supported}"
110            ),
111        }
112    }
113}
114
115impl std::error::Error for LogError {
116    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
117        match self {
118            Self::Io(e) => Some(e),
119            _ => None,
120        }
121    }
122}
123
124impl From<io::Error> for LogError {
125    fn from(e: io::Error) -> Self {
126        Self::Io(e)
127    }
128}
129
130/// Appends envelopes to any [`Write`]: a file, a buffer, a compressor.
131///
132/// Wrap a file in [`std::io::BufWriter`] yourself when you write many envelopes: the writer does not
133/// buffer. [`flush`](Self::flush) after the last write, or [`finish`](Self::finish), so the tail
134/// is not left in a buffer when the program ends.
135///
136/// # Examples
137///
138/// See the module [`record`](crate::record).
139pub struct EnvelopeWriter<W: Write> {
140    out: W,
141    written: u64,
142}
143
144impl<W: Write> EnvelopeWriter<W> {
145    /// Starts a new log in `out`: writes the header line.
146    ///
147    /// # Errors
148    ///
149    /// [`LogError::Io`] when the header cannot be written.
150    pub fn new(mut out: W) -> Result<Self, LogError> {
151        let header = Header {
152            brokapi_log: LOG_FORMAT,
153            envelope_schema: ENVELOPE_SCHEMA,
154        };
155        serde_json::to_writer(&mut out, &header).map_err(json_io)?;
156        out.write_all(b"\n")?;
157        Ok(Self { out, written: 0 })
158    }
159
160    /// Appends one envelope as one line.
161    ///
162    /// # Errors
163    ///
164    /// [`LogError::Io`] when the write fails. The line may be half written then: do not append to
165    /// the same file after an error, start a new one.
166    pub fn write(&mut self, envelope: &Envelope) -> Result<(), LogError> {
167        serde_json::to_writer(&mut self.out, envelope).map_err(json_io)?;
168        self.out.write_all(b"\n")?;
169        self.written += 1;
170        Ok(())
171    }
172
173    /// How many envelopes were written (the header is not counted).
174    pub fn written(&self) -> u64 {
175        self.written
176    }
177
178    /// Flushes the underlying writer.
179    ///
180    /// # Errors
181    ///
182    /// [`LogError::Io`] when the flush fails.
183    pub fn flush(&mut self) -> Result<(), LogError> {
184        Ok(self.out.flush()?)
185    }
186
187    /// Flushes and gives the underlying writer back.
188    ///
189    /// # Errors
190    ///
191    /// [`LogError::Io`] when the flush fails.
192    pub fn finish(mut self) -> Result<W, LogError> {
193        self.out.flush()?;
194        Ok(self.out)
195    }
196}
197
198fn json_io(e: serde_json::Error) -> LogError {
199    LogError::Io(io::Error::other(e))
200}
201
202/// Reads envelopes back from any [`BufRead`], in the order they were written.
203///
204/// An iterator of `Result<Envelope, LogError>`. After the first error the iterator ends: what
205/// follows a bad line cannot be trusted to be in order.
206///
207/// # Examples
208///
209/// See the module [`record`](crate::record).
210pub struct EnvelopeReader<R: BufRead> {
211    input: R,
212    line: u64,
213    failed: bool,
214    buf: String,
215}
216
217impl<R: BufRead> EnvelopeReader<R> {
218    /// Opens a log: reads and checks the header line.
219    ///
220    /// # Errors
221    ///
222    /// - [`LogError::Format`] when the first line is not a log header (an empty file included).
223    /// - [`LogError::Schema`] when the file is of a newer format or holds newer envelopes.
224    /// - [`LogError::Io`] when the read fails.
225    pub fn new(mut input: R) -> Result<Self, LogError> {
226        let mut first = String::new();
227        if input.read_line(&mut first)? == 0 {
228            return Err(LogError::Format {
229                line: 1,
230                message: "the file is empty, a log starts with a header line".into(),
231            });
232        }
233        let header: Header =
234            serde_json::from_str(first.trim_end()).map_err(|e| LogError::Format {
235                line: 1,
236                message: format!("not a log header: {e}"),
237            })?;
238        if header.brokapi_log > LOG_FORMAT {
239            return Err(LogError::Schema {
240                found: header.brokapi_log,
241                supported: LOG_FORMAT,
242            });
243        }
244        if header.envelope_schema > ENVELOPE_SCHEMA {
245            return Err(LogError::Schema {
246                found: u32::from(header.envelope_schema),
247                supported: u32::from(ENVELOPE_SCHEMA),
248            });
249        }
250        Ok(Self {
251            input,
252            line: 1,
253            failed: false,
254            buf: String::new(),
255        })
256    }
257}
258
259impl<R: BufRead> Iterator for EnvelopeReader<R> {
260    type Item = Result<Envelope, LogError>;
261
262    fn next(&mut self) -> Option<Self::Item> {
263        if self.failed {
264            return None;
265        }
266        loop {
267            self.buf.clear();
268            match self.input.read_line(&mut self.buf) {
269                Ok(0) => return None,
270                Ok(_) => {}
271                Err(e) => {
272                    self.failed = true;
273                    return Some(Err(e.into()));
274                }
275            }
276            self.line += 1;
277            let text = self.buf.trim_end();
278            if text.is_empty() {
279                continue;
280            }
281            return Some(match serde_json::from_str::<Envelope>(text) {
282                Ok(env) if env.schema > ENVELOPE_SCHEMA => {
283                    self.failed = true;
284                    Err(LogError::Schema {
285                        found: u32::from(env.schema),
286                        supported: u32::from(ENVELOPE_SCHEMA),
287                    })
288                }
289                Ok(env) => Ok(env),
290                Err(e) => {
291                    self.failed = true;
292                    Err(LogError::Format {
293                        line: self.line,
294                        message: e.to_string(),
295                    })
296                }
297            });
298        }
299    }
300}
301
302/// How fast [`replay`] plays.
303///
304/// # Examples
305///
306/// ```
307/// use brokapi_runtime::record::ReplaySpeed;
308///
309/// assert_eq!(ReplaySpeed::default(), ReplaySpeed::Instant);
310/// ```
311#[derive(Debug, Clone, Copy, PartialEq, Default)]
312pub enum ReplaySpeed {
313    /// As fast as the consumer reads. For backtests and tests.
314    #[default]
315    Instant,
316    /// At the recorded pace divided by `factor`: `1.0` is real time, `10.0` is ten times faster.
317    /// The pace is the difference of `rx_ts` of neighbouring envelopes. A pause is never longer
318    /// than `max_gap`, so a night between two sessions does not stop the replay.
319    Paced {
320        /// The speed-up. Must be above zero.
321        factor: f64,
322        /// The longest single pause.
323        max_gap: Duration,
324    },
325}
326
327/// What a finished [`replay`] did.
328///
329/// # Examples
330///
331/// ```
332/// use brokapi_runtime::record::ReplayReport;
333///
334/// let r = ReplayReport::default();
335/// assert_eq!((r.sent, r.stopped_by_consumer), (0, false));
336/// ```
337#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
338pub struct ReplayReport {
339    /// Envelopes sent into the channel.
340    pub sent: u64,
341    /// `true` when the replay ended because the receiver was dropped, not because the log ended.
342    pub stopped_by_consumer: bool,
343}
344
345/// Sends envelopes into a channel, in order, like a connector would.
346///
347/// The difference from a live connector is the one that matters for research: `replay` waits for
348/// room in the channel ([`mpsc::Sender::send`]), so a slow consumer slows the replay down and no
349/// event is lost. A live connector cannot wait and applies the loss policy of the subscription
350/// instead. Code that handles `Overflow` therefore needs its own test: a replay never produces one.
351///
352/// The envelopes keep their `source` and `sub`, so the consumer routes them as it did live.
353///
354/// # Errors
355///
356/// The first [`LogError`] from `envelopes`. Envelopes sent before it stay sent.
357///
358/// # Examples
359///
360/// ```
361/// use std::sync::Arc;
362/// use std::time::Instant;
363/// use brokapi_core::{Envelope, Event, Timestamps, UnixNanos};
364/// use brokapi_runtime::record::{ReplaySpeed, replay};
365///
366/// # #[tokio::main(flavor = "current_thread")]
367/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
368/// let ts = |n| Timestamps::new(UnixNanos::from_nanos(n), Instant::now());
369/// let log = vec![
370///     Envelope::new(Arc::from("bcs-md"), ts(1), Event::Balances(vec![])),
371///     Envelope::new(Arc::from("bcs-md"), ts(2), Event::Balances(vec![])),
372/// ];
373/// let (tx, mut rx) = tokio::sync::mpsc::channel(8);
374/// let report = replay(log.into_iter().map(Ok), &tx, ReplaySpeed::Instant).await?;
375/// assert_eq!(report.sent, 2);
376/// assert!(rx.recv().await.is_some());
377/// # Ok(()) }
378/// ```
379pub async fn replay<I>(
380    envelopes: I,
381    tx: &mpsc::Sender<Envelope>,
382    speed: ReplaySpeed,
383) -> Result<ReplayReport, LogError>
384where
385    I: IntoIterator<Item = Result<Envelope, LogError>>,
386{
387    let mut report = ReplayReport::default();
388    let mut previous: Option<i64> = None;
389    for item in envelopes {
390        let env = item?;
391        if let ReplaySpeed::Paced { factor, max_gap } = speed {
392            let now = env.ts.rx_ts.as_nanos();
393            if let Some(prev) = previous {
394                let gap = now.saturating_sub(prev).max(0) as u64;
395                if factor > 0.0 && gap > 0 {
396                    let pause = Duration::from_nanos((gap as f64 / factor) as u64).min(max_gap);
397                    tokio::time::sleep(pause).await;
398                }
399            }
400            previous = Some(now);
401        }
402        if tx.send(env).await.is_err() {
403            report.stopped_by_consumer = true;
404            return Ok(report);
405        }
406        report.sent += 1;
407    }
408    Ok(report)
409}
410
411#[cfg(test)]
412mod tests {
413    use std::sync::Arc;
414    use std::time::Instant;
415
416    use brokapi_core::{
417        BookLevel, Event, InstrumentId, InstrumentKey, NativeId, OrderBook, Price, Qty,
418        SubscriptionId, Timestamps, UnixNanos,
419    };
420
421    use super::*;
422
423    fn instrument() -> InstrumentId {
424        InstrumentId::new(
425            InstrumentKey::new("SBER", "TQBR"),
426            NativeId::TickerClass {
427                ticker: "SBER".into(),
428                class_code: "TQBR".into(),
429            },
430        )
431    }
432
433    fn book_env(n: i64, price: &str) -> Envelope {
434        let ts = Timestamps::new(UnixNanos::from_nanos(n), Instant::now())
435            .with_broker_ts(UnixNanos::from_nanos(n - 1));
436        let level = BookLevel {
437            price: price.parse::<Price>().unwrap(),
438            qty: Qty::from(10),
439        };
440        Envelope::new(
441            Arc::from("bcs-md"),
442            ts,
443            Event::OrderBook(OrderBook::Snapshot {
444                instrument: instrument(),
445                bids: vec![level.clone()],
446                asks: vec![],
447                ts,
448            }),
449        )
450        .with_sub(SubscriptionId::new(3))
451    }
452
453    fn write_all(envs: &[Envelope]) -> Vec<u8> {
454        let mut w = EnvelopeWriter::new(Vec::new()).unwrap();
455        for e in envs {
456            w.write(e).unwrap();
457        }
458        assert_eq!(w.written(), envs.len() as u64);
459        w.finish().unwrap()
460    }
461
462    #[test]
463    fn a_round_trip_keeps_every_field_and_the_exact_decimals() {
464        let envs = vec![book_env(10, "244.30"), book_env(20, "0.0001")];
465        let bytes = write_all(&envs);
466        let back: Vec<Envelope> = EnvelopeReader::new(bytes.as_slice())
467            .unwrap()
468            .collect::<Result<_, _>>()
469            .unwrap();
470        assert_eq!(back, envs);
471        // 244.30 is not 244.3 as text: the trailing zero survives.
472        let text = String::from_utf8(bytes).unwrap();
473        assert!(text.contains("\"244.30\""), "{text}");
474    }
475
476    #[test]
477    fn a_log_starts_with_a_header() {
478        let bytes = write_all(&[]);
479        let first = String::from_utf8(bytes).unwrap();
480        assert_eq!(first.trim_end(), r#"{"brokapi_log":1,"envelope_schema":1}"#);
481    }
482
483    #[test]
484    fn an_empty_file_and_a_foreign_file_are_refused() {
485        let empty = EnvelopeReader::new(&b""[..]).err().unwrap();
486        assert!(matches!(empty, LogError::Format { line: 1, .. }));
487        let foreign = EnvelopeReader::new(&b"{\"a\":1}\n"[..]).err().unwrap();
488        assert!(matches!(foreign, LogError::Format { line: 1, .. }));
489    }
490
491    #[test]
492    fn a_newer_format_or_schema_is_refused() {
493        let newer_log = br#"{"brokapi_log":2,"envelope_schema":1}"#;
494        let err = EnvelopeReader::new(&newer_log[..]).err().unwrap();
495        assert!(matches!(
496            err,
497            LogError::Schema {
498                found: 2,
499                supported: 1
500            }
501        ));
502        let newer_env = br#"{"brokapi_log":1,"envelope_schema":9}"#;
503        assert!(matches!(
504            EnvelopeReader::new(&newer_env[..]).err().unwrap(),
505            LogError::Schema { found: 9, .. }
506        ));
507    }
508
509    #[test]
510    fn a_bad_line_stops_the_read_and_names_the_line() {
511        let mut bytes = write_all(&[book_env(1, "1")]);
512        bytes.extend_from_slice(b"{broken\n");
513        bytes.extend_from_slice(&write_all(&[book_env(2, "2")])[39..]);
514        let mut reader = EnvelopeReader::new(bytes.as_slice()).unwrap();
515        assert!(reader.next().unwrap().is_ok());
516        let err = reader.next().unwrap().unwrap_err();
517        assert!(matches!(err, LogError::Format { line: 3, .. }), "{err}");
518        assert!(reader.next().is_none(), "nothing after a bad line");
519    }
520
521    #[test]
522    fn blank_lines_are_skipped() {
523        let mut bytes = write_all(&[book_env(1, "1")]);
524        bytes.extend_from_slice(b"\n\n");
525        let n = EnvelopeReader::new(bytes.as_slice()).unwrap().count();
526        assert_eq!(n, 1);
527    }
528
529    #[tokio::test]
530    async fn replay_sends_in_order_and_waits_for_a_slow_consumer() {
531        let envs: Vec<Envelope> = (1..=50).map(|n| book_env(n, "1")).collect();
532        let bytes = write_all(&envs);
533        let (tx, mut rx) = mpsc::channel(2); // far smaller than the log
534        let reader = EnvelopeReader::new(bytes.as_slice()).unwrap();
535        let sender = async { replay(reader, &tx, ReplaySpeed::Instant).await };
536        let receiver = async {
537            let mut got = Vec::new();
538            while got.len() < 50 {
539                got.push(rx.recv().await.unwrap());
540                tokio::task::yield_now().await;
541            }
542            got
543        };
544        let (report, got) = tokio::join!(sender, receiver);
545        assert_eq!(report.unwrap().sent, 50);
546        assert_eq!(got, envs, "nothing lost, order kept");
547    }
548
549    #[tokio::test]
550    async fn replay_stops_quietly_when_the_consumer_goes_away() {
551        let envs: Vec<Envelope> = (1..=5).map(|n| book_env(n, "1")).collect();
552        let (tx, rx) = mpsc::channel(1);
553        drop(rx);
554        let report = replay(envs.into_iter().map(Ok), &tx, ReplaySpeed::Instant)
555            .await
556            .unwrap();
557        assert!(report.stopped_by_consumer);
558        assert_eq!(report.sent, 0);
559    }
560
561    #[tokio::test]
562    async fn replay_passes_a_read_error_on() {
563        let items: Vec<Result<Envelope, LogError>> = vec![
564            Ok(book_env(1, "1")),
565            Err(LogError::Format {
566                line: 3,
567                message: "x".into(),
568            }),
569        ];
570        let (tx, mut rx) = mpsc::channel(8);
571        let err = replay(items, &tx, ReplaySpeed::Instant).await.unwrap_err();
572        assert!(matches!(err, LogError::Format { line: 3, .. }));
573        assert!(rx.recv().await.is_some(), "what came before is sent");
574    }
575
576    #[tokio::test]
577    async fn paced_replay_waits_and_cuts_long_gaps() {
578        // 200 ms recorded, 100x faster: about 2 ms. A gap of one hour is cut to max_gap.
579        let envs = vec![
580            book_env(0, "1"),
581            book_env(200_000_000, "1"),
582            book_env(3_600_000_000_000, "1"),
583        ];
584        let (tx, mut rx) = mpsc::channel(8);
585        let started = Instant::now();
586        let speed = ReplaySpeed::Paced {
587            factor: 100.0,
588            max_gap: Duration::from_millis(30),
589        };
590        replay(envs.into_iter().map(Ok), &tx, speed).await.unwrap();
591        let took = started.elapsed();
592        assert!(
593            took >= Duration::from_millis(30),
594            "the long gap waits: {took:?}"
595        );
596        assert!(took < Duration::from_secs(2), "but is cut: {took:?}");
597        let mut n = 0;
598        while rx.try_recv().is_ok() {
599            n += 1;
600        }
601        assert_eq!(n, 3);
602    }
603}
brokapi_runtime/record.rs

brokapi_runtime/
record.rs
