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    /// Order-book depth snapshot stream.
36    #[serde(alias = "BookDepth10")]
37    BookDepth,
38    /// Quote (level-1 bid/ask) stream.
39    Quote,
40    /// Trade (last sale) stream.
41    Trade,
42    /// Bar (OHLCV aggregate) stream.
43    Bar,
44}
45
46impl DataClass {
47    /// Returns the canonical string representation of this data class.
48    #[must_use]
49    pub const fn as_str(self) -> &'static str {
50        match self {
51            Self::BookDeltas => "BookDeltas",
52            Self::BookDepth => "BookDepth",
53            Self::Quote => "Quote",
54            Self::Trade => "Trade",
55            Self::Bar => "Bar",
56        }
57    }
58
59    /// Returns the stable token hashed into [`compute_dict_hash`].
60    ///
61    /// The depth token predates the canonical rename; keeping it fixed means hashes recorded by
62    /// pre-rename builds keep verifying.
63    const fn persisted_hash_token(self) -> &'static str {
64        match self {
65            Self::BookDeltas => "BookDeltas",
66            Self::BookDepth => "BookDepth10",
67            Self::Quote => "Quote",
68            Self::Trade => "Trade",
69            Self::Bar => "Bar",
70        }
71    }
72}
73
74impl Display for DataClass {
75    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
76        f.write_str(self.as_str())
77    }
78}
79
80impl std::str::FromStr for DataClass {
81    type Err = String;
82
83    fn from_str(s: &str) -> Result<Self, Self::Err> {
84        match s {
85            "BookDeltas" => Ok(Self::BookDeltas),
86            // Legacy spelling written by markers recorded before the canonical rename
87            "BookDepth" | "BookDepth10" => Ok(Self::BookDepth),
88            "Quote" => Ok(Self::Quote),
89            "Trade" => Ok(Self::Trade),
90            "Bar" => Ok(Self::Bar),
91            other => Err(format!("unknown DataClass, was `{other}`")),
92        }
93    }
94}
95
96/// A slot index identifying a registered market-data stream.
97pub type StreamSlot = u32;
98
99/// The cursor position within a single market-data stream slot.
100#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
101pub struct StreamCursor {
102    /// The stream slot index.
103    pub slot: StreamSlot,
104    /// The highest `ts_init` observed so far in this slot.
105    #[serde(with = "wire::nanos_as_u64")]
106    pub ts_init_hi: UnixNanos,
107    /// The number of records observed so far in this slot.
108    pub count: u64,
109}
110
111/// A snapshot of the cursor positions for all active market-data streams at a marker point.
112#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
113pub struct DataCursorSnapshot {
114    /// Monotonic sequence of this marker within the sidecar.
115    pub marker_seq: u64,
116    /// The event-store sequence before which this snapshot was taken.
117    pub event_seq_before: u64,
118    /// The `ts_init` at the point this snapshot was taken.
119    #[serde(with = "wire::nanos_as_u64")]
120    pub ts_init: UnixNanos,
121    /// The cursors for every stream slot that advanced since the previous snapshot.
122    pub advanced: Vec<StreamCursor>,
123}
124
125/// A high-fidelity per-record marker capturing per-record identity within a stream slot.
126#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
127pub struct HiFiMarker {
128    /// Monotonic sequence of this marker within the sidecar.
129    pub marker_seq: u64,
130    /// The event-store sequence before which this marker was recorded.
131    pub event_seq_before: u64,
132    /// The stream slot index for this record.
133    pub slot: StreamSlot,
134    /// The domain timestamp of the record (`ts_event`).
135    #[serde(with = "wire::nanos_as_u64")]
136    pub ts_event: UnixNanos,
137    /// The ingestion timestamp of the record (`ts_init`).
138    #[serde(with = "wire::nanos_as_u64")]
139    pub ts_init: UnixNanos,
140    /// Ordinal among records sharing the same `ts_init` within a slot.
141    pub same_ts_ordinal: u32,
142    /// A 32-byte fingerprint of the record's content.
143    pub record_fingerprint: [u8; 32],
144}
145
146/// The reason a gap exists in the sidecar marker sequence.
147#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
148pub enum MarkerGapReason {
149    /// The marker ring-buffer overflowed; some markers were dropped.
150    Overflow,
151    /// The marker writer was closed before flushing.
152    WriterClosed,
153}
154
155/// A gap in the sidecar marker sequence.
156#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
157pub struct MarkerGap {
158    /// The first marker sequence number missing from the sequence.
159    pub from_marker_seq: u64,
160    /// The last marker sequence number missing from the sequence.
161    pub to_marker_seq: u64,
162    /// The reason this gap was recorded.
163    pub reason: MarkerGapReason,
164}
165
166/// A registry entry mapping a stream slot to its data class and instrument identifier.
167#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
168pub struct StreamDictEntry {
169    /// The stream slot index.
170    pub slot: StreamSlot,
171    /// The data class of this stream.
172    pub data_cls: DataClass,
173    /// The instrument identifier string for this stream.
174    pub identifier: String,
175}
176
177/// Computes the canonical BLAKE3 hash of a [`DataCursorSnapshot`].
178///
179/// The hash is domain-separated by a crate-internal prefix, writes numeric fields big-endian
180/// in declared order, and length-prefixes the variable-length cursor list so two distinct
181/// snapshots cannot frame to the same byte stream. Store the returned bytes alongside the
182/// record; do not add a hash field to the struct itself.
183#[must_use]
184pub fn compute_marker_hash(snapshot: &DataCursorSnapshot) -> [u8; 32] {
185    let mut hasher = blake3::Hasher::new();
186    hasher.update(MARKER_HASH_DOMAIN);
187    hasher.update(&snapshot.marker_seq.to_be_bytes());
188    hasher.update(&snapshot.event_seq_before.to_be_bytes());
189    hasher.update(&snapshot.ts_init.as_u64().to_be_bytes());
190    hasher.update(&(snapshot.advanced.len() as u64).to_be_bytes());
191    for cursor in &snapshot.advanced {
192        hasher.update(&cursor.slot.to_be_bytes());
193        hasher.update(&cursor.ts_init_hi.as_u64().to_be_bytes());
194        hasher.update(&cursor.count.to_be_bytes());
195    }
196    *hasher.finalize().as_bytes()
197}
198
199/// Computes the canonical BLAKE3 hash of a [`HiFiMarker`].
200///
201/// The hash is domain-separated by a crate-internal prefix and uses big-endian fixed-width
202/// framing for every field. Store the returned bytes alongside the record; do not add a hash
203/// field to the struct itself.
204#[must_use]
205pub fn compute_hifi_hash(marker: &HiFiMarker) -> [u8; 32] {
206    let mut hasher = blake3::Hasher::new();
207    hasher.update(HIFI_HASH_DOMAIN);
208    hasher.update(&marker.marker_seq.to_be_bytes());
209    hasher.update(&marker.event_seq_before.to_be_bytes());
210    hasher.update(&marker.slot.to_be_bytes());
211    hasher.update(&marker.ts_event.as_u64().to_be_bytes());
212    hasher.update(&marker.ts_init.as_u64().to_be_bytes());
213    hasher.update(&marker.same_ts_ordinal.to_be_bytes());
214    hasher.update(&marker.record_fingerprint);
215    *hasher.finalize().as_bytes()
216}
217
218/// Computes the canonical BLAKE3 hash of a [`StreamDictEntry`].
219///
220/// The hash is domain-separated by a crate-internal prefix, writes the numeric `slot`
221/// big-endian, and length-prefixes the data-class and identifier strings so a slot remapped to
222/// a different class or identifier cannot frame to the same byte stream. Store the returned
223/// bytes alongside the record; do not add a hash field to the struct itself.
224#[must_use]
225pub fn compute_dict_hash(entry: &StreamDictEntry) -> [u8; 32] {
226    let mut hasher = blake3::Hasher::new();
227    hasher.update(DICT_HASH_DOMAIN);
228    hasher.update(&entry.slot.to_be_bytes());
229    let class = entry.data_cls.persisted_hash_token().as_bytes();
230    hasher.update(&(class.len() as u64).to_be_bytes());
231    hasher.update(class);
232    let identifier = entry.identifier.as_bytes();
233    hasher.update(&(identifier.len() as u64).to_be_bytes());
234    hasher.update(identifier);
235    *hasher.finalize().as_bytes()
236}
237
238/// Computes the canonical BLAKE3 hash of a [`MarkerGap`].
239///
240/// The hash is domain-separated by a crate-internal prefix and uses big-endian fixed-width
241/// framing for the sequence bounds plus a one-byte reason discriminant. Store the returned bytes
242/// alongside the record; do not add a hash field to the struct itself.
243#[must_use]
244pub fn compute_gap_hash(gap: &MarkerGap) -> [u8; 32] {
245    let mut hasher = blake3::Hasher::new();
246    hasher.update(GAP_HASH_DOMAIN);
247    hasher.update(&gap.from_marker_seq.to_be_bytes());
248    hasher.update(&gap.to_marker_seq.to_be_bytes());
249    let reason = match gap.reason {
250        MarkerGapReason::Overflow => 0u8,
251        MarkerGapReason::WriterClosed => 1u8,
252    };
253    hasher.update(&[reason]);
254    *hasher.finalize().as_bytes()
255}
256
257#[cfg(test)]
258mod tests {
259    use std::{fmt::Write, str::FromStr};
260
261    use rstest::rstest;
262    use serde::Deserialize;
263
264    use super::*;
265
266    #[rstest]
267    fn data_class_roundtrips_to_str() {
268        let variants = [
269            (DataClass::BookDeltas, "BookDeltas"),
270            (DataClass::BookDepth, "BookDepth"),
271            (DataClass::Quote, "Quote"),
272            (DataClass::Trade, "Trade"),
273            (DataClass::Bar, "Bar"),
274        ];
275
276        for (variant, expected) in variants {
277            assert_eq!(variant.as_str(), expected, "as_str for {variant:?}");
278            assert_eq!(variant.to_string(), expected, "Display for {variant:?}");
279            assert_eq!(
280                DataClass::from_str(expected).unwrap(),
281                variant,
282                "from_str for {expected}"
283            );
284        }
285    }
286
287    fn baseline_snapshot() -> DataCursorSnapshot {
288        DataCursorSnapshot {
289            marker_seq: 1,
290            event_seq_before: 42,
291            ts_init: UnixNanos::from(1_700_000_000_000_000_000),
292            advanced: vec![
293                StreamCursor {
294                    slot: 0,
295                    ts_init_hi: UnixNanos::from(1_700_000_000_000_000_001),
296                    count: 7,
297                },
298                StreamCursor {
299                    slot: 1,
300                    ts_init_hi: UnixNanos::from(1_700_000_000_000_000_002),
301                    count: 3,
302                },
303            ],
304        }
305    }
306
307    fn baseline_hifi() -> HiFiMarker {
308        HiFiMarker {
309            marker_seq: 1,
310            event_seq_before: 42,
311            slot: 0,
312            ts_event: UnixNanos::from(1_700_000_000_000_000_000),
313            ts_init: UnixNanos::from(1_700_000_000_000_000_001),
314            same_ts_ordinal: 0,
315            record_fingerprint: [0xABu8; 32],
316        }
317    }
318
319    fn baseline_dict() -> StreamDictEntry {
320        StreamDictEntry {
321            slot: 3,
322            data_cls: DataClass::Quote,
323            identifier: "ETHUSDT.BINANCE".to_string(),
324        }
325    }
326
327    fn baseline_gap() -> MarkerGap {
328        MarkerGap {
329            from_marker_seq: 5,
330            to_marker_seq: 9,
331            reason: MarkerGapReason::Overflow,
332        }
333    }
334
335    fn hex32(bytes: &[u8; 32]) -> String {
336        let mut out = String::with_capacity(64);
337        for byte in bytes {
338            write!(out, "{byte:02x}").expect("writing to a String is infallible");
339        }
340        out
341    }
342
343    #[rstest]
344    fn marker_hash_is_deterministic() {
345        let snap = baseline_snapshot();
346        let h1 = compute_marker_hash(&snap);
347        let h2 = compute_marker_hash(&snap);
348
349        assert_eq!(h1, h2);
350
351        // Pinned wire-format vector. Any change to domain, field order, or endianness flips
352        // this value.
353        let hex = hex32(&h1);
354        assert_eq!(
355            hex, "898bc3efdaf0edd9167a38a1c3060c9b4dc051658ea2f6132004bed78a481c47",
356            "marker hash wire format changed"
357        );
358    }
359
360    #[rstest]
361    fn hifi_hash_is_deterministic() {
362        let marker = baseline_hifi();
363        let h1 = compute_hifi_hash(&marker);
364        let h2 = compute_hifi_hash(&marker);
365
366        assert_eq!(h1, h2);
367
368        let hex = hex32(&h1);
369        assert_eq!(
370            hex, "06542408380d8815ef783b9dbde6b3e3ffdf05605bb17e83ad48474557457517",
371            "hifi hash wire format changed"
372        );
373    }
374
375    #[rstest]
376    fn dict_hash_is_deterministic() {
377        let entry = baseline_dict();
378        let h1 = compute_dict_hash(&entry);
379        let h2 = compute_dict_hash(&entry);
380
381        assert_eq!(h1, h2);
382
383        // Pinned wire-format vector. Any change to domain, field order, or framing flips this.
384        let hex = hex32(&h1);
385        assert_eq!(
386            hex, "24e702c5ae20b832ad6907676919fa18a89b79e97dde9df7e1de454191f42fda",
387            "dict hash wire format changed"
388        );
389    }
390
391    #[rstest]
392    fn gap_hash_is_deterministic() {
393        let gap = baseline_gap();
394        let h1 = compute_gap_hash(&gap);
395        let h2 = compute_gap_hash(&gap);
396
397        assert_eq!(h1, h2);
398
399        // Pinned wire-format vector. Any change to domain, field order, or framing flips this.
400        let hex = hex32(&h1);
401        assert_eq!(
402            hex, "ec1ae0ea813e9971155c6277e95c43de72da6f22ca1832f072aadd9b91f5a3ec",
403            "gap hash wire format changed"
404        );
405    }
406
407    #[rstest]
408    fn marker_record_codec_roundtrip() {
409        // DataCursorSnapshot
410        let snap = baseline_snapshot();
411        let bytes = crate::codec::encode_to_vec(&snap).expect("encode");
412        let decoded =
413            crate::codec::decode_from_slice::<DataCursorSnapshot>(&bytes).expect("decode");
414        assert_eq!(snap, decoded);
415
416        // HiFiMarker
417        let hifi = baseline_hifi();
418        let bytes = crate::codec::encode_to_vec(&hifi).expect("encode");
419        let decoded = crate::codec::decode_from_slice::<HiFiMarker>(&bytes).expect("decode");
420        assert_eq!(hifi, decoded);
421
422        // MarkerGap
423        let gap = MarkerGap {
424            from_marker_seq: 5,
425            to_marker_seq: 10,
426            reason: MarkerGapReason::Overflow,
427        };
428        let bytes = crate::codec::encode_to_vec(&gap).expect("encode");
429        let decoded = crate::codec::decode_from_slice::<MarkerGap>(&bytes).expect("decode");
430        assert_eq!(gap, decoded);
431
432        // StreamDictEntry
433        let dict = StreamDictEntry {
434            slot: 2,
435            data_cls: DataClass::Bar,
436            identifier: "BTCUSDT-PERP.BINANCE".to_string(),
437        };
438        let bytes = crate::codec::encode_to_vec(&dict).expect("encode");
439        let decoded = crate::codec::decode_from_slice::<StreamDictEntry>(&bytes).expect("decode");
440        assert_eq!(dict, decoded);
441    }
442
443    #[rstest]
444    #[case::quote_lowercase("quote")]
445    #[case::empty("")]
446    #[case::trailing_s("Quotes")]
447    #[case::partial("BookDep")]
448    fn data_class_from_str_rejects_unknown(#[case] input: &str) {
449        let err = DataClass::from_str(input).unwrap_err();
450
451        assert!(
452            err.contains(input),
453            "error should name the rejected input, was `{err}`"
454        );
455    }
456
457    #[rstest]
458    fn dict_hash_for_depth_class_uses_pre_rename_token() {
459        let entry = StreamDictEntry {
460            slot: 7,
461            data_cls: DataClass::BookDepth,
462            identifier: "BTCUSDT-PERP.BINANCE".to_string(),
463        };
464
465        // Pinned so the hashed class token cannot drift from what pre-rename
466        // builds recorded; see `DataClass::persisted_hash_token`.
467        assert_eq!(
468            compute_dict_hash(&entry),
469            [
470                0x9c, 0x2e, 0x1e, 0xd8, 0x46, 0xe0, 0xa5, 0x24, 0xea, 0x3d, 0xb2, 0x49, 0x48, 0xbf,
471                0x66, 0xf7, 0x00, 0x8b, 0x91, 0x21, 0x4a, 0x60, 0x9d, 0x75, 0xca, 0x07, 0x0c, 0x1e,
472                0xbf, 0xe2, 0xa7, 0x79,
473            ]
474        );
475    }
476
477    #[rstest]
478    fn data_class_from_str_accepts_legacy_depth10_spelling() {
479        assert_eq!(
480            DataClass::from_str("BookDepth10").unwrap(),
481            DataClass::BookDepth
482        );
483        assert_eq!(
484            DataClass::from_str("BookDepth").unwrap(),
485            DataClass::BookDepth
486        );
487    }
488
489    #[rstest]
490    fn data_class_serde_accepts_legacy_depth10_spelling() {
491        let legacy =
492            serde::de::value::StrDeserializer::<serde::de::value::Error>::new("BookDepth10");
493        assert_eq!(
494            DataClass::deserialize(legacy).unwrap(),
495            DataClass::BookDepth
496        );
497        let canonical =
498            serde::de::value::StrDeserializer::<serde::de::value::Error>::new("BookDepth");
499        assert_eq!(
500            DataClass::deserialize(canonical).unwrap(),
501            DataClass::BookDepth
502        );
503    }
504
505    #[rstest]
506    #[case::marker_seq(|s: &mut DataCursorSnapshot| s.marker_seq = 99)]
507    #[case::event_seq_before(|s: &mut DataCursorSnapshot| s.event_seq_before = 99)]
508    #[case::ts_init(|s: &mut DataCursorSnapshot| s.ts_init = UnixNanos::from(1))]
509    #[case::cursor_slot(|s: &mut DataCursorSnapshot| s.advanced[0].slot = 256)]
510    #[case::cursor_ts_init_hi(|s: &mut DataCursorSnapshot| s.advanced[0].ts_init_hi = UnixNanos::from(1))]
511    #[case::cursor_count(|s: &mut DataCursorSnapshot| s.advanced[0].count = 999)]
512    #[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 }))]
513    #[case::cursor_order(|s: &mut DataCursorSnapshot| s.advanced.reverse())]
514    fn every_marker_field_affects_hash(#[case] mutate: fn(&mut DataCursorSnapshot)) {
515        let base = baseline_snapshot();
516        let mut mutated = base.clone();
517        mutate(&mut mutated);
518
519        assert_ne!(compute_marker_hash(&base), compute_marker_hash(&mutated));
520    }
521
522    #[rstest]
523    #[case::marker_seq(|m: &mut HiFiMarker| m.marker_seq = 99)]
524    #[case::event_seq_before(|m: &mut HiFiMarker| m.event_seq_before = 99)]
525    #[case::slot(|m: &mut HiFiMarker| m.slot = 256)]
526    #[case::ts_event(|m: &mut HiFiMarker| m.ts_event = UnixNanos::from(1))]
527    #[case::ts_init(|m: &mut HiFiMarker| m.ts_init = UnixNanos::from(1))]
528    #[case::same_ts_ordinal(|m: &mut HiFiMarker| m.same_ts_ordinal = 256)]
529    #[case::fingerprint(|m: &mut HiFiMarker| m.record_fingerprint[0] ^= 0x01)]
530    fn every_hifi_field_affects_hash(#[case] mutate: fn(&mut HiFiMarker)) {
531        let base = baseline_hifi();
532        let mut mutated = base.clone();
533        mutate(&mut mutated);
534
535        assert_ne!(compute_hifi_hash(&base), compute_hifi_hash(&mutated));
536    }
537
538    #[rstest]
539    #[case::slot(|e: &mut StreamDictEntry| e.slot = 99)]
540    #[case::data_cls(|e: &mut StreamDictEntry| e.data_cls = DataClass::Trade)]
541    #[case::identifier(|e: &mut StreamDictEntry| e.identifier = "BTCUSDT.BINANCE".to_string())]
542    fn every_dict_field_affects_hash(#[case] mutate: fn(&mut StreamDictEntry)) {
543        let base = baseline_dict();
544        let mut mutated = base.clone();
545        mutate(&mut mutated);
546
547        assert_ne!(compute_dict_hash(&base), compute_dict_hash(&mutated));
548    }
549
550    #[rstest]
551    #[case::from(|g: &mut MarkerGap| g.from_marker_seq = 99)]
552    #[case::to(|g: &mut MarkerGap| g.to_marker_seq = 99)]
553    #[case::reason(|g: &mut MarkerGap| g.reason = MarkerGapReason::WriterClosed)]
554    fn every_gap_field_affects_hash(#[case] mutate: fn(&mut MarkerGap)) {
555        let base = baseline_gap();
556        let mut mutated = base.clone();
557        mutate(&mut mutated);
558
559        assert_ne!(compute_gap_hash(&base), compute_gap_hash(&mutated));
560    }
561
562    #[rstest]
563    fn marker_hash_handles_empty_advanced() {
564        let empty = DataCursorSnapshot {
565            marker_seq: 1,
566            event_seq_before: 42,
567            ts_init: UnixNanos::from(1_700_000_000_000_000_000),
568            advanced: vec![],
569        };
570
571        assert_eq!(compute_marker_hash(&empty), compute_marker_hash(&empty));
572        assert_ne!(
573            compute_marker_hash(&empty),
574            compute_marker_hash(&baseline_snapshot())
575        );
576    }
577}
nautilus_event_store/markers/marker.rs

nautilus_event_store/markers/
marker.rs
