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
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
//! WAL segment export and replay (CDC / PITR / incremental backup) over the
//! log-first segmented WAL (decision B: the log stores expanded op-sets).
use super::{wal, Ordering, StorageError, Store};
use crate::backend::StorageBackend;
use crate::log_store;
use std::sync::Arc;
impl Store {
/// Open this store's segmented WAL and replay any tail beyond the shard
/// materializations into the backends. Returns the log and the highest
/// recovered commit index. The record header carries the routing shard and
/// LSN, so replay is direct (no expander needed) and idempotent across a
/// crash: a record at or below a shard's durable `LAST_APPLIED` is skipped.
pub(crate) fn open_log_and_replay(
&self,
path: &std::path::Path,
backends: &[Option<Arc<dyn StorageBackend>>],
materialization: &[u64],
) -> Result<(log_store::LogStore, u64), StorageError> {
// The WAL lives beside the redb file(s) as `<file>.wal/` so each store
// has its own log (mirroring the `<file>.layout.json` convention).
let wal_dir = {
let mut os = path.as_os_str().to_os_string();
os.push(".wal");
std::path::PathBuf::from(os)
};
let mut log =
log_store::LogStore::open(&wal_dir, 0, self.resource_config.log_segment_bytes)
.map_err(|error| {
StorageError::Durability(format!("open WAL at {}: {error}", wal_dir.display()))
})?;
// A fresh WAL (no existing segments) beside a materialization that
// already has a checkpoint — e.g. a store opened from a restored
// base backup with no accompanying WAL directory — must continue
// the commit sequence from that checkpoint, not restart at LSN 1
// and collide with LSNs the materialization already has. A no-op
// when the WAL has real history: see `seed_next_lsn_if_empty`.
let highest_checkpoint = materialization.iter().copied().max().unwrap_or(0);
log.seed_next_lsn_if_empty(highest_checkpoint.saturating_add(1));
let wal_durable = log.durable_lsn();
if wal_durable == 0 {
return Ok((log, 0));
}
let from = materialization
.iter()
.copied()
.min()
.unwrap_or(0)
.saturating_add(1);
let records = log
.read_from(from)
.map_err(|error| StorageError::Durability(format!("read WAL for recovery: {error}")))?;
let mut expected = from;
// A chained commit (large-transaction support) spans several
// consecutive physical records sharing one LSN; `ChainAssembler`
// reconstructs the whole commit from them so it is applied exactly
// once, atomically, never per-fragment (gap/skip detection below
// operates on the reassembled commit's LSN, not each fragment's).
let mut assembler = wal::ChainAssembler::default();
let deriver = self.deriver();
for (framed_lsn, raw) in records {
let (frame_header, _) = wal::decode_wal_record(&raw).map_err(|error| {
StorageError::Durability(format!("decode WAL record {framed_lsn}: {error}"))
})?;
if frame_header.lsn != framed_lsn {
return Err(StorageError::Durability(format!(
"WAL recovery LSN mismatch: frame {framed_lsn}, header {}",
frame_header.lsn
)));
}
let (header, body) = match assembler.feed(&raw, 0).map_err(|error| {
StorageError::Durability(format!(
"reassemble chained WAL record {framed_lsn}: {error}"
))
})? {
wal::ChainFeed::Pending => continue,
wal::ChainFeed::Complete(header, body) => (header, body),
};
let (_, ops, _command) = wal::decode_logical_body(header, &body).map_err(|error| {
StorageError::Durability(format!("decode WAL record {framed_lsn}: {error}"))
})?;
if header.lsn < expected {
continue; // already materialized on this shard
}
if header.lsn > expected {
return Err(StorageError::Durability(format!(
"WAL recovery gap: expected LSN {expected}, found {}",
header.lsn
)));
}
expected = header.lsn.saturating_add(1);
let shard = usize::try_from(header.shard_id).map_err(|_| {
StorageError::Durability(format!(
"WAL shard {} does not fit usize",
header.shard_id
))
})?;
let already = materialization.get(shard).copied().unwrap_or(0);
if header.lsn <= already {
continue;
}
if let Some(Some(backend)) = backends.get(shard) {
if deriver.is_none() {
self.underived_replay.store(true, Ordering::SeqCst);
}
Self::apply_derived_replay(deriver.as_ref(), backend.as_ref(), &ops, header.lsn)?;
}
}
// `LogStore::open` (above) already truncated away any trailing
// incomplete chain from the segment files themselves, so every
// record `read_from` can return must belong to a complete chain.
// This is a correctness invariant, not a recoverable condition: if
// it ever fires, recovery's own atomicity guarantee has a bug.
debug_assert!(
!assembler.is_open(),
"WAL replay ended with an open chain; LogStore::open should have truncated it"
);
Ok((log, wal_durable))
}
/// Export the committed WAL as a checksummed segment stream (manifest +
/// framed op-set records), bounded to the inclusive LSN range
/// `[from_lsn, to_lsn]`. This is the CDC / PITR / incremental-backup cursor
///. Records come from the segmented WAL, which is the durable truth.
///
/// # Errors
/// Returns [`StorageError::Durability`] if the WAL cannot be read or the
/// segment cannot be written.
///
/// Lock poisoning fences the store until reopen.
pub fn export_wal_segments<W: std::io::Write>(
&self,
writer: &mut W,
from_lsn: u64,
to_lsn: u64,
) -> Result<u64, StorageError> {
self.check_health()?;
let _guard = self.health.guard(self.publish_lock.lock())?;
let mut records: Vec<(u64, Vec<u8>)> = {
let guard = self.health.guard(self.log.lock())?;
match guard.as_ref() {
Some(log) => log.read_from(from_lsn).map_err(|error| {
StorageError::Durability(format!("read WAL for export: {error}"))
})?,
None => Vec::new(),
}
};
records.retain(|(lsn, _)| *lsn <= to_lsn);
// Fail closed: the segment must cover the requested inclusive range
// `[from_lsn, to_lsn]` exactly, with no gaps. A missing record means
// retention pruned it or the range is beyond the committed log;
// certifying the export as complete would let a restore silently skip
// commits. `from_lsn > to_lsn` is an empty range and needs no coverage.
//
// A chained commit (large-transaction support) contributes
// several consecutive physical records sharing one LSN, so coverage
// and contiguity are checked over the *distinct* logical LSNs, not
// the raw physical record count — `records` is already in physical
// (and therefore chain-grouped) order, so consecutive duplicates are
// exactly one chain's fragments.
if from_lsn <= to_lsn {
let mut distinct_lsns: Vec<u64> = Vec::new();
for (lsn, _) in &records {
if distinct_lsns.last() != Some(lsn) {
distinct_lsns.push(*lsn);
}
}
let expected = to_lsn - from_lsn + 1;
let covered = distinct_lsns.len() as u64 == expected
&& distinct_lsns.first() == Some(&from_lsn)
&& distinct_lsns.last() == Some(&to_lsn);
let contiguous = distinct_lsns.windows(2).all(|pair| pair[1] == pair[0] + 1);
if !covered || !contiguous {
return Err(StorageError::Durability(format!(
"WAL export range [{from_lsn}, {to_lsn}] is not fully covered \
({} distinct commits, {} physical records); refusing to certify \
an incomplete segment",
distinct_lsns.len(),
records.len()
)));
}
}
let refs: Vec<(u64, &[u8])> = records
.iter()
.map(|(lsn, payload)| (*lsn, payload.as_slice()))
.collect();
wal::write_wal_segment(writer, 0, &refs)
.map_err(|e| StorageError::Durability(e.to_string()))?;
Ok(records.len() as u64)
}
/// Replay a WAL segment stream produced by [`Store::export_wal_segments`]
/// onto this store (V2 PITR). The store must already hold the state at the
/// segment's base commit (a physical checkpoint, or an equivalent store);
/// each decoded op-set is applied directly in LSN order. Returns the final
/// commit index.
///
/// # Errors
/// Returns [`StorageError`] if the segment is truncated/corrupt or an
/// op-set cannot be applied.
pub fn replay_wal_segments<R: std::io::Read>(
&self,
reader: &mut R,
) -> Result<u64, StorageError> {
self.check_health()?;
self.replay_wal_segments_to(reader, None, None)
}
/// [`Store::replay_wal_segments`] bounded to `(from_lsn, to_lsn]`: records
/// at or below `from_lsn` are already in the base and are skipped, and
/// records past `to_lsn` stop replay. Reads consecutive segments from
/// `reader` until clean EOF.
///
/// # Errors
/// Returns [`StorageError`] if a segment is truncated/corrupt or an op-set
/// cannot be applied.
pub fn replay_wal_segments_to<R: std::io::Read>(
&self,
reader: &mut R,
from_lsn: Option<u64>,
to_lsn: Option<u64>,
) -> Result<u64, StorageError> {
self.check_health()?;
let mut last = self.committed_version();
// Reject an LSN gap — leading or interior. A missing segment must not be
// silently compacted by renumbering, which would produce a database that
// never existed. The base LSN is the caller's `from_lsn`, or the store's
// current commit index when unbounded; the first applied record must be
// its immediate successor, so a segment that starts above the base (a
// leading gap) fails closed instead of silently dropping commits.
let base_lsn = from_lsn.unwrap_or_else(|| self.committed_version());
let mut expected = base_lsn.saturating_add(1);
let mut applied_lsn = base_lsn;
// A chained commit (large-transaction support) contributes
// several consecutive records sharing one LSN; reassemble before
// applying, exactly as `open_log_and_replay` does, so a chain is
// never applied per-fragment.
let mut assembler = wal::ChainAssembler::default();
'segments: while let Some((_manifest, records)) =
wal::read_wal_segment(reader).map_err(|e| StorageError::Durability(e.to_string()))?
{
for record in records {
let (frame_header, _) = wal::decode_wal_record(&record)
.map_err(|e| StorageError::Durability(e.to_string()))?;
if from_lsn.is_some_and(|bound| frame_header.lsn <= bound) {
continue; // already present in the base
}
if to_lsn.is_some_and(|bound| frame_header.lsn > bound) {
break 'segments;
}
let (header, body) = match assembler
.feed(&record, 0)
.map_err(StorageError::Durability)?
{
wal::ChainFeed::Pending => continue,
wal::ChainFeed::Complete(header, body) => (header, body),
};
let (_, ops, _command) =
wal::decode_logical_body(header, &body).map_err(StorageError::Durability)?;
if header.lsn != expected {
return Err(StorageError::Durability(format!(
"WAL replay is missing LSN {expected} before {}",
header.lsn
)));
}
expected = header.lsn.saturating_add(1);
applied_lsn = header.lsn;
let shard = usize::try_from(header.shard_id).map_err(|_| {
StorageError::Durability(format!(
"WAL shard {} does not fit usize",
header.shard_id
))
})?;
if let Some(backend) = self.backend(shard) {
Self::apply_derived_replay(
self.deriver().as_ref(),
backend.as_ref(),
&ops,
header.lsn,
)?;
}
last = header.lsn;
}
}
// A bounded replay that stops before `to_lsn` is incomplete; certifying
// it would let a restore silently skip the tail of the range.
if let Some(bound) = to_lsn {
if applied_lsn < bound {
return Err(StorageError::Durability(format!(
"WAL replay stopped at LSN {applied_lsn} before reaching to_lsn {bound}"
)));
}
}
// Advance the store watermark to the replayed tail so a restored store
// serves the recovered commits.
if applied_lsn > self.committed_version() {
self.next_commit_index
.store(applied_lsn.saturating_add(1), Ordering::SeqCst);
self.published.store(applied_lsn, Ordering::SeqCst);
}
Ok(last)
}
/// Apply one replayed WAL record (base ops only) to a backend together
/// with the ops the installed deriver derives against the backend's
/// current state, exactly as live publication does.
fn apply_derived_replay(
deriver: Option<&Arc<dyn crate::Deriver>>,
backend: &dyn StorageBackend,
ops: &[crate::Op],
lsn: u64,
) -> Result<(), StorageError> {
let all = match deriver {
Some(_) => {
let before = crate::derive::BackendRead(backend.read_view()?);
Self::derive_ops(deriver, lsn, ops, &before)?
}
None => None,
};
backend
.apply(all.as_deref().unwrap_or(ops), lsn)
.map_err(StorageError::Durability)
}
}