nautilus_event_store/capture/
adapter.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//! The bus capture adapter.
17//!
18//! [`BusCaptureAdapter`] is the seam between the message bus and the
19//! [`EventStoreWriter`]. The kernel calls [`BusCaptureAdapter::capture`] inside its bus
20//! dispatch wrappers, immediately before the message reaches downstream handlers, so
21//! every captured entry is durably submitted *before* a subscriber observes it. The
22//! adapter consults the [`EncoderRegistry`] allow-list to decide whether to capture, and
23//! converts the typed message into an [`EntryDraft`] for the writer.
24//!
25//! No-drop contract: any [`SubmitError`] from the writer fires the adapter's halt
26//! callback exactly once (so kernel fail-stop runs even when the writer's own halt path
27//! has not, such as when a caller closes the writer externally) and surfaces as
28//! [`CaptureError::Submit`]. Subsequent capture calls short-circuit with
29//! [`CaptureError::Halted`] without re-entering the writer.
30//!
31//! Under `cfg(madsim)` the writer's `submit` is a synchronous in-thread commit, so the
32//! adapter exposes the same surface and no thread-scheduling differences leak into
33//! tests.
34
35use std::{
36    collections::VecDeque,
37    fmt::Debug,
38    sync::{
39        Arc,
40        atomic::{AtomicBool, AtomicU64, Ordering},
41    },
42};
43
44use ahash::AHashSet;
45use nautilus_core::{UUID4, UnixNanos};
46use parking_lot::Mutex;
47
48use crate::{
49    capture::{encoder::EncodeError, registry::EncoderRegistry},
50    entry::Topic,
51    headers::Headers,
52    writer::{EntryDraft, EventStoreWriter, HaltCallback, HaltReason, SubmitError},
53};
54
55// Duplicate dispatches of the same message land within one engine cycle (endpoint send
56// followed by topic publish), so a small dedup window suffices and keeps memory flat.
57const RECENT_IDENTITY_CAPACITY: usize = 128;
58
59/// Errors returned by [`BusCaptureAdapter::capture`].
60///
61/// Each variant maps to a SPEC-named failure mode at the dispatch boundary; the kernel's
62/// fail-stop callback is the system response to [`CaptureError::Submit`] and
63/// [`CaptureError::Halted`].
64#[derive(Debug, thiserror::Error)]
65pub enum CaptureError {
66    /// The encoder rejected the message.
67    #[error("encode failure: {0}")]
68    Encode(#[from] EncodeError),
69    /// The writer rejected the submit.
70    ///
71    /// The adapter halt callback has fired before this error returns, so the kernel
72    /// fail-stop path is already in motion when the caller observes it.
73    #[error("writer submit failed: {0}")]
74    Submit(#[from] SubmitError),
75    /// A prior capture observed a writer failure and the adapter has fail-stopped.
76    ///
77    /// The halt callback fired on the original failure; subsequent captures short-circuit
78    /// without re-entering the writer to keep the no-drop contract intact (a stuck or
79    /// closed writer must not silently swallow captures).
80    #[error("capture adapter halted")]
81    Halted,
82}
83
84/// Captures bus traffic and forwards encoded entries to the [`EventStoreWriter`].
85///
86/// One adapter instance owns one writer; the kernel constructs the adapter after spawning
87/// the writer, then registers it with the bus dispatch wrappers. The adapter is `Send +
88/// Sync` so it can be shared between bus subscribers, but in practice the message bus is
89/// single-threaded and the adapter lives on the engine thread.
90pub struct BusCaptureAdapter {
91    writer: Arc<EventStoreWriter>,
92    registry: Arc<EncoderRegistry>,
93    halt: HaltCallback,
94    halted: AtomicBool,
95    submit_counter: Option<Arc<AtomicU64>>,
96    recent_identities: Mutex<RecentIdentities>,
97}
98
99// Insertion-ordered set of recently captured message identities with FIFO eviction.
100#[derive(Debug, Default)]
101struct RecentIdentities {
102    order: VecDeque<UUID4>,
103    seen: AHashSet<UUID4>,
104}
105
106impl RecentIdentities {
107    // Returns false when `identity` was already noted; records it otherwise.
108    fn note_fresh(&mut self, identity: UUID4) -> bool {
109        if self.seen.contains(&identity) {
110            return false;
111        }
112
113        if self.order.len() == RECENT_IDENTITY_CAPACITY
114            && let Some(evicted) = self.order.pop_front()
115        {
116            self.seen.remove(&evicted);
117        }
118        self.order.push_back(identity);
119        self.seen.insert(identity);
120        true
121    }
122}
123
124impl Debug for BusCaptureAdapter {
125    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
126        f.debug_struct(stringify!(BusCaptureAdapter))
127            .field("registered_encoders", &self.registry.len())
128            .field("halted", &self.halted.load(Ordering::Acquire))
129            .finish_non_exhaustive()
130    }
131}
132
133impl BusCaptureAdapter {
134    /// Constructs a new adapter over `writer`, `registry`, and `halt`.
135    ///
136    /// `halt` is the adapter-level fail-stop callback. The writer carries its own halt
137    /// callback for backend and backpressure failures; the adapter callback fires on any
138    /// submit error so [`SubmitError::Closed`] (which can originate outside the writer's
139    /// own halt path, e.g. an external close) still reaches the kernel.
140    #[must_use]
141    pub fn new(
142        writer: Arc<EventStoreWriter>,
143        registry: Arc<EncoderRegistry>,
144        halt: HaltCallback,
145    ) -> Self {
146        Self {
147            writer,
148            registry,
149            halt,
150            halted: AtomicBool::new(false),
151            submit_counter: None,
152            recent_identities: Mutex::new(RecentIdentities::default()),
153        }
154    }
155
156    /// Shares an entry-submit counter with the data-marker capture path.
157    #[must_use]
158    pub fn with_submit_counter(mut self, submit_counter: Arc<AtomicU64>) -> Self {
159        self.submit_counter = Some(submit_counter);
160        self
161    }
162
163    /// Returns whether the adapter has fail-stopped.
164    #[must_use]
165    pub fn is_halted(&self) -> bool {
166        self.halted.load(Ordering::Acquire)
167    }
168
169    /// Returns the encoder allow-list this adapter consults.
170    #[must_use]
171    pub fn registry(&self) -> &EncoderRegistry {
172        &self.registry
173    }
174
175    /// Returns the wrapped writer's current durable high-watermark.
176    #[must_use]
177    pub fn high_watermark(&self) -> u64 {
178        self.writer.high_watermark()
179    }
180
181    /// Captures a state-affecting bus message.
182    ///
183    /// Looks up the encoder for `T`, builds an [`EntryDraft`], and forwards it to the
184    /// writer. Returns `Ok(false)` when the type has no registered encoder so the adapter
185    /// can be wired into bus dispatch paths that carry a mix of state-affecting and
186    /// non-state-affecting messages without surfacing per-message errors, and when the
187    /// message's registered identity was already captured on another dispatch hop (the
188    /// same order event reaches the tap via the portfolio endpoint send and the strategy
189    /// topic publish; the log records it once).
190    ///
191    /// `topic` is the bus topic the message was dispatched on, `headers` are the
192    /// dispatch-time correlation headers (defaulting to [`Headers::empty`] until header
193    /// propagation lands across all message types), and `ts_init` is the domain
194    /// timestamp from `AtomicTime` (typically the message's own `ts_init` field).
195    ///
196    /// # Errors
197    ///
198    /// Returns:
199    ///
200    /// - [`CaptureError::Halted`] when a prior capture already observed a writer failure
201    ///   and the adapter has fail-stopped.
202    /// - [`CaptureError::Encode`] when the registered encoder rejects the message.
203    /// - [`CaptureError::Submit`] when the writer rejects the submit; the adapter halt
204    ///   callback fires before this error returns.
205    pub fn capture<T: 'static>(
206        &self,
207        topic: Topic,
208        message: &T,
209        headers: Headers,
210        ts_init: UnixNanos,
211    ) -> Result<bool, CaptureError> {
212        self.capture_any(topic, message as &dyn std::any::Any, headers, ts_init)
213    }
214
215    /// Type-erased counterpart to [`Self::capture`].
216    ///
217    /// Bus dispatch hands messages to the tap as `&dyn Any` because the static type is
218    /// not in scope at the registration site. This method dispatches on the concrete
219    /// type behind the trait object and follows the same fail-stop semantics as
220    /// [`Self::capture`].
221    ///
222    /// # Errors
223    ///
224    /// See [`Self::capture`].
225    pub fn capture_any(
226        &self,
227        topic: Topic,
228        message: &dyn std::any::Any,
229        headers: Headers,
230        ts_init: UnixNanos,
231    ) -> Result<bool, CaptureError> {
232        if self.halted.load(Ordering::Acquire) {
233            return Err(CaptureError::Halted);
234        }
235
236        // Encode before noting the identity: a rejected encode must be re-attempted
237        // on the message's next dispatch hop, not dropped as a duplicate.
238        let Some((payload_type, encoded)) = self.registry.encode_any(message)? else {
239            return Ok(false);
240        };
241
242        if let Some(identity) = self.registry.identity_for_any(message)
243            && !self.note_fresh_identity(identity)
244        {
245            return Ok(false);
246        }
247
248        let draft = EntryDraft {
249            headers,
250            topic,
251            payload_type,
252            payload: encoded.payload,
253            ts_init,
254            index_keys: encoded.index_keys,
255        };
256
257        match self.writer.submit(draft) {
258            Ok(()) => {
259                if let Some(submit_counter) = self.submit_counter.as_ref() {
260                    submit_counter.fetch_add(1, Ordering::AcqRel);
261                }
262                Ok(true)
263            }
264            Err(e) => {
265                self.fail_stop(&e);
266                Err(CaptureError::Submit(e))
267            }
268        }
269    }
270
271    fn note_fresh_identity(&self, identity: UUID4) -> bool {
272        self.recent_identities.lock().note_fresh(identity)
273    }
274
275    fn fail_stop(&self, err: &SubmitError) {
276        if self
277            .halted
278            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
279            .is_ok()
280        {
281            (self.halt)(halt_reason_from_submit(err));
282        }
283    }
284}
285
286/// Maps a [`SubmitError`] onto the [`HaltReason`] the adapter signals to its kernel.
287///
288/// [`SubmitError::HaltSignaled`] preserves the writer-side stall measurement so the
289/// kernel sees the same backpressure context the writer's own halt callback would carry.
290/// [`SubmitError::Closed`] surfaces as a backend error since the writer is no longer
291/// accepting work and the cause is opaque to the adapter (could be external close,
292/// crashed writer thread, or a terminal disk error already reported separately).
293fn halt_reason_from_submit(err: &SubmitError) -> HaltReason {
294    match err {
295        SubmitError::HaltSignaled {
296            stalled_for,
297            threshold,
298        } => HaltReason::BackpressureStall {
299            stalled_for: *stalled_for,
300            threshold: *threshold,
301        },
302        SubmitError::Closed => HaltReason::BackendError("event store writer closed".to_string()),
303    }
304}
305
306#[cfg(test)]
307mod tests {
308    use std::{
309        sync::{
310            Arc,
311            atomic::{AtomicU64, AtomicUsize, Ordering},
312        },
313        time::Duration,
314    };
315
316    use bytes::Bytes;
317    use indexmap::IndexMap;
318    use nautilus_core::{UUID4, UnixNanos, time::get_atomic_clock_static};
319    use parking_lot::Mutex;
320    use rstest::{fixture, rstest};
321    use ustr::Ustr;
322
323    use super::*;
324    use crate::{
325        backend::{AppendEntry, EventStore, IndexKey, IndexKind, MemoryBackend, ScanDirection},
326        capture::encoder::EncodedPayload,
327        entry::EventStoreEntry,
328        error::EventStoreError,
329        manifest::{RegisteredComponents, RunManifest, RunStatus},
330        writer::WriterConfig,
331    };
332
333    #[derive(Debug)]
334    struct StubCommand {
335        client_order_id: String,
336    }
337
338    #[derive(Debug)]
339    struct StubEvent {
340        client_order_id: String,
341        venue_order_id: String,
342    }
343
344    #[derive(Debug)]
345    struct UnknownMessage;
346
347    #[derive(Debug)]
348    struct FailingMessage;
349
350    #[derive(Debug)]
351    struct StubIdentifiedCommand {
352        id: UUID4,
353        payload: String,
354    }
355
356    fn manifest(run_id: &str) -> RunManifest {
357        RunManifest {
358            run_id: run_id.to_string(),
359            parent_run_id: None,
360            instance_id: "trader-001".to_string(),
361            binary_hash: "deadbeef".to_string(),
362            schema_version: 1,
363            crate_versions: "feedface".to_string(),
364            feature_flags: Vec::new(),
365            adapter_versions: IndexMap::new(),
366            config_hash: "cafebabe".to_string(),
367            registered_components: RegisteredComponents::default(),
368            seed: None,
369            start_ts_init: UnixNanos::from(0),
370            end_ts_init: None,
371            high_watermark: 0,
372            status: RunStatus::Running,
373        }
374    }
375
376    fn stub_registry() -> Arc<EncoderRegistry> {
377        let mut registry = EncoderRegistry::new();
378        registry.register::<StubCommand, _>(Ustr::from("StubCommand"), |c| {
379            Ok(EncodedPayload::new(
380                Bytes::copy_from_slice(c.client_order_id.as_bytes()),
381                vec![IndexKey::new(
382                    IndexKind::ClientOrderId,
383                    c.client_order_id.clone(),
384                )],
385            ))
386        });
387        registry.register::<StubEvent, _>(Ustr::from("StubEvent"), |e| {
388            Ok(EncodedPayload::new(
389                Bytes::copy_from_slice(e.client_order_id.as_bytes()),
390                vec![
391                    IndexKey::new(IndexKind::ClientOrderId, e.client_order_id.clone()),
392                    IndexKey::new(IndexKind::VenueOrderId, e.venue_order_id.clone()),
393                ],
394            ))
395        });
396        registry.register::<FailingMessage, _>(Ustr::from("FailingMessage"), |_| {
397            Err(EncodeError::Serialize(
398                "encoder rejected message".to_string(),
399            ))
400        });
401        Arc::new(registry)
402    }
403
404    #[fixture]
405    fn captured_halt() -> (HaltCallback, Arc<Mutex<Vec<HaltReason>>>) {
406        let captured: Arc<Mutex<Vec<HaltReason>>> = Arc::new(Mutex::new(Vec::new()));
407        let captured_for_cb = Arc::clone(&captured);
408        let halt: HaltCallback = Arc::new(move |reason| {
409            captured_for_cb.lock().push(reason);
410        });
411        (halt, captured)
412    }
413
414    fn writer_with_open_run(
415        run_id: &str,
416        halt: HaltCallback,
417    ) -> (Arc<EventStoreWriter>, Arc<Mutex<MemoryBackend>>) {
418        let backend_arc: Arc<Mutex<MemoryBackend>> = Arc::new(Mutex::new(MemoryBackend::new()));
419        backend_arc
420            .lock()
421            .open_run(manifest(run_id))
422            .expect("open run");
423
424        let wrapper = SharedMemory(Arc::clone(&backend_arc));
425        let writer = EventStoreWriter::spawn(
426            Box::new(wrapper),
427            get_atomic_clock_static(),
428            halt,
429            WriterConfig::default(),
430        )
431        .expect("spawn");
432        (Arc::new(writer), backend_arc)
433    }
434
435    /// Wraps a shared `MemoryBackend` so the writer thread can append while the test
436    /// reads the same instance from the engine thread.
437    #[derive(Debug)]
438    struct SharedMemory(Arc<Mutex<MemoryBackend>>);
439
440    impl EventStore for SharedMemory {
441        fn open_run(&mut self, _: RunManifest) -> Result<(), EventStoreError> {
442            unreachable!("test wrapper does not forward open_run")
443        }
444
445        fn append_batch(&mut self, entries: &[AppendEntry]) -> Result<u64, EventStoreError> {
446            self.0.lock().append_batch(entries)
447        }
448
449        fn scan_range(
450            &self,
451            from: u64,
452            to: u64,
453            direction: ScanDirection,
454        ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
455            self.0.lock().scan_range(from, to, direction)
456        }
457
458        fn scan_seq(&self, seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
459            self.0.lock().scan_seq(seq)
460        }
461
462        fn lookup(&self, kind: IndexKind, key: &str) -> Result<Option<u64>, EventStoreError> {
463            self.0.lock().lookup(kind, key)
464        }
465
466        fn iter_index_keys(&self, kind: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
467            self.0.lock().iter_index_keys(kind)
468        }
469
470        fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
471            self.0.lock().seal(status)
472        }
473
474        fn manifest(&self) -> Result<RunManifest, EventStoreError> {
475            self.0.lock().manifest()
476        }
477
478        fn high_watermark(&self) -> Result<u64, EventStoreError> {
479            self.0.lock().high_watermark()
480        }
481    }
482
483    fn drain(writer: &Arc<EventStoreWriter>, target_hwm: u64) {
484        let mut waited = Duration::ZERO;
485        let deadline = Duration::from_secs(2);
486        while writer.high_watermark() < target_hwm && waited < deadline {
487            std::thread::sleep(Duration::from_millis(5));
488            waited += Duration::from_millis(5);
489        }
490        assert!(
491            writer.high_watermark() >= target_hwm,
492            "writer high_watermark {} did not reach {target_hwm} within {:?}",
493            writer.high_watermark(),
494            deadline,
495        );
496    }
497
498    #[rstest]
499    fn capture_records_registered_command_and_returns_true(
500        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
501    ) {
502        let (halt, captured) = captured_halt;
503        let (writer, backend) = writer_with_open_run("run-cmd", Arc::clone(&halt));
504        let adapter = BusCaptureAdapter::new(Arc::clone(&writer), stub_registry(), halt);
505
506        let cmd = StubCommand {
507            client_order_id: "O-1".to_string(),
508        };
509        let captured_flag = adapter
510            .capture::<StubCommand>(
511                Topic::from("exec.command.SubmitOrder"),
512                &cmd,
513                Headers::empty(),
514                UnixNanos::from(100),
515            )
516            .expect("capture");
517
518        assert!(captured_flag);
519        drain(&writer, 1);
520
521        let backend = backend.lock();
522        let entry = backend.scan_seq(1).expect("scan").expect("present");
523        assert_eq!(entry.payload_type.as_str(), "StubCommand");
524        assert_eq!(entry.topic.as_ref(), "exec.command.SubmitOrder");
525        assert_eq!(entry.payload.as_ref(), b"O-1");
526
527        let seq = backend
528            .lookup(IndexKind::ClientOrderId, "O-1")
529            .expect("lookup")
530            .expect("indexed");
531        assert_eq!(seq, 1);
532
533        assert!(captured.lock().is_empty());
534        assert!(!adapter.is_halted());
535    }
536
537    #[rstest]
538    fn capture_returns_false_for_unknown_type(
539        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
540    ) {
541        let (halt, _captured) = captured_halt;
542        let (writer, _backend) = writer_with_open_run("run-unknown", Arc::clone(&halt));
543        let adapter = BusCaptureAdapter::new(Arc::clone(&writer), stub_registry(), halt);
544
545        let captured_flag = adapter
546            .capture::<UnknownMessage>(
547                Topic::from("data.market.unknown"),
548                &UnknownMessage,
549                Headers::empty(),
550                UnixNanos::from(50),
551            )
552            .expect("capture");
553
554        assert!(!captured_flag);
555        assert_eq!(writer.high_watermark(), 0);
556        assert!(!adapter.is_halted());
557    }
558
559    #[rstest]
560    fn submit_counter_increments_on_each_captured_entry(
561        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
562    ) {
563        let (halt, _captured) = captured_halt;
564        let (writer, _backend) = writer_with_open_run("run-submit-counter", Arc::clone(&halt));
565        let submit_counter = Arc::new(AtomicU64::new(1));
566        let adapter = BusCaptureAdapter::new(Arc::clone(&writer), stub_registry(), halt)
567            .with_submit_counter(Arc::clone(&submit_counter));
568
569        adapter
570            .capture::<StubCommand>(
571                Topic::from("exec.command.SubmitOrder"),
572                &StubCommand {
573                    client_order_id: "O-counter-1".to_string(),
574                },
575                Headers::empty(),
576                UnixNanos::from(100),
577            )
578            .expect("first capture");
579        adapter
580            .capture::<UnknownMessage>(
581                Topic::from("data.market.unknown"),
582                &UnknownMessage,
583                Headers::empty(),
584                UnixNanos::from(101),
585            )
586            .expect("unknown type");
587        adapter
588            .capture::<StubEvent>(
589                Topic::from("exec.event.OrderFilled"),
590                &StubEvent {
591                    client_order_id: "O-counter-1".to_string(),
592                    venue_order_id: "V-counter-1".to_string(),
593                },
594                Headers::empty(),
595                UnixNanos::from(102),
596            )
597            .expect("second capture");
598
599        assert_eq!(submit_counter.load(Ordering::Acquire), 3);
600    }
601
602    #[rstest]
603    fn capture_records_event_indices_atomically(
604        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
605    ) {
606        let (halt, _captured) = captured_halt;
607        let (writer, backend) = writer_with_open_run("run-event", Arc::clone(&halt));
608        let adapter = BusCaptureAdapter::new(Arc::clone(&writer), stub_registry(), halt);
609
610        let event = StubEvent {
611            client_order_id: "O-2".to_string(),
612            venue_order_id: "V-9".to_string(),
613        };
614        adapter
615            .capture::<StubEvent>(
616                Topic::from("exec.event.OrderFilled"),
617                &event,
618                Headers::empty(),
619                UnixNanos::from(200),
620            )
621            .expect("capture");
622        drain(&writer, 1);
623
624        let backend = backend.lock();
625        let by_client = backend
626            .lookup(IndexKind::ClientOrderId, "O-2")
627            .expect("lookup")
628            .expect("indexed");
629        let by_venue = backend
630            .lookup(IndexKind::VenueOrderId, "V-9")
631            .expect("lookup")
632            .expect("indexed");
633        assert_eq!(by_client, 1);
634        assert_eq!(by_venue, 1);
635    }
636
637    #[rstest]
638    fn capture_propagates_encoder_error_without_halting(
639        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
640    ) {
641        // An encoder failure is the encoder's contract violation, not a writer fail-stop:
642        // the caller should see CaptureError::Encode but the adapter must stay live so a
643        // subsequent capture for an allow-listed type still goes through.
644        let (halt, captured) = captured_halt;
645        let (writer, backend) = writer_with_open_run("run-encode-err", Arc::clone(&halt));
646        let adapter = BusCaptureAdapter::new(Arc::clone(&writer), stub_registry(), halt);
647
648        let err = adapter
649            .capture::<FailingMessage>(
650                Topic::from("exec.command.Failing"),
651                &FailingMessage,
652                Headers::empty(),
653                UnixNanos::from(500),
654            )
655            .expect_err("encoder must reject");
656
657        match err {
658            CaptureError::Encode(EncodeError::Serialize(msg)) => {
659                assert!(msg.contains("rejected"), "msg was: {msg}");
660            }
661            other => panic!("expected Encode(Serialize), was {other:?}"),
662        }
663        assert!(
664            !adapter.is_halted(),
665            "encoder failure must not fail-stop the adapter",
666        );
667        assert!(captured.lock().is_empty());
668
669        // Subsequent capture for a registered type still works.
670        adapter
671            .capture::<StubCommand>(
672                Topic::from("exec.command.SubmitOrder"),
673                &StubCommand {
674                    client_order_id: "O-after-encode-err".to_string(),
675                },
676                Headers::empty(),
677                UnixNanos::from(501),
678            )
679            .expect("capture after encoder error");
680        drain(&writer, 1);
681        let backend = backend.lock();
682        assert_eq!(backend.high_watermark().expect("hwm"), 1);
683    }
684
685    #[rstest]
686    fn capture_dedupes_second_dispatch_hop_by_identity(
687        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
688    ) {
689        let (halt, _captured) = captured_halt;
690        let (writer, backend) = writer_with_open_run("run-dedup", Arc::clone(&halt));
691
692        let mut registry = EncoderRegistry::new();
693        registry.register::<StubIdentifiedCommand, _>(Ustr::from("StubIdentified"), |c| {
694            Ok(EncodedPayload::new(
695                Bytes::copy_from_slice(c.payload.as_bytes()),
696                Vec::new(),
697            ))
698        });
699        registry.register_identity::<StubIdentifiedCommand, _>(|c| Some(c.id));
700        let adapter = BusCaptureAdapter::new(Arc::clone(&writer), Arc::new(registry), halt);
701
702        let command = StubIdentifiedCommand {
703            id: UUID4::new(),
704            payload: "queued-command".to_string(),
705        };
706        let first = adapter
707            .capture::<StubIdentifiedCommand>(
708                Topic::from("DataEngine.queue_execute"),
709                &command,
710                Headers::empty(),
711                UnixNanos::from(100),
712            )
713            .expect("first hop");
714        let second = adapter
715            .capture::<StubIdentifiedCommand>(
716                Topic::from("DataEngine.execute"),
717                &command,
718                Headers::empty(),
719                UnixNanos::from(101),
720            )
721            .expect("second hop");
722
723        assert!(first, "first dispatch hop must capture");
724        assert!(
725            !second,
726            "second dispatch hop of the same identity must dedupe"
727        );
728        drain(&writer, 1);
729        let backend = backend.lock();
730        assert_eq!(backend.high_watermark().expect("hwm"), 1);
731    }
732
733    #[rstest]
734    fn capture_retries_encode_on_next_hop_after_encoder_failure(
735        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
736    ) {
737        // The identity is noted only after a successful encode, so the next hop
738        // re-attempts a failed encode instead of deduping it.
739        let (halt, _captured) = captured_halt;
740        let (writer, backend) = writer_with_open_run("run-encode-retry", Arc::clone(&halt));
741
742        let attempts = Arc::new(AtomicUsize::new(0));
743        let attempts_for_encoder = Arc::clone(&attempts);
744        let mut registry = EncoderRegistry::new();
745        registry.register::<StubIdentifiedCommand, _>(Ustr::from("StubIdentified"), move |c| {
746            let attempt = attempts_for_encoder.fetch_add(1, Ordering::AcqRel);
747            if attempt == 0 {
748                return Err(EncodeError::Serialize(
749                    "transient encoder failure".to_string(),
750                ));
751            }
752            Ok(EncodedPayload::new(
753                Bytes::copy_from_slice(c.payload.as_bytes()),
754                Vec::new(),
755            ))
756        });
757        registry.register_identity::<StubIdentifiedCommand, _>(|c| Some(c.id));
758        let adapter = BusCaptureAdapter::new(Arc::clone(&writer), Arc::new(registry), halt);
759
760        let command = StubIdentifiedCommand {
761            id: UUID4::new(),
762            payload: "retry-me".to_string(),
763        };
764        let err = adapter
765            .capture::<StubIdentifiedCommand>(
766                Topic::from("DataEngine.queue_execute"),
767                &command,
768                Headers::empty(),
769                UnixNanos::from(100),
770            )
771            .expect_err("first encode must fail");
772        assert!(matches!(err, CaptureError::Encode(_)));
773
774        let retried = adapter
775            .capture::<StubIdentifiedCommand>(
776                Topic::from("DataEngine.execute"),
777                &command,
778                Headers::empty(),
779                UnixNanos::from(101),
780            )
781            .expect("second hop re-attempts encode");
782
783        assert!(retried, "encode retry must capture, was deduped");
784        assert_eq!(attempts.load(Ordering::Acquire), 2);
785        drain(&writer, 1);
786        let backend = backend.lock();
787        assert_eq!(backend.high_watermark().expect("hwm"), 1);
788    }
789
790    #[rstest]
791    #[case::backpressure(
792        SubmitError::HaltSignaled {
793            stalled_for: Duration::from_millis(750),
794            threshold: Duration::from_millis(250),
795        },
796        HaltReason::BackpressureStall {
797            stalled_for: Duration::from_millis(750),
798            threshold: Duration::from_millis(250),
799        },
800    )]
801    #[case::closed(
802        SubmitError::Closed,
803        HaltReason::BackendError("event store writer closed".to_string()),
804    )]
805    fn halt_reason_from_submit_preserves_failure_context(
806        #[case] err: SubmitError,
807        #[case] expected: HaltReason,
808    ) {
809        let actual = halt_reason_from_submit(&err);
810
811        match (actual, expected) {
812            (
813                HaltReason::BackpressureStall {
814                    stalled_for: a_s,
815                    threshold: a_t,
816                },
817                HaltReason::BackpressureStall {
818                    stalled_for: e_s,
819                    threshold: e_t,
820                },
821            ) => {
822                assert_eq!(a_s, e_s);
823                assert_eq!(a_t, e_t);
824            }
825            (HaltReason::BackendError(a), HaltReason::BackendError(e)) => {
826                assert_eq!(a, e);
827            }
828            (actual, expected) => {
829                panic!("variant mismatch: actual={actual:?} expected={expected:?}")
830            }
831        }
832    }
833
834    #[rstest]
835    fn submit_failure_halts_adapter_and_fires_callback_once(
836        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
837    ) {
838        // A halted writer surfaces SubmitError::Closed; the adapter must mirror that
839        // into a single halt-callback firing and then short-circuit subsequent captures
840        // without forwarding further submits.
841        let (halt, captured) = captured_halt;
842        let (writer, _backend) = writer_with_open_run("run-halt", Arc::clone(&halt));
843
844        // Close the writer behind the adapter's back so the next submit returns Closed.
845        let writer_clone = Arc::clone(&writer);
846        // Build the adapter before the close so it owns a strong ref the close path
847        // doesn't see.
848        let adapter = BusCaptureAdapter::new(writer_clone, stub_registry(), halt);
849
850        // Drop one of the outer Arc clones, then force a graceful close on the writer
851        // by unwrapping. We can't unwrap because the adapter holds a clone, so emulate a
852        // closed writer with a separate test that simulates the failure path through
853        // a stub. We use a stub writer adapter instead to keep this test deterministic.
854        drop(writer);
855
856        // Build a fresh adapter wired to a stub that always returns SubmitError::Closed
857        // so we exercise the halt path without depending on writer-internal lifecycle.
858        let halt_for_stub: HaltCallback = adapter_halt_for(&captured);
859        let stub_adapter = StubFailAdapter::new(halt_for_stub);
860
861        let err = stub_adapter
862            .capture::<StubCommand>(
863                Topic::from("exec.command.SubmitOrder"),
864                &StubCommand {
865                    client_order_id: "O-fail".to_string(),
866                },
867                Headers::empty(),
868                UnixNanos::from(1),
869            )
870            .expect_err("first submit fails");
871        assert!(matches!(err, CaptureError::Submit(SubmitError::Closed)));
872        assert!(stub_adapter.is_halted());
873        assert_eq!(captured.lock().len(), 1);
874
875        let err2 = stub_adapter
876            .capture::<StubCommand>(
877                Topic::from("exec.command.SubmitOrder"),
878                &StubCommand {
879                    client_order_id: "O-fail-2".to_string(),
880                },
881                Headers::empty(),
882                UnixNanos::from(2),
883            )
884            .expect_err("second submit short-circuits");
885        assert!(matches!(err2, CaptureError::Halted));
886        assert_eq!(
887            captured.lock().len(),
888            1,
889            "halt callback must not refire after the first failure",
890        );
891
892        // Drop the adapter so its writer Arc is released.
893        drop(adapter);
894    }
895
896    fn adapter_halt_for(captured: &Arc<Mutex<Vec<HaltReason>>>) -> HaltCallback {
897        let captured_for_cb = Arc::clone(captured);
898        Arc::new(move |reason| {
899            captured_for_cb.lock().push(reason);
900        })
901    }
902
903    /// Stand-in for [`BusCaptureAdapter`] that mirrors its halt-state machine but
904    /// always sees [`SubmitError::Closed`] from a synthetic writer. Lets the halt-path
905    /// test stay deterministic without racing against a real writer's shutdown sequence.
906    struct StubFailAdapter {
907        registry: Arc<EncoderRegistry>,
908        halt: HaltCallback,
909        halted: AtomicBool,
910    }
911
912    impl StubFailAdapter {
913        fn new(halt: HaltCallback) -> Self {
914            Self {
915                registry: stub_registry(),
916                halt,
917                halted: AtomicBool::new(false),
918            }
919        }
920
921        fn is_halted(&self) -> bool {
922            self.halted.load(Ordering::Acquire)
923        }
924
925        fn capture<T: 'static>(
926            &self,
927            _topic: Topic,
928            message: &T,
929            _headers: Headers,
930            _ts_init: UnixNanos,
931        ) -> Result<bool, CaptureError> {
932            if self.halted.load(Ordering::Acquire) {
933                return Err(CaptureError::Halted);
934            }
935            let Some((_pt, _encoded)) = self.registry.encode(message)? else {
936                return Ok(false);
937            };
938            let err = SubmitError::Closed;
939
940            if self
941                .halted
942                .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
943                .is_ok()
944            {
945                (self.halt)(super::halt_reason_from_submit(&err));
946            }
947            Err(CaptureError::Submit(err))
948        }
949    }
950}
nautilus_event_store/capture/adapter.rs

nautilus_event_store/capture/
adapter.rs
