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

nautilus_event_store/writer/
mod.rs
