nodedb_wal/
record.rs

1//! WAL record format.
2//!
3//! On-disk layout (all fields little-endian):
4//!
5//! ```text
6//! ┌──────────┬─────────────────┬────────────┬─────┬───────────┬───────────┬─────────────┬─────────┐
7//! │  magic   │ format_version  │ record_type│ lsn │ tenant_id │ vshard_id │ payload_len │ crc32c  │
8//! │  4 bytes │    2 bytes      │   2 bytes  │ 8B  │   4 bytes │  2 bytes  │   4 bytes   │ 4 bytes │
9//! └──────────┴─────────────────┴────────────┴─────┴───────────┴───────────┴─────────────┴─────────┘
10//! Total header: 30 bytes
11//! Followed by: [payload_len bytes of payload]
12//! ```
13
14use crate::error::{Result, WalError};
15
16/// Magic number identifying a NodeDB WAL record.
17pub const WAL_MAGIC: u32 = 0x5359_4E57; // "SYNW" in ASCII
18
19/// Current WAL format version.
20pub const WAL_FORMAT_VERSION: u16 = 1;
21
22/// Maximum WAL record payload size (64 MiB). Distinct from cluster RPC's limit.
23pub const MAX_WAL_PAYLOAD_SIZE: usize = 64 * 1024 * 1024;
24
25/// Size of the record header in bytes.
26pub const HEADER_SIZE: usize = 30;
27
28/// WAL record header (fixed 30 bytes).
29#[derive(Debug, Clone, Copy, PartialEq, Eq)]
30pub struct RecordHeader {
31    /// Magic number (`WAL_MAGIC`).
32    pub magic: u32,
33
34    /// Format version for forward/backward compatibility.
35    pub format_version: u16,
36
37    /// Record type discriminant.
38    pub record_type: u16,
39
40    /// Log Sequence Number — monotonically increasing, globally unique.
41    pub lsn: u64,
42
43    /// Tenant ID for multi-tenant isolation.
44    pub tenant_id: u32,
45
46    /// Virtual shard ID for routing.
47    pub vshard_id: u16,
48
49    /// Length of the payload following this header.
50    pub payload_len: u32,
51
52    /// CRC32C of the header (excluding this field) + payload.
53    pub crc32c: u32,
54}
55
56impl RecordHeader {
57    /// Serialize the header to a byte buffer.
58    pub fn to_bytes(&self) -> [u8; HEADER_SIZE] {
59        let mut buf = [0u8; HEADER_SIZE];
60        buf[0..4].copy_from_slice(&self.magic.to_le_bytes());
61        buf[4..6].copy_from_slice(&self.format_version.to_le_bytes());
62        buf[6..8].copy_from_slice(&self.record_type.to_le_bytes());
63        buf[8..16].copy_from_slice(&self.lsn.to_le_bytes());
64        buf[16..20].copy_from_slice(&self.tenant_id.to_le_bytes());
65        buf[20..22].copy_from_slice(&self.vshard_id.to_le_bytes());
66        buf[22..26].copy_from_slice(&self.payload_len.to_le_bytes());
67        buf[26..30].copy_from_slice(&self.crc32c.to_le_bytes());
68        buf
69    }
70
71    /// Deserialize a header from a byte buffer.
72    pub fn from_bytes(buf: &[u8; HEADER_SIZE]) -> Self {
73        Self {
74            magic: u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]),
75            format_version: u16::from_le_bytes([buf[4], buf[5]]),
76            record_type: u16::from_le_bytes([buf[6], buf[7]]),
77            lsn: u64::from_le_bytes([
78                buf[8], buf[9], buf[10], buf[11], buf[12], buf[13], buf[14], buf[15],
79            ]),
80            tenant_id: u32::from_le_bytes([buf[16], buf[17], buf[18], buf[19]]),
81            vshard_id: u16::from_le_bytes([buf[20], buf[21]]),
82            payload_len: u32::from_le_bytes([buf[22], buf[23], buf[24], buf[25]]),
83            crc32c: u32::from_le_bytes([buf[26], buf[27], buf[28], buf[29]]),
84        }
85    }
86
87    /// Compute the CRC32C over the header (excluding the crc32c field) + payload.
88    pub fn compute_checksum(&self, payload: &[u8]) -> u32 {
89        let header_bytes = self.to_bytes();
90        // Hash everything except the last 4 bytes (the crc32c field itself).
91        let mut digest = crc32c::crc32c(&header_bytes[..HEADER_SIZE - 4]);
92        digest = crc32c::crc32c_append(digest, payload);
93        digest
94    }
95
96    /// Get the logical record type (with encryption flag stripped).
97    pub fn logical_record_type(&self) -> u16 {
98        self.record_type & !ENCRYPTED_FLAG
99    }
100
101    /// Validate this header's magic and version.
102    pub fn validate(&self, offset: u64) -> Result<()> {
103        if self.magic != WAL_MAGIC {
104            return Err(WalError::InvalidMagic {
105                offset,
106                expected: WAL_MAGIC,
107                actual: self.magic,
108            });
109        }
110        if self.format_version > WAL_FORMAT_VERSION {
111            return Err(WalError::UnsupportedVersion {
112                version: self.format_version,
113                supported: WAL_FORMAT_VERSION,
114            });
115        }
116        Ok(())
117    }
118}
119
120/// Record type discriminants.
121///
122/// Types 0-255 are reserved for NodeDB core.
123/// Types 256+ are available for NodeDB specific records.
124///
125/// Bit 15 (0x8000) marks a record as **required** — unknown required records
126/// cause a replay failure. Unknown records without bit 15 set are safely skipped.
127#[derive(Debug, Clone, Copy, PartialEq, Eq)]
128#[repr(u16)]
129pub enum RecordType {
130    /// No-op / padding record (skipped during replay).
131    Noop = 0,
132
133    /// Generic key-value write.
134    Put = 1 | 0x8000,
135
136    /// Generic key deletion.
137    Delete = 2 | 0x8000,
138
139    /// Vector engine: insert/update embedding.
140    VectorPut = 10 | 0x8000,
141
142    /// Vector engine: soft-delete a vector by internal ID.
143    VectorDelete = 11 | 0x8000,
144
145    /// Vector engine: set HNSW index parameters for a collection.
146    VectorParams = 12 | 0x8000,
147
148    /// CRDT engine: delta application.
149    CrdtDelta = 20 | 0x8000,
150
151    /// Timeseries engine: metric sample batch.
152    TimeseriesBatch = 30,
153
154    /// Timeseries engine: log entry batch.
155    LogBatch = 31,
156
157    /// Atomic transaction: wraps multiple sub-records into a single WAL
158    /// group. On replay, either all sub-records apply or none.
159    /// Payload: MessagePack-encoded `Vec<(record_type: u16, payload: Vec<u8>)>`.
160    Transaction = 50 | 0x8000,
161
162    /// Checkpoint marker — indicates a consistent snapshot point.
163    Checkpoint = 100 | 0x8000,
164}
165
166impl RecordType {
167    /// Whether this record type is required (must be understood for correct replay).
168    pub fn is_required(raw: u16) -> bool {
169        raw & 0x8000 != 0
170    }
171
172    /// Convert a raw u16 to a known RecordType, or None if unknown.
173    pub fn from_raw(raw: u16) -> Option<Self> {
174        match raw {
175            0 => Some(Self::Noop),
176            x if x == 1 | 0x8000 => Some(Self::Put),
177            x if x == 2 | 0x8000 => Some(Self::Delete),
178            x if x == 10 | 0x8000 => Some(Self::VectorPut),
179            x if x == 11 | 0x8000 => Some(Self::VectorDelete),
180            x if x == 12 | 0x8000 => Some(Self::VectorParams),
181            x if x == 20 | 0x8000 => Some(Self::CrdtDelta),
182            x if x == 50 | 0x8000 => Some(Self::Transaction),
183            30 => Some(Self::TimeseriesBatch),
184            31 => Some(Self::LogBatch),
185            x if x == 100 | 0x8000 => Some(Self::Checkpoint),
186            _ => None,
187        }
188    }
189}
190
191/// A complete WAL record: header + payload.
192#[derive(Debug, Clone)]
193pub struct WalRecord {
194    pub header: RecordHeader,
195    pub payload: Vec<u8>,
196}
197
198impl WalRecord {
199    /// Create a new WAL record with computed CRC32C.
200    ///
201    /// If `encryption_key` is provided, the payload is encrypted before
202    /// CRC computation. The ciphertext includes a 16-byte auth tag.
203    pub fn new(
204        record_type: u16,
205        lsn: u64,
206        tenant_id: u32,
207        vshard_id: u16,
208        payload: Vec<u8>,
209        encryption_key: Option<&crate::crypto::WalEncryptionKey>,
210    ) -> Result<Self> {
211        if payload.len() > MAX_WAL_PAYLOAD_SIZE {
212            return Err(WalError::PayloadTooLarge {
213                size: payload.len(),
214                max: MAX_WAL_PAYLOAD_SIZE,
215            });
216        }
217
218        // Encrypt if key provided.
219        let (final_payload, encrypted) = if let Some(key) = encryption_key {
220            // Build a temporary header for AAD (crc32c is 0 during encryption).
221            let temp_header = RecordHeader {
222                magic: WAL_MAGIC,
223                format_version: WAL_FORMAT_VERSION,
224                record_type,
225                lsn,
226                tenant_id,
227                vshard_id,
228                payload_len: 0, // Will be updated after encryption.
229                crc32c: 0,
230            };
231            let header_bytes = temp_header.to_bytes();
232            let ciphertext = key.encrypt(lsn, &header_bytes, &payload)?;
233            (ciphertext, true)
234        } else {
235            (payload, false)
236        };
237
238        // Set bit 14 in record_type to indicate encryption.
239        let record_type = if encrypted {
240            record_type | ENCRYPTED_FLAG
241        } else {
242            record_type
243        };
244
245        let mut header = RecordHeader {
246            magic: WAL_MAGIC,
247            format_version: WAL_FORMAT_VERSION,
248            record_type,
249            lsn,
250            tenant_id,
251            vshard_id,
252            payload_len: final_payload.len() as u32,
253            crc32c: 0,
254        };
255
256        header.crc32c = header.compute_checksum(&final_payload);
257
258        Ok(Self {
259            header,
260            payload: final_payload,
261        })
262    }
263
264    /// Decrypt the payload if the record is encrypted.
265    ///
266    /// `epoch` is the encryption epoch from the WAL segment header.
267    /// Returns the plaintext payload. If not encrypted, returns the payload as-is.
268    pub fn decrypt_payload(
269        &self,
270        epoch: &[u8; 4],
271        encryption_key: Option<&crate::crypto::WalEncryptionKey>,
272    ) -> Result<Vec<u8>> {
273        if !self.is_encrypted() {
274            return Ok(self.payload.clone());
275        }
276
277        let key = encryption_key.ok_or_else(|| WalError::EncryptionError {
278            detail: "record is encrypted but no decryption key provided".into(),
279        })?;
280
281        // Reconstruct the header bytes used as AAD (with the encrypted flag stripped
282        // from record_type, and payload_len=0, crc32c=0 — same as during encryption).
283        let mut aad_header = self.header;
284        aad_header.record_type &= !ENCRYPTED_FLAG;
285        aad_header.payload_len = 0;
286        aad_header.crc32c = 0;
287        let header_bytes = aad_header.to_bytes();
288
289        key.decrypt(epoch, self.header.lsn, &header_bytes, &self.payload)
290    }
291
292    /// Decrypt the payload using a key ring (supports dual-key rotation).
293    ///
294    /// `epoch` is the encryption epoch from the WAL segment header.
295    /// Tries the current key first, then falls back to the previous key.
296    /// Returns the plaintext payload. If not encrypted, returns the payload as-is.
297    pub fn decrypt_payload_ring(
298        &self,
299        epoch: &[u8; 4],
300        ring: Option<&crate::crypto::KeyRing>,
301    ) -> Result<Vec<u8>> {
302        if !self.is_encrypted() {
303            return Ok(self.payload.clone());
304        }
305
306        let ring = ring.ok_or_else(|| WalError::EncryptionError {
307            detail: "record is encrypted but no decryption key ring provided".into(),
308        })?;
309
310        let mut aad_header = self.header;
311        aad_header.record_type &= !ENCRYPTED_FLAG;
312        aad_header.payload_len = 0;
313        aad_header.crc32c = 0;
314        let header_bytes = aad_header.to_bytes();
315
316        ring.decrypt(epoch, self.header.lsn, &header_bytes, &self.payload)
317    }
318
319    /// Whether this record's payload is encrypted.
320    pub fn is_encrypted(&self) -> bool {
321        self.header.record_type & ENCRYPTED_FLAG != 0
322    }
323
324    /// Get the logical record type (with encryption flag stripped).
325    pub fn logical_record_type(&self) -> u16 {
326        self.header.record_type & !ENCRYPTED_FLAG
327    }
328
329    /// Verify the CRC32C checksum.
330    pub fn verify_checksum(&self) -> Result<()> {
331        let expected = self.header.crc32c;
332        let actual = self.header.compute_checksum(&self.payload);
333        if expected != actual {
334            return Err(WalError::ChecksumMismatch {
335                lsn: self.header.lsn,
336                expected,
337                actual,
338            });
339        }
340        Ok(())
341    }
342
343    /// Total size on disk: header + payload.
344    pub fn wire_size(&self) -> usize {
345        HEADER_SIZE + self.payload.len()
346    }
347}
348
349/// Bit 14 in record_type signals that the payload is AES-256-GCM encrypted.
350/// This is separate from bit 15 (required flag).
351pub const ENCRYPTED_FLAG: u16 = 0x4000;
352
353#[cfg(test)]
354mod tests {
355    use super::*;
356
357    #[test]
358    fn header_roundtrip() {
359        let header = RecordHeader {
360            magic: WAL_MAGIC,
361            format_version: WAL_FORMAT_VERSION,
362            record_type: RecordType::Put as u16,
363            lsn: 42,
364            tenant_id: 7,
365            vshard_id: 3,
366            payload_len: 100,
367            crc32c: 0xDEAD_BEEF,
368        };
369
370        let bytes = header.to_bytes();
371        let decoded = RecordHeader::from_bytes(&bytes);
372        assert_eq!(header, decoded);
373    }
374
375    #[test]
376    fn checksum_roundtrip() {
377        let payload = b"hello nodedb";
378        let record =
379            WalRecord::new(RecordType::Put as u16, 1, 0, 0, payload.to_vec(), None).unwrap();
380
381        record.verify_checksum().unwrap();
382    }
383
384    #[test]
385    fn checksum_detects_corruption() {
386        let payload = b"hello nodedb";
387        let mut record =
388            WalRecord::new(RecordType::Put as u16, 1, 0, 0, payload.to_vec(), None).unwrap();
389
390        // Corrupt one byte.
391        record.payload[0] ^= 0xFF;
392
393        assert!(matches!(
394            record.verify_checksum(),
395            Err(WalError::ChecksumMismatch { .. })
396        ));
397    }
398
399    #[test]
400    fn invalid_magic_detected() {
401        let header = RecordHeader {
402            magic: 0xBAD0_F00D,
403            format_version: WAL_FORMAT_VERSION,
404            record_type: 0,
405            lsn: 0,
406            tenant_id: 0,
407            vshard_id: 0,
408            payload_len: 0,
409            crc32c: 0,
410        };
411
412        assert!(matches!(
413            header.validate(0),
414            Err(WalError::InvalidMagic { .. })
415        ));
416    }
417
418    #[test]
419    fn payload_too_large_rejected() {
420        let big_payload = vec![0u8; MAX_WAL_PAYLOAD_SIZE + 1];
421        assert!(matches!(
422            WalRecord::new(RecordType::Put as u16, 1, 0, 0, big_payload, None),
423            Err(WalError::PayloadTooLarge { .. })
424        ));
425    }
426
427    #[test]
428    fn record_type_required_flag() {
429        assert!(RecordType::is_required(RecordType::Put as u16));
430        assert!(RecordType::is_required(RecordType::Delete as u16));
431        assert!(RecordType::is_required(RecordType::Checkpoint as u16));
432        assert!(!RecordType::is_required(RecordType::Noop as u16));
433        assert!(!RecordType::is_required(RecordType::TimeseriesBatch as u16));
434        assert!(!RecordType::is_required(RecordType::LogBatch as u16));
435    }
436}
nodedb_wal/record.rs

nodedb_wal/
record.rs
