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
294            .lock()
295            .expect("shared marker")
296            .open_run(manifest())
297            .expect("open marker run");
298
299        let writer = MarkerWriter::spawn(
300            Box::new(wrapper),
301            get_atomic_clock_static(),
302            MarkerWriterConfig {
303                channel_capacity: 100,
304                max_batch: 1,
305                max_latency: Duration::from_millis(1),
306            },
307        )
308        .expect("spawn writer");
309
310        (
311            DataMarkerCapture::new(registry(), writer, submit_counter, config),
312            shared,
313        )
314    }
315
316    fn snapshots(shared: &SharedMemoryMarkerState) -> Vec<DataCursorSnapshot> {
317        shared
318            .lock()
319            .expect("shared marker")
320            .scan_snapshots()
321            .expect("scan snapshots")
322    }
323
324    fn hifi(shared: &SharedMemoryMarkerState) -> Vec<HiFiMarker> {
325        shared
326            .lock()
327            .expect("shared marker")
328            .scan_hifi()
329            .expect("scan hifi")
330    }
331
332    fn dict(shared: &SharedMemoryMarkerState) -> Vec<StreamDictEntry> {
333        shared
334            .lock()
335            .expect("shared marker")
336            .scan_dict()
337            .expect("scan dict")
338    }
339
340    #[rstest]
341    fn event_seq_before_tracks_submit_counter() {
342        let submit_counter = Arc::new(AtomicU64::new(5));
343        let cfg = config(vec!["ETHUSDT.BINANCE".to_string()], Duration::from_secs(1));
344        let (mut capture, shared) = open_capture(&cfg, Arc::clone(&submit_counter));
345        let topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
346
347        capture.observe_publish(
348            topic,
349            &quote("ETHUSDT.BINANCE", 10, 20),
350            UnixNanos::from(20),
351        );
352        submit_counter.store(6, Ordering::Release);
353        capture.on_entry_submitted(UnixNanos::from(30));
354        capture.close();
355
356        let snapshots = snapshots(&shared);
357        let hifi = hifi(&shared);
358
359        assert_eq!(hifi.len(), 1);
360        assert_eq!(hifi[0].event_seq_before, 5);
361        assert_eq!(snapshots.len(), 1);
362        assert_eq!(snapshots[0].event_seq_before, 6);
363    }
364
365    #[rstest]
366    fn snapshot_written_on_entry_boundary() {
367        let submit_counter = Arc::new(AtomicU64::new(1));
368        let cfg = config(Vec::new(), Duration::from_secs(1));
369        let (mut capture, shared) = open_capture(&cfg, submit_counter);
370        let topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
371
372        capture.observe_publish(
373            topic,
374            &quote("ETHUSDT.BINANCE", 10, 20),
375            UnixNanos::from(20),
376        );
377        capture.on_entry_submitted(UnixNanos::from(30));
378        capture.close();
379
380        let snapshots = snapshots(&shared);
381        let dict = dict(&shared);
382
383        assert_eq!(snapshots.len(), 1);
384        assert_eq!(snapshots[0].marker_seq, 1);
385        assert_eq!(snapshots[0].event_seq_before, 1);
386        assert_eq!(snapshots[0].advanced[0].count, 1);
387        assert_eq!(
388            dict,
389            vec![StreamDictEntry {
390                slot: 0,
391                data_cls: DataClass::Quote,
392                identifier: "ETHUSDT.BINANCE".to_string(),
393            }]
394        );
395    }
396
397    #[rstest]
398    fn safety_flush_emits_without_entry() {
399        let submit_counter = Arc::new(AtomicU64::new(0));
400        let cfg = config(Vec::new(), Duration::from_nanos(10));
401        let (mut capture, shared) = open_capture(&cfg, submit_counter);
402        let topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
403
404        capture.observe_publish(
405            topic,
406            &quote("ETHUSDT.BINANCE", 10, 20),
407            UnixNanos::from(20),
408        );
409        capture.maybe_safety_flush(UnixNanos::from(30));
410        capture.close();
411
412        let snapshots = snapshots(&shared);
413
414        assert_eq!(snapshots.len(), 1);
415        assert_eq!(snapshots[0].event_seq_before, 0);
416        assert_eq!(snapshots[0].ts_init, UnixNanos::from(30));
417    }
418
419    #[rstest]
420    fn safety_flush_waits_for_interval_after_entry_boundary() {
421        let submit_counter = Arc::new(AtomicU64::new(1));
422        let cfg = config(Vec::new(), Duration::from_nanos(10));
423        let (mut capture, shared) = open_capture(&cfg, submit_counter);
424        let topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
425
426        capture.observe_publish(
427            topic,
428            &quote("ETHUSDT.BINANCE", 10, 20),
429            UnixNanos::from(20),
430        );
431        capture.on_entry_submitted(UnixNanos::from(100));
432        capture.observe_publish(
433            topic,
434            &quote("ETHUSDT.BINANCE", 30, 40),
435            UnixNanos::from(40),
436        );
437        capture.maybe_safety_flush(UnixNanos::from(109));
438        capture.maybe_safety_flush(UnixNanos::from(110));
439        capture.close();
440
441        let snapshots = snapshots(&shared);
442
443        assert_eq!(snapshots.len(), 2);
444        assert_eq!(snapshots[0].ts_init, UnixNanos::from(100));
445        assert_eq!(snapshots[0].advanced[0].count, 1);
446        assert_eq!(snapshots[1].ts_init, UnixNanos::from(110));
447        assert_eq!(snapshots[1].advanced[0].count, 2);
448    }
449
450    #[rstest]
451    fn hifi_marker_emitted_for_configured_instrument() {
452        let submit_counter = Arc::new(AtomicU64::new(2));
453        let cfg = config(vec!["ETHUSDT.BINANCE".to_string()], Duration::from_secs(1));
454        let (mut capture, shared) = open_capture(&cfg, submit_counter);
455        let eth_topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
456        let btc_topic: Topic = "data.quotes.BINANCE.BTCUSDT".into();
457
458        capture.observe_publish(
459            eth_topic,
460            &quote("ETHUSDT.BINANCE", 10, 20),
461            UnixNanos::from(20),
462        );
463        capture.observe_publish(
464            btc_topic,
465            &quote("BTCUSDT.BINANCE", 30, 40),
466            UnixNanos::from(40),
467        );
468        capture.close();
469
470        let hifi = hifi(&shared);
471
472        assert_eq!(hifi.len(), 1);
473        assert_eq!(hifi[0].marker_seq, 1);
474        assert_eq!(hifi[0].event_seq_before, 2);
475        assert_eq!(hifi[0].slot, 0);
476        assert_eq!(hifi[0].ts_event, UnixNanos::from(10));
477        assert_eq!(hifi[0].ts_init, UnixNanos::from(20));
478        assert_eq!(hifi[0].same_ts_ordinal, 0);
479        assert_eq!(hifi[0].record_fingerprint, [7; 32]);
480    }
481
482    #[rstest]
483    fn hifi_same_ts_ordinal_increments() {
484        let submit_counter = Arc::new(AtomicU64::new(0));
485        let cfg = config(vec!["ETHUSDT.BINANCE".to_string()], Duration::from_secs(1));
486        let (mut capture, shared) = open_capture(&cfg, submit_counter);
487        let topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
488
489        capture.observe_publish(
490            topic,
491            &quote("ETHUSDT.BINANCE", 10, 20),
492            UnixNanos::from(20),
493        );
494        capture.observe_publish(
495            topic,
496            &quote("ETHUSDT.BINANCE", 11, 20),
497            UnixNanos::from(20),
498        );
499        capture.close();
500
501        let hifi = hifi(&shared);
502
503        assert_eq!(hifi.len(), 2);
504        assert_eq!(
505            hifi.iter()
506                .map(|marker| marker.same_ts_ordinal)
507                .collect::<Vec<_>>(),
508            vec![0, 1]
509        );
510        assert_eq!(
511            hifi.iter()
512                .map(|marker| marker.marker_seq)
513                .collect::<Vec<_>>(),
514            vec![1, 2]
515        );
516    }
517
518    #[rstest]
519    fn unregistered_type_is_ignored() {
520        let submit_counter = Arc::new(AtomicU64::new(0));
521        let cfg = config(Vec::new(), Duration::from_nanos(1));
522        let (mut capture, shared) = open_capture(&cfg, submit_counter);
523        let topic: Topic = "data.quotes.BINANCE.ETHUSDT".into();
524
525        capture.observe_publish(topic, &IgnoredMessage, UnixNanos::from(20));
526        capture.maybe_safety_flush(UnixNanos::from(30));
527        capture.close();
528
529        assert!(snapshots(&shared).is_empty());
530        assert!(hifi(&shared).is_empty());
531        assert!(dict(&shared).is_empty());
532    }
533}
nautilus_event_store/markers/capture.rs

nautilus_event_store/markers/
capture.rs
