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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
//! 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();
// Preserve valid rows so concurrent readers through another handle
// never lose a live name while this derived index is reconciled.
for (id, state) in self.branches.iter() {
if seen.insert(state.name.clone())
&& self.branch_names.get(&state.name) != Some(id)
{
self.branch_names.insert(&state.name, &id);
}
}
let stale: Vec<_> = self
.branch_names
.keys()
.filter(|name| !seen.contains(name))
.collect();
for name in stale {
self.branch_names.remove(&name);
}
}
/// 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();
self.tree.register_deferred_reclaims();
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(());
}
// A legacy crash can persist a commit/branch row while losing the
// allocator increment that precedes it. Reusing an existing id would
// overwrite reachable history (and can create a self-parented
// commit), so advance both allocators past every stored id before
// any new id can be issued. Mirrors the NodeId repair in
// `PersistentBTree::deserialize`.
self.repair_id_allocators();
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(())
}
/// Advances the commit/branch id allocators past every stored row.
///
/// Writable restores only: read-only callers return before this point.
/// The allocator rows are advanced before the repair can return, so a
/// subsequent `commit`/`create_branch` cannot reuse an existing id.
fn repair_id_allocators(&mut self) {
let max_commit = self
.commits
.iter()
.map(|(id, _)| id)
.max()
.unwrap_or(NO_COMMIT.0);
let max_branch = self
.branches
.iter()
.map(|(id, _)| id)
.max()
.unwrap_or(NO_COMMIT.0);
let current_commit = self.next_commit.get_value();
let current_branch = self.next_branch.get_value();
let next_commit = current_commit.max(max_commit.saturating_add(1));
let next_branch = current_branch.max(max_branch.saturating_add(1));
let commit_changed = next_commit != current_commit;
let branch_changed = next_branch != current_branch;
if commit_changed {
*self.next_commit.get_mut() = next_commit;
}
if branch_changed {
*self.next_branch.get_mut() = next_branch;
}
if commit_changed || branch_changed {
self.fence(|m| {
m.next_commit.sync_wal();
m.next_branch.sync_wal();
});
}
}
/// 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.0)
.ok_or(VsdbError::CommitNotFound { commit_id: id.0 })?;
if c.id != id {
return Err(VsdbError::Corrupt {
detail: format!("commit row {} carries id {}", id.0, c.id.0),
});
}
for &parent in &c.parents {
if parent == NO_COMMIT {
return Err(VsdbError::Corrupt {
detail: format!(
"commit {} lists the sentinel as a parent",
id.0
),
});
}
if parent.0 >= id.0 {
return Err(VsdbError::Corrupt {
detail: format!(
"commit {} lists non-older parent {}",
id.0, parent.0
),
});
}
*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.fence(|m| {
m.tree.nodes.sync_wal();
m.commits.sync_wal();
m.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.0)
.expect("validated commit disappeared");
if c.ref_count != correct {
c.ref_count = correct;
self.commits.insert(&id.0, &c);
}
}
let all_ids: Vec<CommitId> =
self.commits.iter().map(|(id, _)| CommitId(id)).collect();
for id in all_ids {
if !ref_counts.contains_key(&id) {
self.commits.remove(&id.0);
}
}
// A durable false marker must never outlive unsynced count repairs.
self.fence(|m| m.commits.sync_wal());
self.end_ref_update();
Ok(())
}
}