nodedb_wal/record/
header.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! WAL record header: fixed 63-byte prefix + constants.
4
5use crate::error::{Result, WalError};
6
7/// Magic number identifying a NodeDB WAL record.
8pub const WAL_MAGIC: u32 = 0x5359_4E57; // "SYNW"
9
10/// Current WAL format version.
11///
12/// Every record header carries this version. The header layout is the one
13/// [`HEADER_SIZE`] describes. A record with any other version does not open.
14pub const WAL_FORMAT_VERSION: u16 = 1;
15
16/// Maximum WAL record payload size (64 MiB). Distinct from cluster RPC's limit.
17pub const MAX_WAL_PAYLOAD_SIZE: usize = 64 * 1024 * 1024;
18
19/// Size of the record header in bytes.
20///
21/// Layout (all little-endian):
22///   magic(4) | format_version(2) | record_type(4) | lsn(8) | tenant_id(8)
23///   | vshard_id(4) | payload_len(4) | database_id(8) | apply_key(8)
24///   | event_source(1) | commit_hlc(8) | crc32c(4)
25///
26/// `database_id` occupies bytes 34–41. `apply_key` occupies bytes 42–49.
27pub const HEADER_SIZE: usize = 63;
28
29/// The event source of a record that carries no row write. Replay rebuilds
30/// no write event from it. A write record carries the code of the source its
31/// write ran with; the WAL stores the code and does not interpret it.
32pub const NO_EVENT_SOURCE: u8 = 0;
33
34/// Bit 14 in `record_type` signals the payload is AES-256-GCM encrypted.
35/// Separate from bit 15 (required flag).
36pub const ENCRYPTED_FLAG: u32 = 0x0000_4000;
37
38/// Bit 15: required-flag. Records with this bit set and an unknown type
39/// must not be silently skipped.
40pub const REQUIRED_FLAG: u32 = 0x0000_8000;
41
42/// WAL record header (fixed 63 bytes).
43#[derive(Debug, Clone, Copy, PartialEq, Eq)]
44pub struct RecordHeader {
45    pub magic: u32,
46    pub format_version: u16,
47    pub record_type: u32,
48    pub lsn: u64,
49    pub tenant_id: u64,
50    pub vshard_id: u32,
51    pub payload_len: u32,
52    /// Database scope for this record, as the raw `u64` of a `DatabaseId`.
53    /// Decoded exactly as written. Covered by CRC32C. Occupies bytes 34–41.
54    pub database_id: u64,
55    /// The idempotency key of the replicated proposal whose apply appended
56    /// this record, `0` for a record no proposal apply appended. The record
57    /// and the key are durable together, so a node recovers which proposals
58    /// it applied from the records themselves. Covered by CRC32C. Occupies
59    /// bytes 42–49.
60    pub apply_key: u64,
61    /// The event source of the row write this record carries, as the writer's
62    /// code. [`NO_EVENT_SOURCE`] for a record that carries no row write.
63    /// Covered by CRC32C. Occupies byte 50.
64    pub event_source: u8,
65    /// HLC wall time, in nanoseconds, at which the write this record belongs
66    /// to committed, or the proposer's stamp of the metadata entry whose
67    /// apply appended it. `0` for a record that belongs to neither: a cluster
68    /// restore judges such a record by its WAL position, never by a clock. A
69    /// restore keeps a record with a nonzero value only when it is below the
70    /// restore point's watermark. Covered by CRC32C. Occupies bytes 51–58.
71    pub commit_hlc: u64,
72    pub crc32c: u32,
73}
74
75impl RecordHeader {
76    pub fn to_bytes(&self) -> [u8; HEADER_SIZE] {
77        let mut buf = [0u8; HEADER_SIZE];
78        buf[0..4].copy_from_slice(&self.magic.to_le_bytes());
79        buf[4..6].copy_from_slice(&self.format_version.to_le_bytes());
80        buf[6..10].copy_from_slice(&self.record_type.to_le_bytes());
81        buf[10..18].copy_from_slice(&self.lsn.to_le_bytes());
82        buf[18..26].copy_from_slice(&self.tenant_id.to_le_bytes());
83        buf[26..30].copy_from_slice(&self.vshard_id.to_le_bytes());
84        buf[30..34].copy_from_slice(&self.payload_len.to_le_bytes());
85        buf[34..42].copy_from_slice(&self.database_id.to_le_bytes());
86        buf[42..50].copy_from_slice(&self.apply_key.to_le_bytes());
87        buf[50] = self.event_source;
88        buf[51..59].copy_from_slice(&self.commit_hlc.to_le_bytes());
89        buf[59..63].copy_from_slice(&self.crc32c.to_le_bytes());
90        buf
91    }
92
93    pub fn from_bytes(buf: &[u8; HEADER_SIZE]) -> Self {
94        Self {
95            magic: u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]),
96            format_version: u16::from_le_bytes([buf[4], buf[5]]),
97            record_type: u32::from_le_bytes([buf[6], buf[7], buf[8], buf[9]]),
98            lsn: u64::from_le_bytes([
99                buf[10], buf[11], buf[12], buf[13], buf[14], buf[15], buf[16], buf[17],
100            ]),
101            tenant_id: u64::from_le_bytes([
102                buf[18], buf[19], buf[20], buf[21], buf[22], buf[23], buf[24], buf[25],
103            ]),
104            vshard_id: u32::from_le_bytes([buf[26], buf[27], buf[28], buf[29]]),
105            payload_len: u32::from_le_bytes([buf[30], buf[31], buf[32], buf[33]]),
106            database_id: u64::from_le_bytes([
107                buf[34], buf[35], buf[36], buf[37], buf[38], buf[39], buf[40], buf[41],
108            ]),
109            apply_key: u64::from_le_bytes([
110                buf[42], buf[43], buf[44], buf[45], buf[46], buf[47], buf[48], buf[49],
111            ]),
112            event_source: buf[50],
113            commit_hlc: u64::from_le_bytes([
114                buf[51], buf[52], buf[53], buf[54], buf[55], buf[56], buf[57], buf[58],
115            ]),
116            crc32c: u32::from_le_bytes([buf[59], buf[60], buf[61], buf[62]]),
117        }
118    }
119
120    /// CRC32C over header (excluding the crc32c field) + payload.
121    ///
122    /// The apply key is included in the CRC so it cannot be silently
123    /// modified without detection.
124    pub fn compute_checksum(&self, payload: &[u8]) -> u32 {
125        let header_bytes = self.to_bytes();
126        let mut digest = crc32c::crc32c(&header_bytes[..HEADER_SIZE - 4]);
127        digest = crc32c::crc32c_append(digest, payload);
128        digest
129    }
130
131    /// Logical record type with the encryption flag stripped.
132    pub fn logical_record_type(&self) -> u32 {
133        self.record_type & !ENCRYPTED_FLAG
134    }
135
136    pub fn validate(&self, offset: u64) -> Result<()> {
137        if self.magic != WAL_MAGIC {
138            return Err(WalError::InvalidMagic {
139                offset,
140                expected: WAL_MAGIC,
141                actual: self.magic,
142            });
143        }
144        if self.format_version != WAL_FORMAT_VERSION {
145            return Err(WalError::UnsupportedVersion {
146                version: self.format_version,
147                supported: WAL_FORMAT_VERSION,
148            });
149        }
150        let payload_len =
151            usize::try_from(self.payload_len).map_err(|_| WalError::PayloadTooLarge {
152                size: usize::MAX,
153                max: MAX_WAL_PAYLOAD_SIZE,
154            })?;
155        if payload_len > MAX_WAL_PAYLOAD_SIZE {
156            return Err(WalError::PayloadTooLarge {
157                size: payload_len,
158                max: MAX_WAL_PAYLOAD_SIZE,
159            });
160        }
161        Ok(())
162    }
163}
164
165#[cfg(test)]
166mod tests {
167    use super::*;
168
169    fn make_header(record_type: u32, vshard_id: u32) -> RecordHeader {
170        RecordHeader {
171            magic: WAL_MAGIC,
172            format_version: WAL_FORMAT_VERSION,
173            record_type,
174            lsn: 42,
175            tenant_id: 7,
176            vshard_id,
177            payload_len: 100,
178            database_id: 0,
179            apply_key: 0,
180            event_source: NO_EVENT_SOURCE,
181            commit_hlc: 0,
182            crc32c: 0xDEAD_BEEF,
183        }
184    }
185
186    #[test]
187    fn header_roundtrip() {
188        let header = make_header(1 | REQUIRED_FLAG, 3);
189        let bytes = header.to_bytes();
190        assert_eq!(header, RecordHeader::from_bytes(&bytes));
191    }
192
193    #[test]
194    fn header_golden_63_bytes_exact_offsets() {
195        // magic at 0..4, format_version at 4..6, record_type at 6..10,
196        // lsn at 10..18, tenant_id at 18..26, vshard_id at 26..30,
197        // payload_len at 30..34, database_id at 34..42, apply_key at 42..50,
198        // event_source at 50, commit_hlc at 51..59, crc32c at 59..63.
199        let header = RecordHeader {
200            magic: WAL_MAGIC,
201            format_version: WAL_FORMAT_VERSION,
202            record_type: 1,
203            lsn: 0x0102_0304_0506_0708,
204            tenant_id: 0xDEAD_BEEF_CAFE_1234,
205            vshard_id: 0xCAFE_BABE,
206            payload_len: 256,
207            database_id: 0xABCD_0000_1234_5678,
208            apply_key: 0,
209            event_source: NO_EVENT_SOURCE,
210            commit_hlc: 0,
211            crc32c: 0x1234_5678,
212        };
213        let b = header.to_bytes();
214        assert_eq!(b.len(), 63);
215        // magic
216        assert_eq!(&b[0..4], &WAL_MAGIC.to_le_bytes());
217        // format_version
218        assert_eq!(&b[4..6], &WAL_FORMAT_VERSION.to_le_bytes());
219        // record_type
220        assert_eq!(&b[6..10], &1u32.to_le_bytes());
221        // lsn
222        assert_eq!(&b[10..18], &0x0102_0304_0506_0708u64.to_le_bytes());
223        // tenant_id (u64, 8 bytes)
224        assert_eq!(&b[18..26], &0xDEAD_BEEF_CAFE_1234u64.to_le_bytes());
225        // vshard_id
226        assert_eq!(&b[26..30], &0xCAFE_BABEu32.to_le_bytes());
227        // payload_len
228        assert_eq!(&b[30..34], &256u32.to_le_bytes());
229        // database_id
230        assert_eq!(&b[34..42], &0xABCD_0000_1234_5678u64.to_le_bytes());
231        // apply_key — zero
232        assert_eq!(&b[42..50], &[0u8; 8]);
233        // event_source
234        assert_eq!(b[50], NO_EVENT_SOURCE);
235        // commit_hlc — zero
236        assert_eq!(&b[51..59], &[0u8; 8]);
237        // crc32c
238        assert_eq!(&b[59..63], &0x1234_5678u32.to_le_bytes());
239    }
240
241    #[test]
242    fn database_id_roundtrip() {
243        // Non-zero database_id survives to_bytes → from_bytes.
244        let header = RecordHeader {
245            magic: WAL_MAGIC,
246            format_version: WAL_FORMAT_VERSION,
247            record_type: 1,
248            lsn: 1,
249            tenant_id: 42,
250            vshard_id: 0,
251            payload_len: 0,
252            database_id: 7,
253            apply_key: 0,
254            event_source: NO_EVENT_SOURCE,
255            commit_hlc: 0,
256            crc32c: 0,
257        };
258        let bytes = header.to_bytes();
259        let decoded = RecordHeader::from_bytes(&bytes);
260        assert_eq!(decoded.database_id, 7);
261    }
262
263    #[test]
264    fn commit_hlc_roundtrip() {
265        let header = RecordHeader {
266            commit_hlc: 1_700_000_000_123_456_789,
267            ..make_header(1, 0)
268        };
269        let decoded = RecordHeader::from_bytes(&header.to_bytes());
270        assert_eq!(decoded.commit_hlc, 1_700_000_000_123_456_789);
271        assert_eq!(decoded, header);
272    }
273
274    #[test]
275    fn database_id_decodes_from_its_own_bytes() {
276        // Bytes 34..42 alone decide `database_id`. The neighbouring fields
277        // are all ones, so a misplaced read would show.
278        let mut raw = [0xFFu8; HEADER_SIZE];
279        raw[34..42].copy_from_slice(&0x0102_0304_0506_0708u64.to_le_bytes());
280        let decoded = RecordHeader::from_bytes(&raw);
281        assert_eq!(decoded.database_id, 0x0102_0304_0506_0708);
282        assert_eq!(decoded.payload_len, u32::MAX);
283        assert_eq!(decoded.apply_key, u64::MAX);
284        assert_eq!(RecordHeader::from_bytes(&decoded.to_bytes()), decoded);
285    }
286
287    #[test]
288    fn tenant_id_above_u32_max_roundtrip() {
289        // Verify u64 tenant_id with a value > u32::MAX is preserved exactly.
290        let tid = u32::MAX as u64 + 1;
291        let header = RecordHeader {
292            magic: WAL_MAGIC,
293            format_version: WAL_FORMAT_VERSION,
294            record_type: 1,
295            lsn: 1,
296            tenant_id: tid,
297            vshard_id: 0,
298            payload_len: 0,
299            database_id: 0,
300            apply_key: 0,
301            event_source: NO_EVENT_SOURCE,
302            commit_hlc: 0,
303            crc32c: 0,
304        };
305        let bytes = header.to_bytes();
306        let decoded = RecordHeader::from_bytes(&bytes);
307        assert_eq!(decoded.tenant_id, tid);
308    }
309
310    #[test]
311    fn invalid_magic_detected() {
312        let mut header = make_header(0, 0);
313        header.magic = 0xBAD0_F00D;
314        assert!(matches!(
315            header.validate(0),
316            Err(WalError::InvalidMagic { .. })
317        ));
318    }
319
320    #[test]
321    fn payload_larger_than_limit_is_rejected_before_reading() {
322        let mut header = make_header(0, 0);
323        header.payload_len = (MAX_WAL_PAYLOAD_SIZE + 1) as u32;
324        assert!(matches!(
325            header.validate(0),
326            Err(WalError::PayloadTooLarge {
327                size,
328                max: MAX_WAL_PAYLOAD_SIZE,
329            }) if size == MAX_WAL_PAYLOAD_SIZE + 1
330        ));
331    }
332
333    #[test]
334    fn unsupported_version_detected() {
335        let mut header = make_header(0, 0);
336        header.format_version = WAL_FORMAT_VERSION + 1;
337        assert!(matches!(
338            header.validate(0),
339            Err(WalError::UnsupportedVersion { .. })
340        ));
341    }
342
343    #[test]
344    fn older_version_rejected() {
345        let mut header = make_header(0, 0);
346        header.format_version = WAL_FORMAT_VERSION - 1;
347        assert!(matches!(
348            header.validate(0),
349            Err(WalError::UnsupportedVersion { version, .. })
350                if version == WAL_FORMAT_VERSION - 1
351        ));
352    }
353
354    #[test]
355    fn large_vshard_id_roundtrip() {
356        // 0x1234_5678 is well above old u16::MAX; ensure no truncation.
357        let header = make_header(1, 0x1234_5678);
358        let bytes = header.to_bytes();
359        let decoded = RecordHeader::from_bytes(&bytes);
360        assert_eq!(decoded.vshard_id, 0x1234_5678u32);
361    }
362
363    #[test]
364    fn encrypted_flag_is_u32() {
365        let header = make_header(1 | ENCRYPTED_FLAG, 0);
366        assert_eq!(header.logical_record_type(), 1);
367        assert!(header.record_type & ENCRYPTED_FLAG != 0);
368    }
369
370    #[test]
371    fn large_record_type_roundtrip() {
372        // 0x0001_0001 has bits above old u16 max set; verify u32 width preserved.
373        let header = make_header(0x0001_0001, 0);
374        let bytes = header.to_bytes();
375        let decoded = RecordHeader::from_bytes(&bytes);
376        assert_eq!(decoded.record_type, 0x0001_0001u32);
377        // Flags in bit-14 and bit-15 positions still work alongside high bits.
378        let with_flags = make_header(0x0001_0001 | ENCRYPTED_FLAG | REQUIRED_FLAG, 0);
379        let bytes2 = with_flags.to_bytes();
380        let decoded2 = RecordHeader::from_bytes(&bytes2);
381        assert_eq!(
382            decoded2.record_type,
383            0x0001_0001 | ENCRYPTED_FLAG | REQUIRED_FLAG
384        );
385        assert_eq!(decoded2.logical_record_type(), 0x0001_0001 | REQUIRED_FLAG);
386    }
387
388    #[test]
389    fn every_event_source_code_roundtrips() {
390        for code in 0..=u8::MAX {
391            let mut header = make_header(1, 0);
392            header.event_source = code;
393            let decoded = RecordHeader::from_bytes(&header.to_bytes());
394            assert_eq!(decoded.event_source, code);
395            assert_eq!(decoded, header);
396        }
397    }
398}
nodedb_wal/record/header.rs

nodedb_wal/record/
header.rs
