nautilus_event_store/markers/
capture.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//! Engine-thread data marker capture component.
17
18use std::{
19    any::Any,
20    sync::{
21        Arc,
22        atomic::{AtomicU64, Ordering},
23    },
24    time::Duration,
25};
26
27use ahash::AHashSet;
28use nautilus_core::UnixNanos;
29use nautilus_system::event_store::DataMarkerConfig;
30
31use crate::{
32    Topic,
33    markers::{
34        CursorState, DataMarkerExtractorRegistry, HiFiMarker, MarkerMsg, MarkerWriter,
35        StreamDictEntry,
36    },
37};
38
39/// Engine-thread component that captures data marker cursors at the bus boundary.
40///
41/// `DataMarkerCapture` owns the in-memory cursor state and the run-local marker sequence. It reads
42/// the shared entry submit counter with acquire ordering so marker records preserve engine-thread
43/// causal order while the writer lane remains non-blocking.
44#[derive(Debug)]
45pub struct DataMarkerCapture {
46    cursor: CursorState,
47    registry: DataMarkerExtractorRegistry,
48    writer: MarkerWriter,
49    submit_counter: Arc<AtomicU64>,
50    marker_seq: u64,
51    hifi: AHashSet<String>,
52    last_flush: UnixNanos,
53    safety_flush_interval: Duration,
54    lane_failed: bool,
55}
56
57impl DataMarkerCapture {
58    /// Creates a capture component over `registry`, `writer`, and the shared submit counter.
59    #[must_use]
60    pub fn new(
61        registry: DataMarkerExtractorRegistry,
62        writer: MarkerWriter,
63        submit_counter: Arc<AtomicU64>,
64        config: &DataMarkerConfig,
65    ) -> Self {
66        Self {
67            cursor: CursorState::new(),
68            registry,
69            writer,
70            submit_counter,
71            marker_seq: 0,
72            hifi: config.high_fidelity.iter().cloned().collect(),
73            last_flush: UnixNanos::default(),
74            safety_flush_interval: config.safety_flush_interval,
75            lane_failed: false,
76        }
77    }
78
79    /// Observes a bus publish and advances the data cursor when an extractor is registered.
80    pub fn observe_publish(&mut self, topic: Topic, message: &dyn Any, _now: UnixNanos) {
81        let Some(extractor) = self.registry.lookup(message) else {
82            return;
83        };
84        let event_seq_before = self.submit_counter.load(Ordering::Acquire);
85        let Some(identifier) = extractor.identifier(message) else {
86            return;
87        };
88        let Some((ts_event, ts_init)) = extractor.timestamps(message) else {
89            return;
90        };
91        let data_class = extractor.data_class();
92        let record_fingerprint = if self.hifi.contains(identifier.as_str()) {
93            let Some(record_fingerprint) = extractor.fingerprint(message) else {
94                return;
95            };
96            Some(record_fingerprint)
97        } else {
98            None
99        };
100
101        let (slot, same_ts_ordinal) = self.cursor.advance(topic, data_class, &identifier, ts_init);
102        self.drain_dict_entries();
103
104        if let Some(record_fingerprint) = record_fingerprint {
105            let marker_seq = self.marker_seq + 1;
106            let marker = HiFiMarker {
107                marker_seq,
108                event_seq_before,
109                slot,
110                ts_event,
111                ts_init,
112                same_ts_ordinal,
113                record_fingerprint,
114            };
115            self.submit_marker(MarkerMsg::HiFi(marker), marker_seq);
116        }
117    }
118
119    /// Emits a cursor snapshot for an event-store entry boundary when data advanced.
120    pub fn on_entry_submitted(&mut self, now: UnixNanos) {
121        self.flush_snapshot(now);
122    }
123
124    /// Emits a cursor snapshot when the safety interval has elapsed and data advanced.
125    pub fn maybe_safety_flush(&mut self, now: UnixNanos) {
126        if now
127            .duration_since(&self.last_flush)
128            .is_some_and(|elapsed| elapsed >= duration_nanos_saturating(self.safety_flush_interval))
129        {
130            self.flush_snapshot(now);
131        }
132    }
133
134    /// Closes the writer lane and seals the marker run.
135    pub fn close(self) {
136        self.writer.close();
137    }
138
139    fn drain_dict_entries(&mut self) {
140        for entry in self.cursor.take_new_dict_entries() {
141            self.submit_dict(entry);
142        }
143    }
144
145    fn submit_dict(&mut self, entry: StreamDictEntry) {
146        if self.writer.put_dict(entry).is_err() {
147            self.note_lane_failure();
148        }
149    }
150
151    fn flush_snapshot(&mut self, now: UnixNanos) {
152        let marker_seq = self.marker_seq + 1;
153        let event_seq_before = self.submit_counter.load(Ordering::Acquire);
154
155        if let Some(snapshot) = self
156            .cursor
157            .build_snapshot(marker_seq, event_seq_before, now)
158        {
159            self.submit_marker(MarkerMsg::Snapshot(snapshot), marker_seq);
160            self.last_flush = now;
161        }
162    }
163
164    fn submit_marker(&mut self, msg: MarkerMsg, marker_seq: u64) {
165        if self.writer.submit(msg, marker_seq).is_err() {
166            self.note_lane_failure();
167        }
168        self.marker_seq = marker_seq;
169    }
170
171    // The marker lane is best-effort by contract, but its death must be visible:
172    // without this latch the run's marker_seq just ends with no diagnostic.
173    fn note_lane_failure(&mut self) {
174        if !self.lane_failed {
175            self.lane_failed = true;
176            log::error!(
177                "Marker writer lane failed; data marker capture is disabled for the rest of the run"
178            );
179        }
180    }
181}
182
183fn duration_nanos_saturating(duration: Duration) -> u64 {
184    u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX)
185}
186
187#[cfg(test)]
188mod tests {
189    use std::{
190        any::Any,
191        sync::{
192            Arc,
193            atomic::{AtomicU64, Ordering},
194        },
195        time::Duration,
196    };
197
198    use nautilus_core::{UnixNanos, time::get_atomic_clock_static};
199    use nautilus_system::event_store::DataMarkerClass;
200    use rstest::rstest;
201
202    use super::{
203        super::test_support::{SharedMemoryMarker, SharedMemoryMarkerState},
204        *,
205    };
206    use crate::{
207        Topic,
208        manifest::RunStatus,
209        markers::{
210            DataClass, DataCursorSnapshot, DataMarkerExtractor, DataMarkerExtractorRegistry,
211            HiFiMarker, MarkerBackend, MarkerManifest, MarkerWriter, MarkerWriterConfig,
212            StreamDictEntry,
213        },
214    };
215
216    #[derive(Debug)]
217    struct TestQuote {
218        identifier: &'static str,
219        ts_event: UnixNanos,
220        ts_init: UnixNanos,
221        fingerprint: [u8; 32],
222    }
223
224    #[derive(Debug)]
225    struct IgnoredMessage;
226
227    #[derive(Debug)]
228    struct TestQuoteExtractor;
229
230    impl DataMarkerExtractor for TestQuoteExtractor {
231        fn data_class(&self) -> DataClass {
232            DataClass::Quote
233        }
234
235        fn identifier(&self, msg: &dyn Any) -> Option<String> {
236            msg.downcast_ref::<TestQuote>()
237                .map(|quote| quote.identifier.to_string())
238        }
239
240        fn timestamps(&self, msg: &dyn Any) -> Option<(UnixNanos, UnixNanos)> {
241            msg.downcast_ref::<TestQuote>()
242                .map(|quote| (quote.ts_event, quote.ts_init))
243        }
244
245        fn fingerprint(&self, msg: &dyn Any) -> Option<[u8; 32]> {
246            msg.downcast_ref::<TestQuote>()
247                .map(|quote| quote.fingerprint)
248        }
249    }
250
251    fn manifest() -> MarkerManifest {
252        MarkerManifest {
253            run_id: "1700000000-phase7".to_string(),
254            enabled_classes: vec![DataClass::Quote],
255            high_fidelity: true,
256            snapshot_count: 0,
257            hifi_count: 0,
258            gap_count: 0,
259            dict_count: 0,
260            status: RunStatus::Running,
261        }
262    }
263
264    fn registry() -> DataMarkerExtractorRegistry {
265        let mut registry = DataMarkerExtractorRegistry::new();
266        registry.register::<TestQuote>(Box::new(TestQuoteExtractor));
267        registry
268    }
269
270    fn config(high_fidelity: Vec<String>, safety_flush_interval: Duration) -> DataMarkerConfig {
271        DataMarkerConfig {
272            classes: vec![DataMarkerClass::Quote],
273            high_fidelity,
274            safety_flush_interval,
275            channel_capacity: 100,
276        }
277    }
278
279    fn quote(identifier: &'static str, ts_event: u64, ts_init: u64) -> TestQuote {
280        TestQuote {
281            identifier,
282            ts_event: UnixNanos::from(ts_event),
283            ts_init: UnixNanos::from(ts_init),
284            fingerprint: [7; 32],
285        }
286    }
287
288    fn open_capture(
289        config: &DataMarkerConfig,
290        submit_counter: Arc<AtomicU64>,
291    ) -> (DataMarkerCapture, SharedMemoryMarkerState) {
292        let (wrapper, shared) = SharedMemoryMarker::new();
293        shared.lock().open_run(manifest()).expect("open marker run");
294
295        let writer = MarkerWriter::spawn(
296            Box::new(wrapper),
297            get_atomic_clock_static(),
298            MarkerWriterConfig {
299                channel_capacity: 100,
300                max_batch: 1,
301                max_latency: Duration::from_millis(1),
302            },
303        )
304        .expect("spawn writer");
305
306        (
307            DataMarkerCapture::new(registry(), writer, submit_counter, config),
308            shared,
309        )
310    }
311
312    fn snapshots(shared: &SharedMemoryMarkerState) -> Vec<DataCursorSnapshot> {
313        shared.lock().scan_snapshots().expect("scan snapshots")
314    }
315
316    fn hifi(shared: &SharedMemoryMarkerState) -> Vec<HiFiMarker> {
317        shared.lock().scan_hifi().expect("scan hifi")
318    }
319
320    fn dict(shared: &SharedMemoryMarkerState) -> Vec<StreamDictEntry> {
321        shared.lock().scan_dict().expect("scan dict")
322    }
323
324    #[rstest]
325    fn event_seq_before_tracks_submit_counter() {
326        let submit_counter = Arc::new(AtomicU64::new(5));
327        let cfg = config(vec!["ETHUSDT.BINANCE".to_string()], Duration::from_secs(1));
328        let (mut capture, shared) = open_capture(&cfg, Arc::clone(&submit_counter));
329        let topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
330
331        capture.observe_publish(
332            topic,
333            &quote("ETHUSDT.BINANCE", 10, 20),
334            UnixNanos::from(20),
335        );
336        submit_counter.store(6, Ordering::Release);
337        capture.on_entry_submitted(UnixNanos::from(30));
338        capture.close();
339
340        let snapshots = snapshots(&shared);
341        let hifi = hifi(&shared);
342
343        assert_eq!(hifi.len(), 1);
344        assert_eq!(hifi[0].event_seq_before, 5);
345        assert_eq!(snapshots.len(), 1);
346        assert_eq!(snapshots[0].event_seq_before, 6);
347    }
348
349    #[rstest]
350    fn snapshot_written_on_entry_boundary() {
351        let submit_counter = Arc::new(AtomicU64::new(1));
352        let cfg = config(Vec::new(), Duration::from_secs(1));
353        let (mut capture, shared) = open_capture(&cfg, submit_counter);
354        let topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
355
356        capture.observe_publish(
357            topic,
358            &quote("ETHUSDT.BINANCE", 10, 20),
359            UnixNanos::from(20),
360        );
361        capture.on_entry_submitted(UnixNanos::from(30));
362        capture.close();
363
364        let snapshots = snapshots(&shared);
365        let dict = dict(&shared);
366
367        assert_eq!(snapshots.len(), 1);
368        assert_eq!(snapshots[0].marker_seq, 1);
369        assert_eq!(snapshots[0].event_seq_before, 1);
370        assert_eq!(snapshots[0].advanced[0].count, 1);
371        assert_eq!(
372            dict,
373            vec![StreamDictEntry {
374                slot: 0,
375                data_cls: DataClass::Quote,
376                identifier: "ETHUSDT.BINANCE".to_string(),
377            }]
378        );
379    }
380
381    #[rstest]
382    fn safety_flush_emits_without_entry() {
383        let submit_counter = Arc::new(AtomicU64::new(0));
384        let cfg = config(Vec::new(), Duration::from_nanos(10));
385        let (mut capture, shared) = open_capture(&cfg, submit_counter);
386        let topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
387
388        capture.observe_publish(
389            topic,
390            &quote("ETHUSDT.BINANCE", 10, 20),
391            UnixNanos::from(20),
392        );
393        capture.maybe_safety_flush(UnixNanos::from(30));
394        capture.close();
395
396        let snapshots = snapshots(&shared);
397
398        assert_eq!(snapshots.len(), 1);
399        assert_eq!(snapshots[0].event_seq_before, 0);
400        assert_eq!(snapshots[0].ts_init, UnixNanos::from(30));
401    }
402
403    #[rstest]
404    fn safety_flush_waits_for_interval_after_entry_boundary() {
405        let submit_counter = Arc::new(AtomicU64::new(1));
406        let cfg = config(Vec::new(), Duration::from_nanos(10));
407        let (mut capture, shared) = open_capture(&cfg, submit_counter);
408        let topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
409
410        capture.observe_publish(
411            topic,
412            &quote("ETHUSDT.BINANCE", 10, 20),
413            UnixNanos::from(20),
414        );
415        capture.on_entry_submitted(UnixNanos::from(100));
416        capture.observe_publish(
417            topic,
418            &quote("ETHUSDT.BINANCE", 30, 40),
419            UnixNanos::from(40),
420        );
421        capture.maybe_safety_flush(UnixNanos::from(109));
422        capture.maybe_safety_flush(UnixNanos::from(110));
423        capture.close();
424
425        let snapshots = snapshots(&shared);
426
427        assert_eq!(snapshots.len(), 2);
428        assert_eq!(snapshots[0].ts_init, UnixNanos::from(100));
429        assert_eq!(snapshots[0].advanced[0].count, 1);
430        assert_eq!(snapshots[1].ts_init, UnixNanos::from(110));
431        assert_eq!(snapshots[1].advanced[0].count, 2);
432    }
433
434    #[rstest]
435    fn hifi_marker_emitted_for_configured_instrument() {
436        let submit_counter = Arc::new(AtomicU64::new(2));
437        let cfg = config(vec!["ETHUSDT.BINANCE".to_string()], Duration::from_secs(1));
438        let (mut capture, shared) = open_capture(&cfg, submit_counter);
439        let eth_topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
440        let btc_topic: Topic = "data.quotes.BINANCE.BTCUSDT".into();
441
442        capture.observe_publish(
443            eth_topic,
444            &quote("ETHUSDT.BINANCE", 10, 20),
445            UnixNanos::from(20),
446        );
447        capture.observe_publish(
448            btc_topic,
449            &quote("BTCUSDT.BINANCE", 30, 40),
450            UnixNanos::from(40),
451        );
452        capture.close();
453
454        let hifi = hifi(&shared);
455
456        assert_eq!(hifi.len(), 1);
457        assert_eq!(hifi[0].marker_seq, 1);
458        assert_eq!(hifi[0].event_seq_before, 2);
459        assert_eq!(hifi[0].slot, 0);
460        assert_eq!(hifi[0].ts_event, UnixNanos::from(10));
461        assert_eq!(hifi[0].ts_init, UnixNanos::from(20));
462        assert_eq!(hifi[0].same_ts_ordinal, 0);
463        assert_eq!(hifi[0].record_fingerprint, [7; 32]);
464    }
465
466    #[rstest]
467    fn hifi_same_ts_ordinal_increments() {
468        let submit_counter = Arc::new(AtomicU64::new(0));
469        let cfg = config(vec!["ETHUSDT.BINANCE".to_string()], Duration::from_secs(1));
470        let (mut capture, shared) = open_capture(&cfg, submit_counter);
471        let topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
472
473        capture.observe_publish(
474            topic,
475            &quote("ETHUSDT.BINANCE", 10, 20),
476            UnixNanos::from(20),
477        );
478        capture.observe_publish(
479            topic,
480            &quote("ETHUSDT.BINANCE", 11, 20),
481            UnixNanos::from(20),
482        );
483        capture.close();
484
485        let hifi = hifi(&shared);
486
487        assert_eq!(hifi.len(), 2);
488        assert_eq!(
489            hifi.iter()
490                .map(|marker| marker.same_ts_ordinal)
491                .collect::<Vec<_>>(),
492            vec![0, 1]
493        );
494        assert_eq!(
495            hifi.iter()
496                .map(|marker| marker.marker_seq)
497                .collect::<Vec<_>>(),
498            vec![1, 2]
499        );
500    }
501
502    #[rstest]
503    fn unregistered_type_is_ignored() {
504        let submit_counter = Arc::new(AtomicU64::new(0));
505        let cfg = config(Vec::new(), Duration::from_nanos(1));
506        let (mut capture, shared) = open_capture(&cfg, submit_counter);
507        let topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
508
509        capture.observe_publish(topic, &IgnoredMessage, UnixNanos::from(20));
510        capture.maybe_safety_flush(UnixNanos::from(30));
511        capture.close();
512
513        assert!(snapshots(&shared).is_empty());
514        assert!(hifi(&shared).is_empty());
515        assert!(dict(&shared).is_empty());
516    }
517}
nautilus_event_store/markers/capture.rs

nautilus_event_store/markers/
capture.rs
