backup/
segment.rs

1//! Tailing `wal.bin`: the next byte range to ship, in whole frames.
2//!
3//! A WAL frame is `[len: u32 LE][crc32: u32 LE][payload: len bytes]`
4//! (`core_storage::wal::encode_record`). A segment always ends on a frame
5//! boundary, so concatenating a generation's segments in offset order is a
6//! valid WAL.
7
8use std::io::{self, Read, Seek, SeekFrom};
9use std::path::Path;
10
11/// Frame header: `len` then `crc32`, both u32 little-endian.
12pub const FRAME_HEADER_LEN: usize = 8;
13
14/// Most bytes one call reads, unless a single frame is larger. A WAL that grew
15/// by more than this since the last tick ships over several segments.
16pub const MAX_SEGMENT_BYTES: u64 = 64 * 1024 * 1024;
17
18/// The prefix of `bytes` made of whole, checksum-valid frames: its length and
19/// how many frames it holds.
20///
21/// Stops at the first frame that is incomplete or whose checksum does not
22/// match — the same rule the engine's own decoder applies.
23pub fn complete_frames(bytes: &[u8]) -> (usize, u64) {
24    let mut pos = 0usize;
25    let mut frames = 0u64;
26    while let Some(len) = frame_len_at(bytes, pos) {
27        let start = pos + FRAME_HEADER_LEN;
28        let Some(end) = start.checked_add(len).filter(|&e| e <= bytes.len()) else {
29            break;
30        };
31        let crc = u32::from_le_bytes(bytes[pos + 4..pos + 8].try_into().unwrap());
32        if crc32fast::hash(&bytes[start..end]) != crc {
33            break;
34        }
35        pos = end;
36        frames += 1;
37    }
38    (pos, frames)
39}
40
41/// Counts frames in a WAL read in arbitrary chunks, by their headers.
42///
43/// For a whole file the engine wrote (an archive), the count equals what its
44/// decoder finds. A trailing partial frame is not counted.
45#[derive(Debug, Default)]
46pub struct FrameCounter {
47    header: [u8; FRAME_HEADER_LEN],
48    have: usize,
49    /// Payload bytes of the current frame still to come.
50    remaining: u64,
51    in_frame: bool,
52    frames: u64,
53}
54
55impl FrameCounter {
56    pub fn feed(&mut self, mut buf: &[u8]) {
57        while !buf.is_empty() {
58            if self.in_frame {
59                let take = (self.remaining as usize).min(buf.len());
60                self.remaining -= take as u64;
61                buf = &buf[take..];
62                if self.remaining == 0 {
63                    self.in_frame = false;
64                    self.frames += 1;
65                }
66            } else {
67                let take = (FRAME_HEADER_LEN - self.have).min(buf.len());
68                self.header[self.have..self.have + take].copy_from_slice(&buf[..take]);
69                self.have += take;
70                buf = &buf[take..];
71                if self.have == FRAME_HEADER_LEN {
72                    self.have = 0;
73                    self.remaining =
74                        u32::from_le_bytes(self.header[..4].try_into().unwrap()) as u64;
75                    if self.remaining == 0 {
76                        self.frames += 1;
77                    } else {
78                        self.in_frame = true;
79                    }
80                }
81            }
82        }
83    }
84
85    pub fn frames(&self) -> u64 {
86        self.frames
87    }
88}
89
90fn frame_len_at(bytes: &[u8], pos: usize) -> Option<usize> {
91    let header = bytes.get(pos..pos + FRAME_HEADER_LEN)?;
92    Some(u32::from_le_bytes(header[..4].try_into().unwrap()) as usize)
93}
94
95/// The next segment of `wal_path` to ship: bytes from `from_byte` up to at most
96/// `durable_len`, cut back to the last complete frame.
97///
98/// `durable_len` is what the engine reports as fsynced
99/// (`SharedDb::durable_wal_position`) — never the file's length, which counts
100/// bytes still in the page cache. Returns `(start, end, bytes)`, or `None`
101/// when there is no whole frame to ship (nothing new, the file is missing or
102/// shorter than `from_byte`, or only part of a frame is there).
103///
104/// # Errors
105///
106/// An I/O error reading the file, or `InvalidData` when the bytes at
107/// `from_byte` are long enough to hold a frame but are not one — the cursor is
108/// not on a frame boundary, or the WAL is corrupt — which a retry cannot fix
109/// and must not be shipped.
110pub fn read_new_segment(
111    wal_path: &Path,
112    from_byte: u64,
113    durable_len: u64,
114) -> io::Result<Option<(u64, u64, Vec<u8>)>> {
115    if durable_len <= from_byte {
116        return Ok(None);
117    }
118    let mut file = match std::fs::File::open(wal_path) {
119        Ok(f) => f,
120        Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(None),
121        Err(e) => return Err(e),
122    };
123    let available = durable_len.min(file.metadata()?.len());
124    if available <= from_byte {
125        return Ok(None);
126    }
127    let want = (available - from_byte).min(MAX_SEGMENT_BYTES);
128    file.seek(SeekFrom::Start(from_byte))?;
129    let mut bytes = read_up_to(&mut file, want)?;
130    // A single frame larger than the cap: read exactly that frame.
131    if let Some(len) = frame_len_at(&bytes, 0) {
132        let frame = (FRAME_HEADER_LEN + len) as u64;
133        if frame > bytes.len() as u64 && from_byte + frame <= available {
134            let rest = read_up_to(&mut file, frame - bytes.len() as u64)?;
135            bytes.extend_from_slice(&rest);
136        }
137    }
138    let (whole, _) = complete_frames(&bytes);
139    if whole == 0 {
140        let first_frame_is_there = frame_len_at(&bytes, 0)
141            .is_some_and(|len| (FRAME_HEADER_LEN + len) as u64 <= bytes.len() as u64);
142        if first_frame_is_there {
143            return Err(io::Error::new(
144                io::ErrorKind::InvalidData,
145                format!("wal.bin at byte {from_byte} is not a valid frame; not shipping it"),
146            ));
147        }
148        return Ok(None);
149    }
150    bytes.truncate(whole);
151    Ok(Some((from_byte, from_byte + whole as u64, bytes)))
152}
153
154fn read_up_to(file: &mut std::fs::File, n: u64) -> io::Result<Vec<u8>> {
155    let mut buf = Vec::with_capacity(n as usize);
156    file.take(n).read_to_end(&mut buf)?;
157    Ok(buf)
158}
159
160#[cfg(test)]
161mod tests {
162    use super::*;
163
164    fn frame(payload: &[u8]) -> Vec<u8> {
165        let mut f = Vec::new();
166        f.extend((payload.len() as u32).to_le_bytes());
167        f.extend(crc32fast::hash(payload).to_le_bytes());
168        f.extend(payload);
169        f
170    }
171
172    #[test]
173    fn complete_frames_counts_whole_valid_frames() {
174        let mut b = frame(b"one");
175        b.extend(frame(b"two"));
176        let two = b.len();
177        b.extend(&frame(b"three")[..5]);
178        assert_eq!(complete_frames(&b), (two, 2));
179        assert_eq!(complete_frames(&[]), (0, 0));
180    }
181
182    #[test]
183    fn the_counter_finds_the_same_frames_in_any_chunking() {
184        let mut b = Vec::new();
185        for p in [
186            &b""[..],
187            b"one",
188            b"a much longer payload than the rest",
189            b"x",
190        ] {
191            b.extend(frame(p));
192        }
193        let (_, whole) = complete_frames(&b);
194        assert_eq!(whole, 4);
195        for chunk in [1, 3, 7, 8, 9, 100] {
196            let mut c = FrameCounter::default();
197            for piece in b.chunks(chunk) {
198                c.feed(piece);
199            }
200            assert_eq!(c.frames(), whole, "chunk {chunk}");
201        }
202        let mut c = FrameCounter::default();
203        c.feed(&b[..b.len() - 1]);
204        assert_eq!(c.frames(), 3, "a partial last frame is not counted");
205    }
206
207    #[test]
208    fn a_bad_checksum_stops_the_scan() {
209        let mut b = frame(b"one");
210        let one = b.len();
211        let mut bad = frame(b"two");
212        let last = bad.len() - 1;
213        bad[last] ^= 0xff;
214        b.extend(bad);
215        assert_eq!(complete_frames(&b), (one, 1));
216    }
217
218    #[test]
219    fn nothing_past_the_durable_length_is_read() {
220        let dir = tempfile::tempdir().unwrap();
221        let p = dir.path().join("wal.bin");
222        let mut b = frame(b"one");
223        let one = b.len() as u64;
224        b.extend(frame(b"two"));
225        std::fs::write(&p, &b).unwrap();
226        let (s, e, bytes) = read_new_segment(&p, 0, one).unwrap().unwrap();
227        assert_eq!((s, e), (0, one));
228        assert_eq!(bytes, frame(b"one"));
229        assert!(read_new_segment(&p, one, one).unwrap().is_none());
230    }
231
232    #[test]
233    fn a_whole_frame_that_fails_its_checksum_is_an_error() {
234        let dir = tempfile::tempdir().unwrap();
235        let p = dir.path().join("wal.bin");
236        let mut b = frame(b"0123456789abcdef");
237        let last = b.len() - 1;
238        b[last] ^= 0xff;
239        std::fs::write(&p, &b).unwrap();
240        let err = read_new_segment(&p, 0, b.len() as u64).unwrap_err();
241        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
242    }
243
244    #[test]
245    fn a_missing_file_has_nothing_to_ship() {
246        let dir = tempfile::tempdir().unwrap();
247        assert!(read_new_segment(&dir.path().join("wal.bin"), 0, 100)
248            .unwrap()
249            .is_none());
250    }
251}
backup/segment.rs

backup/
segment.rs
