nautilus_event_store/reader/
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//! Reader for the event store.
17//!
18//! The reader composes over the locked [`EventStore`] trait. It owns one backend instance,
19//! exposes range scans as a chunked iterator, single-`seq` lookups, secondary-index
20//! lookups, and manifest access. The backend the reader receives may be a still-open run
21//! (the writer's backend) or a sealed run produced via
22//! [`crate::backend::RedbBackend::open_sealed`].
23//!
24//! Run iteration across a base directory lives on the redb backend itself
25//! ([`crate::backend::RedbBackend::list_runs`]) because it depends on the on-disk file
26//! layout; the in-memory backend has no analog and reads its single open run in place.
27
28use std::{collections::VecDeque, fmt::Debug};
29
30use crate::{
31    backend::{EventStore, IndexKind, ScanDirection},
32    entry::EventStoreEntry,
33    error::EventStoreError,
34    manifest::RunManifest,
35    snapshot::SnapshotAnchor,
36};
37
38/// Default number of entries materialized per chunked `scan_range` call.
39///
40/// Chosen so a forensics scan of a multi-million-entry run keeps the live working set
41/// bounded while amortizing the per-call transaction overhead. Tune through
42/// [`EventStoreReader::scan_range_chunked`] when a workload prefers different bounds.
43pub const DEFAULT_SCAN_CHUNK_SIZE: u64 = 1_024;
44
45/// Replay bounds derived from the latest cache snapshot anchor.
46#[derive(Clone, Debug, PartialEq, Eq)]
47pub struct SnapshotReplayPlan {
48    /// Latest cache snapshot anchor, or `None` when restore must replay from the start.
49    pub anchor: Option<SnapshotAnchor>,
50    /// First event-store seq to replay after the cache snapshot restore.
51    pub from_seq: u64,
52    /// Current durable high-watermark for the run.
53    pub to_seq: u64,
54}
55
56impl SnapshotReplayPlan {
57    /// Returns whether there are no entries to replay for this plan.
58    #[must_use]
59    pub const fn is_empty(&self) -> bool {
60        self.from_seq > self.to_seq
61    }
62}
63
64/// Read-only handle over an [`EventStore`] backend.
65///
66/// The reader is the canonical entry point for read-only replay, audit, and verifier
67/// scans: it never mutates the backend (no `append_batch` surface) and it tolerates
68/// running and sealed backends uniformly.
69#[derive(Debug)]
70pub struct EventStoreReader<B> {
71    backend: B,
72}
73
74impl<B: EventStore> EventStoreReader<B> {
75    /// Wraps `backend` for read-only access.
76    #[must_use]
77    pub const fn new(backend: B) -> Self {
78        Self { backend }
79    }
80
81    /// Returns the manifest of the open run.
82    ///
83    /// # Errors
84    ///
85    /// Returns [`EventStoreError::Backend`] when no run is open.
86    pub fn manifest(&self) -> Result<RunManifest, EventStoreError> {
87        self.backend.manifest()
88    }
89
90    /// Returns the largest `seq` durably acknowledged for the open run.
91    ///
92    /// # Errors
93    ///
94    /// Returns [`EventStoreError::Backend`] when no run is open.
95    pub fn high_watermark(&self) -> Result<u64, EventStoreError> {
96        self.backend.high_watermark()
97    }
98
99    /// Reads a single entry by `seq`.
100    ///
101    /// Returns `None` when `seq == 0` or `seq` exceeds the current high-watermark.
102    ///
103    /// # Errors
104    ///
105    /// Returns [`EventStoreError::HashMismatch`] when the recomputed entry hash diverges
106    /// from the stored value, [`EventStoreError::Gap`] when the row is missing inside
107    /// the high-watermark, and [`EventStoreError::Backend`] for unclassified backend
108    /// failures.
109    pub fn scan_seq(&self, seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
110        self.backend.scan_seq(seq)
111    }
112
113    /// Looks up the first `seq` recorded under the given index key.
114    ///
115    /// # Errors
116    ///
117    /// Returns [`EventStoreError::Backend`] for unclassified backend failures.
118    pub fn lookup(&self, kind: IndexKind, key: &str) -> Result<Option<u64>, EventStoreError> {
119        self.backend.lookup(kind, key)
120    }
121
122    /// Returns the latest snapshot anchor for the run.
123    ///
124    /// Returns `Ok(None)` when no snapshot anchor has been recorded yet.
125    ///
126    /// # Errors
127    ///
128    /// Returns [`EventStoreError::Backend`] when no run is open or the backend does not
129    /// support snapshot anchors, and [`EventStoreError::Corrupted`] when a stored anchor
130    /// cannot decode.
131    pub fn latest_snapshot_anchor(&self) -> Result<Option<SnapshotAnchor>, EventStoreError> {
132        self.backend.latest_snapshot_anchor()
133    }
134
135    /// Builds the restore replay bounds from the latest snapshot anchor.
136    ///
137    /// Restore callers fetch and validate the cache-owned snapshot blob first, then
138    /// replay entries in `[from_seq, to_seq]`. When an anchor exists, `from_seq` is
139    /// `anchor.high_watermark + 1`; without an anchor, restore replays from seq `1`.
140    ///
141    /// # Errors
142    ///
143    /// Returns [`EventStoreError::Backend`] when no run is open or the backend does not
144    /// support snapshot anchors, and [`EventStoreError::Corrupted`] when the stored
145    /// anchor points past the durable high-watermark.
146    pub fn snapshot_replay_plan(&self) -> Result<SnapshotReplayPlan, EventStoreError> {
147        let anchor = self.latest_snapshot_anchor()?;
148        let to_seq = self.high_watermark()?;
149        let from_seq = match anchor.as_ref() {
150            Some(anchor) if anchor.high_watermark > to_seq => {
151                return Err(EventStoreError::Corrupted(format!(
152                    "snapshot anchor high_watermark {} exceeds durable high_watermark {to_seq}",
153                    anchor.high_watermark,
154                )));
155            }
156            Some(anchor) => anchor.high_watermark.saturating_add(1),
157            None => 1,
158        };
159
160        Ok(SnapshotReplayPlan {
161            anchor,
162            from_seq,
163            to_seq,
164        })
165    }
166
167    /// Scans the forward replay tail after the latest snapshot anchor.
168    ///
169    /// This pairs [`Self::snapshot_replay_plan`] with the actual event iterator used by
170    /// restore: entries start at `anchor.high_watermark + 1` when an anchor exists, or
171    /// at seq `1` when no cache snapshot has been anchored.
172    ///
173    /// # Errors
174    ///
175    /// See [`Self::snapshot_replay_plan`].
176    pub fn scan_snapshot_replay_tail(
177        &self,
178    ) -> Result<(SnapshotReplayPlan, RangeScan<'_>), EventStoreError> {
179        let plan = self.snapshot_replay_plan()?;
180        let scan = self.scan_range(plan.from_seq, plan.to_seq, ScanDirection::Forward);
181        Ok((plan, scan))
182    }
183
184    /// Scans entries by `seq` over the inclusive range `[from, to]`.
185    ///
186    /// The returned iterator pulls [`DEFAULT_SCAN_CHUNK_SIZE`] entries at a time from the
187    /// backend so a multi-million-entry forensics scan keeps the working set bounded
188    /// while still amortizing per-transaction overhead. The iterator yields one entry
189    /// per call; backend errors surface as `Some(Err(...))` and terminate the scan.
190    #[must_use]
191    pub fn scan_range(&self, from: u64, to: u64, direction: ScanDirection) -> RangeScan<'_> {
192        RangeScan::new(&self.backend, from, to, direction, DEFAULT_SCAN_CHUNK_SIZE)
193    }
194
195    /// Variant of [`Self::scan_range`] with a caller-chosen chunk size.
196    ///
197    /// `chunk_size == 0` is normalized to `1`; the reader never asks the backend for a
198    /// degenerate empty window because the chunk window is the only progress signal the
199    /// iterator advances on.
200    #[must_use]
201    pub fn scan_range_chunked(
202        &self,
203        from: u64,
204        to: u64,
205        direction: ScanDirection,
206        chunk_size: u64,
207    ) -> RangeScan<'_> {
208        RangeScan::new(&self.backend, from, to, direction, chunk_size.max(1))
209    }
210
211    /// Returns the underlying backend, consuming the reader.
212    #[must_use]
213    pub fn into_inner(self) -> B {
214        self.backend
215    }
216
217    /// Returns a reference to the underlying backend.
218    #[must_use]
219    pub const fn backend(&self) -> &B {
220        &self.backend
221    }
222}
223
224/// Lazy iterator over a `seq` range, materialized in chunks.
225///
226/// Created by [`EventStoreReader::scan_range`] and
227/// [`EventStoreReader::scan_range_chunked`]. The iterator owns no transaction lifetime:
228/// each chunk opens a fresh [`EventStore::scan_range`] call against the backend, so a
229/// long-running scan is not held open against a writer's commit cadence.
230pub struct RangeScan<'a> {
231    backend: &'a dyn EventStore,
232    direction: ScanDirection,
233    chunk_size: u64,
234    cursor: u64,
235    end: u64,
236    buffer: VecDeque<EventStoreEntry>,
237    has_more: bool,
238    // Reverse scans must clamp their starting cursor against the durable
239    // high-watermark on the first fetch. Without that step, a `to` value
240    // above the watermark (a forensics caller passing an open upper bound,
241    // or simply `to = u64::MAX`) makes the first chunk lie wholly above
242    // the durable rows; the backend clips that chunk to an empty Vec and
243    // the iterator would terminate before reading the rows below the
244    // watermark. Forward scans are not affected: an empty forward chunk
245    // genuinely means we have walked past the watermark.
246    reverse_clamped: bool,
247}
248
249impl Debug for RangeScan<'_> {
250    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
251        f.debug_struct(stringify!(RangeScan))
252            .field("direction", &self.direction)
253            .field("chunk_size", &self.chunk_size)
254            .field("cursor", &self.cursor)
255            .field("end", &self.end)
256            .field("buffered", &self.buffer.len())
257            .field("has_more", &self.has_more)
258            .field("reverse_clamped", &self.reverse_clamped)
259            .finish()
260    }
261}
262
263impl<'a> RangeScan<'a> {
264    fn new(
265        backend: &'a dyn EventStore,
266        from: u64,
267        to: u64,
268        direction: ScanDirection,
269        chunk_size: u64,
270    ) -> Self {
271        // Mirror the backend's empty-range conventions so the iterator never makes a
272        // first call into the backend for a degenerate window. `from == 0` is reserved
273        // (seq is 1-based); `from > to` is an empty range.
274        let valid = from != 0 && from <= to;
275        let (cursor, end) = if valid {
276            match direction {
277                ScanDirection::Forward => (from, to),
278                ScanDirection::Reverse => (to, from),
279            }
280        } else {
281            (0, 0)
282        };
283
284        Self {
285            backend,
286            direction,
287            chunk_size: chunk_size.max(1),
288            cursor,
289            end,
290            buffer: VecDeque::new(),
291            has_more: valid,
292            reverse_clamped: false,
293        }
294    }
295
296    fn fetch_chunk(&mut self) -> Option<Result<(), EventStoreError>> {
297        if !self.has_more {
298            return None;
299        }
300
301        if matches!(self.direction, ScanDirection::Reverse) && !self.reverse_clamped {
302            match self.backend.high_watermark() {
303                Ok(hwm) => {
304                    if hwm == 0 || hwm < self.end {
305                        self.has_more = false;
306                        return Some(Ok(()));
307                    }
308                    self.cursor = self.cursor.min(hwm);
309                    self.reverse_clamped = true;
310                }
311                Err(e) => {
312                    self.has_more = false;
313                    return Some(Err(e));
314                }
315            }
316        }
317        let (chunk_lo, chunk_hi) = match self.direction {
318            ScanDirection::Forward => {
319                let lo = self.cursor;
320                let hi = lo
321                    .saturating_add(self.chunk_size)
322                    .saturating_sub(1)
323                    .min(self.end);
324                (lo, hi)
325            }
326            ScanDirection::Reverse => {
327                let hi = self.cursor;
328                let lo = hi
329                    .saturating_sub(self.chunk_size.saturating_sub(1))
330                    .max(self.end);
331                (lo, hi)
332            }
333        };
334
335        match self.backend.scan_range(chunk_lo, chunk_hi, self.direction) {
336            Ok(entries) => {
337                if entries.is_empty() {
338                    // The backend clipped the window to its high-watermark or the run is
339                    // shorter than the requested range; either way no further chunks
340                    // will yield rows.
341                    self.has_more = false;
342                    return Some(Ok(()));
343                }
344
345                match self.direction {
346                    ScanDirection::Forward => {
347                        // Advance from the last seq actually returned, not the
348                        // requested window end: the backend clips the window to its
349                        // high-watermark, and on a still-open run entries committed
350                        // between chunk fetches would otherwise be skipped silently.
351                        let last_seq = entries.last().map_or(chunk_hi, |entry| entry.seq);
352
353                        if last_seq >= self.end {
354                            self.has_more = false;
355                        } else {
356                            self.cursor = last_seq + 1;
357                        }
358                    }
359                    ScanDirection::Reverse => {
360                        if chunk_lo <= self.end {
361                            self.has_more = false;
362                        } else {
363                            self.cursor = chunk_lo - 1;
364                        }
365                    }
366                }
367                self.buffer.extend(entries);
368                Some(Ok(()))
369            }
370            Err(e) => {
371                self.has_more = false;
372                Some(Err(e))
373            }
374        }
375    }
376}
377
378impl Iterator for RangeScan<'_> {
379    type Item = Result<EventStoreEntry, EventStoreError>;
380
381    fn next(&mut self) -> Option<Self::Item> {
382        loop {
383            if let Some(entry) = self.buffer.pop_front() {
384                return Some(Ok(entry));
385            }
386
387            match self.fetch_chunk() {
388                Some(Ok(())) => {}
389                Some(Err(e)) => return Some(Err(e)),
390                None => return None,
391            }
392        }
393    }
394}
395
396#[cfg(test)]
397mod tests {
398    use bytes::Bytes;
399    use indexmap::IndexMap;
400    use nautilus_core::UnixNanos;
401    use rstest::{fixture, rstest};
402    use ustr::Ustr;
403
404    use super::*;
405    use crate::{
406        backend::{AppendEntry, IndexKey, MemoryBackend},
407        compute_entry_hash,
408        entry::Topic,
409        headers::Headers,
410        manifest::{RegisteredComponents, RunManifest, RunStatus},
411    };
412
413    fn manifest(run_id: &str) -> RunManifest {
414        RunManifest {
415            run_id: run_id.to_string(),
416            parent_run_id: None,
417            instance_id: "trader-001".to_string(),
418            binary_hash: "deadbeef".to_string(),
419            schema_version: 1,
420            crate_versions: "feedface".to_string(),
421            feature_flags: Vec::new(),
422            adapter_versions: IndexMap::new(),
423            config_hash: "cafebabe".to_string(),
424            registered_components: RegisteredComponents::default(),
425            seed: None,
426            start_ts_init: UnixNanos::from(0),
427            end_ts_init: None,
428            high_watermark: 0,
429            status: RunStatus::Running,
430        }
431    }
432
433    fn build_entry(seq: u64, ts_init: u64) -> EventStoreEntry {
434        let topic: Topic = "exec.command.SubmitOrder".into();
435        let payload_type = Ustr::from("SubmitOrder");
436        let payload = Bytes::from_static(b"\x01\x02\x03\x04");
437        let headers = Headers::empty();
438        let ts_publish = UnixNanos::from(ts_init + 1);
439        let ts_init = UnixNanos::from(ts_init);
440        let hash = compute_entry_hash(
441            seq,
442            ts_init,
443            ts_publish,
444            topic.as_ref(),
445            payload_type.as_str(),
446            &payload,
447            &headers,
448        );
449
450        EventStoreEntry::new(
451            hash,
452            seq,
453            headers,
454            topic,
455            payload_type,
456            payload,
457            ts_init,
458            ts_publish,
459        )
460    }
461
462    fn append_with(seq: u64, ts_init: u64, index_keys: Vec<IndexKey>) -> AppendEntry {
463        AppendEntry::new(build_entry(seq, ts_init), index_keys)
464    }
465
466    /// Appends two entries after serving the first chunk, emulating a writer
467    /// committing between a chunked scan's fetches against a still-open run.
468    #[derive(Debug)]
469    struct GrowingBackend {
470        inner: std::cell::RefCell<MemoryBackend>,
471        grown: std::cell::Cell<bool>,
472    }
473
474    impl EventStore for GrowingBackend {
475        fn open_run(&mut self, _: RunManifest) -> Result<(), EventStoreError> {
476            unreachable!("test wrapper does not forward open_run")
477        }
478
479        fn append_batch(&mut self, entries: &[AppendEntry]) -> Result<u64, EventStoreError> {
480            self.inner.borrow_mut().append_batch(entries)
481        }
482
483        fn scan_range(
484            &self,
485            from: u64,
486            to: u64,
487            direction: ScanDirection,
488        ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
489            let result = self.inner.borrow().scan_range(from, to, direction);
490
491            if !self.grown.replace(true) {
492                self.inner
493                    .borrow_mut()
494                    .append_batch(&[
495                        append_with(3, 103, Vec::new()),
496                        append_with(4, 104, Vec::new()),
497                    ])
498                    .expect("grow run between chunks");
499            }
500            result
501        }
502
503        fn scan_seq(&self, seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
504            self.inner.borrow().scan_seq(seq)
505        }
506
507        fn lookup(&self, kind: IndexKind, key: &str) -> Result<Option<u64>, EventStoreError> {
508            self.inner.borrow().lookup(kind, key)
509        }
510
511        fn iter_index_keys(&self, kind: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
512            self.inner.borrow().iter_index_keys(kind)
513        }
514
515        fn seal(&mut self, status: RunStatus) -> Result<(), EventStoreError> {
516            self.inner.borrow_mut().seal(status)
517        }
518
519        fn manifest(&self) -> Result<RunManifest, EventStoreError> {
520            self.inner.borrow().manifest()
521        }
522
523        fn high_watermark(&self) -> Result<u64, EventStoreError> {
524            self.inner.borrow().high_watermark()
525        }
526    }
527
528    fn populated(count: u64) -> EventStoreReader<MemoryBackend> {
529        let mut backend = MemoryBackend::new();
530        backend.open_run(manifest("run-reader")).expect("open run");
531        let batch: Vec<AppendEntry> = (1..=count)
532            .map(|seq| append_with(seq, 100 + seq, Vec::new()))
533            .collect();
534        backend.append_batch(&batch).expect("append");
535        EventStoreReader::new(backend)
536    }
537
538    #[derive(Debug)]
539    struct AnchorPastWatermarkBackend;
540
541    impl EventStore for AnchorPastWatermarkBackend {
542        fn open_run(&mut self, _manifest: RunManifest) -> Result<(), EventStoreError> {
543            Ok(())
544        }
545
546        fn append_batch(&mut self, _entries: &[AppendEntry]) -> Result<u64, EventStoreError> {
547            Ok(1)
548        }
549
550        fn scan_range(
551            &self,
552            _from: u64,
553            _to: u64,
554            _direction: ScanDirection,
555        ) -> Result<Vec<EventStoreEntry>, EventStoreError> {
556            Ok(Vec::new())
557        }
558
559        fn scan_seq(&self, _seq: u64) -> Result<Option<EventStoreEntry>, EventStoreError> {
560            Ok(None)
561        }
562
563        fn lookup(&self, _kind: IndexKind, _key: &str) -> Result<Option<u64>, EventStoreError> {
564            Ok(None)
565        }
566
567        fn iter_index_keys(&self, _kind: IndexKind) -> Result<Vec<(String, u64)>, EventStoreError> {
568            Ok(Vec::new())
569        }
570
571        fn record_snapshot_anchor(
572            &mut self,
573            _anchor: SnapshotAnchor,
574        ) -> Result<(), EventStoreError> {
575            Ok(())
576        }
577
578        fn latest_snapshot_anchor(&self) -> Result<Option<SnapshotAnchor>, EventStoreError> {
579            Ok(Some(SnapshotAnchor::new(
580                2,
581                "cache://snapshots/run-reader/2",
582                "blake3:abc",
583            )))
584        }
585
586        fn seal(&mut self, _status: RunStatus) -> Result<(), EventStoreError> {
587            Ok(())
588        }
589
590        fn manifest(&self) -> Result<RunManifest, EventStoreError> {
591            Ok(manifest("run-anchor-past-watermark"))
592        }
593
594        fn high_watermark(&self) -> Result<u64, EventStoreError> {
595            Ok(1)
596        }
597    }
598
599    #[fixture]
600    fn reader_with_three() -> EventStoreReader<MemoryBackend> {
601        populated(3)
602    }
603
604    #[rstest]
605    fn manifest_delegates_to_backend(reader_with_three: EventStoreReader<MemoryBackend>) {
606        let m = reader_with_three.manifest().expect("manifest");
607
608        assert_eq!(m.run_id, "run-reader");
609        assert_eq!(m.high_watermark, 3);
610    }
611
612    #[rstest]
613    fn high_watermark_delegates_to_backend(reader_with_three: EventStoreReader<MemoryBackend>) {
614        assert_eq!(reader_with_three.high_watermark().expect("hwm"), 3);
615    }
616
617    #[rstest]
618    fn latest_snapshot_anchor_delegates_to_backend() {
619        let mut backend = MemoryBackend::new();
620        backend.open_run(manifest("run-anchor")).expect("open run");
621        backend
622            .append_batch(&[append_with(1, 101, Vec::new())])
623            .expect("append");
624        let anchor = SnapshotAnchor::new(1, "cache://snapshots/run-anchor/1", "blake3:abc");
625        backend
626            .record_snapshot_anchor(anchor.clone())
627            .expect("record anchor");
628        let reader = EventStoreReader::new(backend);
629
630        assert_eq!(
631            reader.latest_snapshot_anchor().expect("latest anchor"),
632            Some(anchor),
633        );
634    }
635
636    #[rstest]
637    fn snapshot_replay_plan_without_anchor_replays_from_start(
638        reader_with_three: EventStoreReader<MemoryBackend>,
639    ) {
640        let plan = reader_with_three
641            .snapshot_replay_plan()
642            .expect("snapshot replay plan");
643
644        assert_eq!(
645            plan,
646            SnapshotReplayPlan {
647                anchor: None,
648                from_seq: 1,
649                to_seq: 3,
650            },
651        );
652        assert!(!plan.is_empty());
653    }
654
655    #[rstest]
656    fn snapshot_replay_plan_with_anchor_starts_after_anchor_watermark() {
657        let mut backend = MemoryBackend::new();
658        backend.open_run(manifest("run-anchor")).expect("open run");
659        backend
660            .append_batch(&[
661                append_with(1, 101, Vec::new()),
662                append_with(2, 102, Vec::new()),
663                append_with(3, 103, Vec::new()),
664            ])
665            .expect("append");
666        let anchor = SnapshotAnchor::new(2, "cache://snapshots/run-anchor/2", "blake3:abc");
667        backend
668            .record_snapshot_anchor(anchor.clone())
669            .expect("record anchor");
670        let reader = EventStoreReader::new(backend);
671
672        let plan = reader.snapshot_replay_plan().expect("snapshot replay plan");
673
674        assert_eq!(
675            plan,
676            SnapshotReplayPlan {
677                anchor: Some(anchor),
678                from_seq: 3,
679                to_seq: 3,
680            },
681        );
682        assert!(!plan.is_empty());
683    }
684
685    #[rstest]
686    fn snapshot_replay_plan_rejects_anchor_past_watermark() {
687        let reader = EventStoreReader::new(AnchorPastWatermarkBackend);
688        let err = reader
689            .snapshot_replay_plan()
690            .expect_err("anchor past watermark must fail");
691
692        match err {
693            EventStoreError::Corrupted(msg) => {
694                assert!(
695                    msg.contains("exceeds durable high_watermark"),
696                    "msg was: {msg}",
697                );
698            }
699            other => panic!("expected Corrupted, was {other:?}"),
700        }
701    }
702
703    #[rstest]
704    fn scan_snapshot_replay_tail_yields_entries_after_anchor() {
705        let mut backend = MemoryBackend::new();
706        backend.open_run(manifest("run-anchor")).expect("open run");
707        backend
708            .append_batch(&[
709                append_with(1, 101, Vec::new()),
710                append_with(2, 102, Vec::new()),
711                append_with(3, 103, Vec::new()),
712            ])
713            .expect("append");
714        backend
715            .record_snapshot_anchor(SnapshotAnchor::new(
716                2,
717                "cache://snapshots/run-anchor/2",
718                "blake3:abc",
719            ))
720            .expect("record anchor");
721        let reader = EventStoreReader::new(backend);
722
723        let (plan, scan) = reader
724            .scan_snapshot_replay_tail()
725            .expect("snapshot replay tail");
726        let seqs: Vec<_> = scan.map(|entry| entry.expect("entry").seq).collect();
727
728        assert_eq!(plan.from_seq, 3);
729        assert_eq!(seqs, vec![3]);
730    }
731
732    #[rstest]
733    fn scan_snapshot_replay_tail_is_empty_when_anchor_matches_watermark() {
734        let mut backend = MemoryBackend::new();
735        backend.open_run(manifest("run-anchor")).expect("open run");
736        backend
737            .append_batch(&[
738                append_with(1, 101, Vec::new()),
739                append_with(2, 102, Vec::new()),
740            ])
741            .expect("append");
742        backend
743            .record_snapshot_anchor(SnapshotAnchor::new(
744                2,
745                "cache://snapshots/run-anchor/2",
746                "blake3:abc",
747            ))
748            .expect("record anchor");
749        let reader = EventStoreReader::new(backend);
750
751        let (plan, scan) = reader
752            .scan_snapshot_replay_tail()
753            .expect("snapshot replay tail");
754        let seqs: Vec<_> = scan.map(|entry| entry.expect("entry").seq).collect();
755
756        assert_eq!(plan.from_seq, 3);
757        assert_eq!(plan.to_seq, 2);
758        assert!(plan.is_empty());
759        assert!(seqs.is_empty());
760    }
761
762    #[rstest]
763    fn scan_seq_returns_committed_entry(reader_with_three: EventStoreReader<MemoryBackend>) {
764        let entry = reader_with_three
765            .scan_seq(2)
766            .expect("scan")
767            .expect("present");
768
769        assert_eq!(entry.seq, 2);
770        assert_eq!(entry.ts_init, UnixNanos::from(102));
771    }
772
773    #[rstest]
774    fn scan_seq_returns_none_outside_watermark(reader_with_three: EventStoreReader<MemoryBackend>) {
775        assert!(reader_with_three.scan_seq(0).expect("scan").is_none());
776        assert!(reader_with_three.scan_seq(99).expect("scan").is_none());
777    }
778
779    #[rstest]
780    fn lookup_finds_recorded_index_key() {
781        let mut backend = MemoryBackend::new();
782        backend.open_run(manifest("run-lookup")).expect("open run");
783        backend
784            .append_batch(&[
785                AppendEntry::new(
786                    build_entry(1, 100),
787                    vec![IndexKey::new(IndexKind::ClientOrderId, "O-1".to_string())],
788                ),
789                AppendEntry::new(
790                    build_entry(2, 101),
791                    vec![IndexKey::new(IndexKind::VenueOrderId, "V-1".to_string())],
792                ),
793            ])
794            .expect("append");
795        let reader = EventStoreReader::new(backend);
796
797        assert_eq!(
798            reader
799                .lookup(IndexKind::ClientOrderId, "O-1")
800                .expect("lookup"),
801            Some(1),
802        );
803        assert_eq!(
804            reader
805                .lookup(IndexKind::VenueOrderId, "V-1")
806                .expect("lookup"),
807            Some(2),
808        );
809        assert!(
810            reader
811                .lookup(IndexKind::ClientOrderId, "missing")
812                .expect("lookup")
813                .is_none(),
814        );
815    }
816
817    #[rstest]
818    fn scan_range_forward_yields_entries_in_order(
819        reader_with_three: EventStoreReader<MemoryBackend>,
820    ) {
821        let seqs: Vec<u64> = reader_with_three
822            .scan_range(1, 3, ScanDirection::Forward)
823            .map(|r| r.expect("entry").seq)
824            .collect();
825
826        assert_eq!(seqs, vec![1, 2, 3]);
827    }
828
829    #[rstest]
830    fn scan_range_reverse_yields_entries_in_reverse(
831        reader_with_three: EventStoreReader<MemoryBackend>,
832    ) {
833        let seqs: Vec<u64> = reader_with_three
834            .scan_range(1, 3, ScanDirection::Reverse)
835            .map(|r| r.expect("entry").seq)
836            .collect();
837
838        assert_eq!(seqs, vec![3, 2, 1]);
839    }
840
841    #[rstest]
842    fn scan_range_window_clips_to_request() {
843        let reader = populated(10);
844
845        let seqs: Vec<u64> = reader
846            .scan_range(4, 7, ScanDirection::Forward)
847            .map(|r| r.expect("entry").seq)
848            .collect();
849
850        assert_eq!(seqs, vec![4, 5, 6, 7]);
851    }
852
853    #[rstest]
854    fn scan_range_chunked_forward_walks_full_range() {
855        // Seven entries with a chunk size of 2 forces the iterator to make four
856        // backend calls (sizes 2, 2, 2, 1) and stitch them into a single forward
857        // sequence without skipping seq 1, 4, or 7.
858        let reader = populated(7);
859
860        let seqs: Vec<u64> = reader
861            .scan_range_chunked(1, 7, ScanDirection::Forward, 2)
862            .map(|r| r.expect("entry").seq)
863            .collect();
864
865        assert_eq!(seqs, vec![1, 2, 3, 4, 5, 6, 7]);
866    }
867
868    #[rstest]
869    fn forward_chunked_scan_includes_entries_committed_between_chunks() {
870        // The backend clips each window to its current high-watermark; the cursor
871        // must advance from the last returned seq so rows committed between chunk
872        // fetches against a still-open run are not silently skipped.
873        let mut inner = MemoryBackend::new();
874        inner.open_run(manifest("run-growing")).expect("open run");
875        inner
876            .append_batch(&[
877                append_with(1, 101, Vec::new()),
878                append_with(2, 102, Vec::new()),
879            ])
880            .expect("seed");
881        let reader = EventStoreReader::new(GrowingBackend {
882            inner: std::cell::RefCell::new(inner),
883            grown: std::cell::Cell::new(false),
884        });
885
886        let seqs: Vec<u64> = reader
887            .scan_range_chunked(1, u64::MAX, ScanDirection::Forward, 10)
888            .map(|r| r.expect("entry").seq)
889            .collect();
890
891        assert_eq!(seqs, vec![1, 2, 3, 4]);
892    }
893
894    #[rstest]
895    fn scan_range_chunked_reverse_walks_full_range() {
896        // Mirror of the forward case: chunk size 2 over seven entries must yield
897        // descending [7, 6, 5, 4, 3, 2, 1] without dropping the chunk-boundary seqs.
898        let reader = populated(7);
899
900        let seqs: Vec<u64> = reader
901            .scan_range_chunked(1, 7, ScanDirection::Reverse, 2)
902            .map(|r| r.expect("entry").seq)
903            .collect();
904
905        assert_eq!(seqs, vec![7, 6, 5, 4, 3, 2, 1]);
906    }
907
908    #[rstest]
909    fn scan_range_clips_to_high_watermark() {
910        // Requesting beyond the watermark must terminate without a Gap error: the
911        // backend reports the empty tail by returning an empty Vec, and the iterator
912        // honors that as end-of-stream.
913        let reader = populated(3);
914
915        let seqs: Vec<u64> = reader
916            .scan_range_chunked(1, 99, ScanDirection::Forward, 2)
917            .map(|r| r.expect("entry").seq)
918            .collect();
919
920        assert_eq!(seqs, vec![1, 2, 3]);
921    }
922
923    #[rstest]
924    fn scan_range_reverse_clips_to_high_watermark() {
925        // Reverse scans whose `to` sits more than one chunk above the high-watermark
926        // must still walk the rows below the watermark. Without the up-front cursor
927        // clamp, the backend would clip the first chunk to an empty Vec (entirely
928        // above the watermark) and the iterator would terminate before yielding any
929        // entry: regression coverage for the open `to` forensics call site.
930        let reader = populated(3);
931
932        let seqs: Vec<u64> = reader
933            .scan_range_chunked(1, 99, ScanDirection::Reverse, 2)
934            .map(|r| r.expect("entry").seq)
935            .collect();
936
937        assert_eq!(seqs, vec![3, 2, 1]);
938    }
939
940    #[rstest]
941    fn scan_range_reverse_with_to_at_u64_max() {
942        // Belt-and-braces: even an open upper bound at `u64::MAX` must yield the
943        // durable rows. Demonstrates the clamp guards against pathological inputs
944        // (a defensive caller, a debug REPL, or a max-sentinel) without spinning.
945        let reader = populated(3);
946
947        let seqs: Vec<u64> = reader
948            .scan_range_chunked(1, u64::MAX, ScanDirection::Reverse, 1)
949            .map(|r| r.expect("entry").seq)
950            .collect();
951
952        assert_eq!(seqs, vec![3, 2, 1]);
953    }
954
955    #[rstest]
956    fn scan_range_reverse_above_watermark_yields_nothing() {
957        // The reverse range sits entirely above the high-watermark; the iterator
958        // must terminate cleanly with zero entries instead of stepping forever.
959        let reader = populated(3);
960
961        let seqs: Vec<u64> = reader
962            .scan_range(10, 20, ScanDirection::Reverse)
963            .map(|r| r.expect("entry").seq)
964            .collect();
965
966        assert!(seqs.is_empty(), "seqs was: {seqs:?}");
967    }
968
969    #[rstest]
970    #[case::inverted(5, 1, ScanDirection::Forward)]
971    #[case::zero_from(0, 5, ScanDirection::Forward)]
972    #[case::inverted_reverse(5, 1, ScanDirection::Reverse)]
973    fn scan_range_empty_bounds_yield_no_entries(
974        #[case] from: u64,
975        #[case] to: u64,
976        #[case] direction: ScanDirection,
977    ) {
978        let reader = populated(3);
979
980        let seqs: Vec<u64> = reader
981            .scan_range(from, to, direction)
982            .map(|r| r.expect("entry").seq)
983            .collect();
984
985        assert!(seqs.is_empty(), "seqs was: {seqs:?}");
986    }
987
988    #[rstest]
989    fn scan_range_propagates_hash_mismatch_error() {
990        // The MemoryBackend's tampered-payload path returns HashMismatch on scan; the
991        // iterator must surface that error and then stop yielding rather than mask the
992        // failure as end-of-stream.
993        let mut backend = MemoryBackend::new();
994        backend.open_run(manifest("run-tamper")).expect("open run");
995        let mut tampered = build_entry(1, 100);
996        tampered.payload = Bytes::from_static(b"\xFF\xFF");
997        backend
998            .append_batch(&[AppendEntry::without_indices(tampered)])
999            .expect("append");
1000        let reader = EventStoreReader::new(backend);
1001
1002        let mut iter = reader.scan_range(1, 1, ScanDirection::Forward);
1003        let first = iter.next().expect("first item");
1004
1005        match first {
1006            Err(EventStoreError::HashMismatch { seq: 1 }) => {}
1007            other => panic!("expected HashMismatch, was {other:?}"),
1008        }
1009        assert!(
1010            iter.next().is_none(),
1011            "iterator must terminate after surfacing the error",
1012        );
1013    }
1014
1015    #[rstest]
1016    fn scan_range_chunk_size_zero_normalizes_to_one() {
1017        // A zero chunk size must not deadlock: it normalizes to 1 so progress is
1018        // guaranteed even under a pathological caller.
1019        let reader = populated(3);
1020
1021        let seqs: Vec<u64> = reader
1022            .scan_range_chunked(1, 3, ScanDirection::Forward, 0)
1023            .map(|r| r.expect("entry").seq)
1024            .collect();
1025
1026        assert_eq!(seqs, vec![1, 2, 3]);
1027    }
1028
1029    #[rstest]
1030    fn into_inner_returns_backend(reader_with_three: EventStoreReader<MemoryBackend>) {
1031        let backend = reader_with_three.into_inner();
1032
1033        assert_eq!(backend.high_watermark().expect("hwm"), 3);
1034    }
1035
1036    #[rstest]
1037    fn lookup_uses_distinct_kinds() {
1038        // Same string under two IndexKinds must return None for the kind that didn't
1039        // record it; the reader's lookup path must not collapse kinds.
1040        let mut backend = MemoryBackend::new();
1041        backend.open_run(manifest("run-kinds")).expect("open run");
1042        let shared_key = "shared-key".to_string();
1043        backend
1044            .append_batch(&[AppendEntry::new(
1045                build_entry(1, 100),
1046                vec![IndexKey::new(IndexKind::ClientOrderId, shared_key.clone())],
1047            )])
1048            .expect("append");
1049        let reader = EventStoreReader::new(backend);
1050
1051        assert_eq!(
1052            reader
1053                .lookup(IndexKind::ClientOrderId, &shared_key)
1054                .expect("lookup"),
1055            Some(1),
1056        );
1057        assert!(
1058            reader
1059                .lookup(IndexKind::VenueOrderId, &shared_key)
1060                .expect("lookup")
1061                .is_none(),
1062        );
1063    }
1064}
nautilus_event_store/reader/mod.rs

nautilus_event_store/reader/
mod.rs
