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
//! WAL entry domain type — separates parsing from application.
//!
//! Extracted from `log_payload.rs` to reduce NLOC.
//!
//! ## WAL versioning (WP-2I)
//!
//! The WAL format is implicitly versioned through its marker bytes. Legacy
//! entries use `LEGACY_STORE_MARKER` / `LEGACY_DELETE_MARKER` (no CRC),
//! while v2 entries use `CRC_STORE_MARKER` / `CRC_DELETE_MARKER` (with
//! CRC32 integrity checking). Future format changes should introduce new
//! marker values, preserving backward-compatible reading of older entries.
//! No separate schema-version header is needed because the per-entry
//! marker already encodes the wire format.
//!
//! ## Torn-tail policy
//!
//! A crash mid-append leaves a partially-written final record. Because the WAL
//! is append-only and replayed sequentially, a truncation can only ever be the
//! last record. [`WalEntry::read`] / [`WalEntry::apply`] therefore signal such a
//! tail by returning `Ok(None)` instead of an error, so the caller stops replay
//! cleanly and keeps every prior entry rather than failing the collection open.
//! This mirrors the vector WAL's `#898` policy (`storage::mmap::wal_replay`).
use super::log_payload_io::{
compute_delete_crc, compute_store_crc, CRC_DELETE_MARKER, CRC_STORE_MARKER,
LEGACY_DELETE_MARKER, LEGACY_STORE_MARKER,
};
use rustc_hash::FxHashMap;
use std::fs::File;
use std::io::{self, BufReader, Read, Seek, SeekFrom};
/// A parsed WAL entry with its file position context.
pub(super) struct WalEntry {
op: WalOp,
/// File position after the marker + ID header.
pos_after_header: u64,
/// Whether this entry uses CRC32 integrity checking.
has_crc: bool,
}
/// The two WAL operations: store (upsert) or delete.
enum WalOp {
Store { id: u64 },
Delete { id: u64 },
}
impl WalEntry {
/// Reads one WAL entry from the reader. Returns `None` on EOF.
pub(super) fn read(reader: &mut BufReader<File>, pos: u64) -> io::Result<Option<Self>> {
let mut marker = [0u8; 1];
if reader.read_exact(&mut marker).is_err() {
return Ok(None);
}
let mut id_bytes = [0u8; 8];
if reader.read_exact(&mut id_bytes).is_err() {
// Torn tail: crashed after the marker but before the full id.
return Ok(None);
}
let id = u64::from_le_bytes(id_bytes);
let pos_after_header = pos + 1 + 8;
let (op, has_crc) = match marker[0] {
LEGACY_STORE_MARKER => (WalOp::Store { id }, false),
LEGACY_DELETE_MARKER => (WalOp::Delete { id }, false),
CRC_STORE_MARKER => (WalOp::Store { id }, true),
CRC_DELETE_MARKER => (WalOp::Delete { id }, true),
_ => {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"Unknown WAL marker",
))
}
};
Ok(Some(Self {
op,
pos_after_header,
has_crc,
}))
}
/// Applies this entry to the index, returning the new file position.
///
/// Returns `Ok(None)` for a torn tail (a record truncated by a crash
/// mid-append) so the caller stops replay cleanly; see the module-level
/// torn-tail policy.
///
/// `wal_end` is the logical end of the WAL being replayed; it bounds the
/// declared payload length so a corrupt length field cannot drive an
/// unbounded allocation (#897/#898).
pub(super) fn apply(
self,
index: &mut FxHashMap<u64, u64>,
reader: &mut BufReader<File>,
wal_end: u64,
) -> io::Result<Option<u64>> {
match self.op {
WalOp::Store { id } => self.apply_store(id, index, reader, wal_end),
WalOp::Delete { id } => Ok(self.apply_delete(id, index, reader)),
}
}
fn apply_store(
&self,
id: u64,
index: &mut FxHashMap<u64, u64>,
reader: &mut BufReader<File>,
wal_end: u64,
) -> io::Result<Option<u64>> {
let len_offset = self.pos_after_header;
let mut len_bytes = [0u8; 4];
if reader.read_exact(&mut len_bytes).is_err() {
// Torn tail: crashed after the id but before the full length field.
return Ok(None);
}
let payload_len = u64::from(u32::from_le_bytes(len_bytes));
// OOM guard (#897/#898): the payload (plus the 4-byte CRC for v2
// records) cannot extend past the WAL end. A length running past EOF is
// a torn tail (crashed before the payload landed) — stop cleanly rather
// than failing the open, and never allocate the oversized buffer.
let payload_start = self.pos_after_header + 4;
let crc_bytes = u64::from(self.has_crc) * 4;
let max_payload = wal_end.saturating_sub(payload_start);
if payload_len.saturating_add(crc_bytes) > max_payload {
return Ok(None);
}
let end_pos = if self.has_crc {
self.apply_store_with_crc(id, payload_len, index, reader, len_offset)?
} else {
let skip = i64::try_from(payload_len)
.map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "Payload too large"))?;
reader.seek(SeekFrom::Current(skip))?;
index.insert(id, len_offset);
self.pos_after_header + 4 + payload_len
};
Ok(Some(end_pos))
}
fn apply_store_with_crc(
&self,
id: u64,
payload_len: u64,
index: &mut FxHashMap<u64, u64>,
reader: &mut BufReader<File>,
len_offset: u64,
) -> io::Result<u64> {
let payload_usize = usize::try_from(payload_len)
.map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "Payload too large"))?;
let mut payload_buf = vec![0u8; payload_usize];
reader.read_exact(&mut payload_buf)?;
let mut crc_bytes = [0u8; 4];
reader.read_exact(&mut crc_bytes)?;
let stored_crc = u32::from_le_bytes(crc_bytes);
let computed_crc = compute_store_crc(id, &payload_buf);
if stored_crc == computed_crc {
index.insert(id, len_offset);
} else {
tracing::warn!(
id,
"WAL CRC mismatch on store entry — skipping corrupted entry"
);
}
Ok(self.pos_after_header + 4 + payload_len + 4)
}
fn apply_delete(
&self,
id: u64,
index: &mut FxHashMap<u64, u64>,
reader: &mut BufReader<File>,
) -> Option<u64> {
if self.has_crc {
let mut crc_bytes = [0u8; 4];
if reader.read_exact(&mut crc_bytes).is_err() {
// Torn tail: crashed after the id but before the CRC.
return None;
}
let stored_crc = u32::from_le_bytes(crc_bytes);
let computed_crc = compute_delete_crc(id);
if stored_crc == computed_crc {
index.remove(&id);
} else {
tracing::warn!(
id,
"WAL CRC mismatch on delete entry — skipping corrupted entry"
);
}
Some(self.pos_after_header + 4)
} else {
index.remove(&id);
Some(self.pos_after_header)
}
}
}