zeppelin_embed/wal/
record.rs

1//! Byte-exact WAL record framing.
2
3use xxhash_rust::xxh3::{Xxh3, xxh3_64};
4
5use super::LogSeq;
6
7/// Bytes before a WAL record payload: length, sequence, and operation.
8pub const RECORD_HEADER_LEN: usize = 14;
9/// Width of the trailing xxh3-64 checksum.
10pub const RECORD_CHECKSUM_LEN: usize = 8;
11/// Smallest encoded WAL record.
12pub const MIN_RECORD_LEN: usize = RECORD_HEADER_LEN + RECORD_CHECKSUM_LEN;
13
14/// One decoded WAL record borrowing its checksummed payload.
15#[derive(Clone, Copy, Debug, Eq, PartialEq)]
16pub struct WalRecord<'a> {
17    /// Monotonic log sequence.
18    pub seq: LogSeq,
19    /// Caller-defined operation identifier.
20    pub op: u16,
21    /// Operation payload bytes.
22    pub payload: &'a [u8],
23}
24
25/// A complete checked record plus its encoded width.
26#[derive(Clone, Copy, Debug, Eq, PartialEq)]
27pub struct DecodedRecord<'a> {
28    /// Checked logical record.
29    pub record: WalRecord<'a>,
30    /// Bytes consumed by this record, including its checksum.
31    pub encoded_len: usize,
32}
33
34/// Record encoding failure.
35#[derive(Clone, Copy, Debug, Eq, PartialEq)]
36pub enum RecordEncodeError {
37    /// The payload cannot fit the permanent u32 length field.
38    PayloadTooLarge(usize),
39}
40
41impl std::fmt::Display for RecordEncodeError {
42    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
43        match self {
44            Self::PayloadTooLarge(length) => {
45                write!(formatter, "WAL payload length {length} exceeds u32")
46            }
47        }
48    }
49}
50
51impl std::error::Error for RecordEncodeError {}
52
53/// Specific validation failure for one WAL record.
54#[derive(Clone, Copy, Debug, Eq, PartialEq)]
55pub enum RecordError {
56    /// The fixed length/sequence/operation header is torn.
57    HeaderTruncated {
58        /// Required fixed header width.
59        needed: usize,
60        /// Bytes available at the record offset.
61        available: usize,
62    },
63    /// The declared payload plus framing cannot fit `usize`.
64    LengthOverflow {
65        /// Persisted u32 payload length.
66        payload_length: u32,
67    },
68    /// The record body or checksum is torn.
69    BodyTruncated {
70        /// Persisted u32 payload length.
71        payload_length: u32,
72        /// Complete framed width implied by the header.
73        needed: usize,
74        /// Bytes available at the record offset.
75        available: usize,
76    },
77    /// The trailing xxh3-64 does not cover the observed record bytes.
78    ChecksumMismatch {
79        /// Persisted checksum.
80        expected: u64,
81        /// Checksum computed from the header and payload.
82        actual: u64,
83        /// Complete framed width used to find a possible successor.
84        record_length: usize,
85    },
86}
87
88impl RecordError {
89    pub(crate) const fn framed_length(self) -> Option<usize> {
90        match self {
91            Self::ChecksumMismatch { record_length, .. } => Some(record_length),
92            Self::HeaderTruncated { .. }
93            | Self::LengthOverflow { .. }
94            | Self::BodyTruncated { .. } => None,
95        }
96    }
97}
98
99impl std::fmt::Display for RecordError {
100    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
101        match self {
102            Self::HeaderTruncated { needed, available } => {
103                write!(
104                    formatter,
105                    "WAL header needs {needed} bytes, got {available}"
106                )
107            }
108            Self::LengthOverflow { payload_length } => {
109                write!(
110                    formatter,
111                    "WAL payload length {payload_length} overflows framing"
112                )
113            }
114            Self::BodyTruncated {
115                payload_length,
116                needed,
117                available,
118            } => write!(
119                formatter,
120                "WAL payload length {payload_length} needs {needed} framed bytes, got {available}"
121            ),
122            Self::ChecksumMismatch {
123                expected, actual, ..
124            } => write!(
125                formatter,
126                "WAL checksum expected {expected:#018x}, computed {actual:#018x}"
127            ),
128        }
129    }
130}
131
132impl std::error::Error for RecordError {}
133
134/// Encodes `[payload_len:u32, seq:u64, op:u16, payload, checksum:u64]`.
135///
136/// The trailing xxh3-64 checksum covers exactly the little-endian payload
137/// length bytes `0..4`, sequence bytes `4..12`, operation bytes `12..14`, and
138/// every payload byte `14..14+payload_len`. It excludes only the checksum
139/// field itself.
140pub fn encode_record(record: WalRecord<'_>) -> Result<Vec<u8>, RecordEncodeError> {
141    let encoded_len = encoded_record_len(record.payload.len())?;
142    let mut encoded = Vec::with_capacity(encoded_len);
143    encode_record_into(record, &mut encoded)?;
144    Ok(encoded)
145}
146
147/// Returns the exact framed width needed by one payload.
148pub(crate) fn encoded_record_len(payload_len: usize) -> Result<usize, RecordEncodeError> {
149    u32::try_from(payload_len).map_err(|_| RecordEncodeError::PayloadTooLarge(payload_len))?;
150    Ok(MIN_RECORD_LEN.saturating_add(payload_len))
151}
152
153/// Encodes one record into caller-reserved storage.
154///
155/// Callers that reserve [`encoded_record_len`] bytes before taking a shared
156/// lock can perform the only payload copy without allocating while locked.
157pub(crate) fn encode_record_into(
158    record: WalRecord<'_>,
159    encoded: &mut Vec<u8>,
160) -> Result<(), RecordEncodeError> {
161    encoded.clear();
162    append_record_into(record, encoded)
163}
164
165/// Appends one encoded record to caller-reserved batch storage.
166pub(crate) fn append_record_into(
167    record: WalRecord<'_>,
168    encoded: &mut Vec<u8>,
169) -> Result<(), RecordEncodeError> {
170    let payload_length = u32::try_from(record.payload.len())
171        .map_err(|_| RecordEncodeError::PayloadTooLarge(record.payload.len()))?;
172    let payload_length_bytes = payload_length.to_le_bytes();
173    let sequence_bytes = record.seq.get().to_le_bytes();
174    let operation_bytes = record.op.to_le_bytes();
175    let mut checksum = Xxh3::new();
176    checksum.update(&payload_length_bytes);
177    checksum.update(&sequence_bytes);
178    checksum.update(&operation_bytes);
179    checksum.update(record.payload);
180    encoded.extend_from_slice(&payload_length_bytes);
181    encoded.extend_from_slice(&sequence_bytes);
182    encoded.extend_from_slice(&operation_bytes);
183    encoded.extend_from_slice(record.payload);
184    encoded.extend_from_slice(&checksum.digest().to_le_bytes());
185    Ok(())
186}
187
188/// Decodes one record and verifies xxh3-64 before returning any fields.
189pub fn decode_record(bytes: &[u8]) -> Result<DecodedRecord<'_>, RecordError> {
190    if bytes.len() < RECORD_HEADER_LEN {
191        return Err(RecordError::HeaderTruncated {
192            needed: RECORD_HEADER_LEN,
193            available: bytes.len(),
194        });
195    }
196    let payload_length = read_u32(bytes, 0)?;
197    let payload_length_usize = usize::try_from(payload_length)
198        .map_err(|_| RecordError::LengthOverflow { payload_length })?;
199    let encoded_len = MIN_RECORD_LEN
200        .checked_add(payload_length_usize)
201        .ok_or(RecordError::LengthOverflow { payload_length })?;
202    if bytes.len() < encoded_len {
203        return Err(RecordError::BodyTruncated {
204            payload_length,
205            needed: encoded_len,
206            available: bytes.len(),
207        });
208    }
209    let payload_end = RECORD_HEADER_LEN.saturating_add(payload_length_usize);
210    let payload = bytes
211        .get(RECORD_HEADER_LEN..payload_end)
212        .ok_or(RecordError::BodyTruncated {
213            payload_length,
214            needed: encoded_len,
215            available: bytes.len(),
216        })?;
217    let expected = read_u64(bytes, payload_end)?;
218    let checksummed = bytes.get(..payload_end).ok_or(RecordError::BodyTruncated {
219        payload_length,
220        needed: encoded_len,
221        available: bytes.len(),
222    })?;
223    let actual = xxh3_64(checksummed);
224    if actual != expected {
225        return Err(RecordError::ChecksumMismatch {
226            expected,
227            actual,
228            record_length: encoded_len,
229        });
230    }
231    Ok(DecodedRecord {
232        record: WalRecord {
233            seq: LogSeq::new(read_u64(bytes, 4)?),
234            op: read_u16(bytes, 12)?,
235            payload,
236        },
237        encoded_len,
238    })
239}
240
241fn read_u16(bytes: &[u8], offset: usize) -> Result<u16, RecordError> {
242    let raw = bytes
243        .get(offset..offset.saturating_add(2))
244        .and_then(|value| <[u8; 2]>::try_from(value).ok())
245        .ok_or(RecordError::HeaderTruncated {
246            needed: RECORD_HEADER_LEN,
247            available: bytes.len(),
248        })?;
249    Ok(u16::from_le_bytes(raw))
250}
251
252fn read_u32(bytes: &[u8], offset: usize) -> Result<u32, RecordError> {
253    let raw = bytes
254        .get(offset..offset.saturating_add(4))
255        .and_then(|value| <[u8; 4]>::try_from(value).ok())
256        .ok_or(RecordError::HeaderTruncated {
257            needed: RECORD_HEADER_LEN,
258            available: bytes.len(),
259        })?;
260    Ok(u32::from_le_bytes(raw))
261}
262
263fn read_u64(bytes: &[u8], offset: usize) -> Result<u64, RecordError> {
264    let raw = bytes
265        .get(offset..offset.saturating_add(8))
266        .and_then(|value| <[u8; 8]>::try_from(value).ok())
267        .ok_or(RecordError::BodyTruncated {
268            payload_length: 0,
269            needed: offset.saturating_add(8),
270            available: bytes.len(),
271        })?;
272    Ok(u64::from_le_bytes(raw))
273}
274
275#[cfg(test)]
276#[allow(clippy::expect_used)]
277mod tests {
278    use super::{WalRecord, encode_record_into, encoded_record_len};
279    use crate::wal::LogSeq;
280
281    #[test]
282    fn caller_reserved_encoding_keeps_its_allocation() {
283        let payload = [0x5a; 1_024];
284        let encoded_len = encoded_record_len(payload.len()).expect("valid payload length");
285        let mut encoded = Vec::with_capacity(encoded_len);
286        let allocation = encoded.as_ptr();
287        let capacity = encoded.capacity();
288
289        encode_record_into(
290            WalRecord {
291                seq: LogSeq::new(7),
292                op: 3,
293                payload: &payload,
294            },
295            &mut encoded,
296        )
297        .expect("encode into reserved allocation");
298
299        assert_eq!(encoded.as_ptr(), allocation);
300        assert_eq!(encoded.capacity(), capacity);
301        assert_eq!(encoded.len(), encoded_len);
302    }
303}
zeppelin_embed/wal/record.rs

zeppelin_embed/wal/
record.rs
