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
//! Durable-log export: crash-log dumps and the streaming-replication
//! follower read path. Both derive the committed
//! command-batch sequence from the WAL's embedded-command envelope
//! rather than a separate journal.
use super::{prefix_range, wal, ReadLimits, ScanProjection, StorageError, Store, LOG_PREFIX};
impl Store {
/// Durable-log export for tests/tools/backups.
pub fn log_bytes(&self) -> Vec<Vec<u8>> {
self.take_log_for_crash()
}
/// Export the decoded commit-batch payloads of the committed command log,
/// ordered by LSN.
///
/// Lock poisoning fences the store until reopen.
pub fn take_log_for_crash(&self) -> Vec<Vec<u8>> {
self.try_take_log_for_crash().unwrap_or_default()
}
/// Fallible crash-log export for recovery and archive callers. Unlike
/// [`Store::take_log_for_crash`], this path never converts a scan,
/// framing, or WAL decode failure into an empty/partial log.
///
/// A durable store derives this from the WAL's embedded-command envelope
///; an in-memory store has no WAL file and keeps the row-based
/// journal ([`log_key`], test mode only).
///
/// # Errors
/// Returns [`StorageError::CorruptLog`] for malformed WAL rows and the
/// underlying storage error for failed backend scans.
///
/// Lock poisoning fences the store until reopen.
pub fn try_take_log_for_crash(&self) -> Result<Vec<Vec<u8>>, StorageError> {
self.check_health()?;
let _guard = self.health.guard(self.publish_lock.lock())?;
if self.is_durable() {
let records = {
let guard = self.health.guard(self.log.lock())?;
match guard.as_ref() {
Some(log) => log.read_from(1).map_err(|error| {
StorageError::Durability(format!("read WAL for crash log: {error}"))
})?,
None => Vec::new(),
}
};
let mut out = Vec::with_capacity(records.len());
// A chained commit (large-transaction support) spans
// several physical records sharing one LSN; reassemble before
// extracting its command, the same way WAL replay does.
let mut assembler = wal::ChainAssembler::default();
for (lsn, raw) in records {
let (header, body) = match assembler
.feed(&raw, 0)
.map_err(|_| StorageError::CorruptLog)?
{
wal::ChainFeed::Pending => continue,
wal::ChainFeed::Complete(header, body) => (header, body),
};
if header.lsn != lsn {
return Err(StorageError::CorruptLog);
}
let (_, _ops, command) = wal::decode_logical_body(header, &body)
.map_err(|_| StorageError::CorruptLog)?;
if let Some(batch) = command {
out.push(rushdb_types::command_codec::encode(&batch));
}
}
return Ok(out);
}
// Every store has at least one backend, so read the log rows
// natively from each shard. Multi-shard: shards are scanned in shard
// order, so the union must be re-sorted by the LSN carried in the WAL
// frame's header.
let mut out: Vec<(u64, Vec<u8>)> = Vec::new();
for backend in self.all_backends() {
let view = backend.read_view()?;
let (ls, le) = prefix_range(LOG_PREFIX);
let rows =
view.current_range(&ls, &le, ReadLimits::default(), ScanProjection::KeyValue)?;
for row in rows {
let (_, value) = row?;
let (header, payload) =
wal::decode_wal_record(&value).map_err(|_| StorageError::CorruptLog)?;
out.push((header.lsn, payload.to_vec()));
}
}
out.sort_by_key(|(lsn, _)| *lsn);
Ok(out.into_iter().map(|(_, payload)| payload).collect())
}
/// Durable-log export of commit-batch payloads from `from_lsn` onward
/// (each paired with its LSN), for a streaming-replication follower
/// to replay via [`Store::apply_replicated_batch_with_payload`].
/// Empty for a non-durable store. Takes the same `publish_lock` as
/// [`Store::try_take_log_for_crash`] (briefly contends with commits —
/// poll at a bounded interval, not a tight loop).
///
/// # Errors
/// Returns [`StorageError::CorruptLog`] for a malformed WAL record and
/// the underlying storage error for a failed WAL read.
pub fn try_log_commands_from(
&self,
from_lsn: u64,
) -> Result<Vec<(u64, Vec<u8>)>, StorageError> {
self.check_health()?;
if !self.is_durable() {
return Ok(Vec::new());
}
let _guard = self.health.guard(self.publish_lock.lock())?;
let records = {
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 replication: {error}"))
})?,
None => Vec::new(),
}
};
let mut out = Vec::with_capacity(records.len());
// Same chain reassembly as `try_take_log_for_crash`: a replicated
// commit too big for one WAL record arrives here as several
// physical records sharing one LSN, and must be reconstructed
// before its command is extracted and shipped to the follower —
// which otherwise receives nothing for that LSN at all, or (worse)
// a `CHAIN_CONTINUE` fragment's partial bytes misread as a complete
// command.
let mut assembler = wal::ChainAssembler::default();
for (lsn, raw) in records {
let (header, body) = match assembler
.feed(&raw, 0)
.map_err(|_| StorageError::CorruptLog)?
{
wal::ChainFeed::Pending => continue,
wal::ChainFeed::Complete(header, body) => (header, body),
};
if header.lsn != lsn {
return Err(StorageError::CorruptLog);
}
let (_, _ops, command) =
wal::decode_logical_body(header, &body).map_err(|_| StorageError::CorruptLog)?;
if let Some(batch) = command {
out.push((lsn, rushdb_types::command_codec::encode(&batch)));
}
}
Ok(out)
}
}