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

nautilus_event_store/markers/
backend.rs
