open_wal/reader.rs
1//! Streaming replay reader (§6, §6.1).
2//!
3//! [`Reader`] is a **lending** iterator: each yielded `&[u8]` borrows the
4//! reader's reused buffer and is valid only until the next [`Reader::next`]
5//! call. This is why it is not a `std::iter::Iterator` (§6.1) — the standard
6//! trait cannot express an item tied to the per-call borrow. The reuse keeps
7//! replay zero-copy and, after warm-up, zero-allocation (§14.7).
8//!
9//! **M4 scope:** the reader follows the log across **multiple segments**. It
10//! holds an immutable borrow of the `Wal`'s `dir` and sorted `segments` list
11//! and opens each segment by path on demand, advancing to the next
12//! higher-`base_lsn` segment when the current one's records end (§15.2
13//! segment-roll following). Crossing a segment boundary opens a file (and may
14//! allocate); the within-segment [`Reader::next`] hot path stays zero-alloc
15//! (§7.5 scopes the zero-alloc guarantee to the no-roll steady state).
16
17use std::fs::File;
18use std::path::Path;
19
20use crate::Lsn;
21use crate::error::Result;
22use crate::record::RECORD_HEADER_SIZE;
23use crate::segment::{self, ScanOutcome};
24
25/// Streaming reader over the log, starting at a caller-chosen LSN.
26///
27/// Holds an immutable borrow of the `Wal`'s directory and segment list (so the
28/// writer's `&mut self` methods are excluded for the reader's lifetime), an
29/// owned `File` for the segment currently being scanned, and a reusable I/O
30/// buffer. Segments past the first are opened lazily as the scan crosses each
31/// boundary.
32pub struct Reader<'w> {
33 dir: &'w Path,
34 /// Sorted (ascending) `base_lsn`s of the segments to replay, oldest first.
35 segments: &'w [Lsn],
36 /// Index into `segments` of the segment `file` is currently open on.
37 seg_idx: usize,
38 /// The currently-open segment file, or `None` before the first record /
39 /// after the stream ends.
40 file: Option<File>,
41 offset: u64,
42 expected_lsn: Lsn,
43 from: Lsn,
44 segment_size: u64,
45 max_record_size: u32,
46 buf: Vec<u8>,
47 done: bool,
48}
49
50impl<'w> Reader<'w> {
51 /// Build a reader replaying `segments` (sorted ascending `base_lsn`s) under
52 /// `dir`, with `first` already opened on `segments[0]`, yielding records
53 /// with `lsn >= from`. The first record's LSN is `segments[0]` (the oldest
54 /// base); continuity carries `expected_lsn` across segment boundaries.
55 pub(crate) fn new(
56 dir: &'w Path,
57 segments: &'w [Lsn],
58 first: File,
59 from: Lsn,
60 segment_size: u64,
61 max_record_size: u32,
62 ) -> Reader<'w> {
63 debug_assert!(!segments.is_empty(), "a Wal always has ≥1 segment");
64 Reader {
65 dir,
66 segments,
67 seg_idx: 0,
68 file: Some(first),
69 offset: segment::HEADER_SIZE,
70 expected_lsn: segments[0],
71 from,
72 segment_size,
73 max_record_size,
74 buf: Vec::new(),
75 done: false,
76 }
77 }
78
79 /// Advance to the next segment in the list, opening its file and resetting
80 /// the scan offset to just past the header. Returns `Ok(false)` when there
81 /// is no next segment (end of log). `expected_lsn` is **not** reset — the
82 /// next segment's first record continues the dense LSN run (recovery
83 /// validated cross-segment continuity, §8.1).
84 fn advance_segment(&mut self) -> Result<bool> {
85 self.seg_idx += 1;
86 if self.seg_idx >= self.segments.len() {
87 self.file = None;
88 return Ok(false);
89 }
90 let name = segment::filename_for(self.segments[self.seg_idx]);
91 self.file = Some(File::open(self.dir.join(name))?);
92 self.offset = segment::HEADER_SIZE;
93 Ok(true)
94 }
95
96 /// Yield the next record at or after `from`, or `None` at end of log.
97 ///
98 /// Lending-style: the returned borrow is tied to `&mut self` and is valid
99 /// only until the following call (§6.1). Records before `from` are skipped.
100 /// When the current segment's records end (sentinel / short read / invalid /
101 /// LSN-continuity break), the reader follows the roll to the next segment;
102 /// at the last segment that ends the stream cleanly. Torn-tail/corruption
103 /// classification is a recovery concern (§8.2), not a live read.
104 //
105 // Deliberately not `std::iter::Iterator::next`: the lending borrow (item
106 // tied to `&mut self`) cannot be expressed by the standard trait (§6.1).
107 #[allow(clippy::should_implement_trait)]
108 pub fn next(&mut self) -> Option<Result<(Lsn, &[u8])>> {
109 // Copy the `Copy` fields so the per-iteration `&mut self.buf` borrow in
110 // `read_record_at` does not conflict with reading `self.file`.
111 let segment_size = self.segment_size;
112 let max_record_size = self.max_record_size;
113
114 loop {
115 if self.done {
116 return None;
117 }
118 let Some(file) = self.file.as_ref() else {
119 self.done = true;
120 return None;
121 };
122
123 match segment::read_record_at(
124 file,
125 self.offset,
126 segment_size,
127 max_record_size,
128 &mut self.buf,
129 ) {
130 Err(e) => {
131 self.done = true;
132 return Some(Err(e));
133 }
134 Ok(ScanOutcome::CleanEnd) | Ok(ScanOutcome::Invalid) => {
135 // End of this segment's records. Follow the roll to the next
136 // segment, if any; otherwise the stream is done. (A sentinel,
137 // short read, or invalid record ends the live stream cleanly;
138 // torn-tail/corruption classification is a recovery concern.)
139 match self.advance_segment() {
140 Ok(true) => continue,
141 Ok(false) => {
142 self.done = true;
143 return None;
144 }
145 Err(e) => {
146 self.done = true;
147 return Some(Err(e));
148 }
149 }
150 }
151 Ok(ScanOutcome::Record {
152 lsn,
153 payload_len,
154 framed_len,
155 }) => {
156 if lsn != self.expected_lsn {
157 // A clean log has dense, in-order LSNs across segments; a
158 // break ends the valid records (defensive — recovery
159 // already validated continuity).
160 self.done = true;
161 return None;
162 }
163 self.offset += framed_len as u64;
164 self.expected_lsn = lsn.next();
165 if lsn < self.from {
166 continue;
167 }
168 let payload = &self.buf[RECORD_HEADER_SIZE..RECORD_HEADER_SIZE + payload_len];
169 return Some(Ok((lsn, payload)));
170 }
171 }
172 }
173 }
174}