nautilus_event_store/markers/
cursor.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//! In-memory per-stream cursor state for the data marker sidecar.
17
18use ahash::AHashMap;
19use nautilus_core::UnixNanos;
20
21use crate::{
22    entry::Topic,
23    markers::{DataClass, DataCursorSnapshot, StreamCursor, StreamDictEntry, StreamSlot},
24};
25
26/// In-memory per-stream cursor state for data marker capture.
27///
28/// Assigns a [`StreamSlot`] lazily on the first observation of a topic, advances each slot's
29/// cursor (cumulative `count` and highest `ts_init`) on every message in O(1) with no I/O, and
30/// builds a [`DataCursorSnapshot`] of the slots that changed since the previous snapshot.
31#[derive(Debug, Default)]
32pub struct CursorState {
33    slots: AHashMap<Topic, StreamSlot>,
34    dict: Vec<StreamDictEntry>,
35    cursors: Vec<Cursor>,
36    dirty: Vec<bool>,
37}
38
39#[derive(Debug, Default)]
40struct Cursor {
41    ts_init_hi: UnixNanos,
42    count: u64,
43    last_ts_init: UnixNanos,
44    same_ts_count: u32,
45}
46
47impl CursorState {
48    /// Creates a new empty [`CursorState`].
49    #[must_use]
50    pub fn new() -> Self {
51        Self::default()
52    }
53
54    /// Advances the cursor for `topic`, assigning a slot on its first observation.
55    ///
56    /// Returns the stream `slot` and the `same_ts_ordinal`: the running ordinal among records
57    /// sharing the current `ts_init` within the slot, which resets to zero when `ts_init`
58    /// advances (only the high-fidelity path reads it). Bumps the slot's cumulative `count`,
59    /// raises its highest `ts_init`, and marks it dirty so the next
60    /// [`build_snapshot`](Self::build_snapshot) includes it. A first observation also queues a
61    /// [`StreamDictEntry`] for [`take_new_dict_entries`](Self::take_new_dict_entries).
62    ///
63    /// # Panics
64    ///
65    /// Panics if the number of distinct streams exceeds `u32::MAX`.
66    pub fn advance(
67        &mut self,
68        topic: Topic,
69        data_cls: DataClass,
70        identifier: &str,
71        ts_init: UnixNanos,
72    ) -> (StreamSlot, u32) {
73        let slot = if let Some(&slot) = self.slots.get(&topic) {
74            slot
75        } else {
76            let slot = u32::try_from(self.cursors.len()).expect("stream slot count fits u32");
77            self.cursors.push(Cursor::default());
78            self.dirty.push(false);
79            self.dict.push(StreamDictEntry {
80                slot,
81                data_cls,
82                identifier: identifier.to_string(),
83            });
84            self.slots.insert(topic, slot);
85            slot
86        };
87
88        let cursor = &mut self.cursors[slot as usize];
89
90        let same_ts_ordinal = if cursor.count == 0 || ts_init != cursor.last_ts_init {
91            cursor.last_ts_init = ts_init;
92            cursor.same_ts_count = 0;
93            0
94        } else {
95            cursor.same_ts_count += 1;
96            cursor.same_ts_count
97        };
98
99        cursor.count += 1;
100        if ts_init > cursor.ts_init_hi {
101            cursor.ts_init_hi = ts_init;
102        }
103        self.dirty[slot as usize] = true;
104
105        (slot, same_ts_ordinal)
106    }
107
108    /// Removes and returns the dict entries queued since the previous call.
109    ///
110    /// Each [`StreamDictEntry`] is produced once, when its slot is first assigned by
111    /// [`advance`](Self::advance).
112    pub fn take_new_dict_entries(&mut self) -> Vec<StreamDictEntry> {
113        std::mem::take(&mut self.dict)
114    }
115
116    /// Builds a snapshot of every slot that advanced since the previous snapshot, clearing the
117    /// dirty set.
118    ///
119    /// Returns `None` when no slot has advanced, so a quiet interval writes nothing.
120    pub fn build_snapshot(
121        &mut self,
122        marker_seq: u64,
123        event_seq_before: u64,
124        ts_init: UnixNanos,
125    ) -> Option<DataCursorSnapshot> {
126        let mut advanced = Vec::new();
127
128        for (slot, (cursor, dirty)) in (0_u32..).zip(self.cursors.iter().zip(self.dirty.iter_mut()))
129        {
130            if *dirty {
131                *dirty = false;
132                advanced.push(StreamCursor {
133                    slot,
134                    ts_init_hi: cursor.ts_init_hi,
135                    count: cursor.count,
136                });
137            }
138        }
139
140        if advanced.is_empty() {
141            None
142        } else {
143            Some(DataCursorSnapshot {
144                marker_seq,
145                event_seq_before,
146                ts_init,
147                advanced,
148            })
149        }
150    }
151}
152
153#[cfg(test)]
154mod tests {
155    use rstest::rstest;
156
157    use super::*;
158
159    #[rstest]
160    fn advance_assigns_slot_lazily_and_writes_dict() {
161        let mut state = CursorState::new();
162        let quotes: Topic = "data.quotes.BINANCE.ETHUSDT".into();
163        let trades: Topic = "data.trades.BINANCE.BTCUSDT".into();
164
165        let (s0, _) = state.advance(
166            quotes,
167            DataClass::Quote,
168            "ETHUSDT.BINANCE",
169            UnixNanos::from(1),
170        );
171        let (s1, _) = state.advance(
172            trades,
173            DataClass::Trade,
174            "BTCUSDT.BINANCE",
175            UnixNanos::from(2),
176        );
177        // A repeat of the first topic reuses its slot rather than allocating a new one.
178        let (s0_again, _) = state.advance(
179            quotes,
180            DataClass::Quote,
181            "ETHUSDT.BINANCE",
182            UnixNanos::from(3),
183        );
184
185        let dict = state.take_new_dict_entries();
186
187        assert_eq!(s0, 0);
188        assert_eq!(s1, 1);
189        assert_eq!(s0_again, 0);
190        assert_eq!(
191            dict,
192            vec![
193                StreamDictEntry {
194                    slot: 0,
195                    data_cls: DataClass::Quote,
196                    identifier: "ETHUSDT.BINANCE".to_string(),
197                },
198                StreamDictEntry {
199                    slot: 1,
200                    data_cls: DataClass::Trade,
201                    identifier: "BTCUSDT.BINANCE".to_string(),
202                },
203            ]
204        );
205        // Dict entries are produced once; a second drain is empty.
206        assert!(state.take_new_dict_entries().is_empty());
207    }
208
209    #[rstest]
210    fn advance_updates_count_and_ts_hi() {
211        let mut state = CursorState::new();
212        let quotes: Topic = "data.quotes.BINANCE.ETHUSDT".into();
213
214        let (slot, _) = state.advance(
215            quotes,
216            DataClass::Quote,
217            "ETHUSDT.BINANCE",
218            UnixNanos::from(100),
219        );
220        state.advance(
221            quotes,
222            DataClass::Quote,
223            "ETHUSDT.BINANCE",
224            UnixNanos::from(200),
225        );
226        // A lower ts_init must not lower the high-water mark.
227        state.advance(
228            quotes,
229            DataClass::Quote,
230            "ETHUSDT.BINANCE",
231            UnixNanos::from(150),
232        );
233
234        let cursor = &state.cursors[slot as usize];
235
236        assert_eq!(cursor.count, 3);
237        assert_eq!(cursor.ts_init_hi, UnixNanos::from(200));
238        assert!(state.dirty[slot as usize]);
239    }
240
241    #[rstest]
242    fn build_snapshot_includes_only_dirty_and_clears() {
243        let mut state = CursorState::new();
244        let a: Topic = "data.quotes.BINANCE.A".into();
245        let b: Topic = "data.quotes.BINANCE.B".into();
246        let c: Topic = "data.quotes.BINANCE.C".into();
247
248        let (sa, _) = state.advance(a, DataClass::Quote, "A.BINANCE", UnixNanos::from(10));
249        state.advance(b, DataClass::Quote, "B.BINANCE", UnixNanos::from(20));
250        let (sc, _) = state.advance(c, DataClass::Quote, "C.BINANCE", UnixNanos::from(30));
251
252        // The first snapshot carries all three streams and clears the dirty set.
253        let first = state
254            .build_snapshot(1, 0, UnixNanos::from(30))
255            .expect("snapshot");
256
257        // Advance only two of the three streams.
258        state.advance(a, DataClass::Quote, "A.BINANCE", UnixNanos::from(40));
259        state.advance(c, DataClass::Quote, "C.BINANCE", UnixNanos::from(50));
260
261        let second = state
262            .build_snapshot(2, 7, UnixNanos::from(50))
263            .expect("snapshot");
264
265        // A follow-up with no new advances writes nothing.
266        let third = state.build_snapshot(3, 7, UnixNanos::from(60));
267
268        assert_eq!(first.advanced.len(), 3);
269        assert_eq!(second.marker_seq, 2);
270        assert_eq!(second.event_seq_before, 7);
271        // Only the two advanced streams appear, carrying their cursor's ts_init_hi and count.
272        assert_eq!(
273            second.advanced,
274            vec![
275                StreamCursor {
276                    slot: sa,
277                    ts_init_hi: UnixNanos::from(40),
278                    count: 2,
279                },
280                StreamCursor {
281                    slot: sc,
282                    ts_init_hi: UnixNanos::from(50),
283                    count: 2,
284                },
285            ]
286        );
287        assert!(third.is_none());
288    }
289
290    // The ordinal counts records within a contiguous run of equal ts_init and resets on any
291    // change. Per the design, the Phase 8 verifier owns detection of a decreasing per-slot
292    // ts_init, so advance treats a non-monotonic decrease as just another run boundary.
293    #[rstest]
294    #[case::repeats_then_advance(vec![100, 100, 100, 200, 200], vec![0, 1, 2, 0, 1])]
295    #[case::non_monotonic_resets_each_change(vec![100, 200, 100], vec![0, 0, 0])]
296    fn same_ts_ordinal_tracks_contiguous_runs(
297        #[case] timestamps: Vec<u64>,
298        #[case] expected: Vec<u32>,
299    ) {
300        let mut state = CursorState::new();
301        let trades: Topic = "data.trades.BINANCE.ETHUSDT".into();
302
303        let ordinals: Vec<u32> = timestamps
304            .into_iter()
305            .map(|ts| {
306                let (_, ordinal) = state.advance(
307                    trades,
308                    DataClass::Trade,
309                    "ETHUSDT.BINANCE",
310                    UnixNanos::from(ts),
311                );
312                ordinal
313            })
314            .collect();
315
316        assert_eq!(ordinals, expected);
317    }
318}
nautilus_event_store/markers/cursor.rs

nautilus_event_store/markers/
cursor.rs
