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 proptest::{prelude::*, test_runner::Config as ProptestConfig};
246    use rstest::rstest;
247
248    use super::*;
249
250    #[rstest]
251    fn data_class_roundtrips_to_str() {
252        let variants = [
253            (DataClass::BookDeltas, "BookDeltas"),
254            (DataClass::BookDepth10, "BookDepth10"),
255            (DataClass::Quote, "Quote"),
256            (DataClass::Trade, "Trade"),
257            (DataClass::Bar, "Bar"),
258        ];
259
260        for (variant, expected) in variants {
261            assert_eq!(variant.as_str(), expected, "as_str for {variant:?}");
262            assert_eq!(variant.to_string(), expected, "Display for {variant:?}");
263            assert_eq!(
264                DataClass::from_str(expected).unwrap(),
265                variant,
266                "from_str for {expected}"
267            );
268        }
269    }
270
271    fn baseline_snapshot() -> DataCursorSnapshot {
272        DataCursorSnapshot {
273            marker_seq: 1,
274            event_seq_before: 42,
275            ts_init: UnixNanos::from(1_700_000_000_000_000_000),
276            advanced: vec![
277                StreamCursor {
278                    slot: 0,
279                    ts_init_hi: UnixNanos::from(1_700_000_000_000_000_001),
280                    count: 7,
281                },
282                StreamCursor {
283                    slot: 1,
284                    ts_init_hi: UnixNanos::from(1_700_000_000_000_000_002),
285                    count: 3,
286                },
287            ],
288        }
289    }
290
291    fn baseline_hifi() -> HiFiMarker {
292        HiFiMarker {
293            marker_seq: 1,
294            event_seq_before: 42,
295            slot: 0,
296            ts_event: UnixNanos::from(1_700_000_000_000_000_000),
297            ts_init: UnixNanos::from(1_700_000_000_000_000_001),
298            same_ts_ordinal: 0,
299            record_fingerprint: [0xABu8; 32],
300        }
301    }
302
303    fn baseline_dict() -> StreamDictEntry {
304        StreamDictEntry {
305            slot: 3,
306            data_cls: DataClass::Quote,
307            identifier: "ETHUSDT.BINANCE".to_string(),
308        }
309    }
310
311    fn baseline_gap() -> MarkerGap {
312        MarkerGap {
313            from_marker_seq: 5,
314            to_marker_seq: 9,
315            reason: MarkerGapReason::Overflow,
316        }
317    }
318
319    fn hex32(bytes: &[u8; 32]) -> String {
320        let mut out = String::with_capacity(64);
321        for byte in bytes {
322            write!(out, "{byte:02x}").expect("writing to a String is infallible");
323        }
324        out
325    }
326
327    #[rstest]
328    fn marker_hash_is_deterministic() {
329        let snap = baseline_snapshot();
330        let h1 = compute_marker_hash(&snap);
331        let h2 = compute_marker_hash(&snap);
332
333        assert_eq!(h1, h2);
334
335        // Pinned wire-format vector. Any change to domain, field order, or endianness flips
336        // this value.
337        let hex = hex32(&h1);
338        assert_eq!(
339            hex, "898bc3efdaf0edd9167a38a1c3060c9b4dc051658ea2f6132004bed78a481c47",
340            "marker hash wire format changed"
341        );
342    }
343
344    #[rstest]
345    fn hifi_hash_is_deterministic() {
346        let marker = baseline_hifi();
347        let h1 = compute_hifi_hash(&marker);
348        let h2 = compute_hifi_hash(&marker);
349
350        assert_eq!(h1, h2);
351
352        let hex = hex32(&h1);
353        assert_eq!(
354            hex, "06542408380d8815ef783b9dbde6b3e3ffdf05605bb17e83ad48474557457517",
355            "hifi hash wire format changed"
356        );
357    }
358
359    #[rstest]
360    fn dict_hash_is_deterministic() {
361        let entry = baseline_dict();
362        let h1 = compute_dict_hash(&entry);
363        let h2 = compute_dict_hash(&entry);
364
365        assert_eq!(h1, h2);
366
367        // Pinned wire-format vector. Any change to domain, field order, or framing flips this.
368        let hex = hex32(&h1);
369        assert_eq!(
370            hex, "24e702c5ae20b832ad6907676919fa18a89b79e97dde9df7e1de454191f42fda",
371            "dict hash wire format changed"
372        );
373    }
374
375    #[rstest]
376    fn gap_hash_is_deterministic() {
377        let gap = baseline_gap();
378        let h1 = compute_gap_hash(&gap);
379        let h2 = compute_gap_hash(&gap);
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, "ec1ae0ea813e9971155c6277e95c43de72da6f22ca1832f072aadd9b91f5a3ec",
387            "gap hash wire format changed"
388        );
389    }
390
391    #[rstest]
392    fn marker_record_codec_roundtrip() {
393        // DataCursorSnapshot
394        let snap = baseline_snapshot();
395        let bytes = crate::codec::encode_to_vec(&snap).expect("encode");
396        let decoded =
397            crate::codec::decode_from_slice::<DataCursorSnapshot>(&bytes).expect("decode");
398        assert_eq!(snap, decoded);
399
400        // HiFiMarker
401        let hifi = baseline_hifi();
402        let bytes = crate::codec::encode_to_vec(&hifi).expect("encode");
403        let decoded = crate::codec::decode_from_slice::<HiFiMarker>(&bytes).expect("decode");
404        assert_eq!(hifi, decoded);
405
406        // MarkerGap
407        let gap = MarkerGap {
408            from_marker_seq: 5,
409            to_marker_seq: 10,
410            reason: MarkerGapReason::Overflow,
411        };
412        let bytes = crate::codec::encode_to_vec(&gap).expect("encode");
413        let decoded = crate::codec::decode_from_slice::<MarkerGap>(&bytes).expect("decode");
414        assert_eq!(gap, decoded);
415
416        // StreamDictEntry
417        let dict = StreamDictEntry {
418            slot: 2,
419            data_cls: DataClass::Bar,
420            identifier: "BTCUSDT-PERP.BINANCE".to_string(),
421        };
422        let bytes = crate::codec::encode_to_vec(&dict).expect("encode");
423        let decoded = crate::codec::decode_from_slice::<StreamDictEntry>(&bytes).expect("decode");
424        assert_eq!(dict, decoded);
425    }
426
427    #[rstest]
428    #[case::quote_lowercase("quote")]
429    #[case::empty("")]
430    #[case::trailing_s("Quotes")]
431    #[case::partial("BookDepth")]
432    fn data_class_from_str_rejects_unknown(#[case] input: &str) {
433        let err = DataClass::from_str(input).unwrap_err();
434
435        assert!(
436            err.contains(input),
437            "error should name the rejected input, was `{err}`"
438        );
439    }
440
441    #[rstest]
442    #[case::marker_seq(|s: &mut DataCursorSnapshot| s.marker_seq = 99)]
443    #[case::event_seq_before(|s: &mut DataCursorSnapshot| s.event_seq_before = 99)]
444    #[case::ts_init(|s: &mut DataCursorSnapshot| s.ts_init = UnixNanos::from(1))]
445    #[case::cursor_slot(|s: &mut DataCursorSnapshot| s.advanced[0].slot = 256)]
446    #[case::cursor_ts_init_hi(|s: &mut DataCursorSnapshot| s.advanced[0].ts_init_hi = UnixNanos::from(1))]
447    #[case::cursor_count(|s: &mut DataCursorSnapshot| s.advanced[0].count = 999)]
448    #[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 }))]
449    #[case::cursor_order(|s: &mut DataCursorSnapshot| s.advanced.reverse())]
450    fn every_marker_field_affects_hash(#[case] mutate: fn(&mut DataCursorSnapshot)) {
451        let base = baseline_snapshot();
452        let mut mutated = base.clone();
453        mutate(&mut mutated);
454
455        assert_ne!(compute_marker_hash(&base), compute_marker_hash(&mutated));
456    }
457
458    #[rstest]
459    #[case::marker_seq(|m: &mut HiFiMarker| m.marker_seq = 99)]
460    #[case::event_seq_before(|m: &mut HiFiMarker| m.event_seq_before = 99)]
461    #[case::slot(|m: &mut HiFiMarker| m.slot = 256)]
462    #[case::ts_event(|m: &mut HiFiMarker| m.ts_event = UnixNanos::from(1))]
463    #[case::ts_init(|m: &mut HiFiMarker| m.ts_init = UnixNanos::from(1))]
464    #[case::same_ts_ordinal(|m: &mut HiFiMarker| m.same_ts_ordinal = 256)]
465    #[case::fingerprint(|m: &mut HiFiMarker| m.record_fingerprint[0] ^= 0x01)]
466    fn every_hifi_field_affects_hash(#[case] mutate: fn(&mut HiFiMarker)) {
467        let base = baseline_hifi();
468        let mut mutated = base.clone();
469        mutate(&mut mutated);
470
471        assert_ne!(compute_hifi_hash(&base), compute_hifi_hash(&mutated));
472    }
473
474    #[rstest]
475    #[case::slot(|e: &mut StreamDictEntry| e.slot = 99)]
476    #[case::data_cls(|e: &mut StreamDictEntry| e.data_cls = DataClass::Trade)]
477    #[case::identifier(|e: &mut StreamDictEntry| e.identifier = "BTCUSDT.BINANCE".to_string())]
478    fn every_dict_field_affects_hash(#[case] mutate: fn(&mut StreamDictEntry)) {
479        let base = baseline_dict();
480        let mut mutated = base.clone();
481        mutate(&mut mutated);
482
483        assert_ne!(compute_dict_hash(&base), compute_dict_hash(&mutated));
484    }
485
486    #[rstest]
487    #[case::from(|g: &mut MarkerGap| g.from_marker_seq = 99)]
488    #[case::to(|g: &mut MarkerGap| g.to_marker_seq = 99)]
489    #[case::reason(|g: &mut MarkerGap| g.reason = MarkerGapReason::WriterClosed)]
490    fn every_gap_field_affects_hash(#[case] mutate: fn(&mut MarkerGap)) {
491        let base = baseline_gap();
492        let mut mutated = base.clone();
493        mutate(&mut mutated);
494
495        assert_ne!(compute_gap_hash(&base), compute_gap_hash(&mutated));
496    }
497
498    #[rstest]
499    fn marker_hash_handles_empty_advanced() {
500        let empty = DataCursorSnapshot {
501            marker_seq: 1,
502            event_seq_before: 42,
503            ts_init: UnixNanos::from(1_700_000_000_000_000_000),
504            advanced: vec![],
505        };
506
507        assert_eq!(compute_marker_hash(&empty), compute_marker_hash(&empty));
508        assert_ne!(
509            compute_marker_hash(&empty),
510            compute_marker_hash(&baseline_snapshot())
511        );
512    }
513
514    proptest! {
515        #![proptest_config(ProptestConfig { cases: 64, ..ProptestConfig::default() })]
516
517        // Any cursor snapshot survives a codec encode/decode unchanged
518        #[rstest]
519        fn prop_marker_snapshot_codec_roundtrip(
520            marker_seq in any::<u64>(),
521            event_seq_before in any::<u64>(),
522            ts_init in any::<u64>(),
523            cursors in proptest::collection::vec((any::<u32>(), any::<u64>(), any::<u64>()), 0..8),
524        ) {
525            let snap = DataCursorSnapshot {
526                marker_seq,
527                event_seq_before,
528                ts_init: UnixNanos::from(ts_init),
529                advanced: cursors
530                    .into_iter()
531                    .map(|(slot, hi, count)| StreamCursor {
532                        slot,
533                        ts_init_hi: UnixNanos::from(hi),
534                        count,
535                    })
536                    .collect(),
537            };
538
539            let bytes = crate::codec::encode_to_vec(&snap).expect("encode");
540            let decoded = crate::codec::decode_from_slice::<DataCursorSnapshot>(&bytes).expect("decode");
541            prop_assert_eq!(snap, decoded);
542        }
543
544        // Any high-fidelity marker survives a codec encode/decode unchanged
545        #[rstest]
546        fn prop_hifi_marker_codec_roundtrip(
547            marker_seq in any::<u64>(),
548            event_seq_before in any::<u64>(),
549            slot in any::<u32>(),
550            ts_event in any::<u64>(),
551            ts_init in any::<u64>(),
552            same_ts_ordinal in any::<u32>(),
553            fingerprint in proptest::array::uniform32(any::<u8>()),
554        ) {
555            let marker = HiFiMarker {
556                marker_seq,
557                event_seq_before,
558                slot,
559                ts_event: UnixNanos::from(ts_event),
560                ts_init: UnixNanos::from(ts_init),
561                same_ts_ordinal,
562                record_fingerprint: fingerprint,
563            };
564
565            let bytes = crate::codec::encode_to_vec(&marker).expect("encode");
566            let decoded = crate::codec::decode_from_slice::<HiFiMarker>(&bytes).expect("decode");
567            prop_assert_eq!(marker, decoded);
568        }
569    }
570}
nautilus_event_store/markers/marker.rs

nautilus_event_store/markers/
marker.rs
