nautilus_event_store/markers/
reader.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//! Read-side cursor folding for data marker sidecars.
17
18use std::fmt::Debug;
19
20use ahash::AHashMap;
21
22use crate::{
23    error::EventStoreError,
24    markers::{MarkerBackend, StreamCursor, StreamDictEntry, StreamSlot},
25};
26
27/// Read-side scanner for a single data marker sidecar.
28///
29/// The reader owns an already-open marker backend. It folds cursor snapshots up to a target
30/// `event_seq_before` and resolves stream slots through the durable stream dictionary.
31pub struct MarkerReader {
32    backend: Box<dyn MarkerBackend>,
33}
34
35impl Debug for MarkerReader {
36    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37        f.debug_struct(stringify!(MarkerReader))
38            .finish_non_exhaustive()
39    }
40}
41
42impl MarkerReader {
43    /// Creates a reader over an already-open marker backend.
44    #[must_use]
45    pub fn new(backend: Box<dyn MarkerBackend>) -> Self {
46        Self { backend }
47    }
48
49    /// Folds cursor snapshots whose `event_seq_before` is at or below `event_seq_before`.
50    ///
51    /// The returned map carries the latest known cursor per stream slot at the target event-store
52    /// boundary. Slots that did not advance in later snapshots keep their previous cursor.
53    ///
54    /// # Errors
55    ///
56    /// Returns [`EventStoreError`] when the backend cannot scan cursor snapshots.
57    pub fn fold_to(
58        &self,
59        event_seq_before: u64,
60    ) -> Result<AHashMap<StreamSlot, StreamCursor>, EventStoreError> {
61        let mut folded = AHashMap::new();
62
63        for snapshot in self.backend.scan_snapshots()? {
64            if snapshot.event_seq_before > event_seq_before {
65                continue;
66            }
67
68            for cursor in snapshot.advanced {
69                folded.insert(cursor.slot, cursor);
70            }
71        }
72
73        Ok(folded)
74    }
75
76    /// Scans the durable stream dictionary into a map keyed by stream slot.
77    ///
78    /// # Errors
79    ///
80    /// Returns [`EventStoreError`] when the backend cannot scan the dictionary.
81    pub fn stream_dictionary(
82        &self,
83    ) -> Result<AHashMap<StreamSlot, StreamDictEntry>, EventStoreError> {
84        Ok(self
85            .backend
86            .scan_dict()?
87            .into_iter()
88            .map(|entry| (entry.slot, entry))
89            .collect())
90    }
91
92    /// Resolves `slot` to its stream dictionary entry.
93    ///
94    /// Returns `None` when the slot is unknown or the backend cannot scan the dictionary.
95    #[must_use]
96    pub fn resolve_slot(&self, slot: StreamSlot) -> Option<StreamDictEntry> {
97        self.stream_dictionary().ok()?.get(&slot).cloned()
98    }
99}
100
101#[cfg(test)]
102mod tests {
103    use nautilus_core::UnixNanos;
104    use rstest::rstest;
105
106    use super::*;
107    use crate::{
108        manifest::RunStatus,
109        markers::{
110            DataClass, DataCursorSnapshot, MarkerBackend, MarkerManifest, MemoryMarkerBackend,
111            StreamCursor, StreamDictEntry, compute_dict_hash, compute_marker_hash,
112        },
113    };
114
115    fn manifest() -> MarkerManifest {
116        MarkerManifest {
117            run_id: "1700000000-reader".to_string(),
118            enabled_classes: vec![DataClass::Quote, DataClass::Trade],
119            high_fidelity: false,
120            snapshot_count: 0,
121            hifi_count: 0,
122            gap_count: 0,
123            dict_count: 0,
124            status: RunStatus::Running,
125        }
126    }
127
128    fn snapshot(
129        marker_seq: u64,
130        event_seq_before: u64,
131        advanced: Vec<StreamCursor>,
132    ) -> DataCursorSnapshot {
133        DataCursorSnapshot {
134            marker_seq,
135            event_seq_before,
136            ts_init: UnixNanos::from(1_700_000_000_000_000_000 + marker_seq),
137            advanced,
138        }
139    }
140
141    fn dict(slot: u32, data_cls: DataClass, identifier: &str) -> StreamDictEntry {
142        StreamDictEntry {
143            slot,
144            data_cls,
145            identifier: identifier.to_string(),
146        }
147    }
148
149    #[rstest]
150    fn fold_cursors_to_event_seq() {
151        let mut backend = MemoryMarkerBackend::new();
152        backend.open_run(manifest()).expect("open run");
153        let quote_dict = dict(0, DataClass::Quote, "ETHUSDT.BINANCE");
154        let trade_dict = dict(1, DataClass::Trade, "BTCUSDT.BINANCE");
155        backend
156            .put_dict(&quote_dict, compute_dict_hash(&quote_dict))
157            .expect("put quote dict");
158        backend
159            .put_dict(&trade_dict, compute_dict_hash(&trade_dict))
160            .expect("put trade dict");
161
162        let s1 = snapshot(
163            1,
164            5,
165            vec![StreamCursor {
166                slot: 0,
167                ts_init_hi: UnixNanos::from(100),
168                count: 1,
169            }],
170        );
171        let s2 = snapshot(
172            2,
173            10,
174            vec![
175                StreamCursor {
176                    slot: 0,
177                    ts_init_hi: UnixNanos::from(300),
178                    count: 3,
179                },
180                StreamCursor {
181                    slot: 1,
182                    ts_init_hi: UnixNanos::from(1_000),
183                    count: 1,
184                },
185            ],
186        );
187        let s3 = snapshot(
188            3,
189            15,
190            vec![StreamCursor {
191                slot: 1,
192                ts_init_hi: UnixNanos::from(2_000),
193                count: 2,
194            }],
195        );
196        backend
197            .append_snapshot(&s1, compute_marker_hash(&s1))
198            .expect("append s1");
199        backend
200            .append_snapshot(&s2, compute_marker_hash(&s2))
201            .expect("append s2");
202        backend
203            .append_snapshot(&s3, compute_marker_hash(&s3))
204            .expect("append s3");
205
206        let reader = MarkerReader::new(Box::new(backend));
207        let folded = reader.fold_to(10).expect("fold");
208
209        assert_eq!(
210            folded.get(&0),
211            Some(&StreamCursor {
212                slot: 0,
213                ts_init_hi: UnixNanos::from(300),
214                count: 3,
215            }),
216        );
217        assert_eq!(
218            folded.get(&1),
219            Some(&StreamCursor {
220                slot: 1,
221                ts_init_hi: UnixNanos::from(1_000),
222                count: 1,
223            }),
224        );
225        assert_eq!(reader.resolve_slot(1), Some(trade_dict));
226    }
227}
nautilus_event_store/markers/reader.rs

nautilus_event_store/markers/
reader.rs
