nautilus-event-store 0.58.0

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! The bus capture adapter.
//!
//! [`BusCaptureAdapter`] is the seam between the message bus and the
//! [`EventStoreWriter`]. The kernel calls [`BusCaptureAdapter::capture`] inside its bus
//! dispatch wrappers, immediately before the message reaches downstream handlers, so
//! every captured entry is durably submitted *before* a subscriber observes it. The
//! adapter consults the [`EncoderRegistry`] allow-list to decide whether to capture, and
//! converts the typed message into an [`EntryDraft`] for the writer.
//!
//! No-drop contract: any [`SubmitError`] from the writer fires the adapter's halt
//! callback exactly once (so kernel fail-stop runs even when the writer's own halt path
//! has not, such as when a caller closes the writer externally) and surfaces as
//! [`CaptureError::Submit`]. Subsequent capture calls short-circuit with
//! [`CaptureError::Halted`] without re-entering the writer.
//!
//! Under `cfg(madsim)` the writer's `submit` is a synchronous in-thread commit, so the
//! adapter exposes the same surface and no thread-scheduling differences leak into
//! tests.

use std::{
    fmt::Debug,
    sync::{
        Arc,
        atomic::{AtomicBool, AtomicU64, Ordering},
    },
};

use nautilus_core::UnixNanos;

use crate::{
    capture::{encoder::EncodeError, registry::EncoderRegistry},
    entry::Topic,
    headers::Headers,
    writer::{EntryDraft, EventStoreWriter, HaltCallback, HaltReason, SubmitError},
};

/// Errors returned by [`BusCaptureAdapter::capture`].
///
/// Each variant maps to a SPEC-named failure mode at the dispatch boundary; the kernel's
/// fail-stop callback is the system response to [`CaptureError::Submit`] and
/// [`CaptureError::Halted`].
#[derive(Debug, thiserror::Error)]
pub enum CaptureError {
    /// The encoder rejected the message.
    #[error("encode failure: {0}")]
    Encode(#[from] EncodeError),
    /// The writer rejected the submit.
    ///
    /// The adapter halt callback has fired before this error returns, so the kernel
    /// fail-stop path is already in motion when the caller observes it.
    #[error("writer submit failed: {0}")]
    Submit(#[from] SubmitError),
    /// A prior capture observed a writer failure and the adapter has fail-stopped.
    ///
    /// The halt callback fired on the original failure; subsequent captures short-circuit
    /// without re-entering the writer to keep the no-drop contract intact (a stuck or
    /// closed writer must not silently swallow captures).
    #[error("capture adapter halted")]
    Halted,
}

/// Captures bus traffic and forwards encoded entries to the [`EventStoreWriter`].
///
/// One adapter instance owns one writer; the kernel constructs the adapter after spawning
/// the writer, then registers it with the bus dispatch wrappers. The adapter is `Send +
/// Sync` so it can be shared between bus subscribers, but in practice the message bus is
/// single-threaded and the adapter lives on the engine thread.
pub struct BusCaptureAdapter {
    writer: Arc<EventStoreWriter>,
    registry: Arc<EncoderRegistry>,
    halt: HaltCallback,
    halted: AtomicBool,
    submit_counter: Option<Arc<AtomicU64>>,
}

impl Debug for BusCaptureAdapter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct(stringify!(BusCaptureAdapter))
            .field("registered_encoders", &self.registry.len())
            .field("halted", &self.halted.load(Ordering::Acquire))
            .finish_non_exhaustive()
    }
}

impl BusCaptureAdapter {
    /// Constructs a new adapter over `writer`, `registry`, and `halt`.
    ///
    /// `halt` is the adapter-level fail-stop callback. The writer carries its own halt
    /// callback for backend and backpressure failures; the adapter callback fires on any
    /// submit error so [`SubmitError::Closed`] (which can originate outside the writer's
    /// own halt path, e.g. an external close) still reaches the kernel.
    #[must_use]
    pub fn new(
        writer: Arc<EventStoreWriter>,
        registry: Arc<EncoderRegistry>,
        halt: HaltCallback,
    ) -> Self {
        Self {
            writer,
            registry,
            halt,
            halted: AtomicBool::new(false),
            submit_counter: None,
        }
    }

    /// Shares an entry-submit counter with the data-marker capture path.
    #[must_use]
    pub fn with_submit_counter(mut self, submit_counter: Arc<AtomicU64>) -> Self {
        self.submit_counter = Some(submit_counter);
        self
    }

    /// Returns whether the adapter has fail-stopped.
    #[must_use]
    pub fn is_halted(&self) -> bool {
        self.halted.load(Ordering::Acquire)
    }

    /// Returns the encoder allow-list this adapter consults.
    #[must_use]
    pub fn registry(&self) -> &EncoderRegistry {
        &self.registry
    }

    /// Returns the wrapped writer's current durable high-watermark.
    #[must_use]
    pub fn high_watermark(&self) -> u64 {
        self.writer.high_watermark()
    }

    /// Captures a state-affecting bus message.
    ///
    /// Looks up the encoder for `T`, builds an [`EntryDraft`], and forwards it to the
    /// writer. Returns `Ok(false)` when the type has no registered encoder so the adapter
    /// can be wired into bus dispatch paths that carry a mix of state-affecting and
    /// non-state-affecting messages without surfacing per-message errors.
    ///
    /// `topic` is the bus topic the message was dispatched on, `headers` are the
    /// dispatch-time correlation headers (defaulting to [`Headers::empty`] until header
    /// propagation lands across all message types), and `ts_init` is the domain
    /// timestamp from `AtomicTime` (typically the message's own `ts_init` field).
    ///
    /// # Errors
    ///
    /// Returns:
    ///
    /// - [`CaptureError::Halted`] when a prior capture already observed a writer failure
    ///   and the adapter has fail-stopped.
    /// - [`CaptureError::Encode`] when the registered encoder rejects the message.
    /// - [`CaptureError::Submit`] when the writer rejects the submit; the adapter halt
    ///   callback fires before this error returns.
    pub fn capture<T: 'static>(
        &self,
        topic: Topic,
        message: &T,
        headers: Headers,
        ts_init: UnixNanos,
    ) -> Result<bool, CaptureError> {
        self.capture_any(topic, message as &dyn std::any::Any, headers, ts_init)
    }

    /// Type-erased counterpart to [`Self::capture`].
    ///
    /// Bus dispatch hands messages to the tap as `&dyn Any` because the static type is
    /// not in scope at the registration site. This method dispatches on the concrete
    /// type behind the trait object and follows the same fail-stop semantics as
    /// [`Self::capture`].
    ///
    /// # Errors
    ///
    /// See [`Self::capture`].
    pub fn capture_any(
        &self,
        topic: Topic,
        message: &dyn std::any::Any,
        headers: Headers,
        ts_init: UnixNanos,
    ) -> Result<bool, CaptureError> {
        if self.halted.load(Ordering::Acquire) {
            return Err(CaptureError::Halted);
        }

        let Some((payload_type, encoded)) = self.registry.encode_any(message)? else {
            return Ok(false);
        };

        let draft = EntryDraft {
            headers,
            topic,
            payload_type,
            payload: encoded.payload,
            ts_init,
            index_keys: encoded.index_keys,
        };

        match self.writer.submit(draft) {
            Ok(()) => {
                if let Some(submit_counter) = self.submit_counter.as_ref() {
                    submit_counter.fetch_add(1, Ordering::AcqRel);
                }
                Ok(true)
            }
            Err(e) => {
                self.fail_stop(&e);
                Err(CaptureError::Submit(e))
            }
        }
    }

    fn fail_stop(&self, err: &SubmitError) {
        if self
            .halted
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_ok()
        {
            (self.halt)(halt_reason_from_submit(err));
        }
    }
}

/// Maps a [`SubmitError`] onto the [`HaltReason`] the adapter signals to its kernel.
///
/// [`SubmitError::HaltSignaled`] preserves the writer-side stall measurement so the
/// kernel sees the same backpressure context the writer's own halt callback would carry.
/// [`SubmitError::Closed`] surfaces as a backend error since the writer is no longer
/// accepting work and the cause is opaque to the adapter (could be external close,
/// crashed writer thread, or a terminal disk error already reported separately).
fn halt_reason_from_submit(err: &SubmitError) -> HaltReason {
    match err {
        SubmitError::HaltSignaled {
            stalled_for,
            threshold,
        } => HaltReason::BackpressureStall {
            stalled_for: *stalled_for,
            threshold: *threshold,
        },
        SubmitError::Closed => HaltReason::BackendError("event store writer closed".to_string()),
    }
}

#[cfg(test)]
mod tests {
    use std::{
        sync::{
            Arc, Mutex,
            atomic::{AtomicU64, Ordering},
        },
        time::Duration,
    };

    use bytes::Bytes;
    use indexmap::IndexMap;
    use nautilus_core::{UnixNanos, time::get_atomic_clock_static};
    use rstest::{fixture, rstest};
    use ustr::Ustr;

    use super::*;
    use crate::{
        backend::{AppendEntry, EventStore, IndexKey, IndexKind, MemoryBackend, ScanDirection},
        capture::encoder::EncodedPayload,
        entry::EventStoreEntry,
        error::EventStoreError,
        manifest::{RegisteredComponents, RunManifest, RunStatus},
        writer::WriterConfig,
    };

    #[derive(Debug)]
    struct StubCommand {
        client_order_id: String,
    }

    #[derive(Debug)]
    struct StubEvent {
        client_order_id: String,
        venue_order_id: String,
    }

    #[derive(Debug)]
    struct UnknownMessage;

    #[derive(Debug)]
    struct FailingMessage;

    fn manifest(run_id: &str) -> RunManifest {
        RunManifest {
            run_id: run_id.to_string(),
            parent_run_id: None,
            instance_id: "trader-001".to_string(),
            binary_hash: "deadbeef".to_string(),
            schema_version: 1,
            crate_versions: "feedface".to_string(),
            feature_flags: Vec::new(),
            adapter_versions: IndexMap::new(),
            config_hash: "cafebabe".to_string(),
            registered_components: RegisteredComponents::default(),
            seed: None,
            start_ts_init: UnixNanos::from(0),
            end_ts_init: None,
            high_watermark: 0,
            status: RunStatus::Running,
        }
    }

    fn stub_registry() -> Arc<EncoderRegistry> {
        let mut registry = EncoderRegistry::new();
        registry.register::<StubCommand, _>(Ustr::from("StubCommand"), |c| {
            Ok(EncodedPayload::new(
                Bytes::copy_from_slice(c.client_order_id.as_bytes()),
                vec![IndexKey::new(
                    IndexKind::ClientOrderId,
                    c.client_order_id.clone(),
                )],
            ))
        });
        registry.register::<StubEvent, _>(Ustr::from("StubEvent"), |e| {
            Ok(EncodedPayload::new(
                Bytes::copy_from_slice(e.client_order_id.as_bytes()),
                vec![
                    IndexKey::new(IndexKind::ClientOrderId, e.client_order_id.clone()),
                    IndexKey::new(IndexKind::VenueOrderId, e.venue_order_id.clone()),
                ],
            ))
        });
        registry.register::<FailingMessage, _>(Ustr::from("FailingMessage"), |_| {
            Err(EncodeError::Serialize(
                "encoder rejected message".to_string(),
            ))
        });
        Arc::new(registry)
    }

    #[fixture]
    fn captured_halt() -> (HaltCallback, Arc<Mutex<Vec<HaltReason>>>) {
        let captured: Arc<Mutex<Vec<HaltReason>>> = Arc::new(Mutex::new(Vec::new()));
        let captured_for_cb = Arc::clone(&captured);
        let halt: HaltCallback = Arc::new(move |reason| {
            captured_for_cb
                .lock()
                .expect("captured halt poisoned")
                .push(reason);
        });
        (halt, captured)
    }

    fn writer_with_open_run(
        run_id: &str,
        halt: HaltCallback,
    ) -> (Arc<EventStoreWriter>, Arc<Mutex<MemoryBackend>>) {
        let backend_arc: Arc<Mutex<MemoryBackend>> = Arc::new(Mutex::new(MemoryBackend::new()));
        backend_arc
            .lock()
            .expect("inner")
            .open_run(manifest(run_id))
            .expect("open run");

        let wrapper = SharedMemory(Arc::clone(&backend_arc));
        let writer = EventStoreWriter::spawn(
            Box::new(wrapper),
            get_atomic_clock_static(),
            halt,
            WriterConfig::default(),
        )
        .expect("spawn");
        (Arc::new(writer), backend_arc)
    }

    /// Wraps a shared `MemoryBackend` so the writer thread can append while the test
    /// reads the same instance from the engine thread.
    #[derive(Debug)]
    struct SharedMemory(Arc<Mutex<MemoryBackend>>);

    impl EventStore for SharedMemory {
        fn open_run(&mut self, _: RunManifest) -> Result<(), EventStoreError> {
            unreachable!("test wrapper does not forward open_run")
        }

        fn append_batch(&mut self, entries: &[AppendEntry]) -> Result<u64, EventStoreError> {
            self.0.lock().expect("shared").append_batch(entries)
        }

        fn scan_range(
            &self,
            from: u64,
            to: u64,
            direction: ScanDirection,
        ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
            self.0
                .lock()
                .expect("shared")
                .scan_range(from, to, direction)
        }

        fn scan_seq(&self, seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
            self.0.lock().expect("shared").scan_seq(seq)
        }

        fn lookup(&self, kind: IndexKind, key: &str) -> Result<Option<u64>, EventStoreError> {
            self.0.lock().expect("shared").lookup(kind, key)
        }

        fn iter_index_keys(&self, kind: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
            self.0.lock().expect("shared").iter_index_keys(kind)
        }

        fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
            self.0.lock().expect("shared").seal(status)
        }

        fn manifest(&self) -> Result<RunManifest, EventStoreError> {
            self.0.lock().expect("shared").manifest()
        }

        fn high_watermark(&self) -> Result<u64, EventStoreError> {
            self.0.lock().expect("shared").high_watermark()
        }
    }

    fn drain(writer: &Arc<EventStoreWriter>, target_hwm: u64) {
        let mut waited = Duration::ZERO;
        let deadline = Duration::from_secs(2);
        while writer.high_watermark() < target_hwm && waited < deadline {
            std::thread::sleep(Duration::from_millis(5));
            waited += Duration::from_millis(5);
        }
        assert!(
            writer.high_watermark() >= target_hwm,
            "writer high_watermark {} did not reach {target_hwm} within {:?}",
            writer.high_watermark(),
            deadline,
        );
    }

    #[rstest]
    fn capture_records_registered_command_and_returns_true(
        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
    ) {
        let (halt, captured) = captured_halt;
        let (writer, backend) = writer_with_open_run("run-cmd", Arc::clone(&halt));
        let adapter = BusCaptureAdapter::new(Arc::clone(&writer), stub_registry(), halt);

        let cmd = StubCommand {
            client_order_id: "O-1".to_string(),
        };
        let captured_flag = adapter
            .capture::<StubCommand>(
                Topic::from("exec.command.SubmitOrder"),
                &cmd,
                Headers::empty(),
                UnixNanos::from(100),
            )
            .expect("capture");

        assert!(captured_flag);
        drain(&writer, 1);

        let backend = backend.lock().expect("backend");
        let entry = backend.scan_seq(1).expect("scan").expect("present");
        assert_eq!(entry.payload_type.as_str(), "StubCommand");
        assert_eq!(entry.topic.as_ref(), "exec.command.SubmitOrder");
        assert_eq!(entry.payload.as_ref(), b"O-1");

        let seq = backend
            .lookup(IndexKind::ClientOrderId, "O-1")
            .expect("lookup")
            .expect("indexed");
        assert_eq!(seq, 1);

        assert!(captured.lock().expect("captured").is_empty());
        assert!(!adapter.is_halted());
    }

    #[rstest]
    fn capture_returns_false_for_unknown_type(
        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
    ) {
        let (halt, _captured) = captured_halt;
        let (writer, _backend) = writer_with_open_run("run-unknown", Arc::clone(&halt));
        let adapter = BusCaptureAdapter::new(Arc::clone(&writer), stub_registry(), halt);

        let captured_flag = adapter
            .capture::<UnknownMessage>(
                Topic::from("data.market.unknown"),
                &UnknownMessage,
                Headers::empty(),
                UnixNanos::from(50),
            )
            .expect("capture");

        assert!(!captured_flag);
        assert_eq!(writer.high_watermark(), 0);
        assert!(!adapter.is_halted());
    }

    #[rstest]
    fn submit_counter_increments_on_each_captured_entry(
        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
    ) {
        let (halt, _captured) = captured_halt;
        let (writer, _backend) = writer_with_open_run("run-submit-counter", Arc::clone(&halt));
        let submit_counter = Arc::new(AtomicU64::new(1));
        let adapter = BusCaptureAdapter::new(Arc::clone(&writer), stub_registry(), halt)
            .with_submit_counter(Arc::clone(&submit_counter));

        adapter
            .capture::<StubCommand>(
                Topic::from("exec.command.SubmitOrder"),
                &StubCommand {
                    client_order_id: "O-counter-1".to_string(),
                },
                Headers::empty(),
                UnixNanos::from(100),
            )
            .expect("first capture");
        adapter
            .capture::<UnknownMessage>(
                Topic::from("data.market.unknown"),
                &UnknownMessage,
                Headers::empty(),
                UnixNanos::from(101),
            )
            .expect("unknown type");
        adapter
            .capture::<StubEvent>(
                Topic::from("exec.event.OrderFilled"),
                &StubEvent {
                    client_order_id: "O-counter-1".to_string(),
                    venue_order_id: "V-counter-1".to_string(),
                },
                Headers::empty(),
                UnixNanos::from(102),
            )
            .expect("second capture");

        assert_eq!(submit_counter.load(Ordering::Acquire), 3);
    }

    #[rstest]
    fn capture_records_event_indices_atomically(
        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
    ) {
        let (halt, _captured) = captured_halt;
        let (writer, backend) = writer_with_open_run("run-event", Arc::clone(&halt));
        let adapter = BusCaptureAdapter::new(Arc::clone(&writer), stub_registry(), halt);

        let event = StubEvent {
            client_order_id: "O-2".to_string(),
            venue_order_id: "V-9".to_string(),
        };
        adapter
            .capture::<StubEvent>(
                Topic::from("exec.event.OrderFilled"),
                &event,
                Headers::empty(),
                UnixNanos::from(200),
            )
            .expect("capture");
        drain(&writer, 1);

        let backend = backend.lock().expect("backend");
        let by_client = backend
            .lookup(IndexKind::ClientOrderId, "O-2")
            .expect("lookup")
            .expect("indexed");
        let by_venue = backend
            .lookup(IndexKind::VenueOrderId, "V-9")
            .expect("lookup")
            .expect("indexed");
        assert_eq!(by_client, 1);
        assert_eq!(by_venue, 1);
    }

    #[rstest]
    fn capture_propagates_encoder_error_without_halting(
        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
    ) {
        // An encoder failure is the encoder's contract violation, not a writer fail-stop:
        // the caller should see CaptureError::Encode but the adapter must stay live so a
        // subsequent capture for an allow-listed type still goes through.
        let (halt, captured) = captured_halt;
        let (writer, backend) = writer_with_open_run("run-encode-err", Arc::clone(&halt));
        let adapter = BusCaptureAdapter::new(Arc::clone(&writer), stub_registry(), halt);

        let err = adapter
            .capture::<FailingMessage>(
                Topic::from("exec.command.Failing"),
                &FailingMessage,
                Headers::empty(),
                UnixNanos::from(500),
            )
            .expect_err("encoder must reject");

        match err {
            CaptureError::Encode(EncodeError::Serialize(msg)) => {
                assert!(msg.contains("rejected"), "msg was: {msg}");
            }
            other => panic!("expected Encode(Serialize), was {other:?}"),
        }
        assert!(
            !adapter.is_halted(),
            "encoder failure must not fail-stop the adapter",
        );
        assert!(captured.lock().expect("captured").is_empty());

        // Subsequent capture for a registered type still works.
        adapter
            .capture::<StubCommand>(
                Topic::from("exec.command.SubmitOrder"),
                &StubCommand {
                    client_order_id: "O-after-encode-err".to_string(),
                },
                Headers::empty(),
                UnixNanos::from(501),
            )
            .expect("capture after encoder error");
        drain(&writer, 1);
        let backend = backend.lock().expect("backend");
        assert_eq!(backend.high_watermark().expect("hwm"), 1);
    }

    #[rstest]
    #[case::backpressure(
        SubmitError::HaltSignaled {
            stalled_for: Duration::from_millis(750),
            threshold: Duration::from_millis(250),
        },
        HaltReason::BackpressureStall {
            stalled_for: Duration::from_millis(750),
            threshold: Duration::from_millis(250),
        },
    )]
    #[case::closed(
        SubmitError::Closed,
        HaltReason::BackendError("event store writer closed".to_string()),
    )]
    fn halt_reason_from_submit_preserves_failure_context(
        #[case] err: SubmitError,
        #[case] expected: HaltReason,
    ) {
        let actual = halt_reason_from_submit(&err);

        match (actual, expected) {
            (
                HaltReason::BackpressureStall {
                    stalled_for: a_s,
                    threshold: a_t,
                },
                HaltReason::BackpressureStall {
                    stalled_for: e_s,
                    threshold: e_t,
                },
            ) => {
                assert_eq!(a_s, e_s);
                assert_eq!(a_t, e_t);
            }
            (HaltReason::BackendError(a), HaltReason::BackendError(e)) => {
                assert_eq!(a, e);
            }
            (actual, expected) => {
                panic!("variant mismatch: actual={actual:?} expected={expected:?}")
            }
        }
    }

    #[rstest]
    fn submit_failure_halts_adapter_and_fires_callback_once(
        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
    ) {
        // A halted writer surfaces SubmitError::Closed; the adapter must mirror that
        // into a single halt-callback firing and then short-circuit subsequent captures
        // without forwarding further submits.
        let (halt, captured) = captured_halt;
        let (writer, _backend) = writer_with_open_run("run-halt", Arc::clone(&halt));

        // Close the writer behind the adapter's back so the next submit returns Closed.
        let writer_clone = Arc::clone(&writer);
        // Build the adapter before the close so it owns a strong ref the close path
        // doesn't see.
        let adapter = BusCaptureAdapter::new(writer_clone, stub_registry(), halt);

        // Drop one of the outer Arc clones, then force a graceful close on the writer
        // by unwrapping. We can't unwrap because the adapter holds a clone, so emulate a
        // closed writer with a separate test that simulates the failure path through
        // a stub. We use a stub writer adapter instead to keep this test deterministic.
        drop(writer);

        // Build a fresh adapter wired to a stub that always returns SubmitError::Closed
        // so we exercise the halt path without depending on writer-internal lifecycle.
        let halt_for_stub: HaltCallback = adapter_halt_for(&captured);
        let stub_adapter = StubFailAdapter::new(halt_for_stub);

        let err = stub_adapter
            .capture::<StubCommand>(
                Topic::from("exec.command.SubmitOrder"),
                &StubCommand {
                    client_order_id: "O-fail".to_string(),
                },
                Headers::empty(),
                UnixNanos::from(1),
            )
            .expect_err("first submit fails");
        assert!(matches!(err, CaptureError::Submit(SubmitError::Closed)));
        assert!(stub_adapter.is_halted());
        assert_eq!(captured.lock().expect("captured").len(), 1);

        let err2 = stub_adapter
            .capture::<StubCommand>(
                Topic::from("exec.command.SubmitOrder"),
                &StubCommand {
                    client_order_id: "O-fail-2".to_string(),
                },
                Headers::empty(),
                UnixNanos::from(2),
            )
            .expect_err("second submit short-circuits");
        assert!(matches!(err2, CaptureError::Halted));
        assert_eq!(
            captured.lock().expect("captured").len(),
            1,
            "halt callback must not refire after the first failure",
        );

        // Drop the adapter so its writer Arc is released.
        drop(adapter);
    }

    fn adapter_halt_for(captured: &Arc<Mutex<Vec<HaltReason>>>) -> HaltCallback {
        let captured_for_cb = Arc::clone(captured);
        Arc::new(move |reason| {
            captured_for_cb
                .lock()
                .expect("captured halt poisoned")
                .push(reason);
        })
    }

    /// Stand-in for [`BusCaptureAdapter`] that mirrors its halt-state machine but
    /// always sees [`SubmitError::Closed`] from a synthetic writer. Lets the halt-path
    /// test stay deterministic without racing against a real writer's shutdown sequence.
    struct StubFailAdapter {
        registry: Arc<EncoderRegistry>,
        halt: HaltCallback,
        halted: AtomicBool,
    }

    impl StubFailAdapter {
        fn new(halt: HaltCallback) -> Self {
            Self {
                registry: stub_registry(),
                halt,
                halted: AtomicBool::new(false),
            }
        }

        fn is_halted(&self) -> bool {
            self.halted.load(Ordering::Acquire)
        }

        fn capture<T: 'static>(
            &self,
            _topic: Topic,
            message: &T,
            _headers: Headers,
            _ts_init: UnixNanos,
        ) -> Result<bool, CaptureError> {
            if self.halted.load(Ordering::Acquire) {
                return Err(CaptureError::Halted);
            }
            let Some((_pt, _encoded)) = self.registry.encode(message)? else {
                return Ok(false);
            };
            let err = SubmitError::Closed;

            if self
                .halted
                .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
                .is_ok()
            {
                (self.halt)(super::halt_reason_from_submit(&err));
            }
            Err(CaptureError::Submit(err))
        }
    }
}