nautilus_event_store/markers/
writer.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//! Non-blocking writer lane for the data marker sidecar.
17
18use std::time::Duration;
19
20use crate::{
21    error::EventStoreError,
22    markers::{DataCursorSnapshot, HiFiMarker, MarkerBackend, MarkerGap, StreamDictEntry},
23};
24
25/// Default channel capacity for markers pending the writer thread.
26pub const DEFAULT_MARKER_CHANNEL_CAPACITY: usize = 10_000;
27/// Default maximum number of markers collected before forcing a flush.
28pub const DEFAULT_MARKER_MAX_BATCH: usize = 100;
29/// Default maximum time a marker batch may accumulate before forcing a flush.
30pub const DEFAULT_MARKER_MAX_LATENCY: Duration = Duration::from_millis(5);
31
32/// Configuration knobs for the data marker writer.
33#[derive(Clone, Debug)]
34pub struct MarkerWriterConfig {
35    /// Capacity of the bounded `sync_channel` between submit and the writer thread.
36    pub channel_capacity: usize,
37    /// Maximum marker messages collected before a flush is forced.
38    pub max_batch: usize,
39    /// Maximum time a marker batch may accumulate before a flush is forced.
40    pub max_latency: Duration,
41}
42
43impl Default for MarkerWriterConfig {
44    fn default() -> Self {
45        Self {
46            channel_capacity: DEFAULT_MARKER_CHANNEL_CAPACITY,
47            max_batch: DEFAULT_MARKER_MAX_BATCH,
48            max_latency: DEFAULT_MARKER_MAX_LATENCY,
49        }
50    }
51}
52
53/// A message sent to the data marker writer.
54#[derive(Debug, Clone, PartialEq, Eq)]
55pub enum MarkerMsg {
56    /// Cursor snapshot marker.
57    Snapshot(DataCursorSnapshot),
58    /// High-fidelity per-record marker.
59    HiFi(HiFiMarker),
60    /// Stream dictionary entry.
61    Dict(StreamDictEntry),
62    /// Closes and seals the marker run after draining older messages.
63    Close,
64    #[doc(hidden)]
65    GapThen { gap: MarkerGap, msg: Box<Self> },
66}
67
68#[cfg(not(madsim))]
69mod imp {
70    use std::{
71        fmt::Debug,
72        sync::{
73            Mutex, PoisonError,
74            atomic::{AtomicBool, AtomicU64, Ordering},
75            mpsc::{self, RecvTimeoutError, SyncSender, TrySendError},
76        },
77        thread::{self, JoinHandle},
78        time::Instant,
79    };
80
81    use nautilus_core::time::AtomicTime;
82
83    use super::{
84        EventStoreError, MarkerBackend, MarkerGap, MarkerMsg, MarkerWriterConfig, StreamDictEntry,
85    };
86    use crate::{
87        manifest::RunStatus,
88        markers::{
89            MarkerGapReason, compute_dict_hash, compute_gap_hash, compute_hifi_hash,
90            compute_marker_hash,
91        },
92    };
93
94    const MARKER_WRITER_THREAD_NAME: &str = "event-store-marker-writer";
95
96    /// Dedicated marker writer thread.
97    pub struct MarkerWriter {
98        tx: Option<SyncSender<MarkerMsg>>,
99        handle: Option<JoinHandle<()>>,
100        last_submitted_seq: AtomicU64,
101        dropped: Mutex<Option<DroppedRange>>,
102        closed: AtomicBool,
103    }
104
105    #[derive(Debug, Clone, Copy)]
106    struct DroppedRange {
107        from_marker_seq: u64,
108        to_marker_seq: u64,
109    }
110
111    impl DroppedRange {
112        const fn new(marker_seq: u64) -> Self {
113            Self {
114                from_marker_seq: marker_seq,
115                to_marker_seq: marker_seq,
116            }
117        }
118
119        fn extend(&mut self, marker_seq: u64) {
120            if marker_seq < self.from_marker_seq {
121                self.from_marker_seq = marker_seq;
122            }
123
124            if marker_seq > self.to_marker_seq {
125                self.to_marker_seq = marker_seq;
126            }
127        }
128
129        const fn gap(self, reason: MarkerGapReason) -> MarkerGap {
130            MarkerGap {
131                from_marker_seq: self.from_marker_seq,
132                to_marker_seq: self.to_marker_seq,
133                reason,
134            }
135        }
136    }
137
138    impl Debug for MarkerWriter {
139        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
140            f.debug_struct(stringify!(MarkerWriter))
141                .field(
142                    "last_submitted_seq",
143                    &self.last_submitted_seq.load(Ordering::Acquire),
144                )
145                .field("closed", &self.closed.load(Ordering::Acquire))
146                .field("tx_attached", &self.tx.is_some())
147                .finish_non_exhaustive()
148        }
149    }
150
151    impl MarkerWriter {
152        /// Spawns the marker writer thread and takes ownership of `backend`.
153        ///
154        /// # Errors
155        ///
156        /// Returns [`EventStoreError::Backend`] when the backend has no open run or when the
157        /// writer thread cannot be spawned.
158        pub fn spawn(
159            backend: Box<dyn MarkerBackend + Send>,
160            _clock: &'static AtomicTime,
161            config: MarkerWriterConfig,
162        ) -> Result<Self, EventStoreError> {
163            backend.manifest()?;
164            // A zero capacity would create a rendezvous channel: try_send only succeeds
165            // while the writer thread is parked in recv, so most submits would overflow
166            // and put_dict would block the engine thread on every new stream.
167            let (tx, rx) = mpsc::sync_channel::<MarkerMsg>(config.channel_capacity.max(1));
168            let config_for_thread = config;
169
170            let handle = thread::Builder::new()
171                .name(MARKER_WRITER_THREAD_NAME.to_string())
172                .spawn(move || run(backend, &rx, &config_for_thread))
173                .map_err(|e| {
174                    EventStoreError::Backend(format!("spawn marker writer thread: {e}"))
175                })?;
176
177            Ok(Self {
178                tx: Some(tx),
179                handle: Some(handle),
180                last_submitted_seq: AtomicU64::new(0),
181                dropped: Mutex::new(None),
182                closed: AtomicBool::new(false),
183            })
184        }
185
186        /// Submits a marker without blocking the caller.
187        ///
188        /// Returns `Ok(false)` when the bounded channel is full; the dropped marker
189        /// sequence is recorded into the next overflow gap.
190        ///
191        /// # Errors
192        ///
193        /// Returns [`EventStoreError::Closed`] when the writer is closed, and
194        /// [`EventStoreError::Backend`] when `msg` is not a `Snapshot` or `HiFi` marker.
195        ///
196        /// # Panics
197        ///
198        /// Panics if the pending overflow range mutex is poisoned.
199        pub fn submit(&self, msg: MarkerMsg, marker_seq: u64) -> Result<bool, EventStoreError> {
200            if self.closed.load(Ordering::Acquire) {
201                return Err(EventStoreError::Closed);
202            }
203
204            if matches!(
205                msg,
206                MarkerMsg::Dict(_) | MarkerMsg::Close | MarkerMsg::GapThen { .. }
207            ) {
208                return Err(EventStoreError::Backend(
209                    "submit accepts Snapshot or HiFi marker messages".to_string(),
210                ));
211            }
212
213            let tx = self.tx.as_ref().ok_or(EventStoreError::Closed)?;
214            let mut dropped = self
215                .dropped
216                .lock()
217                .expect("marker dropped range mutex poisoned");
218            let outbound = if let Some(range) = *dropped {
219                MarkerMsg::GapThen {
220                    gap: range.gap(MarkerGapReason::Overflow),
221                    msg: Box::new(msg),
222                }
223            } else {
224                msg
225            };
226
227            match tx.try_send(outbound) {
228                Ok(()) => {
229                    *dropped = None;
230                    self.last_submitted_seq.store(marker_seq, Ordering::Release);
231                    Ok(true)
232                }
233                Err(TrySendError::Full(_)) => {
234                    extend_dropped_range(&mut dropped, marker_seq);
235                    Ok(false)
236                }
237                Err(TrySendError::Disconnected(_)) => {
238                    self.closed.store(true, Ordering::Release);
239                    Err(EventStoreError::Closed)
240                }
241            }
242        }
243
244        /// Submits a stream dictionary entry to the writer.
245        ///
246        /// Dictionary entries are one-time stream metadata, so this path waits for channel
247        /// capacity instead of dropping and gap-accounting them.
248        ///
249        /// # Errors
250        ///
251        /// Returns [`EventStoreError::Closed`] when the writer is closed.
252        pub fn put_dict(&self, entry: StreamDictEntry) -> Result<bool, EventStoreError> {
253            if self.closed.load(Ordering::Acquire) {
254                return Err(EventStoreError::Closed);
255            }
256
257            let tx = self.tx.as_ref().ok_or(EventStoreError::Closed)?;
258            if tx.send(MarkerMsg::Dict(entry)).is_ok() {
259                Ok(true)
260            } else {
261                self.closed.store(true, Ordering::Release);
262                Err(EventStoreError::Closed)
263            }
264        }
265
266        /// Drains pending markers and seals the marker run.
267        pub fn close(mut self) {
268            self.close_inner();
269        }
270
271        // Shared by close() and Drop: an implicit drop must still convert the pending
272        // overflow range into its WriterClosed gap record and drain buffered markers,
273        // or the verifier sees an unexplained marker_seq hole.
274        fn close_inner(&mut self) {
275            self.closed.store(true, Ordering::Release);
276
277            if let Some(tx) = self.tx.take() {
278                let close_msg = {
279                    let mut dropped = self.dropped.lock().unwrap_or_else(PoisonError::into_inner);
280                    if let Some(range) = dropped.take() {
281                        MarkerMsg::GapThen {
282                            gap: range.gap(MarkerGapReason::WriterClosed),
283                            msg: Box::new(MarkerMsg::Close),
284                        }
285                    } else {
286                        MarkerMsg::Close
287                    }
288                };
289                let _ = tx.send(close_msg);
290                drop(tx);
291            }
292
293            if let Some(handle) = self.handle.take()
294                && handle.join().is_err()
295            {
296                log::error!("Marker writer thread panicked");
297            }
298        }
299    }
300
301    impl Drop for MarkerWriter {
302        fn drop(&mut self) {
303            self.close_inner();
304        }
305    }
306
307    fn extend_dropped_range(dropped: &mut Option<DroppedRange>, marker_seq: u64) {
308        if let Some(range) = dropped {
309            range.extend(marker_seq);
310        } else {
311            *dropped = Some(DroppedRange::new(marker_seq));
312        }
313    }
314
315    fn run(
316        mut backend: Box<dyn MarkerBackend + Send>,
317        rx: &mpsc::Receiver<MarkerMsg>,
318        config: &MarkerWriterConfig,
319    ) {
320        let max_batch = config.max_batch.max(1);
321        let mut batch = Vec::with_capacity(max_batch);
322
323        while let Ok(first) = rx.recv() {
324            let mut should_close = push_batch(&mut batch, first);
325            let mut disconnected = false;
326            let started = Instant::now();
327
328            while !should_close && batch.len() < max_batch {
329                let Some(remaining) = config.max_latency.checked_sub(started.elapsed()) else {
330                    break;
331                };
332
333                if remaining.is_zero() {
334                    break;
335                }
336
337                match rx.recv_timeout(remaining) {
338                    Ok(msg) => should_close = push_batch(&mut batch, msg),
339                    Err(RecvTimeoutError::Timeout) => break,
340                    Err(RecvTimeoutError::Disconnected) => {
341                        disconnected = true;
342                        break;
343                    }
344                }
345            }
346
347            if let Err(e) = write_batch(backend.as_mut(), batch.drain(..)) {
348                log::error!(
349                    "Marker writer fail-stopped, marker capture is disabled for the rest of the run: {e}"
350                );
351                return;
352            }
353
354            if should_close {
355                if let Err(e) = backend.seal(RunStatus::Ended) {
356                    log::error!("Failed to seal marker run on close: {e}");
357                }
358                return;
359            }
360
361            if disconnected {
362                return;
363            }
364        }
365    }
366
367    fn push_batch(batch: &mut Vec<MarkerMsg>, msg: MarkerMsg) -> bool {
368        match msg {
369            MarkerMsg::Close => true,
370            other => {
371                let should_close = closes_after_msg(&other);
372                batch.push(other);
373                should_close
374            }
375        }
376    }
377
378    fn closes_after_msg(msg: &MarkerMsg) -> bool {
379        match msg {
380            MarkerMsg::Close => true,
381            MarkerMsg::GapThen { msg, .. } => closes_after_msg(msg),
382            MarkerMsg::Snapshot(_) | MarkerMsg::HiFi(_) | MarkerMsg::Dict(_) => false,
383        }
384    }
385
386    fn write_batch(
387        backend: &mut dyn MarkerBackend,
388        batch: impl IntoIterator<Item = MarkerMsg>,
389    ) -> Result<(), EventStoreError> {
390        for msg in batch {
391            write_msg(backend, msg)?;
392        }
393        Ok(())
394    }
395
396    fn write_msg(backend: &mut dyn MarkerBackend, msg: MarkerMsg) -> Result<(), EventStoreError> {
397        match msg {
398            MarkerMsg::Snapshot(snapshot) => {
399                backend.append_snapshot(&snapshot, compute_marker_hash(&snapshot))
400            }
401            MarkerMsg::HiFi(marker) => backend.append_hifi(&marker, compute_hifi_hash(&marker)),
402            MarkerMsg::Dict(entry) => backend.put_dict(&entry, compute_dict_hash(&entry)),
403            MarkerMsg::GapThen { gap, msg } => {
404                backend.append_gap(&gap, compute_gap_hash(&gap))?;
405                write_msg(backend, *msg)
406            }
407            MarkerMsg::Close => Ok(()),
408        }
409    }
410}
411
412#[cfg(madsim)]
413mod imp {
414    use std::{
415        fmt::Debug,
416        sync::{
417            Mutex,
418            atomic::{AtomicU64, Ordering},
419        },
420    };
421
422    use nautilus_core::time::AtomicTime;
423
424    use super::{EventStoreError, MarkerBackend, MarkerMsg, MarkerWriterConfig, StreamDictEntry};
425    use crate::{
426        manifest::RunStatus,
427        markers::{compute_dict_hash, compute_hifi_hash, compute_marker_hash},
428    };
429
430    /// Synchronous marker writer used under simulation.
431    pub struct MarkerWriter {
432        inner: Mutex<Inner>,
433        last_submitted_seq: AtomicU64,
434    }
435
436    struct Inner {
437        backend: Box<dyn MarkerBackend + Send>,
438        closed: bool,
439    }
440
441    impl Debug for MarkerWriter {
442        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
443            f.debug_struct(stringify!(MarkerWriter))
444                .field(
445                    "last_submitted_seq",
446                    &self.last_submitted_seq.load(Ordering::Acquire),
447                )
448                .finish_non_exhaustive()
449        }
450    }
451
452    impl MarkerWriter {
453        /// Constructs a synchronous marker writer over `backend`.
454        ///
455        /// # Errors
456        ///
457        /// Returns [`EventStoreError::Backend`] when the backend has no open run.
458        #[expect(
459            clippy::needless_pass_by_value,
460            reason = "synchronous writer keeps ownership of the marker backend"
461        )]
462        pub fn spawn(
463            backend: Box<dyn MarkerBackend + Send>,
464            _clock: &'static AtomicTime,
465            _config: MarkerWriterConfig,
466        ) -> Result<Self, EventStoreError> {
467            backend.manifest()?;
468            Ok(Self {
469                inner: Mutex::new(Inner {
470                    backend,
471                    closed: false,
472                }),
473                last_submitted_seq: AtomicU64::new(0),
474            })
475        }
476
477        /// Commits the marker synchronously.
478        ///
479        /// # Errors
480        ///
481        /// Returns [`EventStoreError::Closed`] when the writer is closed, and
482        /// [`EventStoreError::Backend`] when `msg` is not a `Snapshot` or `HiFi` marker.
483        ///
484        /// # Panics
485        ///
486        /// Panics if the writer mutex is poisoned.
487        pub fn submit(&self, msg: MarkerMsg, marker_seq: u64) -> Result<bool, EventStoreError> {
488            let mut inner = self.inner.lock().expect("marker writer mutex poisoned");
489
490            if inner.closed {
491                return Err(EventStoreError::Closed);
492            }
493
494            match msg {
495                MarkerMsg::Snapshot(snapshot) => inner
496                    .backend
497                    .append_snapshot(&snapshot, compute_marker_hash(&snapshot))?,
498                MarkerMsg::HiFi(marker) => inner
499                    .backend
500                    .append_hifi(&marker, compute_hifi_hash(&marker))?,
501                MarkerMsg::Dict(_) | MarkerMsg::Close | MarkerMsg::GapThen { .. } => {
502                    return Err(EventStoreError::Backend(
503                        "submit accepts Snapshot or HiFi marker messages".to_string(),
504                    ));
505                }
506            }
507
508            self.last_submitted_seq.store(marker_seq, Ordering::Release);
509            Ok(true)
510        }
511
512        /// Commits the stream dictionary entry synchronously.
513        ///
514        /// # Errors
515        ///
516        /// Returns [`EventStoreError::Closed`] when the writer is closed.
517        ///
518        /// # Panics
519        ///
520        /// Panics if the writer mutex is poisoned.
521        pub fn put_dict(&self, entry: StreamDictEntry) -> Result<bool, EventStoreError> {
522            let mut inner = self.inner.lock().expect("marker writer mutex poisoned");
523
524            if inner.closed {
525                return Err(EventStoreError::Closed);
526            }
527
528            inner.backend.put_dict(&entry, compute_dict_hash(&entry))?;
529            Ok(true)
530        }
531
532        /// Seals the marker run.
533        ///
534        /// # Panics
535        ///
536        /// Panics if the writer mutex is poisoned.
537        pub fn close(self) {
538            let mut inner = self.inner.lock().expect("marker writer mutex poisoned");
539
540            if !inner.closed {
541                let _ = inner.backend.seal(RunStatus::Ended);
542                inner.closed = true;
543            }
544        }
545    }
546}
547
548pub use imp::MarkerWriter;
549
550#[cfg(test)]
551#[cfg(not(madsim))]
552mod tests {
553    use std::{
554        sync::{
555            Arc, Mutex,
556            atomic::{AtomicUsize, Ordering},
557        },
558        time::{Duration, Instant},
559    };
560
561    use nautilus_core::{UnixNanos, time::get_atomic_clock_static};
562    use rstest::rstest;
563
564    use super::{super::test_support::SharedMemoryMarker, *};
565    use crate::{
566        error::EventStoreError,
567        manifest::RunStatus,
568        markers::{
569            DataClass, MarkerGap, MarkerGapReason, MarkerManifest, MemoryMarkerBackend,
570            StreamCursor, StreamDictEntry,
571        },
572    };
573
574    fn manifest(run_id: &str) -> MarkerManifest {
575        MarkerManifest {
576            run_id: run_id.to_string(),
577            enabled_classes: vec![DataClass::Quote, DataClass::Trade],
578            high_fidelity: true,
579            snapshot_count: 0,
580            hifi_count: 0,
581            gap_count: 0,
582            dict_count: 0,
583            status: RunStatus::Running,
584        }
585    }
586
587    fn snapshot(marker_seq: u64) -> DataCursorSnapshot {
588        DataCursorSnapshot {
589            marker_seq,
590            event_seq_before: marker_seq.saturating_sub(1),
591            ts_init: UnixNanos::from(1_700_000_000_000_000_000 + marker_seq),
592            advanced: vec![StreamCursor {
593                slot: 0,
594                ts_init_hi: UnixNanos::from(1_700_000_000_000_000_000 + marker_seq),
595                count: marker_seq,
596            }],
597        }
598    }
599
600    fn hifi(marker_seq: u64) -> HiFiMarker {
601        HiFiMarker {
602            marker_seq,
603            event_seq_before: marker_seq.saturating_sub(1),
604            slot: 0,
605            ts_event: UnixNanos::from(1_700_000_000_000_000_100 + marker_seq),
606            ts_init: UnixNanos::from(1_700_000_000_000_000_200 + marker_seq),
607            same_ts_ordinal: 0,
608            record_fingerprint: [7u8; 32],
609        }
610    }
611
612    #[derive(Debug)]
613    struct BlockingMarkerBackend {
614        inner: Arc<Mutex<MemoryMarkerBackend>>,
615        gate: Arc<(Mutex<bool>, std::sync::Condvar)>,
616        appends_seen: Arc<AtomicUsize>,
617    }
618
619    impl BlockingMarkerBackend {
620        fn new(
621            inner: Arc<Mutex<MemoryMarkerBackend>>,
622            gate: Arc<(Mutex<bool>, std::sync::Condvar)>,
623            appends_seen: Arc<AtomicUsize>,
624        ) -> Self {
625            Self {
626                inner,
627                gate,
628                appends_seen,
629            }
630        }
631
632        fn wait_for_release(&self) {
633            let (lock, cvar) = &*self.gate;
634            let mut released = lock.lock().expect("gate poisoned");
635
636            while !*released {
637                released = cvar.wait(released).expect("gate wait");
638            }
639        }
640    }
641
642    impl MarkerBackend for BlockingMarkerBackend {
643        fn open_run(&mut self, _: MarkerManifest) -> Result<(), EventStoreError> {
644            unreachable!("test wrapper does not forward open_run")
645        }
646
647        fn append_snapshot(
648            &mut self,
649            snapshot: &DataCursorSnapshot,
650            hash: [u8; 32],
651        ) -> Result<(), EventStoreError> {
652            self.appends_seen.fetch_add(1, Ordering::SeqCst);
653            self.wait_for_release();
654            self.inner
655                .lock()
656                .expect("inner marker poisoned")
657                .append_snapshot(snapshot, hash)
658        }
659
660        fn append_hifi(
661            &mut self,
662            marker: &HiFiMarker,
663            hash: [u8; 32],
664        ) -> Result<(), EventStoreError> {
665            self.appends_seen.fetch_add(1, Ordering::SeqCst);
666            self.wait_for_release();
667            self.inner
668                .lock()
669                .expect("inner marker poisoned")
670                .append_hifi(marker, hash)
671        }
672
673        fn append_gap(&mut self, gap: &MarkerGap, hash: [u8; 32]) -> Result<(), EventStoreError> {
674            self.appends_seen.fetch_add(1, Ordering::SeqCst);
675            self.wait_for_release();
676            self.inner
677                .lock()
678                .expect("inner marker poisoned")
679                .append_gap(gap, hash)
680        }
681
682        fn put_dict(
683            &mut self,
684            entry: &StreamDictEntry,
685            hash: [u8; 32],
686        ) -> Result<(), EventStoreError> {
687            self.inner
688                .lock()
689                .expect("inner marker poisoned")
690                .put_dict(entry, hash)
691        }
692
693        fn scan_snapshots(&self) -> Result<Vec<DataCursorSnapshot>, EventStoreError> {
694            self.inner
695                .lock()
696                .expect("inner marker poisoned")
697                .scan_snapshots()
698        }
699
700        fn scan_hifi(&self) -> Result<Vec<HiFiMarker>, EventStoreError> {
701            self.inner
702                .lock()
703                .expect("inner marker poisoned")
704                .scan_hifi()
705        }
706
707        fn scan_gaps(&self) -> Result<Vec<MarkerGap>, EventStoreError> {
708            self.inner
709                .lock()
710                .expect("inner marker poisoned")
711                .scan_gaps()
712        }
713
714        fn scan_dict(&self) -> Result<Vec<StreamDictEntry>, EventStoreError> {
715            self.inner
716                .lock()
717                .expect("inner marker poisoned")
718                .scan_dict()
719        }
720
721        fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
722            self.inner
723                .lock()
724                .expect("inner marker poisoned")
725                .seal(status)
726        }
727
728        fn manifest(&self) -> Result<MarkerManifest, EventStoreError> {
729            self.inner.lock().expect("inner marker poisoned").manifest()
730        }
731    }
732
733    fn wait_until(mut predicate: impl FnMut() -> bool, label: &str) {
734        let start = Instant::now();
735
736        while !predicate() {
737            assert!(
738                start.elapsed() < Duration::from_millis(500),
739                "timed out waiting for {label}"
740            );
741            std::thread::sleep(Duration::from_millis(5));
742        }
743    }
744
745    #[rstest]
746    fn submitted_snapshots_reach_backend() {
747        let (wrapper, shared) = SharedMemoryMarker::new();
748        shared
749            .lock()
750            .expect("shared marker")
751            .open_run(manifest("run-snapshots"))
752            .expect("open marker run");
753        let config = MarkerWriterConfig {
754            channel_capacity: 16,
755            max_batch: 2,
756            max_latency: Duration::from_secs(30),
757        };
758
759        let writer = MarkerWriter::spawn(Box::new(wrapper), get_atomic_clock_static(), config)
760            .expect("spawn marker writer");
761        let s1 = snapshot(1);
762        let s2 = snapshot(2);
763
764        assert!(
765            writer
766                .submit(MarkerMsg::Snapshot(s1.clone()), s1.marker_seq)
767                .expect("submit first")
768        );
769        assert!(
770            writer
771                .submit(MarkerMsg::Snapshot(s2.clone()), s2.marker_seq)
772                .expect("submit second")
773        );
774        writer.close();
775
776        let backend = shared.lock().expect("shared marker");
777        assert_eq!(
778            backend.scan_snapshots().expect("scan snapshots"),
779            vec![s1, s2]
780        );
781    }
782
783    #[rstest]
784    fn put_dict_waits_for_capacity_and_persists_metadata() {
785        let inner = Arc::new(Mutex::new(MemoryMarkerBackend::new()));
786        inner
787            .lock()
788            .expect("inner marker")
789            .open_run(manifest("run-dict-capacity"))
790            .expect("open marker run");
791        let gate = Arc::new((Mutex::new(false), std::sync::Condvar::new()));
792        let appends_seen = Arc::new(AtomicUsize::new(0));
793        let backend = BlockingMarkerBackend::new(
794            Arc::clone(&inner),
795            Arc::clone(&gate),
796            Arc::clone(&appends_seen),
797        );
798
799        let writer = MarkerWriter::spawn(
800            Box::new(backend),
801            get_atomic_clock_static(),
802            MarkerWriterConfig {
803                channel_capacity: 1,
804                max_batch: 1,
805                max_latency: Duration::from_secs(30),
806            },
807        )
808        .expect("spawn marker writer");
809
810        let first = snapshot(1);
811        assert!(
812            writer
813                .submit(MarkerMsg::Snapshot(first.clone()), first.marker_seq)
814                .expect("submit first")
815        );
816        wait_until(
817            || appends_seen.load(Ordering::SeqCst) == 1,
818            "writer to block in backend append",
819        );
820
821        let second = snapshot(2);
822        assert!(
823            writer
824                .submit(MarkerMsg::Snapshot(second.clone()), second.marker_seq)
825                .expect("submit second")
826        );
827
828        let entry = StreamDictEntry {
829            slot: 1,
830            data_cls: DataClass::Trade,
831            identifier: "BTCUSDT.BINANCE".to_string(),
832        };
833        let gate_for_release = Arc::clone(&gate);
834
835        let release = std::thread::spawn(move || {
836            std::thread::sleep(Duration::from_millis(20));
837            let (lock, cvar) = &*gate_for_release;
838            *lock.lock().expect("gate") = true;
839            cvar.notify_all();
840        });
841
842        assert!(writer.put_dict(entry.clone()).expect("put dict"));
843        release.join().expect("release gate");
844        writer.close();
845
846        let backend = inner.lock().expect("inner marker");
847        assert_eq!(backend.scan_dict().expect("scan dict"), vec![entry]);
848        assert_eq!(
849            backend
850                .scan_snapshots()
851                .expect("scan snapshots")
852                .into_iter()
853                .map(|snapshot| snapshot.marker_seq)
854                .collect::<Vec<_>>(),
855            vec![1, 2]
856        );
857    }
858
859    #[rstest]
860    fn spawn_requires_open_marker_run() {
861        let backend = MemoryMarkerBackend::new();
862
863        let err = MarkerWriter::spawn(
864            Box::new(backend),
865            get_atomic_clock_static(),
866            MarkerWriterConfig::default(),
867        )
868        .expect_err("spawn must reject a backend with no open run");
869
870        match err {
871            EventStoreError::Backend(msg) => {
872                assert!(msg.contains("no run open"), "msg was: {msg}");
873            }
874            other => panic!("expected Backend, was {other:?}"),
875        }
876    }
877
878    #[rstest]
879    fn latency_window_flushes_marker_before_close() {
880        let (wrapper, shared) = SharedMemoryMarker::new();
881        shared
882            .lock()
883            .expect("shared marker")
884            .open_run(manifest("run-latency"))
885            .expect("open marker run");
886
887        let writer = MarkerWriter::spawn(
888            Box::new(wrapper),
889            get_atomic_clock_static(),
890            MarkerWriterConfig {
891                channel_capacity: 16,
892                max_batch: 100,
893                max_latency: Duration::from_millis(20),
894            },
895        )
896        .expect("spawn marker writer");
897        let snap = snapshot(1);
898
899        assert!(
900            writer
901                .submit(MarkerMsg::Snapshot(snap.clone()), snap.marker_seq)
902                .expect("submit")
903        );
904        wait_until(
905            || {
906                shared
907                    .lock()
908                    .expect("shared marker")
909                    .scan_snapshots()
910                    .expect("scan")
911                    .len()
912                    == 1
913            },
914            "latency flush",
915        );
916        writer.close();
917
918        assert_eq!(
919            shared
920                .lock()
921                .expect("shared marker")
922                .scan_snapshots()
923                .expect("scan snapshots"),
924            vec![snap]
925        );
926    }
927
928    #[rstest]
929    fn overflow_drops_and_records_single_gap() {
930        let inner = Arc::new(Mutex::new(MemoryMarkerBackend::new()));
931        inner
932            .lock()
933            .expect("inner marker")
934            .open_run(manifest("run-overflow"))
935            .expect("open marker run");
936        let gate = Arc::new((Mutex::new(false), std::sync::Condvar::new()));
937        let appends_seen = Arc::new(AtomicUsize::new(0));
938        let backend = BlockingMarkerBackend::new(
939            Arc::clone(&inner),
940            Arc::clone(&gate),
941            Arc::clone(&appends_seen),
942        );
943
944        let writer = MarkerWriter::spawn(
945            Box::new(backend),
946            get_atomic_clock_static(),
947            MarkerWriterConfig {
948                channel_capacity: 1,
949                max_batch: 1,
950                max_latency: Duration::from_secs(30),
951            },
952        )
953        .expect("spawn marker writer");
954
955        let first = snapshot(1);
956        assert!(
957            writer
958                .submit(MarkerMsg::Snapshot(first.clone()), first.marker_seq)
959                .expect("submit first")
960        );
961        wait_until(
962            || appends_seen.load(Ordering::SeqCst) == 1,
963            "writer to block in backend append",
964        );
965
966        let second = snapshot(2);
967        assert!(
968            writer
969                .submit(MarkerMsg::Snapshot(second.clone()), second.marker_seq)
970                .expect("submit second")
971        );
972
973        let start = Instant::now();
974
975        for marker_seq in 3..=5 {
976            let dropped = snapshot(marker_seq);
977            assert!(
978                !writer
979                    .submit(MarkerMsg::Snapshot(dropped), marker_seq)
980                    .expect("submit drop")
981            );
982        }
983        assert!(
984            start.elapsed() < Duration::from_millis(100),
985            "overflow submits must not block"
986        );
987
988        let (lock, cvar) = &*gate;
989        *lock.lock().expect("gate") = true;
990        cvar.notify_all();
991
992        wait_until(
993            || {
994                inner
995                    .lock()
996                    .expect("inner marker")
997                    .scan_snapshots()
998                    .expect("scan")
999                    .len()
1000                    == 2
1001            },
1002            "first two snapshots to drain",
1003        );
1004
1005        let sixth = snapshot(6);
1006        assert!(
1007            writer
1008                .submit(MarkerMsg::Snapshot(sixth.clone()), sixth.marker_seq)
1009                .expect("submit after overflow")
1010        );
1011        writer.close();
1012
1013        let backend = inner.lock().expect("inner marker");
1014        let gaps = backend.scan_gaps().expect("scan gaps");
1015        assert_eq!(
1016            gaps,
1017            vec![MarkerGap {
1018                from_marker_seq: 3,
1019                to_marker_seq: 5,
1020                reason: MarkerGapReason::Overflow,
1021            }]
1022        );
1023        assert_eq!(
1024            backend
1025                .scan_snapshots()
1026                .expect("scan snapshots")
1027                .into_iter()
1028                .map(|snapshot| snapshot.marker_seq)
1029                .collect::<Vec<_>>(),
1030            vec![1, 2, 6]
1031        );
1032    }
1033
1034    #[rstest]
1035    fn overflow_gap_precedes_next_hifi_marker() {
1036        let inner = Arc::new(Mutex::new(MemoryMarkerBackend::new()));
1037        inner
1038            .lock()
1039            .expect("inner marker")
1040            .open_run(manifest("run-overflow-hifi"))
1041            .expect("open marker run");
1042        let gate = Arc::new((Mutex::new(false), std::sync::Condvar::new()));
1043        let appends_seen = Arc::new(AtomicUsize::new(0));
1044        let backend = BlockingMarkerBackend::new(
1045            Arc::clone(&inner),
1046            Arc::clone(&gate),
1047            Arc::clone(&appends_seen),
1048        );
1049
1050        let writer = MarkerWriter::spawn(
1051            Box::new(backend),
1052            get_atomic_clock_static(),
1053            MarkerWriterConfig {
1054                channel_capacity: 1,
1055                max_batch: 1,
1056                max_latency: Duration::from_secs(30),
1057            },
1058        )
1059        .expect("spawn marker writer");
1060
1061        let first = snapshot(1);
1062        assert!(
1063            writer
1064                .submit(MarkerMsg::Snapshot(first.clone()), first.marker_seq)
1065                .expect("submit first")
1066        );
1067        wait_until(
1068            || appends_seen.load(Ordering::SeqCst) == 1,
1069            "writer to block in backend append",
1070        );
1071
1072        let second = snapshot(2);
1073        assert!(
1074            writer
1075                .submit(MarkerMsg::Snapshot(second.clone()), second.marker_seq)
1076                .expect("submit second")
1077        );
1078
1079        for marker_seq in 3..=5 {
1080            let dropped = snapshot(marker_seq);
1081            assert!(
1082                !writer
1083                    .submit(MarkerMsg::Snapshot(dropped), marker_seq)
1084                    .expect("submit drop")
1085            );
1086        }
1087
1088        let (lock, cvar) = &*gate;
1089        *lock.lock().expect("gate") = true;
1090        cvar.notify_all();
1091
1092        wait_until(
1093            || {
1094                inner
1095                    .lock()
1096                    .expect("inner marker")
1097                    .scan_snapshots()
1098                    .expect("scan")
1099                    .len()
1100                    == 2
1101            },
1102            "first two snapshots to drain",
1103        );
1104
1105        let marker = hifi(6);
1106        assert!(
1107            writer
1108                .submit(MarkerMsg::HiFi(marker.clone()), marker.marker_seq)
1109                .expect("submit hifi after overflow")
1110        );
1111        writer.close();
1112
1113        let backend = inner.lock().expect("inner marker");
1114        assert_eq!(
1115            backend.scan_gaps().expect("scan gaps"),
1116            vec![MarkerGap {
1117                from_marker_seq: 3,
1118                to_marker_seq: 5,
1119                reason: MarkerGapReason::Overflow,
1120            }]
1121        );
1122        assert_eq!(backend.scan_hifi().expect("scan hifi"), vec![marker]);
1123    }
1124
1125    #[rstest]
1126    fn close_records_pending_drop_as_writer_closed_gap() {
1127        let inner = Arc::new(Mutex::new(MemoryMarkerBackend::new()));
1128        inner
1129            .lock()
1130            .expect("inner marker")
1131            .open_run(manifest("run-close-gap"))
1132            .expect("open marker run");
1133        let gate = Arc::new((Mutex::new(false), std::sync::Condvar::new()));
1134        let appends_seen = Arc::new(AtomicUsize::new(0));
1135        let backend = BlockingMarkerBackend::new(
1136            Arc::clone(&inner),
1137            Arc::clone(&gate),
1138            Arc::clone(&appends_seen),
1139        );
1140
1141        let writer = MarkerWriter::spawn(
1142            Box::new(backend),
1143            get_atomic_clock_static(),
1144            MarkerWriterConfig {
1145                channel_capacity: 1,
1146                max_batch: 1,
1147                max_latency: Duration::from_secs(30),
1148            },
1149        )
1150        .expect("spawn marker writer");
1151
1152        let first = snapshot(1);
1153        assert!(
1154            writer
1155                .submit(MarkerMsg::Snapshot(first.clone()), first.marker_seq)
1156                .expect("submit first")
1157        );
1158        wait_until(
1159            || appends_seen.load(Ordering::SeqCst) == 1,
1160            "writer to block in backend append",
1161        );
1162
1163        let second = snapshot(2);
1164        assert!(
1165            writer
1166                .submit(MarkerMsg::Snapshot(second.clone()), second.marker_seq)
1167                .expect("submit second")
1168        );
1169
1170        for marker_seq in 3..=4 {
1171            let dropped = snapshot(marker_seq);
1172            assert!(
1173                !writer
1174                    .submit(MarkerMsg::Snapshot(dropped), marker_seq)
1175                    .expect("submit drop")
1176            );
1177        }
1178
1179        let close_thread = std::thread::spawn(move || writer.close());
1180        std::thread::sleep(Duration::from_millis(20));
1181        let (lock, cvar) = &*gate;
1182        *lock.lock().expect("gate") = true;
1183        cvar.notify_all();
1184        close_thread.join().expect("close thread");
1185
1186        let backend = inner.lock().expect("inner marker");
1187        assert_eq!(
1188            backend.scan_gaps().expect("scan gaps"),
1189            vec![MarkerGap {
1190                from_marker_seq: 3,
1191                to_marker_seq: 4,
1192                reason: MarkerGapReason::WriterClosed,
1193            }]
1194        );
1195        assert_eq!(
1196            backend
1197                .scan_snapshots()
1198                .expect("scan snapshots")
1199                .into_iter()
1200                .map(|snapshot| snapshot.marker_seq)
1201                .collect::<Vec<_>>(),
1202            vec![1, 2]
1203        );
1204        assert_eq!(
1205            backend.manifest().expect("manifest").status,
1206            RunStatus::Ended
1207        );
1208    }
1209
1210    #[rstest]
1211    fn drop_records_pending_drop_as_writer_closed_gap() {
1212        // An implicit drop (panic unwind, owner teardown that skips close) must still
1213        // convert the pending overflow range into its WriterClosed gap record; losing
1214        // it leaves the verifier an unexplained marker_seq hole.
1215        let inner = Arc::new(Mutex::new(MemoryMarkerBackend::new()));
1216        inner
1217            .lock()
1218            .expect("inner marker")
1219            .open_run(manifest("run-drop-gap"))
1220            .expect("open marker run");
1221        let gate = Arc::new((Mutex::new(false), std::sync::Condvar::new()));
1222        let appends_seen = Arc::new(AtomicUsize::new(0));
1223        let backend = BlockingMarkerBackend::new(
1224            Arc::clone(&inner),
1225            Arc::clone(&gate),
1226            Arc::clone(&appends_seen),
1227        );
1228
1229        let writer = MarkerWriter::spawn(
1230            Box::new(backend),
1231            get_atomic_clock_static(),
1232            MarkerWriterConfig {
1233                channel_capacity: 1,
1234                max_batch: 1,
1235                max_latency: Duration::from_secs(30),
1236            },
1237        )
1238        .expect("spawn marker writer");
1239
1240        let first = snapshot(1);
1241        assert!(
1242            writer
1243                .submit(MarkerMsg::Snapshot(first.clone()), first.marker_seq)
1244                .expect("submit first")
1245        );
1246        wait_until(
1247            || appends_seen.load(Ordering::SeqCst) == 1,
1248            "writer to block in backend append",
1249        );
1250
1251        let second = snapshot(2);
1252        assert!(
1253            writer
1254                .submit(MarkerMsg::Snapshot(second.clone()), second.marker_seq)
1255                .expect("submit second")
1256        );
1257
1258        let dropped = snapshot(3);
1259        assert!(
1260            !writer
1261                .submit(MarkerMsg::Snapshot(dropped), 3)
1262                .expect("submit drop")
1263        );
1264
1265        let drop_thread = std::thread::spawn(move || drop(writer));
1266        std::thread::sleep(Duration::from_millis(20));
1267        let (lock, cvar) = &*gate;
1268        *lock.lock().expect("gate") = true;
1269        cvar.notify_all();
1270        drop_thread.join().expect("drop thread");
1271
1272        let backend = inner.lock().expect("inner marker");
1273        assert_eq!(
1274            backend.scan_gaps().expect("scan gaps"),
1275            vec![MarkerGap {
1276                from_marker_seq: 3,
1277                to_marker_seq: 3,
1278                reason: MarkerGapReason::WriterClosed,
1279            }]
1280        );
1281        assert_eq!(
1282            backend
1283                .scan_snapshots()
1284                .expect("scan snapshots")
1285                .into_iter()
1286                .map(|snapshot| snapshot.marker_seq)
1287                .collect::<Vec<_>>(),
1288            vec![1, 2]
1289        );
1290        assert_eq!(
1291            backend.manifest().expect("manifest").status,
1292            RunStatus::Ended
1293        );
1294    }
1295
1296    #[rstest]
1297    fn close_drains_and_seals() {
1298        let (wrapper, shared) = SharedMemoryMarker::new();
1299        shared
1300            .lock()
1301            .expect("shared marker")
1302            .open_run(manifest("run-close"))
1303            .expect("open marker run");
1304        let config = MarkerWriterConfig {
1305            channel_capacity: 16,
1306            max_batch: 10,
1307            max_latency: Duration::from_secs(30),
1308        };
1309
1310        let writer = MarkerWriter::spawn(Box::new(wrapper), get_atomic_clock_static(), config)
1311            .expect("spawn marker writer");
1312        let marker = hifi(1);
1313
1314        assert!(
1315            writer
1316                .submit(MarkerMsg::HiFi(marker.clone()), marker.marker_seq)
1317                .expect("submit hifi")
1318        );
1319        writer.close();
1320
1321        let backend = shared.lock().expect("shared marker");
1322        assert_eq!(backend.scan_hifi().expect("scan hifi"), vec![marker]);
1323        assert_eq!(
1324            backend.manifest().expect("manifest").status,
1325            RunStatus::Ended
1326        );
1327    }
1328}
1329
1330#[cfg(test)]
1331#[cfg(madsim)]
1332mod madsim_tests {
1333    use nautilus_core::{UnixNanos, time::get_atomic_clock_static};
1334    use rstest::rstest;
1335
1336    use super::{super::test_support::SharedMemoryMarker, *};
1337    use crate::{
1338        manifest::RunStatus,
1339        markers::{DataClass, MarkerManifest, StreamCursor},
1340    };
1341
1342    fn manifest(run_id: &str) -> MarkerManifest {
1343        MarkerManifest {
1344            run_id: run_id.to_string(),
1345            enabled_classes: vec![DataClass::Quote],
1346            high_fidelity: false,
1347            snapshot_count: 0,
1348            hifi_count: 0,
1349            gap_count: 0,
1350            dict_count: 0,
1351            status: RunStatus::Running,
1352        }
1353    }
1354
1355    fn snapshot(marker_seq: u64) -> DataCursorSnapshot {
1356        DataCursorSnapshot {
1357            marker_seq,
1358            event_seq_before: marker_seq.saturating_sub(1),
1359            ts_init: UnixNanos::from(1_700_000_000_000_000_000 + marker_seq),
1360            advanced: vec![StreamCursor {
1361                slot: 0,
1362                ts_init_hi: UnixNanos::from(1_700_000_000_000_000_000 + marker_seq),
1363                count: marker_seq,
1364            }],
1365        }
1366    }
1367
1368    #[rstest]
1369    fn submit_persists_synchronously_and_close_seals() {
1370        let (wrapper, shared) = SharedMemoryMarker::new();
1371        shared
1372            .lock()
1373            .expect("shared marker")
1374            .open_run(manifest("run-madsim"))
1375            .expect("open marker run");
1376
1377        let writer = MarkerWriter::spawn(
1378            Box::new(wrapper),
1379            get_atomic_clock_static(),
1380            MarkerWriterConfig::default(),
1381        )
1382        .expect("spawn marker writer");
1383        let snap = snapshot(1);
1384
1385        assert!(
1386            writer
1387                .submit(MarkerMsg::Snapshot(snap.clone()), snap.marker_seq)
1388                .expect("submit")
1389        );
1390        assert_eq!(
1391            shared
1392                .lock()
1393                .expect("shared marker")
1394                .scan_snapshots()
1395                .expect("scan snapshots"),
1396            vec![snap]
1397        );
1398
1399        writer.close();
1400
1401        assert_eq!(
1402            shared
1403                .lock()
1404                .expect("shared marker")
1405                .manifest()
1406                .expect("manifest")
1407                .status,
1408            RunStatus::Ended
1409        );
1410    }
1411}
nautilus_event_store/markers/writer.rs

nautilus_event_store/markers/
writer.rs
