1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
//! Reference-count recovery for the commit DAG and B+ tree node pool.
//!
//! Called after deserialisation and on crash recovery.
use std::collections::{HashMap, HashSet};
use crate::{
basic::persistent_btree::{EMPTY_ROOT, NodeId},
common::error::{Result, VsdbError},
};
use super::{CommitId, NO_COMMIT, map::VerMap};
impl<K, V> VerMap<K, V> {
pub(crate) fn rebuild_branch_name_index(&mut self) {
let mut seen = HashSet::new();
self.branch_names.clear();
for (id, state) in self.branches.iter() {
if seen.insert(state.name.clone()) {
self.branch_names.insert(&state.name, &id);
}
}
}
/// Rebuilds the B+ tree's in-memory ref-count map from the
/// current set of live roots (all commit roots + dirty roots).
///
/// Called after every deserialization path (serde, from_meta) because a
/// newly restored node-pool runtime starts without reference counts.
pub(crate) fn rebuild_tree_ref_counts(&mut self) {
// No old durable pointer may survive a sweep that retires its nodes.
// The helper is a no-op for read-only restoration.
self.sync_storage();
let mut live_roots: Vec<NodeId> =
self.commits.iter().map(|(_, c)| c.root).collect();
for (_, s) in self.branches.iter() {
if s.dirty_root != EMPTY_ROOT {
live_roots.push(s.dirty_root);
}
}
self.tree.rebuild_ref_counts(&live_roots);
}
pub(crate) fn repair_commit_ref_counts_if_needed(&mut self) -> Result<()> {
// Validate even clean/read-only restores before changing any index or
// reclaiming anything. Missing HEADs/parents are corruption, not proof
// that the remaining stored history is unreachable.
self.validate_commit_graph()?;
// Read-only restores must not rewrite the branch-name index or
// crash-recovery metadata. Queries use the authoritative branch
// table directly where needed, while tree runtime state is rebuilt
// separately below without changing durable data.
if self.namespace().is_read_only() {
return Ok(());
}
self.rebuild_branch_name_index();
if self.gc_dirty.get_value()
|| self.commits.iter().any(|(_, c)| c.ref_count == 0)
{
self.rebuild_ref_counts()?;
}
Ok(())
}
/// Counts every branch HEAD and reachable parent edge without writing.
/// A missing reachable record must be reported before any orphan cleanup.
pub(crate) fn validate_commit_graph(&self) -> Result<HashMap<CommitId, u32>> {
let mut visited = HashSet::new();
let mut ref_counts: HashMap<CommitId, u32> = HashMap::new();
let mut queue: Vec<CommitId> = Vec::new();
for (_, s) in self.branches.iter() {
if s.head != NO_COMMIT {
*ref_counts.entry(s.head).or_insert(0) += 1;
queue.push(s.head);
}
}
while let Some(id) = queue.pop() {
if !visited.insert(id) {
continue;
}
let c = self
.commits
.get(&id)
.ok_or(VsdbError::CommitNotFound { commit_id: id })?;
for &parent in &c.parents {
if parent != NO_COMMIT {
*ref_counts.entry(parent).or_insert(0) += 1;
queue.push(parent);
}
}
}
Ok(ref_counts)
}
/// Rebuilds all commit ref_counts from scratch by walking all
/// live branches. Hard-deletes any unreachable commits only after the
/// entire reachable graph has been validated.
///
/// Called on crash recovery (`gc_dirty == true`) or when migrating
/// from pre-ref-count data (`ref_count == 0` on all commits).
pub(crate) fn rebuild_ref_counts(&mut self) -> Result<()> {
let ref_counts = self.validate_commit_graph()?;
// A legacy zero-count repair may begin with a false marker. Pin true
// before changing counts, and settle the graph before deleting rows.
self.begin_ref_update();
self.tree.nodes.sync_wal();
self.commits.sync_wal();
self.branches.sync_wal();
for (&id, &correct) in &ref_counts {
// The full graph was validated above and structural writers are
// serialized, so every reachable commit still exists.
let mut c = self.commits.get(&id).expect("validated commit disappeared");
if c.ref_count != correct {
c.ref_count = correct;
self.commits.insert(&id, &c);
}
}
let all_ids: Vec<u64> = self.commits.iter().map(|(id, _)| id).collect();
for id in all_ids {
if !ref_counts.contains_key(&id) {
self.commits.remove(&id);
}
}
// A durable false marker must never outlive unsynced count repairs.
self.commits.sync_wal();
self.end_ref_update();
Ok(())
}
}