reddb_server/storage/compact/
scrub.rs

1use super::*;
2
3/// A failed tree is unverified; the number of affected records is unknown.
4#[derive(Debug)]
5pub struct PagedTreeFinding {
6    pub collection: String,
7    pub tree: String,
8    pub root_page: u32,
9    pub reason: String,
10}
11
12/// This is a tree-only diagnostic, not a complete store integrity check.
13#[derive(Debug, Default)]
14pub struct PagedTreeScrubReport {
15    pub trees_verified: u64,
16    pub records_verified: u64,
17    pub findings: Vec<PagedTreeFinding>,
18}
19
20/// Inspect trees in a closed, checkpointed paged store without writing to it.
21/// Vault, physical assets, logical index correctness and replica state are not
22/// verified. The same layout and WAL restrictions as offline compaction apply.
23pub fn scrub_paged_trees(path: &Path) -> Result<PagedTreeScrubReport> {
24    let source = Arc::new(Source::open(path)?);
25    let catalog = Catalog::read(&source)?;
26    ensure(
27        source.header.version < f::file_format::PAGE_DISK_RESIDENCY_FILE_VERSION
28            || catalog.resume.is_some(),
29        "prepared disk-resident file has no resume catalog",
30    )?;
31    let mut report = PagedTreeScrubReport::default();
32    for collection in &catalog.roots {
33        let state = catalog
34            .resume
35            .as_ref()
36            .and_then(|r| r.collections.get(&collection.collection));
37        if catalog.resume.is_some() {
38            ensure(state.is_some(), "collection missing from resume catalog")?;
39        }
40        if let Some(state) = state {
41            ensure(
42                state.root_page == collection.root_page,
43                "collection root differs from resume catalog",
44            )?;
45        }
46        inspect_tree(
47            &source,
48            &mut report,
49            &collection.collection,
50            "collection",
51            collection.root_page,
52            state.map(|s| s.entities),
53        );
54    }
55    if let Some(resume) = &catalog.resume {
56        ensure(
57            resume.collections.len() == catalog.roots.len(),
58            "resume catalog has unknown collections",
59        )?;
60        for (name, state) in &resume.collections {
61            if let Some(root) = state.key_index_root {
62                inspect_tree(&source, &mut report, name, "key index", root, None);
63            }
64            for (index, state) in &state.secondary_indices {
65                inspect_tree(
66                    &source,
67                    &mut report,
68                    name,
69                    &format!("secondary index {index}"),
70                    state.root_page,
71                    None,
72                );
73            }
74        }
75    }
76    Ok(report)
77}
78
79fn inspect_tree(
80    source: &Arc<Source>,
81    report: &mut PagedTreeScrubReport,
82    collection: &str,
83    name: &str,
84    root: u32,
85    declared_count: Option<u64>,
86) {
87    let result = (|| -> Result<u64> {
88        if root == 0 {
89            ensure(
90                declared_count.unwrap_or(0) == 0,
91                "empty tree has nonzero declared count",
92            )?;
93            return Ok(0);
94        }
95        let tree = BTree::with_root(Arc::clone(source), root);
96        let (expected, _) = inventory_tree(source, &tree, true)?;
97        let mut cursor = tree.cursor_first()?;
98        let mut previous: Option<Vec<u8>> = None;
99        let mut count = 0u64;
100        while let Some((key, _value)) = cursor.next_with_budget(32 * 1024 * 1024)? {
101            ensure(
102                previous.as_ref().is_none_or(|old| old < &key),
103                "tree keys are not strictly ordered",
104            )?;
105            previous = Some(key);
106            count += 1;
107            ensure(
108                count <= expected,
109                "tree cursor exceeds structural inventory",
110            )?;
111        }
112        ensure(count == expected, "tree cursor omits owned records")?;
113        if let Some(declared) = declared_count {
114            ensure(
115                count == declared,
116                "collection count disagrees with resume catalog",
117            )?;
118        }
119        Ok(count)
120    })();
121    match result {
122        Ok(count) => {
123            report.trees_verified += 1;
124            report.records_verified += count;
125        }
126        Err(error) => report.findings.push(PagedTreeFinding {
127            collection: collection.to_owned(),
128            tree: name.to_owned(),
129            root_page: root,
130            reason: error.to_string(),
131        }),
132    }
133}
reddb_server/storage/compact/scrub.rs

reddb_server/storage/compact/
scrub.rs
