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

nautilus_event_store/reader/
mod.rs
