reddb_file/
scrub.rs

1use std::fs::File;
2use std::io::{Read, Seek, SeekFrom};
3use std::path::Path;
4use std::time::Instant;
5
6use crate::{
7    RdbFileError, RdbFileResult, EMBEDDED_RDB_SUPERBLOCK_0_OFFSET,
8    EMBEDDED_RDB_SUPERBLOCK_1_OFFSET, EMBEDDED_RDB_SUPERBLOCK_SIZE,
9};
10
11const CHECKSUM_LEN: usize = 4;
12const SUPERBLOCK_MAGIC: &[u8; 8] = b"RDBSBLK1";
13
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct StorageScrubFinding {
16    pub zone_kind: String,
17    pub physical_identity: String,
18    pub collection: Option<String>,
19    pub expected_checksum: Option<String>,
20    pub actual_checksum: Option<String>,
21    pub fault_class: Option<String>,
22}
23
24#[derive(Debug, Clone, Default, PartialEq, Eq)]
25pub struct StorageScrubVerifiedCounters {
26    pub superblock: u64,
27    pub manifest: u64,
28    pub wal: u64,
29    pub page: u64,
30    pub segment_chunk: u64,
31}
32
33#[derive(Debug, Clone, Default, PartialEq, Eq)]
34pub struct StorageScrubReport {
35    pub findings: Vec<StorageScrubFinding>,
36    pub verified: StorageScrubVerifiedCounters,
37    pub objects_verified: u64,
38    pub total_objects: u64,
39    pub bytes_read: u64,
40    pub duration_ms: u64,
41    pub next_cursor: usize,
42    pub complete: bool,
43}
44
45#[derive(Debug, Clone, Copy)]
46struct ParsedSuperblock {
47    generation: u64,
48    manifest_offset: u64,
49    manifest_len: u64,
50    manifest_checksum: u32,
51    wal_recovery_boundary: u64,
52    snapshot_offset: u64,
53    snapshot_bytes: u64,
54    snapshot_checksum: u32,
55}
56
57pub fn scrub_embedded_store(
58    path: &Path,
59    start_cursor: usize,
60    max_objects: usize,
61) -> RdbFileResult<StorageScrubReport> {
62    let started = Instant::now();
63    let mut file = File::open(path)?;
64    let file_len = file.metadata()?.len();
65    let mut report = StorageScrubReport {
66        total_objects: 5,
67        next_cursor: start_cursor,
68        ..StorageScrubReport::default()
69    };
70
71    let mut superblocks = Vec::new();
72    // A resumed pass (cursor past a superblock) still needs the superblock
73    // authority that an earlier tick verified: reload it silently — no
74    // findings, no verified counters — since the tick that verified it
75    // already reported any corruption. Without this, every zone after the
76    // split would report a false missing-authority finding.
77    for copy_index in 0..start_cursor.min(2) {
78        if let Some(parsed) = load_superblock(&mut file, copy_index as u8)? {
79            superblocks.push(parsed);
80        }
81    }
82    let end_cursor = start_cursor.saturating_add(max_objects.max(1)).min(5);
83    for object_index in start_cursor..end_cursor {
84        match object_index {
85            0 => verify_superblock(&mut file, 0, &mut report, &mut superblocks)?,
86            1 => verify_superblock(&mut file, 1, &mut report, &mut superblocks)?,
87            2 => verify_manifest(&mut file, &mut report, &superblocks, file_len)?,
88            3 => verify_snapshot(&mut file, &mut report, &superblocks, file_len)?,
89            4 => verify_wal_boundary(&mut file, &mut report, &superblocks, file_len)?,
90            _ => {}
91        }
92        report.next_cursor = object_index + 1;
93    }
94
95    report.complete = report.next_cursor >= 5;
96    report.duration_ms = started.elapsed().as_millis().try_into().unwrap_or(u64::MAX);
97    Ok(report)
98}
99
100/// Read and parse a superblock copy for use as zone authority, reporting
101/// nothing: a checksum-invalid copy yields `None` exactly as it yields no
102/// authority in the verifying path.
103fn load_superblock(file: &mut File, copy_index: u8) -> RdbFileResult<Option<ParsedSuperblock>> {
104    let offset = match copy_index {
105        0 => EMBEDDED_RDB_SUPERBLOCK_0_OFFSET,
106        1 => EMBEDDED_RDB_SUPERBLOCK_1_OFFSET,
107        _ => unreachable!(),
108    };
109    let mut bytes = vec![0u8; EMBEDDED_RDB_SUPERBLOCK_SIZE as usize];
110    file.seek(SeekFrom::Start(offset))?;
111    file.read_exact(&mut bytes)?;
112    let stored = checksum_trailer(&bytes)?;
113    if stored != crc32(&bytes[..bytes.len() - CHECKSUM_LEN]) {
114        return Ok(None);
115    }
116    Ok(parse_superblock(copy_index, &bytes))
117}
118
119fn verify_superblock(
120    file: &mut File,
121    copy_index: u8,
122    report: &mut StorageScrubReport,
123    superblocks: &mut Vec<ParsedSuperblock>,
124) -> RdbFileResult<()> {
125    let offset = match copy_index {
126        0 => EMBEDDED_RDB_SUPERBLOCK_0_OFFSET,
127        1 => EMBEDDED_RDB_SUPERBLOCK_1_OFFSET,
128        _ => unreachable!(),
129    };
130    let mut bytes = vec![0u8; EMBEDDED_RDB_SUPERBLOCK_SIZE as usize];
131    file.seek(SeekFrom::Start(offset))?;
132    file.read_exact(&mut bytes)?;
133    report.bytes_read += bytes.len() as u64;
134    report.objects_verified += 1;
135
136    let stored = checksum_trailer(&bytes)?;
137    let actual = crc32(&bytes[..bytes.len() - CHECKSUM_LEN]);
138    let identity = format!("superblock:{copy_index}");
139    if stored != actual {
140        report.findings.push(StorageScrubFinding {
141            zone_kind: "superblock".to_string(),
142            physical_identity: identity,
143            collection: None,
144            expected_checksum: Some(hex32(stored)),
145            actual_checksum: Some(hex32(actual)),
146            fault_class: Some("bit-rot-evidence".to_string()),
147        });
148        return Ok(());
149    }
150
151    if let Some(parsed) = parse_superblock(copy_index, &bytes) {
152        superblocks.push(parsed);
153    }
154    report.verified.superblock += 1;
155    Ok(())
156}
157
158fn verify_manifest(
159    file: &mut File,
160    report: &mut StorageScrubReport,
161    superblocks: &[ParsedSuperblock],
162    file_len: u64,
163) -> RdbFileResult<()> {
164    report.objects_verified += 1;
165    let Some(superblock) = newest_superblock(superblocks) else {
166        report
167            .findings
168            .push(missing_authority("manifest", "manifest"));
169        return Ok(());
170    };
171    if superblock.manifest_len < CHECKSUM_LEN as u64
172        || superblock
173            .manifest_offset
174            .saturating_add(superblock.manifest_len)
175            > file_len
176    {
177        report.findings.push(StorageScrubFinding {
178            zone_kind: "manifest".to_string(),
179            physical_identity: "manifest".to_string(),
180            collection: None,
181            expected_checksum: Some(hex32(superblock.manifest_checksum)),
182            actual_checksum: None,
183            fault_class: Some("torn-write-evidence".to_string()),
184        });
185        return Ok(());
186    }
187    let mut bytes = vec![0u8; superblock.manifest_len as usize];
188    file.seek(SeekFrom::Start(superblock.manifest_offset))?;
189    file.read_exact(&mut bytes)?;
190    report.bytes_read += bytes.len() as u64;
191    let actual = crc32(&bytes[..bytes.len() - CHECKSUM_LEN]);
192    if actual != superblock.manifest_checksum {
193        report.findings.push(StorageScrubFinding {
194            zone_kind: "manifest".to_string(),
195            physical_identity: "manifest".to_string(),
196            collection: None,
197            expected_checksum: Some(hex32(superblock.manifest_checksum)),
198            actual_checksum: Some(hex32(actual)),
199            fault_class: Some("bit-rot-evidence".to_string()),
200        });
201        return Ok(());
202    }
203    report.verified.manifest += 1;
204    Ok(())
205}
206
207fn verify_snapshot(
208    file: &mut File,
209    report: &mut StorageScrubReport,
210    superblocks: &[ParsedSuperblock],
211    file_len: u64,
212) -> RdbFileResult<()> {
213    report.objects_verified += 1;
214    let Some(superblock) = newest_superblock(superblocks) else {
215        report.findings.push(missing_authority("page", "snapshot"));
216        return Ok(());
217    };
218    if superblock.snapshot_bytes == 0 {
219        report.verified.page += 1;
220        return Ok(());
221    }
222    if superblock
223        .snapshot_offset
224        .saturating_add(superblock.snapshot_bytes)
225        > file_len
226    {
227        report.findings.push(StorageScrubFinding {
228            zone_kind: "page".to_string(),
229            physical_identity: "snapshot".to_string(),
230            collection: None,
231            expected_checksum: Some(hex32(superblock.snapshot_checksum)),
232            actual_checksum: None,
233            fault_class: Some("torn-write-evidence".to_string()),
234        });
235        return Ok(());
236    }
237    let mut bytes = vec![0u8; superblock.snapshot_bytes as usize];
238    file.seek(SeekFrom::Start(superblock.snapshot_offset))?;
239    file.read_exact(&mut bytes)?;
240    report.bytes_read += bytes.len() as u64;
241    let actual = crc32(&bytes);
242    if actual != superblock.snapshot_checksum {
243        report.findings.push(StorageScrubFinding {
244            zone_kind: "page".to_string(),
245            physical_identity: "snapshot".to_string(),
246            collection: None,
247            expected_checksum: Some(hex32(superblock.snapshot_checksum)),
248            actual_checksum: Some(hex32(actual)),
249            fault_class: Some("bit-rot-evidence".to_string()),
250        });
251        return Ok(());
252    }
253    report.verified.page += 1;
254    Ok(())
255}
256
257fn verify_wal_boundary(
258    _file: &mut File,
259    report: &mut StorageScrubReport,
260    superblocks: &[ParsedSuperblock],
261    file_len: u64,
262) -> RdbFileResult<()> {
263    report.objects_verified += 1;
264    let Some(superblock) = newest_superblock(superblocks) else {
265        report
266            .findings
267            .push(missing_authority("wal", "embedded-wal"));
268        return Ok(());
269    };
270    if superblock.wal_recovery_boundary > file_len {
271        report.findings.push(StorageScrubFinding {
272            zone_kind: "wal".to_string(),
273            physical_identity: "embedded-wal".to_string(),
274            collection: None,
275            expected_checksum: None,
276            actual_checksum: None,
277            fault_class: Some("torn-write-evidence".to_string()),
278        });
279        return Ok(());
280    }
281    report.verified.wal += 1;
282    Ok(())
283}
284
285fn parse_superblock(copy_index: u8, bytes: &[u8]) -> Option<ParsedSuperblock> {
286    if bytes.len() != EMBEDDED_RDB_SUPERBLOCK_SIZE as usize || &bytes[..8] != SUPERBLOCK_MAGIC {
287        return None;
288    }
289    let stored_copy_index = bytes[12];
290    if stored_copy_index != copy_index {
291        return None;
292    }
293    Some(ParsedSuperblock {
294        generation: read_u64(bytes, 13)?,
295        manifest_offset: read_u64(bytes, 25)?,
296        manifest_len: read_u64(bytes, 33)?,
297        manifest_checksum: read_u32(bytes, 41)?,
298        wal_recovery_boundary: read_u64(bytes, 61)?,
299        // offset 69..77 is wal_live_bytes (superblock v2); the census does
300        // not need it, but the snapshot fields sit 8 bytes further out.
301        snapshot_offset: read_u64(bytes, 77)?,
302        snapshot_bytes: read_u64(bytes, 85)?,
303        snapshot_checksum: read_u32(bytes, 93)?,
304    })
305}
306
307fn newest_superblock(superblocks: &[ParsedSuperblock]) -> Option<ParsedSuperblock> {
308    superblocks
309        .iter()
310        .max_by_key(|superblock| superblock.generation)
311        .copied()
312}
313
314fn missing_authority(zone_kind: &str, physical_identity: &str) -> StorageScrubFinding {
315    StorageScrubFinding {
316        zone_kind: zone_kind.to_string(),
317        physical_identity: physical_identity.to_string(),
318        collection: None,
319        expected_checksum: None,
320        actual_checksum: None,
321        fault_class: Some("missing-checksum-authority".to_string()),
322    }
323}
324
325fn checksum_trailer(bytes: &[u8]) -> RdbFileResult<u32> {
326    if bytes.len() < CHECKSUM_LEN {
327        return Err(RdbFileError::InvalidOperation(
328            "checksum trailer requires at least four bytes".to_string(),
329        ));
330    }
331    Ok(u32::from_le_bytes(
332        bytes[bytes.len() - CHECKSUM_LEN..].try_into().unwrap(),
333    ))
334}
335
336fn read_u32(bytes: &[u8], offset: usize) -> Option<u32> {
337    Some(u32::from_le_bytes(
338        bytes.get(offset..offset + 4)?.try_into().ok()?,
339    ))
340}
341
342fn read_u64(bytes: &[u8], offset: usize) -> Option<u64> {
343    Some(u64::from_le_bytes(
344        bytes.get(offset..offset + 8)?.try_into().ok()?,
345    ))
346}
347
348fn crc32(bytes: &[u8]) -> u32 {
349    let mut hasher = crc32fast::Hasher::new();
350    hasher.update(bytes);
351    hasher.finalize()
352}
353
354fn hex32(value: u32) -> String {
355    format!("crc32:{value:08x}")
356}
reddb_file/scrub.rs

reddb_file/
scrub.rs
