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

nautilus_event_store/writer/
mod.rs
