nautilus_event_store/markers/
backend.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 backend trait and in-memory backend for the data marker sidecar.
17
18use std::fmt::Debug;
19
20use indexmap::IndexMap;
21use serde::{Deserialize, Serialize};
22
23use crate::{
24    error::EventStoreError,
25    manifest::{RunId, RunStatus},
26    markers::{DataClass, DataCursorSnapshot, HiFiMarker, MarkerGap, StreamDictEntry, StreamSlot},
27};
28
29/// Per-run manifest for a data marker sidecar file.
30///
31/// Mirrors [`RunManifest`](crate::manifest::RunManifest) for the marker sidecar: it links the
32/// marker file to its run, records the enabled data classes and whether high-fidelity capture is
33/// active, tracks per-table counts, and carries the sealed lifecycle status. It is a distinct
34/// struct, not a `RunManifest`, because the marker file is sealed independently of the entry run.
35#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
36pub struct MarkerManifest {
37    /// The id of the run this marker file belongs to.
38    pub run_id: RunId,
39    /// The data classes enabled for marker capture on this run.
40    pub enabled_classes: Vec<DataClass>,
41    /// Whether high-fidelity per-record markers are recorded for this run.
42    pub high_fidelity: bool,
43    /// The number of cursor snapshots recorded.
44    pub snapshot_count: u64,
45    /// The number of high-fidelity markers recorded.
46    pub hifi_count: u64,
47    /// The number of marker gaps recorded.
48    pub gap_count: u64,
49    /// The number of distinct stream dictionary entries recorded.
50    pub dict_count: u64,
51    /// The lifecycle state of this marker file.
52    pub status: RunStatus,
53}
54
55impl MarkerManifest {
56    /// Returns `true` once `status` is anything other than [`RunStatus::Running`].
57    #[must_use]
58    pub const fn is_sealed(&self) -> bool {
59        !matches!(self.status, RunStatus::Running)
60    }
61}
62
63/// A durable marker record returned with the integrity hash stored beside it.
64#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
65pub struct StoredMarkerRecord<T> {
66    /// The decoded durable marker record.
67    pub record: T,
68    /// The stored 32-byte integrity hash for `record`.
69    pub hash: [u8; 32],
70}
71
72/// A durable backend for the data marker sidecar.
73///
74/// Backends persist cursor snapshots, high-fidelity markers, gaps, and the slot dictionary for
75/// one open run at a time, mirroring the [`EventStore`](crate::backend::EventStore) trait for the
76/// marker file. Each durable record is appended with its precomputed integrity hash so a verifier
77/// can recompute and detect tampering. The marker path never blocks trading, so overflow is
78/// recorded as a [`MarkerGap`] by the writer rather than failing the caller here.
79pub trait MarkerBackend: Debug + Send {
80    /// Opens a run for marker capture with the supplied manifest.
81    ///
82    /// The status is normalized to [`RunStatus::Running`] and the per-table counts are zeroed, so
83    /// a caller can pass a freshly built or reused manifest without pre-clearing it.
84    ///
85    /// # Errors
86    ///
87    /// Returns an error if the backend fails.
88    fn open_run(&mut self, manifest: MarkerManifest) -> Result<(), EventStoreError>;
89
90    /// Appends a cursor snapshot with its precomputed integrity hash.
91    ///
92    /// # Errors
93    ///
94    /// Returns an error if the run is sealed, no run is open, or the backend fails.
95    fn append_snapshot(
96        &mut self,
97        snapshot: &DataCursorSnapshot,
98        hash: [u8; 32],
99    ) -> Result<(), EventStoreError>;
100
101    /// Appends a high-fidelity marker with its precomputed integrity hash.
102    ///
103    /// # Errors
104    ///
105    /// Returns an error if [`append_snapshot`](Self::append_snapshot) fails.
106    fn append_hifi(&mut self, marker: &HiFiMarker, hash: [u8; 32]) -> Result<(), EventStoreError>;
107
108    /// Appends a gap covering a dropped range of marker sequences, with its precomputed integrity
109    /// hash.
110    ///
111    /// # Errors
112    ///
113    /// Returns an error if [`append_snapshot`](Self::append_snapshot) fails.
114    fn append_gap(&mut self, gap: &MarkerGap, hash: [u8; 32]) -> Result<(), EventStoreError>;
115
116    /// Records a slot dictionary entry with its precomputed integrity hash, write-once by slot.
117    ///
118    /// A second call for an already-recorded slot is ignored, so the first observed
119    /// `slot -> (data_cls, identifier)` mapping wins and cannot be remapped.
120    ///
121    /// # Errors
122    ///
123    /// Returns an error if [`append_snapshot`](Self::append_snapshot) fails.
124    fn put_dict(&mut self, entry: &StreamDictEntry, hash: [u8; 32]) -> Result<(), EventStoreError>;
125
126    /// Scans all cursor snapshots in ascending `marker_seq` order.
127    ///
128    /// # Errors
129    ///
130    /// Returns an error if no run is open or the backend fails.
131    fn scan_snapshots(&self) -> Result<Vec<DataCursorSnapshot>, EventStoreError>;
132
133    /// Scans cursor snapshots with stored integrity hashes when the backend exposes them.
134    ///
135    /// Returns `Ok(None)` for backends that can only expose decoded records.
136    ///
137    /// # Errors
138    ///
139    /// Returns an error if [`scan_snapshots`](Self::scan_snapshots) fails.
140    fn scan_snapshot_records(
141        &self,
142    ) -> Result<Option<Vec<StoredMarkerRecord<DataCursorSnapshot>>>, EventStoreError> {
143        Ok(None)
144    }
145
146    /// Scans all high-fidelity markers in ascending `marker_seq` order.
147    ///
148    /// # Errors
149    ///
150    /// Returns an error if [`scan_snapshots`](Self::scan_snapshots) fails.
151    fn scan_hifi(&self) -> Result<Vec<HiFiMarker>, EventStoreError>;
152
153    /// Scans high-fidelity markers with stored integrity hashes when the backend exposes them.
154    ///
155    /// Returns `Ok(None)` for backends that can only expose decoded records.
156    ///
157    /// # Errors
158    ///
159    /// Returns an error if [`scan_hifi`](Self::scan_hifi) fails.
160    fn scan_hifi_records(
161        &self,
162    ) -> Result<Option<Vec<StoredMarkerRecord<HiFiMarker>>>, EventStoreError> {
163        Ok(None)
164    }
165
166    /// Scans all recorded gaps in ascending `from_marker_seq` order.
167    ///
168    /// # Errors
169    ///
170    /// Returns an error if [`scan_snapshots`](Self::scan_snapshots) fails.
171    fn scan_gaps(&self) -> Result<Vec<MarkerGap>, EventStoreError>;
172
173    /// Scans marker gaps with stored integrity hashes when the backend exposes them.
174    ///
175    /// Returns `Ok(None)` for backends that can only expose decoded records.
176    ///
177    /// # Errors
178    ///
179    /// Returns an error if [`scan_gaps`](Self::scan_gaps) fails.
180    fn scan_gap_records(
181        &self,
182    ) -> Result<Option<Vec<StoredMarkerRecord<MarkerGap>>>, EventStoreError> {
183        Ok(None)
184    }
185
186    /// Scans all slot dictionary entries in ascending slot order.
187    ///
188    /// # Errors
189    ///
190    /// Returns an error if [`scan_snapshots`](Self::scan_snapshots) fails.
191    fn scan_dict(&self) -> Result<Vec<StreamDictEntry>, EventStoreError>;
192
193    /// Scans dictionary entries with stored integrity hashes when the backend exposes them.
194    ///
195    /// Returns `Ok(None)` for backends that can only expose decoded records.
196    ///
197    /// # Errors
198    ///
199    /// Returns an error if [`scan_dict`](Self::scan_dict) fails.
200    fn scan_dict_records(
201        &self,
202    ) -> Result<Option<Vec<StoredMarkerRecord<StreamDictEntry>>>, EventStoreError> {
203        Ok(None)
204    }
205
206    /// Seals the open run with the given terminal status.
207    ///
208    /// # Errors
209    ///
210    /// Returns an error if `status` is [`RunStatus::Running`] or no run is open, or if the run is
211    /// already sealed.
212    fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError>;
213
214    /// Returns the current marker manifest.
215    ///
216    /// # Errors
217    ///
218    /// Returns an error if no run is open.
219    fn manifest(&self) -> Result<MarkerManifest, EventStoreError>;
220}
221
222/// In-memory implementation of [`MarkerBackend`].
223///
224/// Stores snapshots, high-fidelity markers, and gaps as `Vec` tables, and the slot dictionary as
225/// an [`IndexMap`] keyed by slot for write-once semantics. Each record's integrity hash is
226/// retained alongside it so the verifier can recompute and compare. The `scan_*` methods return
227/// records in their documented sort order regardless of append order, matching the keyed
228/// persistent backend. One instance owns at most one open run at a time; opening a new run
229/// replaces the previous one (markers never block boot). Used by writer and reader unit tests and
230/// by the `cfg(madsim)` simulation lane.
231#[derive(Debug, Default)]
232pub struct MemoryMarkerBackend {
233    state: Option<RunState>,
234}
235
236#[derive(Debug)]
237struct RunState {
238    manifest: MarkerManifest,
239    snapshots: Vec<(DataCursorSnapshot, [u8; 32])>,
240    hifi: Vec<(HiFiMarker, [u8; 32])>,
241    gaps: Vec<(MarkerGap, [u8; 32])>,
242    dict: IndexMap<StreamSlot, (StreamDictEntry, [u8; 32])>,
243}
244
245impl MemoryMarkerBackend {
246    /// Creates a new empty [`MemoryMarkerBackend`] with no run open.
247    #[must_use]
248    pub fn new() -> Self {
249        Self::default()
250    }
251
252    fn state(&self) -> Result<&RunState, EventStoreError> {
253        self.state
254            .as_ref()
255            .ok_or_else(|| EventStoreError::Backend("no run open".to_string()))
256    }
257
258    fn state_mut(&mut self) -> Result<&mut RunState, EventStoreError> {
259        self.state
260            .as_mut()
261            .ok_or_else(|| EventStoreError::Backend("no run open".to_string()))
262    }
263
264    fn writable_state(&mut self) -> Result<&mut RunState, EventStoreError> {
265        let state = self.state_mut()?;
266        if state.manifest.is_sealed() {
267            return Err(EventStoreError::Closed);
268        }
269        Ok(state)
270    }
271}
272
273impl MarkerBackend for MemoryMarkerBackend {
274    fn open_run(&mut self, mut manifest: MarkerManifest) -> Result<(), EventStoreError> {
275        // Markers never block boot, so a previous unsealed run is replaced rather than surfacing
276        // CrashedPredecessor the way the entry backend does.
277        manifest.status = RunStatus::Running;
278        manifest.snapshot_count = 0;
279        manifest.hifi_count = 0;
280        manifest.gap_count = 0;
281        manifest.dict_count = 0;
282
283        self.state = Some(RunState {
284            manifest,
285            snapshots: Vec::new(),
286            hifi: Vec::new(),
287            gaps: Vec::new(),
288            dict: IndexMap::new(),
289        });
290        Ok(())
291    }
292
293    fn append_snapshot(
294        &mut self,
295        snapshot: &DataCursorSnapshot,
296        hash: [u8; 32],
297    ) -> Result<(), EventStoreError> {
298        let state = self.writable_state()?;
299        state.snapshots.push((snapshot.clone(), hash));
300        state.manifest.snapshot_count = state.snapshots.len() as u64;
301        Ok(())
302    }
303
304    fn append_hifi(&mut self, marker: &HiFiMarker, hash: [u8; 32]) -> Result<(), EventStoreError> {
305        let state = self.writable_state()?;
306        state.hifi.push((marker.clone(), hash));
307        state.manifest.hifi_count = state.hifi.len() as u64;
308        Ok(())
309    }
310
311    fn append_gap(&mut self, gap: &MarkerGap, hash: [u8; 32]) -> Result<(), EventStoreError> {
312        let state = self.writable_state()?;
313        state.gaps.push((gap.clone(), hash));
314        state.manifest.gap_count = state.gaps.len() as u64;
315        Ok(())
316    }
317
318    fn put_dict(&mut self, entry: &StreamDictEntry, hash: [u8; 32]) -> Result<(), EventStoreError> {
319        let state = self.writable_state()?;
320        // Write-once by slot: the first observed mapping wins, so a later re-put cannot remap a
321        // slot to a different class or identifier. The count tracks distinct slots, so a duplicate
322        // re-put leaves it unchanged.
323        state
324            .dict
325            .entry(entry.slot)
326            .or_insert_with(|| (entry.clone(), hash));
327        state.manifest.dict_count = state.dict.len() as u64;
328        Ok(())
329    }
330
331    fn scan_snapshots(&self) -> Result<Vec<DataCursorSnapshot>, EventStoreError> {
332        let out = self
333            .scan_snapshot_records()?
334            .unwrap_or_default()
335            .into_iter()
336            .map(|stored| stored.record)
337            .collect();
338        Ok(out)
339    }
340
341    fn scan_snapshot_records(
342        &self,
343    ) -> Result<Option<Vec<StoredMarkerRecord<DataCursorSnapshot>>>, EventStoreError> {
344        let mut out: Vec<StoredMarkerRecord<DataCursorSnapshot>> = self
345            .state()?
346            .snapshots
347            .iter()
348            .map(|(record, hash)| StoredMarkerRecord {
349                record: record.clone(),
350                hash: *hash,
351            })
352            .collect();
353        out.sort_by_key(|stored| stored.record.marker_seq);
354        Ok(Some(out))
355    }
356
357    fn scan_hifi(&self) -> Result<Vec<HiFiMarker>, EventStoreError> {
358        let out = self
359            .scan_hifi_records()?
360            .unwrap_or_default()
361            .into_iter()
362            .map(|stored| stored.record)
363            .collect();
364        Ok(out)
365    }
366
367    fn scan_hifi_records(
368        &self,
369    ) -> Result<Option<Vec<StoredMarkerRecord<HiFiMarker>>>, EventStoreError> {
370        let mut out: Vec<StoredMarkerRecord<HiFiMarker>> = self
371            .state()?
372            .hifi
373            .iter()
374            .map(|(record, hash)| StoredMarkerRecord {
375                record: record.clone(),
376                hash: *hash,
377            })
378            .collect();
379        out.sort_by_key(|stored| stored.record.marker_seq);
380        Ok(Some(out))
381    }
382
383    fn scan_gaps(&self) -> Result<Vec<MarkerGap>, EventStoreError> {
384        let out = self
385            .scan_gap_records()?
386            .unwrap_or_default()
387            .into_iter()
388            .map(|stored| stored.record)
389            .collect();
390        Ok(out)
391    }
392
393    fn scan_gap_records(
394        &self,
395    ) -> Result<Option<Vec<StoredMarkerRecord<MarkerGap>>>, EventStoreError> {
396        let mut out: Vec<StoredMarkerRecord<MarkerGap>> = self
397            .state()?
398            .gaps
399            .iter()
400            .map(|(record, hash)| StoredMarkerRecord {
401                record: record.clone(),
402                hash: *hash,
403            })
404            .collect();
405        out.sort_by_key(|stored| stored.record.from_marker_seq);
406        Ok(Some(out))
407    }
408
409    fn scan_dict(&self) -> Result<Vec<StreamDictEntry>, EventStoreError> {
410        let out = self
411            .scan_dict_records()?
412            .unwrap_or_default()
413            .into_iter()
414            .map(|stored| stored.record)
415            .collect();
416        Ok(out)
417    }
418
419    fn scan_dict_records(
420        &self,
421    ) -> Result<Option<Vec<StoredMarkerRecord<StreamDictEntry>>>, EventStoreError> {
422        let mut out: Vec<StoredMarkerRecord<StreamDictEntry>> = self
423            .state()?
424            .dict
425            .values()
426            .map(|(record, hash)| StoredMarkerRecord {
427                record: record.clone(),
428                hash: *hash,
429            })
430            .collect();
431        out.sort_by_key(|stored| stored.record.slot);
432        Ok(Some(out))
433    }
434
435    fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
436        let state = self.state_mut()?;
437
438        // Running is not a terminal state; accepting it would leave the manifest unsealed while
439        // returning Ok, so subsequent appends would not see Closed.
440        if matches!(status, RunStatus::Running) {
441            return Err(EventStoreError::Backend(
442                "seal status must be a terminal state, was Running".to_string(),
443            ));
444        }
445
446        if state.manifest.is_sealed() {
447            return Err(EventStoreError::Closed);
448        }
449
450        state.manifest.status = status;
451        Ok(())
452    }
453
454    fn manifest(&self) -> Result<MarkerManifest, EventStoreError> {
455        Ok(self.state()?.manifest.clone())
456    }
457}
458
459#[cfg(test)]
460mod tests {
461    use nautilus_core::UnixNanos;
462    use rstest::{fixture, rstest};
463
464    use super::*;
465    use crate::markers::{
466        MarkerGapReason, StreamCursor, compute_dict_hash, compute_gap_hash, compute_hifi_hash,
467        compute_marker_hash,
468    };
469
470    fn manifest() -> MarkerManifest {
471        MarkerManifest {
472            run_id: "1700000000-aaaa1111".to_string(),
473            enabled_classes: vec![DataClass::Quote, DataClass::Trade],
474            high_fidelity: false,
475            snapshot_count: 0,
476            hifi_count: 0,
477            gap_count: 0,
478            dict_count: 0,
479            status: RunStatus::Running,
480        }
481    }
482
483    fn snapshot(marker_seq: u64, event_seq_before: u64) -> DataCursorSnapshot {
484        DataCursorSnapshot {
485            marker_seq,
486            event_seq_before,
487            ts_init: UnixNanos::from(1_700_000_000_000_000_000 + marker_seq),
488            advanced: vec![StreamCursor {
489                slot: 0,
490                ts_init_hi: UnixNanos::from(1_700_000_000_000_000_000 + marker_seq),
491                count: marker_seq,
492            }],
493        }
494    }
495
496    fn hifi(marker_seq: u64) -> HiFiMarker {
497        HiFiMarker {
498            marker_seq,
499            event_seq_before: 10,
500            slot: 0,
501            ts_event: UnixNanos::from(1),
502            ts_init: UnixNanos::from(2),
503            same_ts_ordinal: 0,
504            record_fingerprint: [0u8; 32],
505        }
506    }
507
508    #[fixture]
509    fn open_backend() -> MemoryMarkerBackend {
510        let mut backend = MemoryMarkerBackend::new();
511        backend.open_run(manifest()).expect("open run");
512        backend
513    }
514
515    #[rstest]
516    fn append_and_scan_snapshots_in_seq_order(mut open_backend: MemoryMarkerBackend) {
517        let s1 = snapshot(1, 10);
518        let s2 = snapshot(2, 20);
519        open_backend
520            .append_snapshot(&s1, compute_marker_hash(&s1))
521            .expect("append 1");
522        open_backend
523            .append_snapshot(&s2, compute_marker_hash(&s2))
524            .expect("append 2");
525
526        let scanned = open_backend.scan_snapshots().expect("scan");
527
528        assert_eq!(scanned, vec![s1, s2]);
529    }
530
531    #[rstest]
532    fn scans_return_marker_seq_order_regardless_of_append_order(
533        mut open_backend: MemoryMarkerBackend,
534    ) {
535        let s2 = snapshot(2, 20);
536        let s1 = snapshot(1, 10);
537        let m2 = hifi(4);
538        let m1 = hifi(3);
539        // Appended out of order: the backend must still scan in ascending marker_seq.
540        open_backend
541            .append_snapshot(&s2, compute_marker_hash(&s2))
542            .expect("snap 2");
543        open_backend
544            .append_snapshot(&s1, compute_marker_hash(&s1))
545            .expect("snap 1");
546        open_backend
547            .append_hifi(&m2, compute_hifi_hash(&m2))
548            .expect("hifi 4");
549        open_backend
550            .append_hifi(&m1, compute_hifi_hash(&m1))
551            .expect("hifi 3");
552
553        assert_eq!(open_backend.scan_snapshots().expect("snaps"), vec![s1, s2]);
554        assert_eq!(open_backend.scan_hifi().expect("hifi"), vec![m1, m2]);
555    }
556
557    #[rstest]
558    fn append_and_scan_hifi(mut open_backend: MemoryMarkerBackend) {
559        let m1 = hifi(1);
560        let m2 = hifi(2);
561        open_backend
562            .append_hifi(&m1, compute_hifi_hash(&m1))
563            .expect("hifi 1");
564        open_backend
565            .append_hifi(&m2, compute_hifi_hash(&m2))
566            .expect("hifi 2");
567
568        let scanned = open_backend.scan_hifi().expect("scan hifi");
569
570        assert_eq!(scanned, vec![m1, m2]);
571    }
572
573    #[rstest]
574    fn dict_entries_are_write_once_by_slot(mut open_backend: MemoryMarkerBackend) {
575        let first = StreamDictEntry {
576            slot: 0,
577            data_cls: DataClass::Quote,
578            identifier: "ETHUSDT.BINANCE".to_string(),
579        };
580        // A second put for slot 0 must not remap the slot to a different class/identifier.
581        let remap = StreamDictEntry {
582            slot: 0,
583            data_cls: DataClass::Trade,
584            identifier: "BTCUSDT.BINANCE".to_string(),
585        };
586        let other = StreamDictEntry {
587            slot: 1,
588            data_cls: DataClass::Trade,
589            identifier: "BTCUSDT.BINANCE".to_string(),
590        };
591
592        open_backend
593            .put_dict(&first, compute_dict_hash(&first))
594            .expect("put 0");
595        open_backend
596            .put_dict(&remap, compute_dict_hash(&remap))
597            .expect("re-put 0");
598        open_backend
599            .put_dict(&other, compute_dict_hash(&other))
600            .expect("put 1");
601
602        let dict = open_backend.scan_dict().expect("scan dict");
603
604        assert_eq!(dict, vec![first, other]);
605        // The duplicate re-put of slot 0 is not double-counted.
606        assert_eq!(open_backend.manifest().expect("manifest").dict_count, 2);
607    }
608
609    #[rstest]
610    fn append_gap_and_scan_gaps(mut open_backend: MemoryMarkerBackend) {
611        let g1 = MarkerGap {
612            from_marker_seq: 5,
613            to_marker_seq: 9,
614            reason: MarkerGapReason::Overflow,
615        };
616        let g2 = MarkerGap {
617            from_marker_seq: 20,
618            to_marker_seq: 20,
619            reason: MarkerGapReason::WriterClosed,
620        };
621        open_backend
622            .append_gap(&g1, compute_gap_hash(&g1))
623            .expect("gap 1");
624        open_backend
625            .append_gap(&g2, compute_gap_hash(&g2))
626            .expect("gap 2");
627
628        let gaps = open_backend.scan_gaps().expect("scan gaps");
629
630        assert_eq!(gaps, vec![g1, g2]);
631    }
632
633    #[rstest]
634    #[case::ended(RunStatus::Ended)]
635    #[case::crashed_recovered(RunStatus::CrashedRecovered)]
636    #[case::quarantined(RunStatus::Quarantined)]
637    fn seal_sets_manifest_status(mut open_backend: MemoryMarkerBackend, #[case] status: RunStatus) {
638        open_backend.seal(status).expect("seal");
639
640        let m = open_backend.manifest().expect("manifest");
641        assert_eq!(m.status, status);
642        assert!(m.is_sealed());
643    }
644
645    #[rstest]
646    fn open_run_normalizes_status_and_zeroes_counts() {
647        let mut backend = MemoryMarkerBackend::new();
648        let mut stale = manifest();
649        stale.status = RunStatus::Ended;
650        stale.snapshot_count = 99;
651        stale.hifi_count = 99;
652        stale.gap_count = 99;
653        stale.dict_count = 99;
654
655        backend.open_run(stale).expect("open");
656
657        let opened = backend.manifest().expect("manifest");
658        assert_eq!(opened.status, RunStatus::Running);
659        assert_eq!(opened.snapshot_count, 0);
660        assert_eq!(opened.hifi_count, 0);
661        assert_eq!(opened.gap_count, 0);
662        assert_eq!(opened.dict_count, 0);
663        // Enabled classes and mode survive the open.
664        assert_eq!(
665            opened.enabled_classes,
666            vec![DataClass::Quote, DataClass::Trade]
667        );
668        assert!(!opened.high_fidelity);
669    }
670
671    #[rstest]
672    fn manifest_counts_track_appends(mut open_backend: MemoryMarkerBackend) {
673        let snap = snapshot(1, 10);
674        let marker = hifi(2);
675        let gap = MarkerGap {
676            from_marker_seq: 3,
677            to_marker_seq: 4,
678            reason: MarkerGapReason::Overflow,
679        };
680        let entry = StreamDictEntry {
681            slot: 0,
682            data_cls: DataClass::Quote,
683            identifier: "ETHUSDT.BINANCE".to_string(),
684        };
685
686        open_backend
687            .append_snapshot(&snap, compute_marker_hash(&snap))
688            .expect("snap");
689        open_backend
690            .append_hifi(&marker, compute_hifi_hash(&marker))
691            .expect("hifi");
692        open_backend
693            .append_gap(&gap, compute_gap_hash(&gap))
694            .expect("gap");
695        open_backend
696            .put_dict(&entry, compute_dict_hash(&entry))
697            .expect("dict");
698
699        let m = open_backend.manifest().expect("manifest");
700        assert_eq!(m.snapshot_count, 1);
701        assert_eq!(m.hifi_count, 1);
702        assert_eq!(m.gap_count, 1);
703        assert_eq!(m.dict_count, 1);
704    }
705
706    #[rstest]
707    fn append_after_seal_returns_closed(mut open_backend: MemoryMarkerBackend) {
708        open_backend.seal(RunStatus::Ended).expect("seal");
709        let snap = snapshot(1, 10);
710
711        let err = open_backend
712            .append_snapshot(&snap, compute_marker_hash(&snap))
713            .expect_err("must reject");
714
715        assert!(matches!(err, EventStoreError::Closed));
716    }
717
718    #[rstest]
719    fn seal_rejects_running_status(mut open_backend: MemoryMarkerBackend) {
720        let err = open_backend
721            .seal(RunStatus::Running)
722            .expect_err("must reject");
723
724        match err {
725            EventStoreError::Backend(msg) => assert!(msg.contains("Running"), "msg was: {msg}"),
726            other => panic!("expected Backend, was {other:?}"),
727        }
728        assert!(!open_backend.manifest().expect("manifest").is_sealed());
729    }
730
731    #[rstest]
732    fn re_seal_returns_closed(mut open_backend: MemoryMarkerBackend) {
733        open_backend.seal(RunStatus::Ended).expect("seal");
734
735        let err = open_backend
736            .seal(RunStatus::Quarantined)
737            .expect_err("re-seal");
738
739        assert!(matches!(err, EventStoreError::Closed));
740    }
741
742    #[rstest]
743    #[case::append_snapshot("append_snapshot")]
744    #[case::append_hifi("append_hifi")]
745    #[case::append_gap("append_gap")]
746    #[case::put_dict("put_dict")]
747    #[case::scan_snapshots("scan_snapshots")]
748    #[case::scan_hifi("scan_hifi")]
749    #[case::scan_gaps("scan_gaps")]
750    #[case::scan_dict("scan_dict")]
751    #[case::seal("seal")]
752    #[case::manifest("manifest")]
753    fn methods_error_when_no_run_open(#[case] op: &str) {
754        let mut backend = MemoryMarkerBackend::new();
755        let snap = snapshot(1, 10);
756        let marker = hifi(1);
757        let gap = MarkerGap {
758            from_marker_seq: 1,
759            to_marker_seq: 2,
760            reason: MarkerGapReason::Overflow,
761        };
762        let entry = StreamDictEntry {
763            slot: 0,
764            data_cls: DataClass::Quote,
765            identifier: "ETHUSDT.BINANCE".to_string(),
766        };
767
768        let err = match op {
769            "append_snapshot" => backend
770                .append_snapshot(&snap, compute_marker_hash(&snap))
771                .unwrap_err(),
772            "append_hifi" => backend
773                .append_hifi(&marker, compute_hifi_hash(&marker))
774                .unwrap_err(),
775            "append_gap" => backend
776                .append_gap(&gap, compute_gap_hash(&gap))
777                .unwrap_err(),
778            "put_dict" => backend
779                .put_dict(&entry, compute_dict_hash(&entry))
780                .unwrap_err(),
781            "scan_snapshots" => backend.scan_snapshots().unwrap_err(),
782            "scan_hifi" => backend.scan_hifi().unwrap_err(),
783            "scan_gaps" => backend.scan_gaps().unwrap_err(),
784            "scan_dict" => backend.scan_dict().unwrap_err(),
785            "seal" => backend.seal(RunStatus::Ended).unwrap_err(),
786            "manifest" => backend.manifest().unwrap_err(),
787            _ => unreachable!(),
788        };
789
790        match err {
791            EventStoreError::Backend(msg) => assert!(msg.contains("no run open"), "msg was: {msg}"),
792            other => panic!("expected Backend, was {other:?}"),
793        }
794    }
795}
nautilus_event_store/markers/backend.rs

nautilus_event_store/markers/
backend.rs
