nodedb_wal/double_write/
recover.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Reading records back out of the double-write buffer.
4//!
5//! Recovery scans **all** slots rather than trusting the file header's
6//! `count` / `write_pos`, which a crash can leave stale or corrupt. Each slot
7//! is self-describing: a checksummed prefix carries its sequence number and
8//! length, and the record's own CRC says whether the bytes are usable.
9
10use crate::error::Result;
11use crate::record::WalRecord;
12
13use super::buffer::DoubleWriteBuffer;
14use super::layout::{DWB_CAPACITY, DWB_SLOT_STRIDE, SLOT_PREFIX_SIZE, SlotPrefix, slot_offset};
15
16impl DoubleWriteBuffer {
17    /// Try to recover a WAL record by LSN from the double-write buffer.
18    ///
19    /// Reads the slots on disk only. A record staged since the last flush is
20    /// not visible here until that flush writes its slot.
21    ///
22    /// When two slots hold the same LSN — a crash between the DWB fsync and
23    /// the WAL fsync leaves an LSN durable here but absent from the WAL, so
24    /// recovery hands that LSN to a *different* record — the copy with the
25    /// highest slot sequence number wins. Returning the older one would
26    /// resurrect a payload that was never acknowledged to any client.
27    pub fn recover_record(&mut self, target_lsn: u64) -> Result<Option<WalRecord>> {
28        let mut best: Option<(u64, WalRecord)> = None;
29        for_each_slot(self, |prefix, record_bytes| {
30            let Some(record) = decode_record(record_bytes) else {
31                return;
32            };
33            if record.header.lsn != target_lsn || record.verify_checksum().is_err() {
34                return;
35            }
36            if best.as_ref().is_none_or(|(seq, _)| prefix.seq > *seq) {
37                best = Some((prefix.seq, record));
38            }
39        })?;
40        Ok(best.map(|(_, record)| record))
41    }
42}
43
44/// Highest slot sequence number readable in the ring, or 0 when it holds none.
45///
46/// The write path resumes above this so a reused sequence number can never let
47/// a stale copy tie with the record that replaced it.
48pub(super) fn scan_max_seq(dwb: &mut DoubleWriteBuffer) -> Result<u64> {
49    let mut max = 0u64;
50    for_each_slot(dwb, |prefix, _| max = max.max(prefix.seq))?;
51    Ok(max)
52}
53
54/// Visit every slot that carries a well-formed prefix, handing the callback
55/// the prefix and the record bytes (WAL header + payload) it frames.
56fn for_each_slot<F>(dwb: &DoubleWriteBuffer, mut visit: F) -> Result<()>
57where
58    F: FnMut(&SlotPrefix, &[u8]),
59{
60    use std::os::unix::io::AsRawFd as _;
61
62    use crate::align::{AlignedBuf, DEFAULT_ALIGNMENT};
63
64    // Under O_DIRECT, reads must also use aligned buffers and aligned
65    // lengths. Read one full aligned slot at a time, then parse.
66    let mut slot = AlignedBuf::new(DWB_SLOT_STRIDE, DEFAULT_ALIGNMENT)?;
67
68    for i in 0..DWB_CAPACITY as u32 {
69        let offset = slot_offset(i);
70        // SAFETY: slot.as_mut_ptr is valid for `capacity()` bytes.
71        let read = unsafe {
72            libc::pread(
73                dwb.file.as_raw_fd(),
74                slot.as_mut_ptr() as *mut libc::c_void,
75                DWB_SLOT_STRIDE,
76                offset as libc::off_t,
77            )
78        };
79        if read <= 0 {
80            continue;
81        }
82        // SAFETY: the kernel populated `read` bytes starting at the buffer.
83        let bytes: &[u8] = unsafe { std::slice::from_raw_parts(slot.as_ptr(), read as usize) };
84
85        let Some(prefix) = SlotPrefix::decode(bytes) else {
86            continue;
87        };
88        let end = SLOT_PREFIX_SIZE + prefix.total_size;
89        if bytes.len() < end {
90            continue;
91        }
92        visit(&prefix, &bytes[SLOT_PREFIX_SIZE..end]);
93    }
94
95    Ok(())
96}
97
98/// Rebuild a record from the bytes a slot frames. `None` when the header is
99/// not a WAL header at all.
100fn decode_record(bytes: &[u8]) -> Option<WalRecord> {
101    use crate::record::{HEADER_SIZE, RecordHeader, WAL_MAGIC};
102
103    if bytes.len() < HEADER_SIZE {
104        return None;
105    }
106    let mut header_buf = [0u8; HEADER_SIZE];
107    header_buf.copy_from_slice(&bytes[..HEADER_SIZE]);
108    let header = RecordHeader::from_bytes(&header_buf);
109    if header.magic != WAL_MAGIC {
110        return None;
111    }
112    Some(WalRecord {
113        header,
114        payload: bytes[HEADER_SIZE..].to_vec(),
115    })
116}
117
118#[cfg(test)]
119mod tests {
120    use std::path::Path;
121
122    use super::*;
123    use crate::double_write::layout::DWB_CAPACITY;
124    use crate::double_write::mode::DwbMode;
125    use crate::double_write::status::DwbMirror;
126    use crate::record::{RecordType, WalRecordArgs};
127
128    fn open_buffered(path: &Path) -> DoubleWriteBuffer {
129        DoubleWriteBuffer::open(path, DwbMode::Buffered).unwrap()
130    }
131
132    fn record(lsn: u64, payload: &[u8]) -> WalRecord {
133        WalRecord::new(WalRecordArgs {
134            record_type: RecordType::Put as u32,
135            lsn,
136            tenant_id: 1,
137            vshard_id: 0,
138            database_id: 0,
139            payload: payload.to_vec(),
140            encryption_key: None,
141            preamble_bytes: None,
142        })
143        .unwrap()
144    }
145
146    fn mirror(dwb: &mut DoubleWriteBuffer, rec: &WalRecord) {
147        assert_eq!(dwb.write_record(rec).unwrap(), DwbMirror::Mirrored);
148    }
149
150    fn mirror_deferred(dwb: &mut DoubleWriteBuffer, rec: &WalRecord) {
151        assert_eq!(dwb.write_record_deferred(rec).unwrap(), DwbMirror::Mirrored);
152    }
153
154    #[test]
155    fn recover_after_wraparound() {
156        let dir = tempfile::tempdir().unwrap();
157        let mut dwb = open_buffered(&dir.path().join("wrap.dwb"));
158
159        let total = DWB_CAPACITY as u64 + 5;
160        for lsn in 1..=total {
161            mirror_deferred(&mut dwb, &record(lsn, format!("wrap-{lsn}").as_bytes()));
162        }
163        dwb.flush().unwrap();
164
165        for lsn in (total - 4)..=total {
166            let recovered = dwb.recover_record(lsn).unwrap().expect("recoverable");
167            assert_eq!(recovered.payload, format!("wrap-{lsn}").into_bytes());
168        }
169
170        for lsn in 1..=5u64 {
171            assert!(
172                dwb.recover_record(lsn).unwrap().is_none(),
173                "LSN {lsn} should have been overwritten by wrap-around"
174            );
175        }
176    }
177
178    /// A crash in the window between the DWB fsync and the WAL fsync leaves an
179    /// LSN durable here but missing from the WAL. Recovery resumes at that LSN
180    /// and a *different* record is written under it, so two CRC-valid slots
181    /// claim it. The one written last is the one the WAL committed.
182    #[test]
183    fn reused_lsn_recovers_the_newer_copy() {
184        let dir = tempfile::tempdir().unwrap();
185        let mut dwb = open_buffered(&dir.path().join("reuse.dwb"));
186
187        mirror(&mut dwb, &record(9, b"never-acknowledged"));
188        mirror(&mut dwb, &record(9, b"after-recovery"));
189
190        let recovered = dwb.recover_record(9).unwrap().expect("recoverable");
191        assert_eq!(
192            recovered.payload, b"after-recovery",
193            "recovery resurrected the unacknowledged copy"
194        );
195    }
196
197    /// The same reuse, but with the ring wrapping between the two copies, so
198    /// the newer one sits at a *lower* slot index than the older one. Pins the
199    /// selection to the sequence number: neither ascending nor descending slot
200    /// order is correct on its own, and this case is the one a descending scan
201    /// would get wrong.
202    #[test]
203    fn reused_lsn_across_a_wrap_recovers_the_newer_copy() {
204        let dir = tempfile::tempdir().unwrap();
205        let mut dwb = open_buffered(&dir.path().join("reuse_wrap.dwb"));
206
207        // Older copy lands in the last slot of the ring.
208        for lsn in 1..DWB_CAPACITY as u64 {
209            mirror_deferred(&mut dwb, &record(lsn, b"filler"));
210        }
211        mirror_deferred(&mut dwb, &record(500, b"never-acknowledged"));
212        // The next write wraps to slot 0.
213        mirror_deferred(&mut dwb, &record(500, b"after-recovery"));
214        dwb.flush().unwrap();
215
216        let recovered = dwb.recover_record(500).unwrap().expect("recoverable");
217        assert_eq!(recovered.payload, b"after-recovery");
218    }
219
220    /// Reopening must not restart the sequence: a fresh copy written after a
221    /// restart still has to outrank the stale one already on disk.
222    #[test]
223    fn reused_lsn_across_a_reopen_recovers_the_newer_copy() {
224        let dir = tempfile::tempdir().unwrap();
225        let path = dir.path().join("reuse_reopen.dwb");
226
227        {
228            let mut dwb = open_buffered(&path);
229            mirror(&mut dwb, &record(3, b"never-acknowledged"));
230        }
231
232        let mut dwb = open_buffered(&path);
233        mirror(&mut dwb, &record(3, b"after-recovery"));
234
235        let recovered = dwb.recover_record(3).unwrap().expect("recoverable");
236        assert_eq!(recovered.payload, b"after-recovery");
237    }
238}
nodedb_wal/double_write/recover.rs

nodedb_wal/double_write/
recover.rs
