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    /// Returns the plaintext payload. If not encrypted, returns the payload as-is.
267    pub fn decrypt_payload(
268        &self,
269        encryption_key: Option<&crate::crypto::WalEncryptionKey>,
270    ) -> Result<Vec<u8>> {
271        if !self.is_encrypted() {
272            return Ok(self.payload.clone());
273        }
274
275        let key = encryption_key.ok_or_else(|| WalError::EncryptionError {
276            detail: "record is encrypted but no decryption key provided".into(),
277        })?;
278
279        // Reconstruct the header bytes used as AAD (with the encrypted flag stripped
280        // from record_type, and payload_len=0, crc32c=0 — same as during encryption).
281        let mut aad_header = self.header;
282        aad_header.record_type &= !ENCRYPTED_FLAG;
283        aad_header.payload_len = 0;
284        aad_header.crc32c = 0;
285        let header_bytes = aad_header.to_bytes();
286
287        key.decrypt(self.header.lsn, &header_bytes, &self.payload)
288    }
289
290    /// Decrypt the payload using a key ring (supports dual-key rotation).
291    ///
292    /// Tries the current key first, then falls back to the previous key.
293    /// Returns the plaintext payload. If not encrypted, returns the payload as-is.
294    pub fn decrypt_payload_ring(&self, ring: Option<&crate::crypto::KeyRing>) -> Result<Vec<u8>> {
295        if !self.is_encrypted() {
296            return Ok(self.payload.clone());
297        }
298
299        let ring = ring.ok_or_else(|| WalError::EncryptionError {
300            detail: "record is encrypted but no decryption key ring provided".into(),
301        })?;
302
303        let mut aad_header = self.header;
304        aad_header.record_type &= !ENCRYPTED_FLAG;
305        aad_header.payload_len = 0;
306        aad_header.crc32c = 0;
307        let header_bytes = aad_header.to_bytes();
308
309        ring.decrypt(self.header.lsn, &header_bytes, &self.payload)
310    }
311
312    /// Whether this record's payload is encrypted.
313    pub fn is_encrypted(&self) -> bool {
314        self.header.record_type & ENCRYPTED_FLAG != 0
315    }
316
317    /// Get the logical record type (with encryption flag stripped).
318    pub fn logical_record_type(&self) -> u16 {
319        self.header.record_type & !ENCRYPTED_FLAG
320    }
321
322    /// Verify the CRC32C checksum.
323    pub fn verify_checksum(&self) -> Result<()> {
324        let expected = self.header.crc32c;
325        let actual = self.header.compute_checksum(&self.payload);
326        if expected != actual {
327            return Err(WalError::ChecksumMismatch {
328                lsn: self.header.lsn,
329                expected,
330                actual,
331            });
332        }
333        Ok(())
334    }
335
336    /// Total size on disk: header + payload.
337    pub fn wire_size(&self) -> usize {
338        HEADER_SIZE + self.payload.len()
339    }
340}
341
342/// Bit 14 in record_type signals that the payload is AES-256-GCM encrypted.
343/// This is separate from bit 15 (required flag).
344pub const ENCRYPTED_FLAG: u16 = 0x4000;
345
346#[cfg(test)]
347mod tests {
348    use super::*;
349
350    #[test]
351    fn header_roundtrip() {
352        let header = RecordHeader {
353            magic: WAL_MAGIC,
354            format_version: WAL_FORMAT_VERSION,
355            record_type: RecordType::Put as u16,
356            lsn: 42,
357            tenant_id: 7,
358            vshard_id: 3,
359            payload_len: 100,
360            crc32c: 0xDEAD_BEEF,
361        };
362
363        let bytes = header.to_bytes();
364        let decoded = RecordHeader::from_bytes(&bytes);
365        assert_eq!(header, decoded);
366    }
367
368    #[test]
369    fn checksum_roundtrip() {
370        let payload = b"hello nodedb";
371        let record =
372            WalRecord::new(RecordType::Put as u16, 1, 0, 0, payload.to_vec(), None).unwrap();
373
374        record.verify_checksum().unwrap();
375    }
376
377    #[test]
378    fn checksum_detects_corruption() {
379        let payload = b"hello nodedb";
380        let mut record =
381            WalRecord::new(RecordType::Put as u16, 1, 0, 0, payload.to_vec(), None).unwrap();
382
383        // Corrupt one byte.
384        record.payload[0] ^= 0xFF;
385
386        assert!(matches!(
387            record.verify_checksum(),
388            Err(WalError::ChecksumMismatch { .. })
389        ));
390    }
391
392    #[test]
393    fn invalid_magic_detected() {
394        let header = RecordHeader {
395            magic: 0xBAD0_F00D,
396            format_version: WAL_FORMAT_VERSION,
397            record_type: 0,
398            lsn: 0,
399            tenant_id: 0,
400            vshard_id: 0,
401            payload_len: 0,
402            crc32c: 0,
403        };
404
405        assert!(matches!(
406            header.validate(0),
407            Err(WalError::InvalidMagic { .. })
408        ));
409    }
410
411    #[test]
412    fn payload_too_large_rejected() {
413        let big_payload = vec![0u8; MAX_WAL_PAYLOAD_SIZE + 1];
414        assert!(matches!(
415            WalRecord::new(RecordType::Put as u16, 1, 0, 0, big_payload, None),
416            Err(WalError::PayloadTooLarge { .. })
417        ));
418    }
419
420    #[test]
421    fn record_type_required_flag() {
422        assert!(RecordType::is_required(RecordType::Put as u16));
423        assert!(RecordType::is_required(RecordType::Delete as u16));
424        assert!(RecordType::is_required(RecordType::Checkpoint as u16));
425        assert!(!RecordType::is_required(RecordType::Noop as u16));
426        assert!(!RecordType::is_required(RecordType::TimeseriesBatch as u16));
427        assert!(!RecordType::is_required(RecordType::LogBatch as u16));
428    }
429}
nodedb_wal/record.rs

nodedb_wal/
record.rs
