nodedb_wal/record/
wal_record.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! `WalRecord` — header + payload with encryption + checksum helpers.
4
5use super::header::{
6    ENCRYPTED_FLAG, HEADER_SIZE, MAX_WAL_PAYLOAD_SIZE, NO_EVENT_SOURCE, RecordHeader,
7    WAL_FORMAT_VERSION, WAL_MAGIC,
8};
9use crate::error::{Result, WalError};
10use crate::preamble::PREAMBLE_SIZE;
11
12/// A complete WAL record: header + payload.
13#[derive(Debug, Clone)]
14pub struct WalRecord {
15    pub header: RecordHeader,
16    pub payload: Vec<u8>,
17}
18
19/// The header fields of a record its caller decides: its type, scope and
20/// event source. The writer assigns the LSN.
21#[derive(Debug, Clone, Copy)]
22pub struct RecordTarget {
23    pub record_type: u32,
24    pub tenant_id: u64,
25    pub vshard_id: u32,
26    pub database_id: u64,
27    /// The event source code of the row write the record carries.
28    /// [`NO_EVENT_SOURCE`] for a record that carries no row write.
29    pub event_source: u8,
30    /// HLC wall time, in nanoseconds, the record's write committed at. See
31    /// [`RecordHeader::commit_hlc`].
32    pub commit_hlc: u64,
33}
34
35/// The header fields that tie a record to the write that appended it.
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37pub struct RecordStamp {
38    /// The idempotency key of the replicated proposal whose apply appended
39    /// the record. `0` when no proposal apply appended it.
40    pub apply_key: u64,
41    /// The event source code of the row write the record carries.
42    pub event_source: u8,
43    /// HLC wall time, in nanoseconds, the record's write committed at. See
44    /// [`RecordHeader::commit_hlc`].
45    pub commit_hlc: u64,
46}
47
48impl RecordStamp {
49    /// No proposal key, no row write, and no commit HLC.
50    pub const NONE: Self = Self {
51        apply_key: 0,
52        event_source: NO_EVENT_SOURCE,
53        commit_hlc: 0,
54    };
55}
56
57/// Parameters for [`WalRecord::new`].
58pub struct WalRecordArgs<'a> {
59    pub record_type: u32,
60    pub lsn: u64,
61    pub tenant_id: u64,
62    pub vshard_id: u32,
63    pub database_id: u64,
64    pub payload: Vec<u8>,
65    pub encryption_key: Option<&'a crate::crypto::WalEncryptionKey>,
66    pub preamble_bytes: Option<&'a [u8; PREAMBLE_SIZE]>,
67}
68
69impl WalRecord {
70    /// Create a new WAL record with computed CRC32C.
71    ///
72    /// If `encryption_key` is provided, the payload is encrypted before
73    /// CRC computation. The ciphertext includes a 16-byte auth tag.
74    ///
75    /// `preamble_bytes` — when encryption is active, the 16-byte segment
76    /// preamble that was written at offset 0 of this segment file. It is
77    /// concatenated with the record header bytes to form the AAD, binding
78    /// the ciphertext to its segment (preamble-swap defense). Pass `None`
79    /// for unencrypted records (the argument is ignored in that case).
80    ///
81    /// `database_id` is stored in header bytes 34-41.
82    pub fn new(args: WalRecordArgs<'_>) -> Result<Self> {
83        Self::new_stamped(args, RecordStamp::NONE)
84    }
85
86    /// [`Self::new`] with the proposal key, event source and commit HLC of
87    /// `stamp`. They ride the header, inside the CRC and the encryption AAD,
88    /// so the record and its stamp are durable together.
89    pub fn new_stamped(args: WalRecordArgs<'_>, stamp: RecordStamp) -> Result<Self> {
90        let RecordStamp {
91            apply_key,
92            event_source,
93            commit_hlc,
94        } = stamp;
95        let WalRecordArgs {
96            record_type,
97            lsn,
98            tenant_id,
99            vshard_id,
100            database_id,
101            payload,
102            encryption_key,
103            preamble_bytes,
104        } = args;
105        if payload.len() > MAX_WAL_PAYLOAD_SIZE {
106            return Err(WalError::PayloadTooLarge {
107                size: payload.len(),
108                max: MAX_WAL_PAYLOAD_SIZE,
109            });
110        }
111
112        let (final_payload, encrypted) = if let Some(key) = encryption_key {
113            let temp_header = RecordHeader {
114                magic: WAL_MAGIC,
115                format_version: WAL_FORMAT_VERSION,
116                record_type,
117                lsn,
118                tenant_id,
119                vshard_id,
120                payload_len: 0,
121                database_id,
122                apply_key,
123                event_source,
124                commit_hlc,
125                crc32c: 0,
126            };
127            let header_bytes = temp_header.to_bytes();
128            // AAD = preamble_bytes || header_bytes — binds ciphertext to both
129            // the segment it lives in and the record header it belongs to.
130            let aad = build_aad(preamble_bytes, &header_bytes);
131            let ciphertext = key.encrypt_aad(lsn, &aad, &payload)?;
132            (ciphertext, true)
133        } else {
134            (payload, false)
135        };
136
137        let record_type = if encrypted {
138            record_type | ENCRYPTED_FLAG
139        } else {
140            record_type
141        };
142
143        let mut header = RecordHeader {
144            magic: WAL_MAGIC,
145            format_version: WAL_FORMAT_VERSION,
146            record_type,
147            lsn,
148            tenant_id,
149            vshard_id,
150            payload_len: final_payload.len() as u32,
151            database_id,
152            apply_key,
153            event_source,
154            commit_hlc,
155            crc32c: 0,
156        };
157
158        header.crc32c = header.compute_checksum(&final_payload);
159
160        Ok(Self {
161            header,
162            payload: final_payload,
163        })
164    }
165
166    /// The idempotency key of the proposal whose apply appended this record,
167    /// `0` when no proposal apply appended it.
168    pub fn apply_key(&self) -> u64 {
169        self.header.apply_key
170    }
171
172    /// The event source code of the row write this record carries.
173    /// [`NO_EVENT_SOURCE`] for a record that carries no row write.
174    pub fn event_source(&self) -> u8 {
175        self.header.event_source
176    }
177
178    /// Decrypt the payload if the record is encrypted.
179    ///
180    /// `epoch` must come from the on-disk segment preamble, not from the
181    /// current in-memory key. `preamble_bytes` must be the same 16-byte
182    /// preamble that was used as part of the AAD during encryption.
183    pub fn decrypt_payload(
184        &self,
185        epoch: &[u8; 4],
186        preamble_bytes: Option<&[u8; PREAMBLE_SIZE]>,
187        encryption_key: Option<&crate::crypto::WalEncryptionKey>,
188    ) -> Result<Vec<u8>> {
189        if !self.is_encrypted() {
190            return Ok(self.payload.clone());
191        }
192
193        let key = encryption_key.ok_or_else(|| WalError::EncryptionError {
194            detail: "record is encrypted but no decryption key provided".into(),
195        })?;
196
197        let mut aad_header = self.header;
198        aad_header.record_type &= !ENCRYPTED_FLAG;
199        aad_header.payload_len = 0;
200        aad_header.crc32c = 0;
201        let header_bytes = aad_header.to_bytes();
202        let aad = build_aad(preamble_bytes, &header_bytes);
203
204        key.decrypt_aad(epoch, self.header.lsn, &aad, &self.payload)
205    }
206
207    /// Decrypt the payload using a key ring (supports dual-key rotation).
208    ///
209    /// `epoch` must come from the on-disk segment preamble. `preamble_bytes`
210    /// must match the preamble bytes written at the start of this segment.
211    pub fn decrypt_payload_ring(
212        &self,
213        epoch: &[u8; 4],
214        preamble_bytes: Option<&[u8; PREAMBLE_SIZE]>,
215        ring: Option<&crate::crypto::KeyRing>,
216    ) -> Result<Vec<u8>> {
217        if !self.is_encrypted() {
218            return Ok(self.payload.clone());
219        }
220
221        let ring = ring.ok_or_else(|| WalError::EncryptionError {
222            detail: "record is encrypted but no decryption key ring provided".into(),
223        })?;
224
225        let mut aad_header = self.header;
226        aad_header.record_type &= !ENCRYPTED_FLAG;
227        aad_header.payload_len = 0;
228        aad_header.crc32c = 0;
229        let header_bytes = aad_header.to_bytes();
230        let aad = build_aad(preamble_bytes, &header_bytes);
231
232        ring.decrypt_aad(epoch, self.header.lsn, &aad, &self.payload)
233    }
234
235    /// Consume an encrypted record and return the equivalent plaintext record.
236    ///
237    /// The result is indistinguishable from a record that was written without
238    /// encryption in the first place: `ENCRYPTED_FLAG` is cleared, `payload_len`
239    /// matches the plaintext, and the CRC is recomputed over the new header and
240    /// payload so [`Self::verify_checksum`] still passes and re-serialization
241    /// stays coherent.
242    ///
243    /// Recomputing the CRC does not weaken any integrity check. The on-disk CRC
244    /// was already verified against the on-disk bytes by the reader before this
245    /// call, and the AES-GCM auth tag independently binds the ciphertext to the
246    /// segment preamble and the record header. Keeping the on-disk CRC after
247    /// rewriting two header fields and the payload would leave a record that
248    /// fails its own checksum.
249    ///
250    /// A record that is not encrypted is returned untouched, so this is safe to
251    /// apply to a mixed stream.
252    pub fn into_decrypted(
253        mut self,
254        epoch: &[u8; 4],
255        preamble_bytes: Option<&[u8; PREAMBLE_SIZE]>,
256        ring: Option<&crate::crypto::KeyRing>,
257    ) -> Result<Self> {
258        if !self.is_encrypted() {
259            return Ok(self);
260        }
261
262        let plaintext = self.decrypt_payload_ring(epoch, preamble_bytes, ring)?;
263        // The ciphertext is longer than the plaintext and its length already
264        // fit `u32` in the header this record was read from, so the conversion
265        // cannot overflow; it is still checked rather than asserted.
266        let payload_len = u32::try_from(plaintext.len()).map_err(|_| WalError::CorruptRecord {
267            lsn: self.header.lsn,
268            detail: "decrypted payload length does not fit the record header".into(),
269        })?;
270
271        self.header.record_type &= !ENCRYPTED_FLAG;
272        self.header.payload_len = payload_len;
273        self.payload = plaintext;
274        self.header.crc32c = self.header.compute_checksum(&self.payload);
275        Ok(self)
276    }
277
278    /// Whether this record's payload is encrypted.
279    pub fn is_encrypted(&self) -> bool {
280        self.header.record_type & ENCRYPTED_FLAG != 0
281    }
282
283    /// Logical record type with the encryption flag stripped.
284    pub fn logical_record_type(&self) -> u32 {
285        self.header.record_type & !ENCRYPTED_FLAG
286    }
287
288    /// Verify the CRC32C checksum.
289    pub fn verify_checksum(&self) -> Result<()> {
290        let expected = self.header.crc32c;
291        let actual = self.header.compute_checksum(&self.payload);
292        if expected != actual {
293            return Err(WalError::ChecksumMismatch {
294                lsn: self.header.lsn,
295                expected,
296                actual,
297            });
298        }
299        Ok(())
300    }
301
302    /// Total size on disk: header + payload.
303    pub fn wire_size(&self) -> usize {
304        HEADER_SIZE + self.payload.len()
305    }
306}
307
308/// Build the AAD buffer: `preamble_bytes || header_bytes`.
309///
310/// When `preamble_bytes` is `None` (no encryption), the AAD
311/// is just the header bytes. When present, the preamble is prepended.
312pub(crate) fn build_aad(
313    preamble_bytes: Option<&[u8; PREAMBLE_SIZE]>,
314    header_bytes: &[u8; HEADER_SIZE],
315) -> Vec<u8> {
316    match preamble_bytes {
317        Some(p) => {
318            let mut aad = Vec::with_capacity(PREAMBLE_SIZE + HEADER_SIZE);
319            aad.extend_from_slice(p);
320            aad.extend_from_slice(header_bytes);
321            aad
322        }
323        None => header_bytes.to_vec(),
324    }
325}
326
327#[cfg(test)]
328mod tests {
329    use super::super::types::RecordType;
330    use super::*;
331
332    #[test]
333    fn checksum_roundtrip() {
334        let payload = b"hello nodedb";
335        let record = WalRecord::new(WalRecordArgs {
336            record_type: RecordType::Put as u32,
337            lsn: 1,
338            tenant_id: 0,
339            vshard_id: 0,
340            database_id: 0,
341            payload: payload.to_vec(),
342            encryption_key: None,
343            preamble_bytes: None,
344        })
345        .unwrap();
346        record.verify_checksum().unwrap();
347    }
348
349    #[test]
350    fn checksum_detects_corruption() {
351        let payload = b"hello nodedb";
352        let mut record = WalRecord::new(WalRecordArgs {
353            record_type: RecordType::Put as u32,
354            lsn: 1,
355            tenant_id: 0,
356            vshard_id: 0,
357            database_id: 0,
358            payload: payload.to_vec(),
359            encryption_key: None,
360            preamble_bytes: None,
361        })
362        .unwrap();
363        record.payload[0] ^= 0xFF;
364        assert!(matches!(
365            record.verify_checksum(),
366            Err(WalError::ChecksumMismatch { .. })
367        ));
368    }
369
370    #[test]
371    fn payload_too_large_rejected() {
372        let big_payload = vec![0u8; MAX_WAL_PAYLOAD_SIZE + 1];
373        assert!(matches!(
374            WalRecord::new(WalRecordArgs {
375                record_type: RecordType::Put as u32,
376                lsn: 1,
377                tenant_id: 0,
378                vshard_id: 0,
379                database_id: 0,
380                payload: big_payload,
381                encryption_key: None,
382                preamble_bytes: None,
383            }),
384            Err(WalError::PayloadTooLarge { .. })
385        ));
386    }
387
388    #[test]
389    fn anchor_payload_in_record() {
390        use super::super::anchor::TimeAnchorPayload;
391        let anchor = TimeAnchorPayload::new(1_700_000_000_000_000_000);
392        let record = WalRecord::new(WalRecordArgs {
393            record_type: RecordType::TimeAnchor as u32,
394            lsn: 42,
395            tenant_id: 0,
396            vshard_id: 0,
397            database_id: 0,
398            payload: anchor.to_bytes().to_vec(),
399            encryption_key: None,
400            preamble_bytes: None,
401        })
402        .unwrap();
403        record.verify_checksum().unwrap();
404        assert_eq!(record.logical_record_type(), RecordType::TimeAnchor as u32);
405        let decoded = TimeAnchorPayload::from_bytes(&record.payload).unwrap();
406        assert_eq!(decoded, anchor);
407    }
408
409    #[test]
410    fn a_stamped_record_keeps_its_event_source_and_checksum() {
411        for code in [NO_EVENT_SOURCE, 1, 2, 3, 4, 5, 6, u8::MAX] {
412            let record = WalRecord::new_stamped(
413                WalRecordArgs {
414                    record_type: 1,
415                    lsn: 9,
416                    tenant_id: 1,
417                    vshard_id: 0,
418                    database_id: 0,
419                    payload: b"row".to_vec(),
420                    encryption_key: None,
421                    preamble_bytes: None,
422                },
423                RecordStamp {
424                    apply_key: 7,
425                    event_source: code,
426                    commit_hlc: 11,
427                },
428            )
429            .expect("record");
430            assert_eq!(record.event_source(), code);
431            assert_eq!(record.apply_key(), 7);
432            record
433                .verify_checksum()
434                .expect("checksum covers the source");
435            let decoded = RecordHeader::from_bytes(&record.header.to_bytes());
436            assert_eq!(decoded.event_source, code);
437            assert_eq!(decoded.commit_hlc, 11);
438        }
439    }
440}
nodedb_wal/record/wal_record.rs

nodedb_wal/record/
wal_record.rs
