nautilus_event_store/markers/
marker.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Durable schema types and canonical content hashes for the data marker sidecar.
17
18use std::fmt::Display;
19
20use nautilus_core::UnixNanos;
21use serde::{Deserialize, Serialize};
22
23use crate::wire;
24
25const MARKER_HASH_DOMAIN: &[u8] = b"nautilus-event-store/marker/v1";
26const HIFI_HASH_DOMAIN: &[u8] = b"nautilus-event-store/hifi/v1";
27const DICT_HASH_DOMAIN: &[u8] = b"nautilus-event-store/dict/v1";
28const GAP_HASH_DOMAIN: &[u8] = b"nautilus-event-store/gap/v1";
29
30/// The class of market-data stream being tracked by a sidecar slot.
31#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
32pub enum DataClass {
33    /// Order-book delta stream.
34    BookDeltas,
35    /// Level-10 order-book snapshot stream.
36    BookDepth10,
37    /// Quote (level-1 bid/ask) stream.
38    Quote,
39    /// Trade (last sale) stream.
40    Trade,
41    /// Bar (OHLCV aggregate) stream.
42    Bar,
43}
44
45impl DataClass {
46    /// Returns the canonical string representation of this data class.
47    #[must_use]
48    pub const fn as_str(self) -> &'static str {
49        match self {
50            Self::BookDeltas => "BookDeltas",
51            Self::BookDepth10 => "BookDepth10",
52            Self::Quote => "Quote",
53            Self::Trade => "Trade",
54            Self::Bar => "Bar",
55        }
56    }
57}
58
59impl Display for DataClass {
60    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
61        f.write_str(self.as_str())
62    }
63}
64
65impl std::str::FromStr for DataClass {
66    type Err = String;
67
68    fn from_str(s: &str) -> Result<Self, Self::Err> {
69        match s {
70            "BookDeltas" => Ok(Self::BookDeltas),
71            "BookDepth10" => Ok(Self::BookDepth10),
72            "Quote" => Ok(Self::Quote),
73            "Trade" => Ok(Self::Trade),
74            "Bar" => Ok(Self::Bar),
75            other => Err(format!("unknown DataClass, was `{other}`")),
76        }
77    }
78}
79
80/// A slot index identifying a registered market-data stream.
81pub type StreamSlot = u32;
82
83/// The cursor position within a single market-data stream slot.
84#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
85pub struct StreamCursor {
86    /// The stream slot index.
87    pub slot: StreamSlot,
88    /// The highest `ts_init` observed so far in this slot.
89    #[serde(with = "wire::nanos_as_u64")]
90    pub ts_init_hi: UnixNanos,
91    /// The number of records observed so far in this slot.
92    pub count: u64,
93}
94
95/// A snapshot of the cursor positions for all active market-data streams at a marker point.
96#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
97pub struct DataCursorSnapshot {
98    /// Monotonic sequence of this marker within the sidecar.
99    pub marker_seq: u64,
100    /// The event-store sequence before which this snapshot was taken.
101    pub event_seq_before: u64,
102    /// The `ts_init` at the point this snapshot was taken.
103    #[serde(with = "wire::nanos_as_u64")]
104    pub ts_init: UnixNanos,
105    /// The cursors for every stream slot that advanced since the previous snapshot.
106    pub advanced: Vec<StreamCursor>,
107}
108
109/// A high-fidelity per-record marker capturing per-record identity within a stream slot.
110#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
111pub struct HiFiMarker {
112    /// Monotonic sequence of this marker within the sidecar.
113    pub marker_seq: u64,
114    /// The event-store sequence before which this marker was recorded.
115    pub event_seq_before: u64,
116    /// The stream slot index for this record.
117    pub slot: StreamSlot,
118    /// The domain timestamp of the record (`ts_event`).
119    #[serde(with = "wire::nanos_as_u64")]
120    pub ts_event: UnixNanos,
121    /// The ingestion timestamp of the record (`ts_init`).
122    #[serde(with = "wire::nanos_as_u64")]
123    pub ts_init: UnixNanos,
124    /// Ordinal among records sharing the same `ts_init` within a slot.
125    pub same_ts_ordinal: u32,
126    /// A 32-byte fingerprint of the record's content.
127    pub record_fingerprint: [u8; 32],
128}
129
130/// The reason a gap exists in the sidecar marker sequence.
131#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
132pub enum MarkerGapReason {
133    /// The marker ring-buffer overflowed; some markers were dropped.
134    Overflow,
135    /// The marker writer was closed before flushing.
136    WriterClosed,
137}
138
139/// A gap in the sidecar marker sequence.
140#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
141pub struct MarkerGap {
142    /// The first marker sequence number missing from the sequence.
143    pub from_marker_seq: u64,
144    /// The last marker sequence number missing from the sequence.
145    pub to_marker_seq: u64,
146    /// The reason this gap was recorded.
147    pub reason: MarkerGapReason,
148}
149
150/// A registry entry mapping a stream slot to its data class and instrument identifier.
151#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
152pub struct StreamDictEntry {
153    /// The stream slot index.
154    pub slot: StreamSlot,
155    /// The data class of this stream.
156    pub data_cls: DataClass,
157    /// The instrument identifier string for this stream.
158    pub identifier: String,
159}
160
161/// Computes the canonical BLAKE3 hash of a [`DataCursorSnapshot`].
162///
163/// The hash is domain-separated by a crate-internal prefix, writes numeric fields big-endian
164/// in declared order, and length-prefixes the variable-length cursor list so two distinct
165/// snapshots cannot frame to the same byte stream. Store the returned bytes alongside the
166/// record; do not add a hash field to the struct itself.
167#[must_use]
168pub fn compute_marker_hash(snapshot: &DataCursorSnapshot) -> [u8; 32] {
169    let mut hasher = blake3::Hasher::new();
170    hasher.update(MARKER_HASH_DOMAIN);
171    hasher.update(&snapshot.marker_seq.to_be_bytes());
172    hasher.update(&snapshot.event_seq_before.to_be_bytes());
173    hasher.update(&snapshot.ts_init.as_u64().to_be_bytes());
174    hasher.update(&(snapshot.advanced.len() as u64).to_be_bytes());
175    for cursor in &snapshot.advanced {
176        hasher.update(&cursor.slot.to_be_bytes());
177        hasher.update(&cursor.ts_init_hi.as_u64().to_be_bytes());
178        hasher.update(&cursor.count.to_be_bytes());
179    }
180    *hasher.finalize().as_bytes()
181}
182
183/// Computes the canonical BLAKE3 hash of a [`HiFiMarker`].
184///
185/// The hash is domain-separated by a crate-internal prefix and uses big-endian fixed-width
186/// framing for every field. Store the returned bytes alongside the record; do not add a hash
187/// field to the struct itself.
188#[must_use]
189pub fn compute_hifi_hash(marker: &HiFiMarker) -> [u8; 32] {
190    let mut hasher = blake3::Hasher::new();
191    hasher.update(HIFI_HASH_DOMAIN);
192    hasher.update(&marker.marker_seq.to_be_bytes());
193    hasher.update(&marker.event_seq_before.to_be_bytes());
194    hasher.update(&marker.slot.to_be_bytes());
195    hasher.update(&marker.ts_event.as_u64().to_be_bytes());
196    hasher.update(&marker.ts_init.as_u64().to_be_bytes());
197    hasher.update(&marker.same_ts_ordinal.to_be_bytes());
198    hasher.update(&marker.record_fingerprint);
199    *hasher.finalize().as_bytes()
200}
201
202/// Computes the canonical BLAKE3 hash of a [`StreamDictEntry`].
203///
204/// The hash is domain-separated by a crate-internal prefix, writes the numeric `slot`
205/// big-endian, and length-prefixes the data-class and identifier strings so a slot remapped to
206/// a different class or identifier cannot frame to the same byte stream. Store the returned
207/// bytes alongside the record; do not add a hash field to the struct itself.
208#[must_use]
209pub fn compute_dict_hash(entry: &StreamDictEntry) -> [u8; 32] {
210    let mut hasher = blake3::Hasher::new();
211    hasher.update(DICT_HASH_DOMAIN);
212    hasher.update(&entry.slot.to_be_bytes());
213    let class = entry.data_cls.as_str().as_bytes();
214    hasher.update(&(class.len() as u64).to_be_bytes());
215    hasher.update(class);
216    let identifier = entry.identifier.as_bytes();
217    hasher.update(&(identifier.len() as u64).to_be_bytes());
218    hasher.update(identifier);
219    *hasher.finalize().as_bytes()
220}
221
222/// Computes the canonical BLAKE3 hash of a [`MarkerGap`].
223///
224/// The hash is domain-separated by a crate-internal prefix and uses big-endian fixed-width
225/// framing for the sequence bounds plus a one-byte reason discriminant. Store the returned bytes
226/// alongside the record; do not add a hash field to the struct itself.
227#[must_use]
228pub fn compute_gap_hash(gap: &MarkerGap) -> [u8; 32] {
229    let mut hasher = blake3::Hasher::new();
230    hasher.update(GAP_HASH_DOMAIN);
231    hasher.update(&gap.from_marker_seq.to_be_bytes());
232    hasher.update(&gap.to_marker_seq.to_be_bytes());
233    let reason = match gap.reason {
234        MarkerGapReason::Overflow => 0u8,
235        MarkerGapReason::WriterClosed => 1u8,
236    };
237    hasher.update(&[reason]);
238    *hasher.finalize().as_bytes()
239}
240
241#[cfg(test)]
242mod tests {
243    use std::{fmt::Write, str::FromStr};
244
245    use rstest::rstest;
246
247    use super::*;
248
249    #[rstest]
250    fn data_class_roundtrips_to_str() {
251        let variants = [
252            (DataClass::BookDeltas, "BookDeltas"),
253            (DataClass::BookDepth10, "BookDepth10"),
254            (DataClass::Quote, "Quote"),
255            (DataClass::Trade, "Trade"),
256            (DataClass::Bar, "Bar"),
257        ];
258
259        for (variant, expected) in variants {
260            assert_eq!(variant.as_str(), expected, "as_str for {variant:?}");
261            assert_eq!(variant.to_string(), expected, "Display for {variant:?}");
262            assert_eq!(
263                DataClass::from_str(expected).unwrap(),
264                variant,
265                "from_str for {expected}"
266            );
267        }
268    }
269
270    fn baseline_snapshot() -> DataCursorSnapshot {
271        DataCursorSnapshot {
272            marker_seq: 1,
273            event_seq_before: 42,
274            ts_init: UnixNanos::from(1_700_000_000_000_000_000),
275            advanced: vec![
276                StreamCursor {
277                    slot: 0,
278                    ts_init_hi: UnixNanos::from(1_700_000_000_000_000_001),
279                    count: 7,
280                },
281                StreamCursor {
282                    slot: 1,
283                    ts_init_hi: UnixNanos::from(1_700_000_000_000_000_002),
284                    count: 3,
285                },
286            ],
287        }
288    }
289
290    fn baseline_hifi() -> HiFiMarker {
291        HiFiMarker {
292            marker_seq: 1,
293            event_seq_before: 42,
294            slot: 0,
295            ts_event: UnixNanos::from(1_700_000_000_000_000_000),
296            ts_init: UnixNanos::from(1_700_000_000_000_000_001),
297            same_ts_ordinal: 0,
298            record_fingerprint: [0xABu8; 32],
299        }
300    }
301
302    fn baseline_dict() -> StreamDictEntry {
303        StreamDictEntry {
304            slot: 3,
305            data_cls: DataClass::Quote,
306            identifier: "ETHUSDT.BINANCE".to_string(),
307        }
308    }
309
310    fn baseline_gap() -> MarkerGap {
311        MarkerGap {
312            from_marker_seq: 5,
313            to_marker_seq: 9,
314            reason: MarkerGapReason::Overflow,
315        }
316    }
317
318    fn hex32(bytes: &[u8; 32]) -> String {
319        let mut out = String::with_capacity(64);
320        for byte in bytes {
321            write!(out, "{byte:02x}").expect("writing to a String is infallible");
322        }
323        out
324    }
325
326    #[rstest]
327    fn marker_hash_is_deterministic() {
328        let snap = baseline_snapshot();
329        let h1 = compute_marker_hash(&snap);
330        let h2 = compute_marker_hash(&snap);
331
332        assert_eq!(h1, h2);
333
334        // Pinned wire-format vector. Any change to domain, field order, or endianness flips
335        // this value.
336        let hex = hex32(&h1);
337        assert_eq!(
338            hex, "898bc3efdaf0edd9167a38a1c3060c9b4dc051658ea2f6132004bed78a481c47",
339            "marker hash wire format changed"
340        );
341    }
342
343    #[rstest]
344    fn hifi_hash_is_deterministic() {
345        let marker = baseline_hifi();
346        let h1 = compute_hifi_hash(&marker);
347        let h2 = compute_hifi_hash(&marker);
348
349        assert_eq!(h1, h2);
350
351        let hex = hex32(&h1);
352        assert_eq!(
353            hex, "06542408380d8815ef783b9dbde6b3e3ffdf05605bb17e83ad48474557457517",
354            "hifi hash wire format changed"
355        );
356    }
357
358    #[rstest]
359    fn dict_hash_is_deterministic() {
360        let entry = baseline_dict();
361        let h1 = compute_dict_hash(&entry);
362        let h2 = compute_dict_hash(&entry);
363
364        assert_eq!(h1, h2);
365
366        // Pinned wire-format vector. Any change to domain, field order, or framing flips this.
367        let hex = hex32(&h1);
368        assert_eq!(
369            hex, "24e702c5ae20b832ad6907676919fa18a89b79e97dde9df7e1de454191f42fda",
370            "dict hash wire format changed"
371        );
372    }
373
374    #[rstest]
375    fn gap_hash_is_deterministic() {
376        let gap = baseline_gap();
377        let h1 = compute_gap_hash(&gap);
378        let h2 = compute_gap_hash(&gap);
379
380        assert_eq!(h1, h2);
381
382        // Pinned wire-format vector. Any change to domain, field order, or framing flips this.
383        let hex = hex32(&h1);
384        assert_eq!(
385            hex, "ec1ae0ea813e9971155c6277e95c43de72da6f22ca1832f072aadd9b91f5a3ec",
386            "gap hash wire format changed"
387        );
388    }
389
390    #[rstest]
391    fn marker_record_codec_roundtrip() {
392        // DataCursorSnapshot
393        let snap = baseline_snapshot();
394        let bytes = crate::codec::encode_to_vec(&snap).expect("encode");
395        let decoded =
396            crate::codec::decode_from_slice::<DataCursorSnapshot>(&bytes).expect("decode");
397        assert_eq!(snap, decoded);
398
399        // HiFiMarker
400        let hifi = baseline_hifi();
401        let bytes = crate::codec::encode_to_vec(&hifi).expect("encode");
402        let decoded = crate::codec::decode_from_slice::<HiFiMarker>(&bytes).expect("decode");
403        assert_eq!(hifi, decoded);
404
405        // MarkerGap
406        let gap = MarkerGap {
407            from_marker_seq: 5,
408            to_marker_seq: 10,
409            reason: MarkerGapReason::Overflow,
410        };
411        let bytes = crate::codec::encode_to_vec(&gap).expect("encode");
412        let decoded = crate::codec::decode_from_slice::<MarkerGap>(&bytes).expect("decode");
413        assert_eq!(gap, decoded);
414
415        // StreamDictEntry
416        let dict = StreamDictEntry {
417            slot: 2,
418            data_cls: DataClass::Bar,
419            identifier: "BTCUSDT-PERP.BINANCE".to_string(),
420        };
421        let bytes = crate::codec::encode_to_vec(&dict).expect("encode");
422        let decoded = crate::codec::decode_from_slice::<StreamDictEntry>(&bytes).expect("decode");
423        assert_eq!(dict, decoded);
424    }
425
426    #[rstest]
427    #[case::quote_lowercase("quote")]
428    #[case::empty("")]
429    #[case::trailing_s("Quotes")]
430    #[case::partial("BookDepth")]
431    fn data_class_from_str_rejects_unknown(#[case] input: &str) {
432        let err = DataClass::from_str(input).unwrap_err();
433
434        assert!(
435            err.contains(input),
436            "error should name the rejected input, was `{err}`"
437        );
438    }
439
440    #[rstest]
441    #[case::marker_seq(|s: &mut DataCursorSnapshot| s.marker_seq = 99)]
442    #[case::event_seq_before(|s: &mut DataCursorSnapshot| s.event_seq_before = 99)]
443    #[case::ts_init(|s: &mut DataCursorSnapshot| s.ts_init = UnixNanos::from(1))]
444    #[case::cursor_slot(|s: &mut DataCursorSnapshot| s.advanced[0].slot = 256)]
445    #[case::cursor_ts_init_hi(|s: &mut DataCursorSnapshot| s.advanced[0].ts_init_hi = UnixNanos::from(1))]
446    #[case::cursor_count(|s: &mut DataCursorSnapshot| s.advanced[0].count = 999)]
447    #[case::extra_cursor(|s: &mut DataCursorSnapshot| s.advanced.push(StreamCursor { slot: 2, ts_init_hi: UnixNanos::from(1_700_000_000_000_000_003), count: 1 }))]
448    #[case::cursor_order(|s: &mut DataCursorSnapshot| s.advanced.reverse())]
449    fn every_marker_field_affects_hash(#[case] mutate: fn(&mut DataCursorSnapshot)) {
450        let base = baseline_snapshot();
451        let mut mutated = base.clone();
452        mutate(&mut mutated);
453
454        assert_ne!(compute_marker_hash(&base), compute_marker_hash(&mutated));
455    }
456
457    #[rstest]
458    #[case::marker_seq(|m: &mut HiFiMarker| m.marker_seq = 99)]
459    #[case::event_seq_before(|m: &mut HiFiMarker| m.event_seq_before = 99)]
460    #[case::slot(|m: &mut HiFiMarker| m.slot = 256)]
461    #[case::ts_event(|m: &mut HiFiMarker| m.ts_event = UnixNanos::from(1))]
462    #[case::ts_init(|m: &mut HiFiMarker| m.ts_init = UnixNanos::from(1))]
463    #[case::same_ts_ordinal(|m: &mut HiFiMarker| m.same_ts_ordinal = 256)]
464    #[case::fingerprint(|m: &mut HiFiMarker| m.record_fingerprint[0] ^= 0x01)]
465    fn every_hifi_field_affects_hash(#[case] mutate: fn(&mut HiFiMarker)) {
466        let base = baseline_hifi();
467        let mut mutated = base.clone();
468        mutate(&mut mutated);
469
470        assert_ne!(compute_hifi_hash(&base), compute_hifi_hash(&mutated));
471    }
472
473    #[rstest]
474    #[case::slot(|e: &mut StreamDictEntry| e.slot = 99)]
475    #[case::data_cls(|e: &mut StreamDictEntry| e.data_cls = DataClass::Trade)]
476    #[case::identifier(|e: &mut StreamDictEntry| e.identifier = "BTCUSDT.BINANCE".to_string())]
477    fn every_dict_field_affects_hash(#[case] mutate: fn(&mut StreamDictEntry)) {
478        let base = baseline_dict();
479        let mut mutated = base.clone();
480        mutate(&mut mutated);
481
482        assert_ne!(compute_dict_hash(&base), compute_dict_hash(&mutated));
483    }
484
485    #[rstest]
486    #[case::from(|g: &mut MarkerGap| g.from_marker_seq = 99)]
487    #[case::to(|g: &mut MarkerGap| g.to_marker_seq = 99)]
488    #[case::reason(|g: &mut MarkerGap| g.reason = MarkerGapReason::WriterClosed)]
489    fn every_gap_field_affects_hash(#[case] mutate: fn(&mut MarkerGap)) {
490        let base = baseline_gap();
491        let mut mutated = base.clone();
492        mutate(&mut mutated);
493
494        assert_ne!(compute_gap_hash(&base), compute_gap_hash(&mutated));
495    }
496
497    #[rstest]
498    fn marker_hash_handles_empty_advanced() {
499        let empty = DataCursorSnapshot {
500            marker_seq: 1,
501            event_seq_before: 42,
502            ts_init: UnixNanos::from(1_700_000_000_000_000_000),
503            advanced: vec![],
504        };
505
506        assert_eq!(compute_marker_hash(&empty), compute_marker_hash(&empty));
507        assert_ne!(
508            compute_marker_hash(&empty),
509            compute_marker_hash(&baseline_snapshot())
510        );
511    }
512}
nautilus_event_store/markers/marker.rs

nautilus_event_store/markers/
marker.rs
