nautilus_event_store/writer/
mod.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//! Dedicated writer for the event store.
17//!
18//! The writer owns a single backend instance and exposes a thread-safe `submit` entry
19//! point. Captured entries enter via a bounded `std::sync::mpsc::sync_channel` and the
20//! writer thread drains them into batched, atomic `append_batch` commits. The
21//! high-watermark only advances on durable acknowledgement; a stalled submit or a
22//! backend-side disk/corruption failure fires the registered halt callback.
23//!
24//! Under `cfg(madsim)` the writer drops the channel and the dedicated thread, mirroring
25//! the logger's simulation pattern: submits commit synchronously on the calling thread so
26//! tests assert against an authoritative in-process log without thread scheduling.
27
28pub mod halt;
29
30// `batcher` carries thread-loop routines gated out under cfg(madsim) since the
31// synchronous path bypasses the channel and the run loop, but `build_append_entry`
32// is reused in both paths so the module stays compiled either way.
33mod batcher;
34
35use std::time::Duration;
36
37use bytes::Bytes;
38pub use halt::{HaltCallback, HaltReason, noop_halt};
39use nautilus_core::UnixNanos;
40
41use crate::{
42    backend::IndexKey,
43    entry::{PayloadType, Topic},
44    headers::Headers,
45    snapshot::SnapshotAnchor,
46};
47
48/// Default channel capacity for entries pending the writer thread.
49pub const DEFAULT_CHANNEL_CAPACITY: usize = 10_000;
50/// Default maximum number of entries collected before forcing a commit.
51pub const DEFAULT_MAX_BATCH_ENTRIES: usize = 100;
52/// Default maximum time a batch may accumulate before forcing a commit.
53pub const DEFAULT_MAX_BATCH_LATENCY: Duration = Duration::from_millis(5);
54/// Default submit-side stall ceiling that fires the halt callback.
55pub const DEFAULT_HALT_THRESHOLD: Duration = Duration::from_millis(250);
56
57/// Configuration knobs for the writer.
58#[derive(Clone, Debug)]
59pub struct WriterConfig {
60    /// Capacity of the bounded `sync_channel` between submit and the writer thread.
61    pub channel_capacity: usize,
62    /// Maximum entries collected before a commit is forced.
63    pub max_batch_entries: usize,
64    /// Maximum time a batch may accumulate before a commit is forced.
65    pub max_batch_latency: Duration,
66    /// Submit-side stall ceiling. A submit that blocks longer than this fires the halt
67    /// callback once and returns [`SubmitError::HaltSignaled`].
68    pub halt_threshold: Duration,
69}
70
71impl Default for WriterConfig {
72    fn default() -> Self {
73        Self {
74            channel_capacity: DEFAULT_CHANNEL_CAPACITY,
75            max_batch_entries: DEFAULT_MAX_BATCH_ENTRIES,
76            max_batch_latency: DEFAULT_MAX_BATCH_LATENCY,
77            halt_threshold: DEFAULT_HALT_THRESHOLD,
78        }
79    }
80}
81
82/// An unsealed entry handed to [`EventStoreWriter::submit`].
83///
84/// `seq`, `ts_publish`, and `entry_hash` are stamped by the writer; everything else is the
85/// captured message identity plus encoder output.
86#[derive(Clone, Debug, PartialEq, Eq)]
87pub struct EntryDraft {
88    /// First-class correlation headers.
89    pub headers: Headers,
90    /// The bus topic the entry was captured on.
91    pub topic: Topic,
92    /// The canonical payload type tag.
93    pub payload_type: PayloadType,
94    /// The encoded payload bytes.
95    pub payload: Bytes,
96    /// The domain timestamp from `AtomicTime`.
97    pub ts_init: UnixNanos,
98    /// Sidecar index keys produced by the encoder.
99    pub index_keys: Vec<IndexKey>,
100}
101
102impl EntryDraft {
103    /// Creates a new [`EntryDraft`] with no sidecar index keys.
104    #[must_use]
105    pub const fn without_indices(
106        headers: Headers,
107        topic: Topic,
108        payload_type: PayloadType,
109        payload: Bytes,
110        ts_init: UnixNanos,
111    ) -> Self {
112        Self {
113            headers,
114            topic,
115            payload_type,
116            payload,
117            ts_init,
118            index_keys: Vec::new(),
119        }
120    }
121}
122
123/// Errors returned by [`EventStoreWriter::submit`].
124#[derive(Debug, thiserror::Error)]
125pub enum SubmitError {
126    /// The writer is shut down or the writer thread has exited.
127    #[error("writer is closed")]
128    Closed,
129    /// The submit blocked longer than the configured halt threshold; the halt callback
130    /// has been fired.
131    #[error("submit stalled for {stalled_for:?}, halt threshold {threshold:?}")]
132    HaltSignaled {
133        /// How long the submit blocked before signaling halt.
134        stalled_for: Duration,
135        /// The configured threshold the stall exceeded.
136        threshold: Duration,
137    },
138}
139
140#[cfg(not(madsim))]
141mod imp {
142    use std::{
143        fmt::Debug,
144        sync::{
145            Arc,
146            atomic::{AtomicBool, AtomicU64, Ordering},
147            mpsc::{self, RecvTimeoutError, SyncSender, TrySendError},
148        },
149        thread::{self, JoinHandle},
150        time::{Duration, Instant},
151    };
152
153    use nautilus_core::time::AtomicTime;
154
155    use super::{
156        EntryDraft, SnapshotAnchor, SubmitError, WriterConfig,
157        batcher::{self, HaltSink, WriterMessage},
158        halt::{self, HaltCallback, HaltReason},
159    };
160    use crate::{backend::EventStore, error::EventStoreError};
161
162    const WRITER_THREAD_NAME: &str = "event-store-writer";
163    const SUBMIT_RETRY_INTERVAL: Duration = Duration::from_micros(100);
164
165    /// The dedicated event store writer.
166    pub struct EventStoreWriter {
167        tx: Option<SyncSender<WriterMessage>>,
168        handle: Option<JoinHandle<()>>,
169        high_watermark: Arc<AtomicU64>,
170        halt: HaltCallback,
171        halt_threshold: Duration,
172        // Shared with the writer thread so any halt fire latches it exactly once;
173        // subsequent submits return Closed instead of re-entering the retry loop.
174        halted: Arc<AtomicBool>,
175        clock: &'static AtomicTime,
176    }
177
178    impl Debug for EventStoreWriter {
179        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
180            f.debug_struct(stringify!(EventStoreWriter))
181                .field(
182                    "high_watermark",
183                    &self.high_watermark.load(Ordering::Acquire),
184                )
185                .field("halt_threshold", &self.halt_threshold)
186                .field("halted", &self.halted.load(Ordering::Acquire))
187                .field("tx_attached", &self.tx.is_some())
188                .finish_non_exhaustive()
189        }
190    }
191
192    impl EventStoreWriter {
193        /// Spawns the writer thread and takes ownership of `backend`.
194        ///
195        /// The backend must already have an open run; the writer reads its current
196        /// high-watermark to seed the next assigned `seq`.
197        ///
198        /// # Errors
199        ///
200        /// Returns an error if the backend has no open run or when the writer thread cannot be
201        /// spawned.
202        pub fn spawn(
203            backend: Box<dyn EventStore + Send>,
204            clock: &'static AtomicTime,
205            halt: HaltCallback,
206            config: WriterConfig,
207        ) -> Result<Self, EventStoreError> {
208            let initial_hwm = backend.high_watermark()?;
209            let high_watermark = Arc::new(AtomicU64::new(initial_hwm));
210            let halted = Arc::new(AtomicBool::new(false));
211            // Zero capacity is a rendezvous channel: a commit stall longer than
212            // the halt threshold would fail-stop instead of being absorbed.
213            let (tx, rx) = mpsc::sync_channel::<WriterMessage>(config.channel_capacity.max(1));
214
215            let watermark_for_thread = Arc::clone(&high_watermark);
216            let halt_for_thread = Arc::clone(&halt);
217            let halted_for_thread = Arc::clone(&halted);
218            let halt_threshold = config.halt_threshold;
219            let config_for_thread = config;
220
221            let handle = thread::Builder::new()
222                .name(WRITER_THREAD_NAME.to_string())
223                .spawn(move || {
224                    batcher::run(
225                        backend,
226                        rx,
227                        config_for_thread,
228                        HaltSink::new(halt_for_thread, halted_for_thread),
229                        watermark_for_thread,
230                        clock,
231                    );
232                })
233                .map_err(|e| EventStoreError::Backend(format!("spawn writer thread: {e}")))?;
234
235            Ok(Self {
236                tx: Some(tx),
237                handle: Some(handle),
238                high_watermark,
239                halt,
240                halt_threshold,
241                halted,
242                clock,
243            })
244        }
245
246        /// Submits a captured entry. Stamps `ts_publish` from the clock at receive time
247        /// and hands the draft to the writer thread.
248        ///
249        /// Blocks (with retry) when the channel is full. If the cumulative wait exceeds
250        /// the halt threshold, signals halt, firing the callback unless an earlier
251        /// condition already did, and returns [`SubmitError::HaltSignaled`];
252        /// subsequent submits return [`SubmitError::Closed`] without blocking.
253        ///
254        /// The halt callback fires exactly once across the submit-side stall path and
255        /// every writer-thread failure path; the first condition to fire wins the
256        /// recorded reason.
257        ///
258        /// # Errors
259        ///
260        /// Returns an error if the writer is shut down, the writer thread has exited, or a prior
261        /// halt fired, or if this submit's stall first crosses the configured halt threshold.
262        pub fn submit(&self, draft: EntryDraft) -> Result<(), SubmitError> {
263            // Refuse further entries once a halt has been signaled, even if the
264            // channel later drains: halt is terminal for the run.
265            if self.halted.load(Ordering::Acquire) {
266                return Err(SubmitError::Closed);
267            }
268
269            let tx = self.tx.as_ref().ok_or(SubmitError::Closed)?;
270            let ts_publish = self.clock.get_time_ns();
271            let pending = WriterMessage::Entry { draft, ts_publish };
272            let start = Instant::now();
273
274            match self.enqueue_with_backpressure(tx, pending, start) {
275                Ok(()) => Ok(()),
276                Err(EnqueueFailure::Stalled(elapsed)) => Err(SubmitError::HaltSignaled {
277                    stalled_for: elapsed,
278                    threshold: self.halt_threshold,
279                }),
280                Err(EnqueueFailure::Closed) => Err(SubmitError::Closed),
281            }
282        }
283
284        /// Returns the largest seq durably acknowledged by the backend.
285        ///
286        /// Updated only after a successful `append_batch` ack; reflects what is safe to
287        /// anchor a snapshot against.
288        #[must_use]
289        pub fn high_watermark(&self) -> u64 {
290            self.high_watermark.load(Ordering::Acquire)
291        }
292
293        /// Flushes pending entries and records a snapshot anchor at the durable
294        /// high-watermark.
295        ///
296        /// The cache owns `blob_ref` and `content_hash`; the writer derives the
297        /// high-watermark only after earlier submitted entries have committed, so the
298        /// anchor never points past durable event-store state.
299        ///
300        /// # Errors
301        ///
302        /// Returns an error if the writer is closed or halted, and forwards backend errors when
303        /// flushing pending entries or recording the anchor fails.
304        pub fn record_snapshot_anchor(
305            &self,
306            blob_ref: impl Into<String>,
307            content_hash: impl Into<String>,
308        ) -> Result<SnapshotAnchor, EventStoreError> {
309            if self.halted.load(Ordering::Acquire) {
310                return Err(EventStoreError::Closed);
311            }
312
313            let tx = self.tx.as_ref().ok_or(EventStoreError::Closed)?;
314            let (ack_tx, ack_rx) = mpsc::sync_channel::<Result<SnapshotAnchor, EventStoreError>>(1);
315            let pending = WriterMessage::RecordSnapshotAnchor {
316                blob_ref: blob_ref.into(),
317                content_hash: content_hash.into(),
318                ack: ack_tx,
319            };
320            let start = Instant::now();
321
322            if let Err(e) = self.enqueue_with_backpressure(tx, pending, start) {
323                match e {
324                    EnqueueFailure::Stalled(elapsed) => {
325                        return Err(EventStoreError::Backend(format!(
326                            "snapshot anchor submit stalled for {elapsed:?}, halt threshold {:?}",
327                            self.halt_threshold,
328                        )));
329                    }
330                    EnqueueFailure::Closed => return Err(EventStoreError::Closed),
331                }
332            }
333
334            match ack_rx.recv_timeout(self.halt_threshold) {
335                Ok(result) => result,
336                Err(RecvTimeoutError::Timeout) => {
337                    let elapsed = start.elapsed();
338                    self.signal_backpressure_stall(elapsed);
339                    Err(EventStoreError::Backend(format!(
340                        "snapshot anchor ack stalled for {elapsed:?}, halt threshold {:?}",
341                        self.halt_threshold,
342                    )))
343                }
344                Err(RecvTimeoutError::Disconnected) => Err(EventStoreError::Backend(
345                    "snapshot anchor ack channel disconnected".to_string(),
346                )),
347            }
348        }
349
350        fn enqueue_with_backpressure(
351            &self,
352            tx: &SyncSender<WriterMessage>,
353            mut pending: WriterMessage,
354            start: Instant,
355        ) -> Result<(), EnqueueFailure> {
356            // Check elapsed before each try_send (including after a sleep) so that a
357            // stall which exceeds the threshold fires halt even when the next attempt
358            // would have succeeded. The first iteration's elapsed is ~0, so it falls
359            // through to try_send.
360            loop {
361                // A halt latched while this submit waited is terminal: refuse the
362                // entry instead of accepting one the doomed writer thread would drop.
363                if self.halted.load(Ordering::Acquire) {
364                    return Err(EnqueueFailure::Closed);
365                }
366
367                let elapsed = start.elapsed();
368
369                if elapsed >= self.halt_threshold {
370                    self.signal_backpressure_stall(elapsed);
371                    return Err(EnqueueFailure::Stalled(elapsed));
372                }
373
374                match tx.try_send(pending) {
375                    Ok(()) => return Ok(()),
376                    Err(TrySendError::Full(returned)) => {
377                        pending = returned;
378                        thread::sleep(SUBMIT_RETRY_INTERVAL);
379                    }
380                    Err(TrySendError::Disconnected(_)) => return Err(EnqueueFailure::Closed),
381                }
382            }
383        }
384
385        fn signal_backpressure_stall(&self, stalled_for: Duration) {
386            halt::fire_once(
387                &self.halt,
388                &self.halted,
389                HaltReason::BackpressureStall {
390                    stalled_for,
391                    threshold: self.halt_threshold,
392                },
393            );
394        }
395
396        /// Drains the channel, commits `run_ended` as the final entry, and seals the
397        /// manifest with [`crate::manifest::RunStatus::Ended`].
398        ///
399        /// Consumes the writer; further submits are unrepresentable.
400        ///
401        /// # Errors
402        ///
403        /// Returns an error if the writer thread fails to commit the final batch, when seal fails,
404        /// or when the writer thread panicked.
405        pub fn close(mut self, run_ended: EntryDraft) -> Result<u64, EventStoreError> {
406            let tx = self
407                .tx
408                .take()
409                .ok_or_else(|| EventStoreError::Backend("writer already closed".to_string()))?;
410
411            let (ack_tx, ack_rx) = mpsc::sync_channel::<Result<u64, EventStoreError>>(1);
412            tx.send(WriterMessage::Close {
413                run_ended,
414                ack: ack_tx,
415            })
416            .map_err(|_| EventStoreError::Backend("writer thread disconnected".to_string()))?;
417            // Drop the producer so the writer's recv loop will see Disconnected after it
418            // finishes Close handling, even if the path through Close returns earlier.
419            drop(tx);
420
421            let result = ack_rx.recv().map_err(|_| {
422                EventStoreError::Backend("writer ack channel disconnected".to_string())
423            })?;
424
425            if let Some(handle) = self.handle.take() {
426                handle
427                    .join()
428                    .map_err(|_| EventStoreError::Backend("writer thread panicked".to_string()))?;
429            }
430            result
431        }
432    }
433
434    enum EnqueueFailure {
435        Stalled(Duration),
436        Closed,
437    }
438
439    impl Drop for EventStoreWriter {
440        fn drop(&mut self) {
441            // Implicit drop without close(): release the producer so the writer thread
442            // exits cleanly, leaving the manifest unsealed so a later open of the same
443            // run observes a CrashedPredecessor.
444            self.tx.take();
445
446            if let Some(handle) = self.handle.take() {
447                let _ = handle.join();
448            }
449        }
450    }
451}
452
453#[cfg(madsim)]
454mod imp {
455    use std::{
456        fmt::Debug,
457        sync::{
458            Arc,
459            atomic::{AtomicU64, Ordering},
460        },
461    };
462
463    use nautilus_core::time::AtomicTime;
464    use parking_lot::Mutex;
465
466    use super::{
467        EntryDraft, SnapshotAnchor, SubmitError, WriterConfig, batcher,
468        halt::{HaltCallback, HaltReason},
469    };
470    use crate::{backend::EventStore, error::EventStoreError, manifest::RunStatus};
471
472    /// Synchronous, in-thread writer used under simulation.
473    ///
474    /// The dedicated thread and bounded channel are dropped: each `submit` commits a
475    /// single-entry batch on the calling thread so tests can assert against the
476    /// authoritative in-process log without thread scheduling.
477    pub struct EventStoreWriter {
478        inner: Mutex<Inner>,
479        high_watermark: Arc<AtomicU64>,
480        halt: HaltCallback,
481        clock: &'static AtomicTime,
482    }
483
484    impl Debug for EventStoreWriter {
485        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
486            f.debug_struct(stringify!(EventStoreWriter))
487                .field(
488                    "high_watermark",
489                    &self.high_watermark.load(Ordering::Acquire),
490                )
491                .finish_non_exhaustive()
492        }
493    }
494
495    struct Inner {
496        backend: Box<dyn EventStore + Send>,
497        next_seq: u64,
498        closed: bool,
499    }
500
501    impl EventStoreWriter {
502        /// Constructs a synchronous writer over `backend`.
503        ///
504        /// # Errors
505        ///
506        /// Returns an error if the backend has no open run.
507        pub fn spawn(
508            backend: Box<dyn EventStore + Send>,
509            clock: &'static AtomicTime,
510            halt: HaltCallback,
511            _config: WriterConfig,
512        ) -> Result<Self, EventStoreError> {
513            let initial_hwm = backend.high_watermark()?;
514            let high_watermark = Arc::new(AtomicU64::new(initial_hwm));
515            let inner = Inner {
516                backend,
517                next_seq: initial_hwm + 1,
518                closed: false,
519            };
520
521            Ok(Self {
522                inner: Mutex::new(inner),
523                high_watermark,
524                halt,
525                clock,
526            })
527        }
528
529        /// Commits `draft` synchronously as a single-entry batch.
530        ///
531        /// # Errors
532        ///
533        /// Returns an error if the writer has been closed or fail-stopped.
534        pub fn submit(&self, draft: EntryDraft) -> Result<(), SubmitError> {
535            let mut inner = self.inner.lock();
536
537            if inner.closed {
538                return Err(SubmitError::Closed);
539            }
540
541            let ts_publish = self.clock.get_time_ns();
542            let seq = inner.next_seq;
543            let append = batcher::build_append_entry(draft, ts_publish, seq);
544
545            match inner.backend.append_batch(std::slice::from_ref(&append)) {
546                Ok(new_hwm) => {
547                    inner.next_seq = seq + 1;
548                    self.high_watermark.store(new_hwm, Ordering::Release);
549                    Ok(())
550                }
551                Err(e) => {
552                    (self.halt)(HaltReason::from_backend_error(&e));
553                    inner.closed = true;
554                    Err(SubmitError::Closed)
555                }
556            }
557        }
558
559        /// Returns the largest seq durably acknowledged by the backend.
560        #[must_use]
561        pub fn high_watermark(&self) -> u64 {
562            self.high_watermark.load(Ordering::Acquire)
563        }
564
565        /// Records a snapshot anchor at the current durable high-watermark.
566        ///
567        /// # Errors
568        ///
569        /// Returns an error if the writer has closed, and forwards backend errors when recording
570        /// the anchor fails.
571        pub fn record_snapshot_anchor(
572            &self,
573            blob_ref: impl Into<String>,
574            content_hash: impl Into<String>,
575        ) -> Result<SnapshotAnchor, EventStoreError> {
576            let mut inner = self.inner.lock();
577
578            if inner.closed {
579                return Err(EventStoreError::Closed);
580            }
581
582            let anchor = SnapshotAnchor::new(
583                self.high_watermark.load(Ordering::Acquire),
584                blob_ref,
585                content_hash,
586            );
587
588            match inner.backend.record_snapshot_anchor(anchor.clone()) {
589                Ok(()) => Ok(anchor),
590                Err(e) => {
591                    (self.halt)(HaltReason::from_backend_error(&e));
592                    inner.closed = true;
593                    Err(e)
594                }
595            }
596        }
597
598        /// Commits `run_ended` synchronously as the final entry and seals the manifest.
599        ///
600        /// # Errors
601        ///
602        /// Returns an error if the final commit or seal fails.
603        pub fn close(self, run_ended: EntryDraft) -> Result<u64, EventStoreError> {
604            let mut inner = self.inner.lock();
605
606            if inner.closed {
607                return Err(EventStoreError::Backend(
608                    "writer already closed".to_string(),
609                ));
610            }
611
612            let ts_publish = self.clock.get_time_ns();
613            let seq = inner.next_seq;
614            let append = batcher::build_append_entry(run_ended, ts_publish, seq);
615
616            match inner.backend.append_batch(std::slice::from_ref(&append)) {
617                Ok(new_hwm) => {
618                    inner.next_seq = seq + 1;
619                    self.high_watermark.store(new_hwm, Ordering::Release);
620                }
621                Err(e) => {
622                    (self.halt)(HaltReason::from_backend_error(&e));
623                    inner.closed = true;
624                    return Err(e);
625                }
626            }
627
628            match inner.backend.seal(RunStatus::Ended) {
629                Ok(()) => {
630                    inner.closed = true;
631                    Ok(self.high_watermark.load(Ordering::Acquire))
632                }
633                Err(e) => {
634                    (self.halt)(HaltReason::from_backend_error(&e));
635                    inner.closed = true;
636                    Err(e)
637                }
638            }
639        }
640    }
641}
642
643pub use imp::EventStoreWriter;
644
645#[cfg(test)]
646#[cfg(not(madsim))]
647mod tests {
648    use std::sync::{
649        Arc,
650        atomic::{AtomicUsize, Ordering},
651    };
652
653    use bytes::Bytes;
654    use indexmap::IndexMap;
655    use nautilus_core::{UnixNanos, time::get_atomic_clock_static};
656    use parking_lot::Mutex;
657    use rstest::{fixture, rstest};
658    use ustr::Ustr;
659
660    use super::*;
661    use crate::{
662        backend::{AppendEntry, EventStore, IndexKind, MemoryBackend, ScanDirection},
663        entry::EventStoreEntry,
664        error::EventStoreError,
665        manifest::{RegisteredComponents, RunManifest, RunStatus},
666    };
667
668    fn manifest(run_id: &str) -> RunManifest {
669        RunManifest {
670            run_id: run_id.to_string(),
671            parent_run_id: None,
672            instance_id: "trader-001".to_string(),
673            binary_hash: "deadbeef".to_string(),
674            schema_version: 1,
675            crate_versions: "feedface".to_string(),
676            feature_flags: Vec::new(),
677            adapter_versions: IndexMap::new(),
678            config_hash: "cafebabe".to_string(),
679            registered_components: RegisteredComponents::default(),
680            seed: None,
681            start_ts_init: UnixNanos::from(0),
682            end_ts_init: None,
683            high_watermark: 0,
684            status: RunStatus::Running,
685        }
686    }
687
688    fn entry_draft(ts_init: u64) -> EntryDraft {
689        EntryDraft {
690            headers: Headers::empty(),
691            topic: "exec.command.SubmitOrder".into(),
692            payload_type: Ustr::from("SubmitOrder"),
693            payload: Bytes::from_static(b"\x01\x02\x03\x04"),
694            ts_init: UnixNanos::from(ts_init),
695            index_keys: Vec::new(),
696        }
697    }
698
699    fn run_ended_draft() -> EntryDraft {
700        EntryDraft {
701            headers: Headers::empty(),
702            topic: "run.lifecycle.RunEnded".into(),
703            payload_type: Ustr::from("RunEnded"),
704            payload: Bytes::new(),
705            ts_init: UnixNanos::from(9_999),
706            index_keys: Vec::new(),
707        }
708    }
709
710    /// Wraps `MemoryBackend` so tests can read the same instance the writer thread
711    /// commits into.
712    #[derive(Debug)]
713    struct SharedMemory(Arc<Mutex<MemoryBackend>>);
714
715    impl SharedMemory {
716        fn new() -> (Self, Arc<Mutex<MemoryBackend>>) {
717            let arc = Arc::new(Mutex::new(MemoryBackend::new()));
718            (Self(Arc::clone(&arc)), arc)
719        }
720    }
721
722    impl EventStore for SharedMemory {
723        fn open_run(&mut self, _: RunManifest) -> Result<(), EventStoreError> {
724            // Tests open the underlying backend directly.
725            unreachable!("test wrapper does not forward open_run")
726        }
727
728        fn append_batch(&mut self, entries: &[AppendEntry]) -> Result<u64, EventStoreError> {
729            self.0.lock().append_batch(entries)
730        }
731
732        fn scan_range(
733            &self,
734            from: u64,
735            to: u64,
736            direction: ScanDirection,
737        ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
738            self.0.lock().scan_range(from, to, direction)
739        }
740
741        fn scan_seq(&self, seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
742            self.0.lock().scan_seq(seq)
743        }
744
745        fn lookup(&self, kind: IndexKind, key: &str) -> Result<Option<u64>, EventStoreError> {
746            self.0.lock().lookup(kind, key)
747        }
748
749        fn iter_index_keys(&self, kind: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
750            self.0.lock().iter_index_keys(kind)
751        }
752
753        fn record_snapshot_anchor(
754            &mut self,
755            anchor: SnapshotAnchor,
756        ) -> Result<(), EventStoreError> {
757            self.0.lock().record_snapshot_anchor(anchor)
758        }
759
760        fn latest_snapshot_anchor(&self) -> Result<Option<SnapshotAnchor>, EventStoreError> {
761            self.0.lock().latest_snapshot_anchor()
762        }
763
764        fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
765            self.0.lock().seal(status)
766        }
767
768        fn manifest(&self) -> Result<RunManifest, EventStoreError> {
769            self.0.lock().manifest()
770        }
771
772        fn high_watermark(&self) -> Result<u64, EventStoreError> {
773            self.0.lock().high_watermark()
774        }
775    }
776
777    /// `EventStore` wrapper that blocks `append_batch` until a release flag flips.
778    #[derive(Debug)]
779    struct BlockingBackend {
780        inner: Arc<Mutex<MemoryBackend>>,
781        gate: Arc<(Mutex<bool>, parking_lot::Condvar)>,
782        appends_seen: Arc<AtomicUsize>,
783    }
784
785    impl BlockingBackend {
786        fn new(
787            inner: Arc<Mutex<MemoryBackend>>,
788            gate: Arc<(Mutex<bool>, parking_lot::Condvar)>,
789            appends_seen: Arc<AtomicUsize>,
790        ) -> Self {
791            Self {
792                inner,
793                gate,
794                appends_seen,
795            }
796        }
797    }
798
799    impl EventStore for BlockingBackend {
800        fn open_run(&mut self, _: RunManifest) -> Result<(), EventStoreError> {
801            unreachable!("test wrapper does not forward open_run")
802        }
803
804        fn append_batch(&mut self, entries: &[AppendEntry]) -> Result<u64, EventStoreError> {
805            self.appends_seen.fetch_add(1, Ordering::SeqCst);
806            let (lock, cvar) = &*self.gate;
807            let mut released = lock.lock();
808
809            while !*released {
810                cvar.wait(&mut released);
811            }
812            self.inner.lock().append_batch(entries)
813        }
814
815        fn scan_range(
816            &self,
817            from: u64,
818            to: u64,
819            direction: ScanDirection,
820        ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
821            self.inner.lock().scan_range(from, to, direction)
822        }
823
824        fn scan_seq(&self, seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
825            self.inner.lock().scan_seq(seq)
826        }
827
828        fn lookup(&self, kind: IndexKind, key: &str) -> Result<Option<u64>, EventStoreError> {
829            self.inner.lock().lookup(kind, key)
830        }
831
832        fn iter_index_keys(&self, kind: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
833            self.inner.lock().iter_index_keys(kind)
834        }
835
836        fn record_snapshot_anchor(
837            &mut self,
838            anchor: SnapshotAnchor,
839        ) -> Result<(), EventStoreError> {
840            self.inner.lock().record_snapshot_anchor(anchor)
841        }
842
843        fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
844            self.inner.lock().seal(status)
845        }
846
847        fn manifest(&self) -> Result<RunManifest, EventStoreError> {
848            self.inner.lock().manifest()
849        }
850
851        fn high_watermark(&self) -> Result<u64, EventStoreError> {
852            self.inner.lock().high_watermark()
853        }
854    }
855
856    /// `EventStore` wrapper that returns `EventStoreError::Disk` for every append.
857    #[derive(Debug, Default)]
858    struct DiskFailureBackend {
859        appends_seen: Arc<AtomicUsize>,
860    }
861
862    impl EventStore for DiskFailureBackend {
863        fn open_run(&mut self, _: RunManifest) -> Result<(), EventStoreError> {
864            Ok(())
865        }
866
867        fn append_batch(&mut self, _: &[AppendEntry]) -> Result<u64, EventStoreError> {
868            self.appends_seen.fetch_add(1, Ordering::SeqCst);
869            Err(EventStoreError::Disk("ENOSPC".to_string()))
870        }
871
872        fn scan_range(
873            &self,
874            _: u64,
875            _: u64,
876            _: ScanDirection,
877        ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
878            Ok(Vec::new())
879        }
880
881        fn scan_seq(&self, _: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
882            Ok(None)
883        }
884
885        fn lookup(&self, _: IndexKind, _: &str) -> Result<Option<u64>, EventStoreError> {
886            Ok(None)
887        }
888
889        fn iter_index_keys(&self, _: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
890            Ok(Vec::new())
891        }
892
893        fn seal(&mut self, _: RunStatus) -> Result<(), EventStoreError> {
894            Ok(())
895        }
896
897        fn manifest(&self) -> Result<RunManifest, EventStoreError> {
898            Err(EventStoreError::Backend("disk failure".to_string()))
899        }
900
901        fn high_watermark(&self) -> Result<u64, EventStoreError> {
902            Ok(0)
903        }
904    }
905
906    /// `EventStore` wrapper that blocks `append_batch` at a gate, then fails with
907    /// `EventStoreError::Disk` once released.
908    #[derive(Debug)]
909    struct GatedDiskFailureBackend {
910        gate: Arc<(Mutex<bool>, parking_lot::Condvar)>,
911        appends_seen: Arc<AtomicUsize>,
912    }
913
914    impl EventStore for GatedDiskFailureBackend {
915        fn open_run(&mut self, _: RunManifest) -> Result<(), EventStoreError> {
916            Ok(())
917        }
918
919        fn append_batch(&mut self, _: &[AppendEntry]) -> Result<u64, EventStoreError> {
920            self.appends_seen.fetch_add(1, Ordering::SeqCst);
921            let (lock, cvar) = &*self.gate;
922            let mut released = lock.lock();
923
924            while !*released {
925                cvar.wait(&mut released);
926            }
927            Err(EventStoreError::Disk("ENOSPC".to_string()))
928        }
929
930        fn scan_range(
931            &self,
932            _: u64,
933            _: u64,
934            _: ScanDirection,
935        ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
936            Ok(Vec::new())
937        }
938
939        fn scan_seq(&self, _: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
940            Ok(None)
941        }
942
943        fn lookup(&self, _: IndexKind, _: &str) -> Result<Option<u64>, EventStoreError> {
944            Ok(None)
945        }
946
947        fn iter_index_keys(&self, _: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
948            Ok(Vec::new())
949        }
950
951        fn seal(&mut self, _: RunStatus) -> Result<(), EventStoreError> {
952            Ok(())
953        }
954
955        fn manifest(&self) -> Result<RunManifest, EventStoreError> {
956            Err(EventStoreError::Backend("disk failure".to_string()))
957        }
958
959        fn high_watermark(&self) -> Result<u64, EventStoreError> {
960            Ok(0)
961        }
962    }
963
964    #[fixture]
965    fn captured_halt() -> (HaltCallback, Arc<Mutex<Vec<HaltReason>>>) {
966        let captured: Arc<Mutex<Vec<HaltReason>>> = Arc::new(Mutex::new(Vec::new()));
967        let captured_for_cb = Arc::clone(&captured);
968        let halt: HaltCallback = Arc::new(move |reason| {
969            captured_for_cb.lock().push(reason);
970        });
971        (halt, captured)
972    }
973
974    #[rstest]
975    fn submit_then_close_records_entries_and_seals(
976        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
977    ) {
978        let (halt, captured) = captured_halt;
979        let (wrapper, shared) = SharedMemory::new();
980        shared.lock().open_run(manifest("run-1")).expect("open");
981
982        let writer = EventStoreWriter::spawn(
983            Box::new(wrapper),
984            get_atomic_clock_static(),
985            halt,
986            WriterConfig::default(),
987        )
988        .expect("spawn");
989
990        for ts in 10_u64..15_u64 {
991            writer.submit(entry_draft(ts)).expect("submit");
992        }
993
994        let final_hwm = writer.close(run_ended_draft()).expect("close");
995
996        // Five drafts plus the RunEnded entry.
997        assert_eq!(final_hwm, 6);
998        let backend = shared.lock();
999        let m = backend.manifest().expect("manifest");
1000        assert_eq!(m.status, RunStatus::Ended);
1001        assert_eq!(m.high_watermark, 6);
1002
1003        let last = backend.scan_seq(6).expect("scan").expect("present");
1004        assert_eq!(last.payload_type.as_str(), "RunEnded");
1005        assert!(captured.lock().is_empty());
1006    }
1007
1008    #[rstest]
1009    fn record_snapshot_anchor_flushes_pending_entries_and_replay_tail_starts_after_anchor(
1010        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
1011    ) {
1012        let (halt, captured) = captured_halt;
1013        let (wrapper, shared) = SharedMemory::new();
1014        shared
1015            .lock()
1016            .open_run(manifest("run-anchor"))
1017            .expect("open");
1018
1019        let writer = EventStoreWriter::spawn(
1020            Box::new(wrapper),
1021            get_atomic_clock_static(),
1022            halt,
1023            WriterConfig::default(),
1024        )
1025        .expect("spawn");
1026
1027        writer.submit(entry_draft(10)).expect("submit first");
1028        writer.submit(entry_draft(11)).expect("submit second");
1029        let anchor = writer
1030            .record_snapshot_anchor("cache://position-snapshots/P-1/0", "blake3:abc")
1031            .expect("record anchor");
1032
1033        assert_eq!(anchor.high_watermark, 2);
1034
1035        writer.submit(entry_draft(12)).expect("submit third");
1036        writer.submit(entry_draft(13)).expect("submit fourth");
1037        let final_hwm = writer.close(run_ended_draft()).expect("close");
1038
1039        let backend = shared.lock();
1040        assert_eq!(
1041            backend.latest_snapshot_anchor().expect("latest anchor"),
1042            Some(anchor.clone()),
1043        );
1044
1045        let tail_seqs: Vec<_> = backend
1046            .scan_range(anchor.high_watermark + 1, final_hwm, ScanDirection::Forward)
1047            .expect("scan tail")
1048            .into_iter()
1049            .map(|entry| entry.seq)
1050            .collect();
1051
1052        assert_eq!(tail_seqs, vec![3, 4, 5]);
1053        assert!(captured.lock().is_empty());
1054    }
1055
1056    #[rstest]
1057    fn batches_respect_max_entries_threshold(
1058        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
1059    ) {
1060        // Tiny batch ceiling (2) plus a large latency window forces the size threshold
1061        // to drive every flush; six entries produce three size-driven commits and one
1062        // close commit.
1063        let (halt, _) = captured_halt;
1064        let inner = Arc::new(Mutex::new(MemoryBackend::new()));
1065        inner.lock().open_run(manifest("run-batch")).expect("open");
1066
1067        let appends_seen = Arc::new(AtomicUsize::new(0));
1068        let gate = Arc::new((Mutex::new(true), parking_lot::Condvar::new()));
1069        let backend = BlockingBackend::new(
1070            Arc::clone(&inner),
1071            Arc::clone(&gate),
1072            Arc::clone(&appends_seen),
1073        );
1074
1075        let config = WriterConfig {
1076            channel_capacity: 16,
1077            max_batch_entries: 2,
1078            max_batch_latency: Duration::from_secs(30),
1079            halt_threshold: Duration::from_secs(30),
1080        };
1081
1082        let clock = get_atomic_clock_static();
1083        let boxed = Box::new(backend);
1084
1085        let writer = EventStoreWriter::spawn(boxed, clock, halt, config).expect("spawn");
1086
1087        for ts in 10_u64..16_u64 {
1088            writer.submit(entry_draft(ts)).expect("submit");
1089        }
1090
1091        let final_hwm = writer.close(run_ended_draft()).expect("close");
1092
1093        // 6 submits + 1 RunEnded == 7 entries, batch=2 -> 4 commits (3 size-driven + 1 close).
1094        assert_eq!(final_hwm, 7);
1095        assert_eq!(appends_seen.load(Ordering::SeqCst), 4);
1096    }
1097
1098    #[rstest]
1099    fn submit_signals_halt_when_stalled_past_threshold(
1100        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
1101    ) {
1102        // Channel capacity 1 with a backend gate held closed forces a stall: the first
1103        // submit fills the buffer, the writer thread blocks inside append_batch, and a
1104        // subsequent submit can never enqueue before the halt threshold fires.
1105        let (halt, captured) = captured_halt;
1106        let inner = Arc::new(Mutex::new(MemoryBackend::new()));
1107        inner.lock().open_run(manifest("run-halt")).expect("open");
1108
1109        let gate = Arc::new((Mutex::new(false), parking_lot::Condvar::new()));
1110        let appends_seen = Arc::new(AtomicUsize::new(0));
1111        let backend = BlockingBackend::new(
1112            Arc::clone(&inner),
1113            Arc::clone(&gate),
1114            Arc::clone(&appends_seen),
1115        );
1116
1117        let halt_threshold = Duration::from_millis(50);
1118        let config = WriterConfig {
1119            channel_capacity: 1,
1120            max_batch_entries: 1,
1121            max_batch_latency: Duration::from_millis(1),
1122            halt_threshold,
1123        };
1124
1125        let clock = get_atomic_clock_static();
1126        let boxed = Box::new(backend);
1127
1128        let writer = EventStoreWriter::spawn(boxed, clock, halt, config).expect("spawn");
1129
1130        // First submit fits in the channel; the writer thread takes it and blocks.
1131        writer.submit(entry_draft(10)).expect("first submit fits");
1132
1133        // Wait long enough for the writer to dequeue and become blocked at the gate.
1134        std::thread::sleep(Duration::from_millis(20));
1135
1136        // Second and third submits saturate the slot; the channel buffer holds one,
1137        // the next one stalls past the halt threshold.
1138        let _ = writer.submit(entry_draft(11));
1139        let stalled = writer.submit(entry_draft(12)).expect_err("must stall");
1140
1141        match stalled {
1142            SubmitError::HaltSignaled { .. } => {}
1143            SubmitError::Closed => panic!("expected HaltSignaled, was Closed"),
1144        }
1145        let captured_reasons = captured.lock();
1146        assert_eq!(
1147            captured_reasons.len(),
1148            1,
1149            "halt callback must fire exactly once",
1150        );
1151        assert_backpressure_stall(captured_reasons.first(), halt_threshold);
1152        drop(captured_reasons);
1153
1154        // After a backpressure halt has fired, subsequent submits must reject without
1155        // re-entering the retry loop, even though the channel and writer thread are
1156        // still alive.
1157        let post_halt = writer
1158            .submit(entry_draft(13))
1159            .expect_err("post-halt submit");
1160
1161        match post_halt {
1162            SubmitError::Closed => {}
1163            SubmitError::HaltSignaled { .. } => {
1164                panic!("expected Closed after halt, was HaltSignaled")
1165            }
1166        }
1167        // The halt callback must not refire on subsequent submits.
1168        assert_eq!(
1169            captured.lock().len(),
1170            1,
1171            "halt callback must not refire after the first stall",
1172        );
1173
1174        // Release the gate so the writer thread can finish and the test can drop the
1175        // writer cleanly.
1176        let (lock, cvar) = &*gate;
1177        *lock.lock() = true;
1178        cvar.notify_all();
1179    }
1180
1181    #[rstest]
1182    fn halt_fires_once_across_stall_and_backend_failure(
1183        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
1184    ) {
1185        // A stall fires first, then the writer thread hits a disk failure: the latch
1186        // must suppress the second fire and keep the first condition's reason.
1187        let (halt, captured) = captured_halt;
1188        let gate = Arc::new((Mutex::new(false), parking_lot::Condvar::new()));
1189        let appends_seen = Arc::new(AtomicUsize::new(0));
1190        let backend = GatedDiskFailureBackend {
1191            gate: Arc::clone(&gate),
1192            appends_seen: Arc::clone(&appends_seen),
1193        };
1194
1195        let halt_threshold = Duration::from_millis(50);
1196        let config = WriterConfig {
1197            channel_capacity: 1,
1198            max_batch_entries: 1,
1199            max_batch_latency: Duration::from_millis(1),
1200            halt_threshold,
1201        };
1202
1203        let clock = get_atomic_clock_static();
1204        let boxed = Box::new(backend);
1205
1206        let writer = EventStoreWriter::spawn(boxed, clock, halt, config).expect("spawn");
1207
1208        writer.submit(entry_draft(10)).expect("first submit fits");
1209
1210        let mut waited = Duration::ZERO;
1211        while appends_seen.load(Ordering::SeqCst) == 0 && waited < Duration::from_secs(2) {
1212            std::thread::sleep(Duration::from_millis(5));
1213            waited += Duration::from_millis(5);
1214        }
1215        assert_eq!(
1216            appends_seen.load(Ordering::SeqCst),
1217            1,
1218            "writer thread did not reach the gated append",
1219        );
1220
1221        // Fill the single channel slot; the next submit stalls past the threshold
1222        let _ = writer.submit(entry_draft(11));
1223        let stalled = writer.submit(entry_draft(12)).expect_err("must stall");
1224        assert!(
1225            matches!(stalled, SubmitError::HaltSignaled { .. }),
1226            "was {stalled:?}",
1227        );
1228
1229        // Release the gate so the append fails with Disk; without the latch this
1230        // fires a second, misclassified halt.
1231        let (lock, cvar) = &*gate;
1232        *lock.lock() = true;
1233        cvar.notify_all();
1234
1235        // Dropping joins the writer thread, so the failure has been observed
1236        drop(writer);
1237
1238        let reasons = captured.lock();
1239        assert_eq!(
1240            reasons.len(),
1241            1,
1242            "halt must fire exactly once across stall and backend failure",
1243        );
1244        assert_backpressure_stall(reasons.first(), halt_threshold);
1245    }
1246
1247    #[rstest]
1248    fn zero_channel_capacity_is_clamped_and_submit_buffers(
1249        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
1250    ) {
1251        let (halt, captured) = captured_halt;
1252        let inner = Arc::new(Mutex::new(MemoryBackend::new()));
1253        inner
1254            .lock()
1255            .open_run(manifest("run-zero-capacity"))
1256            .expect("open");
1257
1258        let gate = Arc::new((Mutex::new(false), parking_lot::Condvar::new()));
1259        let appends_seen = Arc::new(AtomicUsize::new(0));
1260        let backend = BlockingBackend::new(
1261            Arc::clone(&inner),
1262            Arc::clone(&gate),
1263            Arc::clone(&appends_seen),
1264        );
1265
1266        let config = WriterConfig {
1267            channel_capacity: 0,
1268            max_batch_entries: 1,
1269            max_batch_latency: Duration::from_millis(1),
1270            halt_threshold: Duration::from_millis(250),
1271        };
1272
1273        let clock = get_atomic_clock_static();
1274        let boxed = Box::new(backend);
1275
1276        let writer = EventStoreWriter::spawn(boxed, clock, halt, config).expect("spawn");
1277
1278        writer.submit(entry_draft(10)).expect("first submit fits");
1279
1280        let mut waited = Duration::ZERO;
1281        while appends_seen.load(Ordering::SeqCst) == 0 && waited < Duration::from_secs(2) {
1282            std::thread::sleep(Duration::from_millis(5));
1283            waited += Duration::from_millis(5);
1284        }
1285        assert_eq!(
1286            appends_seen.load(Ordering::SeqCst),
1287            1,
1288            "writer thread did not reach the gated append",
1289        );
1290
1291        // Release the gate before asserting so a regression fails instead of
1292        // hanging the writer join.
1293        let second_submit = writer.submit(entry_draft(11));
1294
1295        let (lock, cvar) = &*gate;
1296        *lock.lock() = true;
1297        cvar.notify_all();
1298
1299        let final_hwm = writer.close(run_ended_draft()).expect("close");
1300        second_submit.expect("second submit must be buffered by the clamped capacity");
1301        assert_eq!(final_hwm, 3);
1302        assert!(captured.lock().is_empty());
1303    }
1304
1305    #[rstest]
1306    fn submit_after_writer_thread_halt_returns_closed(
1307        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
1308    ) {
1309        // A writer-thread halt latches the shared flag; post-halt submits must
1310        // reject rather than be accepted and silently dropped.
1311        let (halt, captured) = captured_halt;
1312        let config = WriterConfig {
1313            max_batch_entries: 1,
1314            ..WriterConfig::default()
1315        };
1316
1317        let writer = EventStoreWriter::spawn(
1318            Box::new(DiskFailureBackend::default()),
1319            get_atomic_clock_static(),
1320            halt,
1321            config,
1322        )
1323        .expect("spawn");
1324
1325        writer.submit(entry_draft(10)).expect("submit accepted");
1326
1327        let mut waited = Duration::ZERO;
1328        while waited < Duration::from_secs(2) {
1329            if !captured.lock().is_empty() {
1330                break;
1331            }
1332            std::thread::sleep(Duration::from_millis(5));
1333            waited += Duration::from_millis(5);
1334        }
1335
1336        let reasons = captured.lock();
1337        assert_eq!(reasons.len(), 1, "writer-thread halt did not fire");
1338        assert!(
1339            matches!(reasons.first(), Some(HaltReason::BackendDisk(_))),
1340            "was {:?}",
1341            reasons.first(),
1342        );
1343        drop(reasons);
1344
1345        let post_halt = writer
1346            .submit(entry_draft(11))
1347            .expect_err("post-halt submit must reject");
1348        assert!(
1349            matches!(post_halt, SubmitError::Closed),
1350            "was {post_halt:?}",
1351        );
1352        assert_eq!(
1353            captured.lock().len(),
1354            1,
1355            "halt must not refire on post-halt submits",
1356        );
1357    }
1358
1359    #[rstest]
1360    fn retrying_submit_returns_closed_after_stall_halt_latches(
1361        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
1362    ) {
1363        // A submit already sleeping in the retry loop when the halt latches must
1364        // return Closed rather than enqueue once the channel drains.
1365        let (halt, captured) = captured_halt;
1366        let inner = Arc::new(Mutex::new(MemoryBackend::new()));
1367        inner
1368            .lock()
1369            .open_run(manifest("run-retry-latch"))
1370            .expect("open");
1371
1372        let gate = Arc::new((Mutex::new(false), parking_lot::Condvar::new()));
1373        let appends_seen = Arc::new(AtomicUsize::new(0));
1374        let backend = BlockingBackend::new(
1375            Arc::clone(&inner),
1376            Arc::clone(&gate),
1377            Arc::clone(&appends_seen),
1378        );
1379
1380        let config = WriterConfig {
1381            channel_capacity: 1,
1382            max_batch_entries: 1,
1383            max_batch_latency: Duration::from_millis(1),
1384            halt_threshold: Duration::from_millis(50),
1385        };
1386
1387        let clock = get_atomic_clock_static();
1388        let writer = Arc::new(
1389            EventStoreWriter::spawn(Box::new(backend), clock, halt, config).expect("spawn"),
1390        );
1391
1392        writer.submit(entry_draft(10)).expect("first submit fits");
1393
1394        let mut waited = Duration::ZERO;
1395        while appends_seen.load(Ordering::SeqCst) == 0 && waited < Duration::from_secs(2) {
1396            std::thread::sleep(Duration::from_millis(5));
1397            waited += Duration::from_millis(5);
1398        }
1399        assert_eq!(
1400            appends_seen.load(Ordering::SeqCst),
1401            1,
1402            "writer thread did not reach the gated append",
1403        );
1404
1405        writer
1406            .submit(entry_draft(11))
1407            .expect("second submit fills the slot");
1408
1409        // This submit stalls past the threshold and latches the halt
1410        let stalled = writer.submit(entry_draft(12)).expect_err("must stall");
1411        assert!(
1412            matches!(stalled, SubmitError::HaltSignaled { .. }),
1413            "was {stalled:?}",
1414        );
1415
1416        // A second submitter now waits in the retry loop while the channel stays full
1417        let writer_for_thread = Arc::clone(&writer);
1418        let retrying = std::thread::spawn(move || writer_for_thread.submit(entry_draft(13)));
1419        std::thread::sleep(Duration::from_millis(20));
1420
1421        // Release the gate: the freed slot must not rescue the retrying submit
1422        let (lock, cvar) = &*gate;
1423        *lock.lock() = true;
1424        cvar.notify_all();
1425
1426        let result = retrying.join().expect("retrying thread panicked");
1427        assert!(matches!(result, Err(SubmitError::Closed)), "was {result:?}");
1428
1429        // The refused entry never commits
1430        let mut waited = Duration::ZERO;
1431        while writer.high_watermark() < 2 && waited < Duration::from_secs(2) {
1432            std::thread::sleep(Duration::from_millis(5));
1433            waited += Duration::from_millis(5);
1434        }
1435        assert_eq!(writer.high_watermark(), 2);
1436        assert_eq!(
1437            captured.lock().len(),
1438            1,
1439            "halt must not refire for the refused submit",
1440        );
1441    }
1442
1443    #[rstest]
1444    fn record_snapshot_anchor_signals_halt_when_ack_stalls(
1445        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
1446    ) {
1447        let (halt, captured) = captured_halt;
1448        let inner = Arc::new(Mutex::new(MemoryBackend::new()));
1449        inner
1450            .lock()
1451            .open_run(manifest("run-anchor-halt"))
1452            .expect("open");
1453
1454        let gate = Arc::new((Mutex::new(false), parking_lot::Condvar::new()));
1455        let appends_seen = Arc::new(AtomicUsize::new(0));
1456        let backend = BlockingBackend::new(
1457            Arc::clone(&inner),
1458            Arc::clone(&gate),
1459            Arc::clone(&appends_seen),
1460        );
1461
1462        let halt_threshold = Duration::from_millis(50);
1463
1464        let writer = EventStoreWriter::spawn(
1465            Box::new(backend),
1466            get_atomic_clock_static(),
1467            halt,
1468            WriterConfig {
1469                channel_capacity: 2,
1470                max_batch_entries: 1,
1471                max_batch_latency: Duration::from_millis(1),
1472                halt_threshold,
1473            },
1474        )
1475        .expect("spawn");
1476
1477        writer.submit(entry_draft(10)).expect("first submit fits");
1478        std::thread::sleep(Duration::from_millis(20));
1479
1480        let err = writer
1481            .record_snapshot_anchor("cache://position-snapshots/P-1/0", "blake3:abc")
1482            .expect_err("snapshot anchor ack must time out");
1483        let post_halt = writer
1484            .submit(entry_draft(11))
1485            .expect_err("post-halt submit");
1486
1487        let (lock, cvar) = &*gate;
1488        *lock.lock() = true;
1489        cvar.notify_all();
1490
1491        match err {
1492            EventStoreError::Backend(msg) => {
1493                assert!(
1494                    msg.contains("snapshot anchor ack stalled"),
1495                    "msg was: {msg}"
1496                );
1497            }
1498            other => panic!("expected Backend, was {other:?}"),
1499        }
1500
1501        match post_halt {
1502            SubmitError::Closed => {}
1503            SubmitError::HaltSignaled { .. } => {
1504                panic!("expected Closed after anchor halt, was HaltSignaled")
1505            }
1506        }
1507
1508        let captured_reasons = captured.lock();
1509        assert_eq!(
1510            captured_reasons.len(),
1511            1,
1512            "halt callback must fire exactly once",
1513        );
1514        assert_backpressure_stall(captured_reasons.first(), halt_threshold);
1515    }
1516
1517    #[rstest]
1518    fn record_snapshot_anchor_signals_halt_when_submit_stalls(
1519        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
1520    ) {
1521        let (halt, captured) = captured_halt;
1522        let inner = Arc::new(Mutex::new(MemoryBackend::new()));
1523        inner
1524            .lock()
1525            .open_run(manifest("run-anchor-submit-halt"))
1526            .expect("open");
1527
1528        let gate = Arc::new((Mutex::new(false), parking_lot::Condvar::new()));
1529        let appends_seen = Arc::new(AtomicUsize::new(0));
1530        let backend = BlockingBackend::new(
1531            Arc::clone(&inner),
1532            Arc::clone(&gate),
1533            Arc::clone(&appends_seen),
1534        );
1535
1536        let halt_threshold = Duration::from_millis(50);
1537
1538        let writer = EventStoreWriter::spawn(
1539            Box::new(backend),
1540            get_atomic_clock_static(),
1541            halt,
1542            WriterConfig {
1543                channel_capacity: 1,
1544                max_batch_entries: 1,
1545                max_batch_latency: Duration::from_secs(30),
1546                halt_threshold,
1547            },
1548        )
1549        .expect("spawn");
1550
1551        writer.submit(entry_draft(10)).expect("first submit fits");
1552        let mut waited = Duration::ZERO;
1553        while appends_seen.load(Ordering::SeqCst) == 0 && waited < Duration::from_secs(1) {
1554            std::thread::sleep(Duration::from_millis(2));
1555            waited += Duration::from_millis(2);
1556        }
1557        assert_eq!(
1558            appends_seen.load(Ordering::SeqCst),
1559            1,
1560            "writer must be blocked inside the first backend append",
1561        );
1562        writer.submit(entry_draft(11)).expect("second submit fits");
1563
1564        let err = writer
1565            .record_snapshot_anchor("cache://position-snapshots/P-1/0", "blake3:abc")
1566            .expect_err("snapshot anchor submit must time out");
1567        let post_halt = writer
1568            .submit(entry_draft(12))
1569            .expect_err("post-halt submit");
1570
1571        let (lock, cvar) = &*gate;
1572        *lock.lock() = true;
1573        cvar.notify_all();
1574
1575        match err {
1576            EventStoreError::Backend(msg) => {
1577                assert!(
1578                    msg.contains("snapshot anchor submit stalled"),
1579                    "msg was: {msg}"
1580                );
1581            }
1582            other => panic!("expected Backend, was {other:?}"),
1583        }
1584
1585        match post_halt {
1586            SubmitError::Closed => {}
1587            SubmitError::HaltSignaled { .. } => {
1588                panic!("expected Closed after anchor halt, was HaltSignaled")
1589            }
1590        }
1591
1592        let captured_reasons = captured.lock();
1593        assert_eq!(
1594            captured_reasons.len(),
1595            1,
1596            "halt callback must fire exactly once",
1597        );
1598        assert_backpressure_stall(captured_reasons.first(), halt_threshold);
1599    }
1600
1601    #[rstest]
1602    fn backend_disk_error_fires_halt_and_closes_writer(
1603        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
1604    ) {
1605        let (halt, captured) = captured_halt;
1606        let backend = DiskFailureBackend::default();
1607
1608        let writer = EventStoreWriter::spawn(
1609            Box::new(backend),
1610            get_atomic_clock_static(),
1611            halt,
1612            WriterConfig {
1613                channel_capacity: 4,
1614                max_batch_entries: 1,
1615                max_batch_latency: Duration::from_millis(1),
1616                halt_threshold: Duration::from_millis(500),
1617            },
1618        )
1619        .expect("spawn");
1620
1621        writer
1622            .submit(entry_draft(10))
1623            .expect("first submit fits in channel before writer fail-stops");
1624
1625        // Wait until the writer fail-stops and the halt fires.
1626        let mut waited = Duration::ZERO;
1627        let deadline = Duration::from_millis(500);
1628        while captured.lock().is_empty() && waited < deadline {
1629            std::thread::sleep(Duration::from_millis(10));
1630            waited += Duration::from_millis(10);
1631        }
1632
1633        let captured_reasons = captured.lock();
1634        assert!(matches!(
1635            captured_reasons.first(),
1636            Some(HaltReason::BackendDisk(_))
1637        ));
1638        drop(captured_reasons);
1639
1640        // Subsequent submits return Closed once the writer thread has exited.
1641        let mut closed_seen = false;
1642
1643        for _ in 0..50 {
1644            match writer.submit(entry_draft(11)) {
1645                Err(SubmitError::Closed) => {
1646                    closed_seen = true;
1647                    break;
1648                }
1649                _ => std::thread::sleep(Duration::from_millis(10)),
1650            }
1651        }
1652        assert!(closed_seen, "submits must surface Closed after fail-stop");
1653
1654        // Close after fail-stop returns an error rather than panicking.
1655        let close_result = writer.close(run_ended_draft());
1656        assert!(close_result.is_err());
1657    }
1658
1659    #[rstest]
1660    fn time_driven_flush_advances_watermark_before_close(
1661        captured_halt: (HaltCallback, Arc<Mutex<Vec<HaltReason>>>),
1662    ) {
1663        // A single submit well below max_batch_entries must still commit on the
1664        // latency window. Without this, a broken recv_timeout deadline would only
1665        // surface at close drain, masking the steady-state batching contract.
1666        let (halt, _) = captured_halt;
1667        let (wrapper, shared) = SharedMemory::new();
1668        shared.lock().open_run(manifest("run-time")).expect("open");
1669
1670        let writer = EventStoreWriter::spawn(
1671            Box::new(wrapper),
1672            get_atomic_clock_static(),
1673            halt,
1674            WriterConfig {
1675                channel_capacity: 32,
1676                max_batch_entries: 100,
1677                max_batch_latency: Duration::from_millis(20),
1678                halt_threshold: Duration::from_secs(30),
1679            },
1680        )
1681        .expect("spawn");
1682
1683        writer.submit(entry_draft(10)).expect("submit");
1684
1685        // Wait long enough that the latency window has elapsed multiple times.
1686        let mut waited = Duration::ZERO;
1687        while writer.high_watermark() == 0 && waited < Duration::from_millis(500) {
1688            std::thread::sleep(Duration::from_millis(5));
1689            waited += Duration::from_millis(5);
1690        }
1691        assert_eq!(
1692            writer.high_watermark(),
1693            1,
1694            "latency window must commit a sub-batch entry before close",
1695        );
1696
1697        let final_hwm = writer.close(run_ended_draft()).expect("close");
1698        assert_eq!(final_hwm, 2);
1699    }
1700
1701    #[rstest]
1702    fn entry_draft_without_indices_constructor() {
1703        let topic: crate::entry::Topic = "exec.command.SubmitOrder".into();
1704        let payload_type = Ustr::from("SubmitOrder");
1705        let payload = Bytes::from_static(b"\x01\x02");
1706        let ts_init = UnixNanos::from(42);
1707        let draft = EntryDraft::without_indices(
1708            Headers::empty(),
1709            topic,
1710            payload_type,
1711            payload.clone(),
1712            ts_init,
1713        );
1714
1715        assert!(draft.headers.is_empty());
1716        assert_eq!(draft.topic.as_ref(), "exec.command.SubmitOrder");
1717        assert_eq!(draft.payload_type.as_str(), "SubmitOrder");
1718        assert_eq!(draft.payload, payload);
1719        assert_eq!(draft.ts_init, ts_init);
1720        assert!(draft.index_keys.is_empty());
1721    }
1722
1723    fn assert_backpressure_stall(reason: Option<&HaltReason>, expected_threshold: Duration) {
1724        match reason {
1725            Some(HaltReason::BackpressureStall {
1726                stalled_for,
1727                threshold,
1728            }) => {
1729                assert!(
1730                    *stalled_for >= expected_threshold,
1731                    "stalled_for {stalled_for:?} must be >= {expected_threshold:?}",
1732                );
1733                assert_eq!(*threshold, expected_threshold);
1734            }
1735            other => panic!("expected BackpressureStall, was {other:?}"),
1736        }
1737    }
1738}
1739
1740#[cfg(test)]
1741#[cfg(madsim)]
1742mod madsim_tests {
1743    use std::sync::Arc;
1744
1745    use bytes::Bytes;
1746    use indexmap::IndexMap;
1747    use nautilus_core::{UnixNanos, time::get_atomic_clock_static};
1748    use parking_lot::Mutex;
1749    use rstest::rstest;
1750    use ustr::Ustr;
1751
1752    use super::*;
1753    use crate::{
1754        backend::{AppendEntry, EventStore, IndexKind, MemoryBackend, ScanDirection},
1755        entry::EventStoreEntry,
1756        error::EventStoreError,
1757        manifest::{RegisteredComponents, RunManifest, RunStatus},
1758    };
1759
1760    fn manifest(run_id: &str) -> RunManifest {
1761        RunManifest {
1762            run_id: run_id.to_string(),
1763            parent_run_id: None,
1764            instance_id: "trader-001".to_string(),
1765            binary_hash: "deadbeef".to_string(),
1766            schema_version: 1,
1767            crate_versions: "feedface".to_string(),
1768            feature_flags: Vec::new(),
1769            adapter_versions: IndexMap::new(),
1770            config_hash: "cafebabe".to_string(),
1771            registered_components: RegisteredComponents::default(),
1772            seed: None,
1773            start_ts_init: UnixNanos::from(0),
1774            end_ts_init: None,
1775            high_watermark: 0,
1776            status: RunStatus::Running,
1777        }
1778    }
1779
1780    fn entry_draft(ts_init: u64) -> EntryDraft {
1781        EntryDraft {
1782            headers: Headers::empty(),
1783            topic: "exec.command.SubmitOrder".into(),
1784            payload_type: Ustr::from("SubmitOrder"),
1785            payload: Bytes::from_static(b"\x01\x02\x03\x04"),
1786            ts_init: UnixNanos::from(ts_init),
1787            index_keys: Vec::new(),
1788        }
1789    }
1790
1791    #[derive(Debug)]
1792    struct SharedMemory(Arc<Mutex<MemoryBackend>>);
1793
1794    impl SharedMemory {
1795        fn new() -> (Self, Arc<Mutex<MemoryBackend>>) {
1796            let arc = Arc::new(Mutex::new(MemoryBackend::new()));
1797            (Self(Arc::clone(&arc)), arc)
1798        }
1799    }
1800
1801    impl EventStore for SharedMemory {
1802        fn open_run(&mut self, _: RunManifest) -> Result<(), EventStoreError> {
1803            unreachable!("test wrapper does not forward open_run")
1804        }
1805
1806        fn append_batch(&mut self, entries: &[AppendEntry]) -> Result<u64, EventStoreError> {
1807            self.0.lock().append_batch(entries)
1808        }
1809
1810        fn scan_range(
1811            &self,
1812            from: u64,
1813            to: u64,
1814            direction: ScanDirection,
1815        ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
1816            self.0.lock().scan_range(from, to, direction)
1817        }
1818
1819        fn scan_seq(&self, seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
1820            self.0.lock().scan_seq(seq)
1821        }
1822
1823        fn lookup(&self, kind: IndexKind, key: &str) -> Result<Option<u64>, EventStoreError> {
1824            self.0.lock().lookup(kind, key)
1825        }
1826
1827        fn iter_index_keys(&self, kind: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
1828            self.0.lock().iter_index_keys(kind)
1829        }
1830
1831        fn record_snapshot_anchor(
1832            &mut self,
1833            anchor: SnapshotAnchor,
1834        ) -> Result<(), EventStoreError> {
1835            self.0.lock().record_snapshot_anchor(anchor)
1836        }
1837
1838        fn latest_snapshot_anchor(&self) -> Result<Option<SnapshotAnchor>, EventStoreError> {
1839            self.0.lock().latest_snapshot_anchor()
1840        }
1841
1842        fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
1843            self.0.lock().seal(status)
1844        }
1845
1846        fn manifest(&self) -> Result<RunManifest, EventStoreError> {
1847            self.0.lock().manifest()
1848        }
1849
1850        fn high_watermark(&self) -> Result<u64, EventStoreError> {
1851            self.0.lock().high_watermark()
1852        }
1853    }
1854
1855    #[rstest]
1856    fn record_snapshot_anchor_records_current_watermark_under_madsim() {
1857        let (wrapper, shared) = SharedMemory::new();
1858        shared
1859            .lock()
1860            .open_run(manifest("run-anchor"))
1861            .expect("open");
1862
1863        let writer = EventStoreWriter::spawn(
1864            Box::new(wrapper),
1865            get_atomic_clock_static(),
1866            noop_halt(),
1867            WriterConfig::default(),
1868        )
1869        .expect("spawn");
1870
1871        writer.submit(entry_draft(10)).expect("submit first");
1872        writer.submit(entry_draft(11)).expect("submit second");
1873        let anchor = writer
1874            .record_snapshot_anchor("cache://position-snapshots/P-1/0", "blake3:abc")
1875            .expect("record anchor");
1876
1877        let backend = shared.lock();
1878        assert_eq!(anchor.high_watermark, 2);
1879        assert_eq!(
1880            backend.latest_snapshot_anchor().expect("latest anchor"),
1881            Some(anchor),
1882        );
1883    }
1884}
nautilus_event_store/writer/mod.rs

nautilus_event_store/writer/
mod.rs
