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
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
use super::*;
impl WALManager {
/// Extract the committed catalog substream from this WAL authority. Every
/// embedded catalog transaction is checked against the shared outer commit
/// marker before its bytes are admitted to the result.
pub(crate) fn read_catalog_transactions(
&self,
from_lsn: u64,
byte_budget: u64,
) -> Result<Vec<u8>> {
let mut pending = rustc_hash::FxHashMap::<i64, Vec<u8>>::default();
let mut output = Vec::new();
self.replay_two_phase(from_lsn, |entry| {
if entry.operation == WALOperationType::CatalogMutation {
if pending.insert(entry.txn_id, entry.data).is_some() {
return Err(Error::internal(format!(
"transaction {} contains more than one catalog mutation record",
entry.txn_id
)));
}
return Ok(());
}
if !entry.is_commit_marker() {
return Ok(());
}
let Some(encoded) = pending.remove(&entry.txn_id) else {
return Ok(());
};
let decoded =
crate::v6::decode_catalog_wal(&encoded, crate::v6::CatalogWalReplayLimits::hard())
.map_err(|error| {
Error::internal(format!(
"catalog WAL record for transaction {} is invalid: {error}",
entry.txn_id
))
})?;
if decoded.incomplete_tail_bytes() != 0 || decoded.transactions().len() != 1 {
return Err(Error::internal(format!(
"catalog WAL record for transaction {} is not one complete transaction",
entry.txn_id
)));
}
let catalog_transaction = &decoded.transactions()[0];
if catalog_transaction.commit_lsn() != entry.lsn {
return Err(Error::internal(format!(
"catalog WAL transaction {} binds commit LSN {}, shared marker is {}",
entry.txn_id,
catalog_transaction.commit_lsn(),
entry.lsn
)));
}
let mut expected_identity = [0_u8; 16];
expected_identity[..8].copy_from_slice(&entry.txn_id.to_le_bytes());
expected_identity[8..].copy_from_slice(&entry.lsn.to_le_bytes());
if catalog_transaction.transaction_id().as_bytes() != expected_identity {
return Err(Error::internal(format!(
"catalog WAL transaction {} has a foreign embedded identity",
entry.txn_id
)));
}
let next_len = output
.len()
.checked_add(encoded.len())
.ok_or_else(|| Error::internal("catalog WAL replay byte count overflow"))?;
if next_len as u64 > byte_budget {
return Err(Error::internal(format!(
"catalog WAL replay exceeds byte budget: {} bytes (maximum {})",
next_len, byte_budget
)));
}
output.extend_from_slice(&encoded);
Ok(())
})?;
if !pending.is_empty() {
return Err(Error::internal(
"committed catalog WAL mutation is missing its shared commit callback",
));
}
Ok(output)
}
/// Two-phase WAL replay for crash recovery
///
/// Phase 1 (Analysis): Scan all entries to identify committed/aborted transactions.
/// Outcomes use a fixed memory budget and spill to a
/// temporary disk hash index for long retained history.
/// Phase 2 (REDO): Re-read WAL and apply only entries from committed transactions
///
/// This ensures that after a crash, only committed transactions are visible.
/// Uncommitted transactions (those without a COMMIT_MARKER) are discarded.
///
/// The common path uses two streaming passes. If the declared outcome
/// memory budget is exceeded, analysis performs one additional validated
/// pass to build the disk index; process RSS remains independent of the
/// number of retained transactions.
pub fn replay_two_phase<F>(&self, from_lsn: u64, callback: F) -> Result<TwoPhaseRecoveryInfo>
where
F: FnMut(WALEntry) -> Result<()>,
{
self.replay_two_phase_with_outcome_observer(from_lsn, callback, |_| Ok(()))
}
pub(crate) fn replay_two_phase_with_outcome_observer<F, O>(
&self,
from_lsn: u64,
mut callback: F,
mut observe_uncommitted: O,
) -> Result<TwoPhaseRecoveryInfo>
where
F: FnMut(WALEntry) -> Result<()>,
O: FnMut(i64) -> Result<()>,
{
// Flush buffer first
self.flush()?;
// The caller owns the replay floor after validating the corresponding
// manifest/snapshot generation. In particular, an explicit zero must
// remain zero when any table manifest has no committed checkpoint;
// A CONTROL slot alone cannot prove table-artifact completeness.
let replay_floor = *self
.replay_floor
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
// A caller may request a historical boundary, but CONTROL has already
// made every earlier record redundant and eligible for retirement.
// Clamp instead of requiring a file which the selected generation no
// longer owns.
let from_lsn = from_lsn.max(replay_floor.lsn());
let generations = Self::collect_validated_generations(
&self.path,
replay_floor.generation().get(),
replay_floor.lsn(),
)?;
Self::validate_required_generation_suffix(&generations, from_lsn)?;
// =====================================================
// Phase 1: Analysis - Identify transaction outcomes.
// Keep a bounded common-case map. Once it crosses the declared
// budget, discard it and build an exact disk-backed index from a
// second validated scan.
// =====================================================
let mut memory_outcomes = rustc_hash::FxHashMap::default();
let mut outcome_markers = 0usize;
let mut needs_spill = false;
let mut last_lsn = from_lsn;
let mut max_transaction_id = self.transaction_high_water.load(Ordering::Acquire);
for generation in &generations {
let mut reader = ValidatedWalReader::open(&generation.path)?;
while let Some(entry) = reader.next_entry()? {
if entry.txn_id == i64::MAX {
return Err(Error::internal(format!(
"WAL transaction ID domain exhausted at LSN {}",
entry.lsn
)));
}
if entry.txn_id > max_transaction_id {
max_transaction_id = entry.txn_id;
}
if !crate::timestamp::observe_persisted_timestamp(entry.timestamp) {
return Err(Error::internal(format!(
"WAL timestamp domain exhausted at LSN {}",
entry.lsn
)));
}
if entry.lsn <= from_lsn {
continue;
}
last_lsn = last_lsn.max(entry.lsn);
let Some(outcome) = Self::marker_outcome(&entry) else {
continue;
};
outcome_markers = outcome_markers
.checked_add(1)
.ok_or_else(|| Error::internal("WAL recovery marker count overflow"))?;
if needs_spill {
continue;
}
match memory_outcomes.get(&entry.txn_id) {
Some(stored) if *stored != outcome => {
return Err(Error::internal(format!(
"WAL transaction {} has both commit and abort outcomes",
entry.txn_id
)))
}
Some(_) => {}
None => {
memory_outcomes.insert(entry.txn_id, outcome);
if memory_outcomes.len() > RECOVERY_OUTCOME_MEMORY_LIMIT {
needs_spill = true;
memory_outcomes.clear();
memory_outcomes.shrink_to_fit();
}
}
}
}
}
self.transaction_high_water
.fetch_max(max_transaction_id, Ordering::AcqRel);
let mut outcomes = if needs_spill {
let mut disk = DiskRecoveryOutcomes::create(&self.path, outcome_markers)?;
for generation in &generations {
Self::scan_wal_outcomes(&generation.path, from_lsn, |txn_id, outcome| {
disk.insert(txn_id, outcome)
})?;
}
RecoveryOutcomes::Disk(disk)
} else {
RecoveryOutcomes::Memory(memory_outcomes)
};
let (committed_transactions, aborted_transactions) = outcomes.counts();
// =====================================================
// Phase 2: REDO - Re-read WAL and apply committed entries
// Streaming approach: read and apply one entry at a time
// =====================================================
let mut applied_count = 0u64;
let mut skipped_count = 0u64;
for generation in &generations {
let mut reader = ValidatedWalReader::open(&generation.path)?;
while let Some(entry) = reader.next_entry()? {
if entry.lsn <= from_lsn {
continue;
}
// Abort markers do not apply data, but the registry must
// retain their non-committed identity and high-water.
if entry.is_abort_marker() {
observe_uncommitted(entry.txn_id)?;
continue;
}
// For commit markers: pass to callback so registry can be updated.
if entry.is_commit_marker() {
if outcomes.get(entry.txn_id)? == Some(RECOVERY_OUTCOME_COMMITTED) {
callback(entry)?;
}
continue;
}
// Apply only committed transactions' data entries.
if outcomes.get(entry.txn_id)? == Some(RECOVERY_OUTCOME_COMMITTED) {
callback(entry)?;
applied_count += 1;
} else {
// Transaction is aborted or in-doubt (no commit marker).
observe_uncommitted(entry.txn_id)?;
skipped_count += 1;
}
}
}
// Update current LSN if we replayed entries
if last_lsn > self.current_lsn.load(Ordering::Acquire) {
self.current_lsn.store(last_lsn, Ordering::Release);
}
Ok(TwoPhaseRecoveryInfo {
last_lsn,
committed_transactions,
aborted_transactions,
applied_entries: applied_count,
skipped_entries: skipped_count,
max_transaction_id,
})
}
#[inline]
pub(super) fn marker_outcome(entry: &WALEntry) -> Option<u8> {
if entry.is_commit_marker() {
Some(RECOVERY_OUTCOME_COMMITTED)
} else if entry.is_abort_marker() {
Some(RECOVERY_OUTCOME_ABORTED)
} else {
None
}
}
pub(super) fn scan_wal_outcomes<F>(
wal_path: &Path,
from_lsn: u64,
mut callback: F,
) -> Result<()>
where
F: FnMut(i64, u8) -> Result<()>,
{
let mut reader = ValidatedWalReader::open(wal_path)?;
while let Some(entry) = reader.next_entry()? {
if entry.lsn <= from_lsn {
continue;
}
if let Some(outcome) = Self::marker_outcome(&entry) {
callback(entry.txn_id, outcome)?;
}
}
Ok(())
}
}