reddb_file/
wal_record.rs

1//! Main WAL record byte contract.
2//!
3//! Runtime code owns the semantic meaning of each record. This module owns the
4//! persisted record tags, body framing, compression tag, term envelope, and
5//! record checksum.
6//!
7//! Main WAL files are a sequence of frames after the file header. Every frame
8//! starts with a stable record type tag, carries a versioned body, and ends with
9//! a CRC32 checksum over the persisted bytes that precede the checksum.
10
11use crate::{WAL_FILE_VERSION, WAL_FILE_VERSION_V2};
12use std::io::{self, Read};
13
14pub const MAIN_WAL_DEFAULT_COMPRESS_THRESHOLD: usize = 256;
15const MAX_MAIN_WAL_PAYLOAD: usize = 256 * 1024 * 1024;
16const MAX_MAIN_WAL_ITEMS: usize = 1_000_000;
17
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19#[repr(u8)]
20pub enum MainWalRecordType {
21    Begin = 1,
22    Commit = 2,
23    Rollback = 3,
24    PageWrite = 4,
25    Checkpoint = 5,
26    PageWriteCompressed = 6,
27    TxCommitBatch = 7,
28    FullPageImage = 8,
29    VectorInsert = 9,
30    ProbabilisticDelta = 10,
31}
32
33impl MainWalRecordType {
34    pub fn from_u8(value: u8) -> Option<Self> {
35        match value {
36            1 => Some(Self::Begin),
37            2 => Some(Self::Commit),
38            3 => Some(Self::Rollback),
39            4 => Some(Self::PageWrite),
40            5 => Some(Self::Checkpoint),
41            6 => Some(Self::PageWriteCompressed),
42            7 => Some(Self::TxCommitBatch),
43            8 => Some(Self::FullPageImage),
44            9 => Some(Self::VectorInsert),
45            10 => Some(Self::ProbabilisticDelta),
46            _ => None,
47        }
48    }
49}
50
51#[derive(Debug, Clone, Copy, PartialEq, Eq)]
52#[repr(u8)]
53pub enum MainWalCompression {
54    None = 0,
55    Zstd = 1,
56}
57
58impl MainWalCompression {
59    fn from_u8(value: u8) -> Option<Self> {
60        match value {
61            0 => Some(Self::None),
62            1 => Some(Self::Zstd),
63            _ => None,
64        }
65    }
66}
67
68#[derive(Debug, Clone, PartialEq)]
69pub enum MainWalRecordFrame {
70    Begin {
71        tx_id: u64,
72    },
73    Commit {
74        tx_id: u64,
75    },
76    Rollback {
77        tx_id: u64,
78    },
79    PageWrite {
80        tx_id: u64,
81        page_id: u32,
82        data: Vec<u8>,
83    },
84    TxCommitBatch {
85        tx_id: u64,
86        actions: Vec<Vec<u8>>,
87    },
88    FullPageImage {
89        tx_id: u64,
90        page_id: u32,
91        ckpt_epoch: u64,
92        data: Vec<u8>,
93    },
94    VectorInsert {
95        collection: String,
96        entity_id: u64,
97        vector: Vec<f32>,
98    },
99    ProbabilisticDelta {
100        kind: u8,
101        operation: u8,
102        name: String,
103        operands: Vec<Vec<u8>>,
104    },
105    Checkpoint {
106        lsn: u64,
107    },
108}
109
110/// Borrowed view of a [`MainWalRecordFrame`].
111///
112/// The append path encodes straight out of the runtime record, so the payload
113/// of a `PageWrite`, `FullPageImage` or `TxCommitBatch` never has to be cloned
114/// into an owned frame first. The encoder works exclusively on this view — the
115/// owned frame encodes by borrowing itself — so both paths emit the same bytes
116/// by construction, not by convention.
117#[derive(Debug, Clone, Copy, PartialEq)]
118pub enum MainWalRecordFrameRef<'a> {
119    Begin {
120        tx_id: u64,
121    },
122    Commit {
123        tx_id: u64,
124    },
125    Rollback {
126        tx_id: u64,
127    },
128    PageWrite {
129        tx_id: u64,
130        page_id: u32,
131        data: &'a [u8],
132    },
133    TxCommitBatch {
134        tx_id: u64,
135        actions: &'a [Vec<u8>],
136    },
137    FullPageImage {
138        tx_id: u64,
139        page_id: u32,
140        ckpt_epoch: u64,
141        data: &'a [u8],
142    },
143    VectorInsert {
144        collection: &'a str,
145        entity_id: u64,
146        vector: &'a [f32],
147    },
148    ProbabilisticDelta {
149        kind: u8,
150        operation: u8,
151        name: &'a str,
152        operands: &'a [Vec<u8>],
153    },
154    Checkpoint {
155        lsn: u64,
156    },
157}
158
159impl<'a> From<&'a MainWalRecordFrame> for MainWalRecordFrameRef<'a> {
160    fn from(frame: &'a MainWalRecordFrame) -> Self {
161        match frame {
162            MainWalRecordFrame::Begin { tx_id } => MainWalRecordFrameRef::Begin { tx_id: *tx_id },
163            MainWalRecordFrame::Commit { tx_id } => MainWalRecordFrameRef::Commit { tx_id: *tx_id },
164            MainWalRecordFrame::Rollback { tx_id } => {
165                MainWalRecordFrameRef::Rollback { tx_id: *tx_id }
166            }
167            MainWalRecordFrame::PageWrite {
168                tx_id,
169                page_id,
170                data,
171            } => MainWalRecordFrameRef::PageWrite {
172                tx_id: *tx_id,
173                page_id: *page_id,
174                data,
175            },
176            MainWalRecordFrame::TxCommitBatch { tx_id, actions } => {
177                MainWalRecordFrameRef::TxCommitBatch {
178                    tx_id: *tx_id,
179                    actions,
180                }
181            }
182            MainWalRecordFrame::FullPageImage {
183                tx_id,
184                page_id,
185                ckpt_epoch,
186                data,
187            } => MainWalRecordFrameRef::FullPageImage {
188                tx_id: *tx_id,
189                page_id: *page_id,
190                ckpt_epoch: *ckpt_epoch,
191                data,
192            },
193            MainWalRecordFrame::VectorInsert {
194                collection,
195                entity_id,
196                vector,
197            } => MainWalRecordFrameRef::VectorInsert {
198                collection,
199                entity_id: *entity_id,
200                vector,
201            },
202            MainWalRecordFrame::ProbabilisticDelta {
203                kind,
204                operation,
205                name,
206                operands,
207            } => MainWalRecordFrameRef::ProbabilisticDelta {
208                kind: *kind,
209                operation: *operation,
210                name,
211                operands,
212            },
213            MainWalRecordFrame::Checkpoint { lsn } => {
214                MainWalRecordFrameRef::Checkpoint { lsn: *lsn }
215            }
216        }
217    }
218}
219
220#[derive(Debug, Clone, Copy, PartialEq, Eq)]
221pub struct MainWalRecordAuthority {
222    pub term: u64,
223    pub ownership_epoch: Option<u64>,
224}
225
226pub fn encode_main_wal_record_frame(frame: &MainWalRecordFrame, term: u64) -> io::Result<Vec<u8>> {
227    encode_main_wal_record_frame_with_authority(
228        frame,
229        MainWalRecordAuthority {
230            term,
231            ownership_epoch: None,
232        },
233    )
234}
235
236pub fn encode_main_wal_record_frame_with_authority(
237    frame: &MainWalRecordFrame,
238    authority: MainWalRecordAuthority,
239) -> io::Result<Vec<u8>> {
240    let mut out = Vec::new();
241    encode_main_wal_record_frame_with_authority_into(frame, authority, &mut out)?;
242    Ok(out)
243}
244
245/// Encode a frame into `out`. Accepts an owned `&MainWalRecordFrame` or a
246/// borrowed [`MainWalRecordFrameRef`] — the append path passes the latter, so
247/// the payload is written straight from the caller's buffers with no copy.
248pub fn encode_main_wal_record_frame_into<'a>(
249    frame: impl Into<MainWalRecordFrameRef<'a>>,
250    term: u64,
251    out: &mut Vec<u8>,
252) -> io::Result<()> {
253    encode_main_wal_record_frame_with_authority_into(
254        frame,
255        MainWalRecordAuthority {
256            term,
257            ownership_epoch: None,
258        },
259        out,
260    )
261}
262
263/// The single encoder. Every other encode entry point funnels here, so the
264/// owned and borrowed paths cannot drift in the bytes they persist.
265pub fn encode_main_wal_record_frame_with_authority_into<'a>(
266    frame: impl Into<MainWalRecordFrameRef<'a>>,
267    authority: MainWalRecordAuthority,
268    out: &mut Vec<u8>,
269) -> io::Result<()> {
270    let start = out.len();
271    match frame.into() {
272        MainWalRecordFrameRef::Begin { tx_id } => {
273            write_type_and_authority(out, MainWalRecordType::Begin, authority);
274            out.extend_from_slice(&tx_id.to_le_bytes());
275        }
276        MainWalRecordFrameRef::Commit { tx_id } => {
277            write_type_and_authority(out, MainWalRecordType::Commit, authority);
278            out.extend_from_slice(&tx_id.to_le_bytes());
279        }
280        MainWalRecordFrameRef::Rollback { tx_id } => {
281            write_type_and_authority(out, MainWalRecordType::Rollback, authority);
282            out.extend_from_slice(&tx_id.to_le_bytes());
283        }
284        MainWalRecordFrameRef::PageWrite {
285            tx_id,
286            page_id,
287            data,
288        } => {
289            if data.len() >= MAIN_WAL_DEFAULT_COMPRESS_THRESHOLD {
290                if let Ok(compressed) = zstd::bulk::compress(data, 3) {
291                    if compressed.len() < data.len() {
292                        write_type_and_authority(
293                            out,
294                            MainWalRecordType::PageWriteCompressed,
295                            authority,
296                        );
297                        out.extend_from_slice(&tx_id.to_le_bytes());
298                        out.extend_from_slice(&page_id.to_le_bytes());
299                        out.push(MainWalCompression::Zstd as u8);
300                        write_u32_len(out, data.len(), "main wal original page length")?;
301                        write_u32_len(out, compressed.len(), "main wal compressed page length")?;
302                        out.extend_from_slice(&compressed);
303                        append_crc(out, start);
304                        return Ok(());
305                    }
306                }
307            }
308
309            write_type_and_authority(out, MainWalRecordType::PageWrite, authority);
310            out.extend_from_slice(&tx_id.to_le_bytes());
311            out.extend_from_slice(&page_id.to_le_bytes());
312            write_u32_len(out, data.len(), "main wal page length")?;
313            out.extend_from_slice(data);
314        }
315        MainWalRecordFrameRef::TxCommitBatch { tx_id, actions } => {
316            write_type_and_authority(out, MainWalRecordType::TxCommitBatch, authority);
317            out.extend_from_slice(&tx_id.to_le_bytes());
318            write_u32_len(out, actions.len(), "main wal action count")?;
319            for action in actions {
320                write_u32_len(out, action.len(), "main wal action length")?;
321                out.extend_from_slice(action);
322            }
323        }
324        MainWalRecordFrameRef::FullPageImage {
325            tx_id,
326            page_id,
327            ckpt_epoch,
328            data,
329        } => {
330            write_type_and_authority(out, MainWalRecordType::FullPageImage, authority);
331            out.extend_from_slice(&tx_id.to_le_bytes());
332            out.extend_from_slice(&page_id.to_le_bytes());
333            out.extend_from_slice(&ckpt_epoch.to_le_bytes());
334            write_u32_len(out, data.len(), "main wal full-page image length")?;
335            out.extend_from_slice(data);
336        }
337        MainWalRecordFrameRef::VectorInsert {
338            collection,
339            entity_id,
340            vector,
341        } => {
342            write_type_and_authority(out, MainWalRecordType::VectorInsert, authority);
343            write_u32_len(out, collection.len(), "main wal collection name length")?;
344            out.extend_from_slice(collection.as_bytes());
345            out.extend_from_slice(&entity_id.to_le_bytes());
346            write_u32_len(out, vector.len(), "main wal vector length")?;
347            for value in vector {
348                out.extend_from_slice(&value.to_le_bytes());
349            }
350        }
351        MainWalRecordFrameRef::ProbabilisticDelta {
352            kind,
353            operation,
354            name,
355            operands,
356        } => {
357            write_type_and_authority(out, MainWalRecordType::ProbabilisticDelta, authority);
358            out.push(kind);
359            out.push(operation);
360            write_u32_len(out, name.len(), "main wal probabilistic name length")?;
361            out.extend_from_slice(name.as_bytes());
362            write_u32_len(out, operands.len(), "main wal probabilistic operand count")?;
363            for operand in operands {
364                write_u32_len(out, operand.len(), "main wal probabilistic operand length")?;
365                out.extend_from_slice(operand);
366            }
367        }
368        MainWalRecordFrameRef::Checkpoint { lsn } => {
369            write_type_and_authority(out, MainWalRecordType::Checkpoint, authority);
370            out.extend_from_slice(&lsn.to_le_bytes());
371        }
372    }
373
374    append_crc(out, start);
375    Ok(())
376}
377
378pub fn decode_main_wal_record_frame<R: Read>(
379    reader: &mut R,
380    format_version: u8,
381    default_term: u64,
382) -> io::Result<Option<(u64, MainWalRecordFrame)>> {
383    Ok(decode_main_wal_record_frame_with_authority(
384        reader,
385        format_version,
386        MainWalRecordAuthority {
387            term: default_term,
388            ownership_epoch: None,
389        },
390    )?
391    .map(|(authority, frame)| (authority.term, frame)))
392}
393
394pub fn decode_main_wal_record_frame_with_authority<R: Read>(
395    reader: &mut R,
396    format_version: u8,
397    default_authority: MainWalRecordAuthority,
398) -> io::Result<Option<(MainWalRecordAuthority, MainWalRecordFrame)>> {
399    let mut checksum_bytes = Vec::new();
400    let mut type_buf = [0u8; 1];
401    match reader.read_exact(&mut type_buf) {
402        Ok(()) => checksum_bytes.extend_from_slice(&type_buf),
403        Err(err) if err.kind() == io::ErrorKind::UnexpectedEof => return Ok(None),
404        Err(err) => return Err(err),
405    }
406
407    let record_type = MainWalRecordType::from_u8(type_buf[0])
408        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Invalid record type"))?;
409
410    let authority = match format_version {
411        WAL_FILE_VERSION => {
412            let term = read_u64_tracked(reader, &mut checksum_bytes)?;
413            let ownership_epoch = read_u64_tracked(reader, &mut checksum_bytes)?;
414            MainWalRecordAuthority {
415                term,
416                ownership_epoch: if ownership_epoch == 0 {
417                    None
418                } else {
419                    Some(ownership_epoch)
420                },
421            }
422        }
423        WAL_FILE_VERSION_V2 => default_authority,
424        _ => {
425            return Err(io::Error::new(
426                io::ErrorKind::InvalidData,
427                format!("Unsupported WAL version: {format_version}"),
428            ));
429        }
430    };
431
432    let frame = match record_type {
433        MainWalRecordType::Begin => MainWalRecordFrame::Begin {
434            tx_id: read_u64_tracked(reader, &mut checksum_bytes)?,
435        },
436        MainWalRecordType::Commit => MainWalRecordFrame::Commit {
437            tx_id: read_u64_tracked(reader, &mut checksum_bytes)?,
438        },
439        MainWalRecordType::Rollback => MainWalRecordFrame::Rollback {
440            tx_id: read_u64_tracked(reader, &mut checksum_bytes)?,
441        },
442        MainWalRecordType::PageWrite => {
443            let tx_id = read_u64_tracked(reader, &mut checksum_bytes)?;
444            let page_id = read_u32_tracked(reader, &mut checksum_bytes)?;
445            let data = read_bytes_tracked(reader, &mut checksum_bytes)?;
446            MainWalRecordFrame::PageWrite {
447                tx_id,
448                page_id,
449                data,
450            }
451        }
452        MainWalRecordType::PageWriteCompressed => {
453            let tx_id = read_u64_tracked(reader, &mut checksum_bytes)?;
454            let page_id = read_u32_tracked(reader, &mut checksum_bytes)?;
455            let compression = read_compression_tracked(reader, &mut checksum_bytes)?;
456            let original_len = read_u32_tracked(reader, &mut checksum_bytes)? as usize;
457            validate_len(original_len, "main WAL original page length")?;
458            let compressed = read_bytes_tracked(reader, &mut checksum_bytes)?;
459            let data = match compression {
460                MainWalCompression::Zstd => {
461                    let mut out = vec![0u8; original_len];
462                    zstd::bulk::decompress_to_buffer(&compressed, &mut out).map_err(|err| {
463                        io::Error::new(
464                            io::ErrorKind::InvalidData,
465                            format!("WAL zstd decompress failed: {err}"),
466                        )
467                    })?;
468                    out
469                }
470                MainWalCompression::None => compressed,
471            };
472            MainWalRecordFrame::PageWrite {
473                tx_id,
474                page_id,
475                data,
476            }
477        }
478        MainWalRecordType::TxCommitBatch => {
479            let tx_id = read_u64_tracked(reader, &mut checksum_bytes)?;
480            let count = read_u32_tracked(reader, &mut checksum_bytes)? as usize;
481            validate_count(count, "main WAL action count")?;
482            let mut actions = Vec::with_capacity(count);
483            for _ in 0..count {
484                actions.push(read_bytes_tracked(reader, &mut checksum_bytes)?);
485            }
486            MainWalRecordFrame::TxCommitBatch { tx_id, actions }
487        }
488        MainWalRecordType::FullPageImage => {
489            let tx_id = read_u64_tracked(reader, &mut checksum_bytes)?;
490            let page_id = read_u32_tracked(reader, &mut checksum_bytes)?;
491            let ckpt_epoch = read_u64_tracked(reader, &mut checksum_bytes)?;
492            let data = read_bytes_tracked(reader, &mut checksum_bytes)?;
493            MainWalRecordFrame::FullPageImage {
494                tx_id,
495                page_id,
496                ckpt_epoch,
497                data,
498            }
499        }
500        MainWalRecordType::VectorInsert => {
501            let collection = String::from_utf8(read_bytes_tracked(reader, &mut checksum_bytes)?)
502                .map_err(|err| {
503                    io::Error::new(
504                        io::ErrorKind::InvalidData,
505                        format!("invalid collection utf8: {err}"),
506                    )
507                })?;
508            let entity_id = read_u64_tracked(reader, &mut checksum_bytes)?;
509            let count = read_u32_tracked(reader, &mut checksum_bytes)? as usize;
510            validate_count(count, "main WAL vector length")?;
511            let mut vector = Vec::with_capacity(count);
512            for _ in 0..count {
513                vector.push(f32::from_le_bytes(read_array_tracked(
514                    reader,
515                    &mut checksum_bytes,
516                )?));
517            }
518            MainWalRecordFrame::VectorInsert {
519                collection,
520                entity_id,
521                vector,
522            }
523        }
524        MainWalRecordType::ProbabilisticDelta => {
525            let kind = read_u8_tracked(reader, &mut checksum_bytes)?;
526            let operation = read_u8_tracked(reader, &mut checksum_bytes)?;
527            let name = String::from_utf8(read_bytes_tracked(reader, &mut checksum_bytes)?)
528                .map_err(|err| {
529                    io::Error::new(
530                        io::ErrorKind::InvalidData,
531                        format!("invalid probabilistic name utf8: {err}"),
532                    )
533                })?;
534            let count = read_u32_tracked(reader, &mut checksum_bytes)? as usize;
535            validate_count(count, "main WAL probabilistic operand count")?;
536            let mut operands = Vec::with_capacity(count);
537            for _ in 0..count {
538                operands.push(read_bytes_tracked(reader, &mut checksum_bytes)?);
539            }
540            MainWalRecordFrame::ProbabilisticDelta {
541                kind,
542                operation,
543                name,
544                operands,
545            }
546        }
547        MainWalRecordType::Checkpoint => MainWalRecordFrame::Checkpoint {
548            lsn: read_u64_tracked(reader, &mut checksum_bytes)?,
549        },
550    };
551
552    let stored_crc = read_u32_untracked(reader)?;
553    if crc32(&checksum_bytes) != stored_crc {
554        return Err(io::Error::new(
555            io::ErrorKind::InvalidData,
556            "WAL record checksum mismatch",
557        ));
558    }
559
560    Ok(Some((authority, frame)))
561}
562
563fn write_type_and_authority(
564    out: &mut Vec<u8>,
565    record_type: MainWalRecordType,
566    authority: MainWalRecordAuthority,
567) {
568    out.push(record_type as u8);
569    out.extend_from_slice(&authority.term.to_le_bytes());
570    out.extend_from_slice(&authority.ownership_epoch.unwrap_or(0).to_le_bytes());
571}
572
573fn write_u32_len(out: &mut Vec<u8>, len: usize, label: &'static str) -> io::Result<()> {
574    let len = u32::try_from(len).map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, label))?;
575    out.extend_from_slice(&len.to_le_bytes());
576    Ok(())
577}
578
579fn append_crc(out: &mut Vec<u8>, start: usize) {
580    let checksum = crc32(&out[start..]);
581    out.extend_from_slice(&checksum.to_le_bytes());
582}
583
584fn crc32(bytes: &[u8]) -> u32 {
585    let mut hasher = crc32fast::Hasher::new();
586    hasher.update(bytes);
587    hasher.finalize()
588}
589
590fn read_compression_tracked<R: Read>(
591    reader: &mut R,
592    checksum_bytes: &mut Vec<u8>,
593) -> io::Result<MainWalCompression> {
594    let value = read_array_tracked::<_, 1>(reader, checksum_bytes)?[0];
595    MainWalCompression::from_u8(value).ok_or_else(|| {
596        io::Error::new(
597            io::ErrorKind::InvalidData,
598            format!("Unknown WAL compression algorithm: {value}"),
599        )
600    })
601}
602
603fn read_bytes_tracked<R: Read>(
604    reader: &mut R,
605    checksum_bytes: &mut Vec<u8>,
606) -> io::Result<Vec<u8>> {
607    let len = read_u32_tracked(reader, checksum_bytes)? as usize;
608    validate_len(len, "main WAL payload length")?;
609    let mut bytes = vec![0u8; len];
610    reader.read_exact(&mut bytes)?;
611    checksum_bytes.extend_from_slice(&bytes);
612    Ok(bytes)
613}
614
615fn validate_len(len: usize, label: &'static str) -> io::Result<()> {
616    if len > MAX_MAIN_WAL_PAYLOAD {
617        return Err(io::Error::new(
618            io::ErrorKind::InvalidData,
619            format!("implausible {label}: {len}"),
620        ));
621    }
622    Ok(())
623}
624
625fn validate_count(count: usize, label: &'static str) -> io::Result<()> {
626    if count > MAX_MAIN_WAL_ITEMS {
627        return Err(io::Error::new(
628            io::ErrorKind::InvalidData,
629            format!("implausible {label}: {count}"),
630        ));
631    }
632    Ok(())
633}
634
635fn read_u64_tracked<R: Read>(reader: &mut R, checksum_bytes: &mut Vec<u8>) -> io::Result<u64> {
636    Ok(u64::from_le_bytes(read_array_tracked(
637        reader,
638        checksum_bytes,
639    )?))
640}
641
642fn read_u32_tracked<R: Read>(reader: &mut R, checksum_bytes: &mut Vec<u8>) -> io::Result<u32> {
643    Ok(u32::from_le_bytes(read_array_tracked(
644        reader,
645        checksum_bytes,
646    )?))
647}
648
649fn read_u8_tracked<R: Read>(reader: &mut R, checksum_bytes: &mut Vec<u8>) -> io::Result<u8> {
650    Ok(read_array_tracked::<_, 1>(reader, checksum_bytes)?[0])
651}
652
653fn read_array_tracked<R: Read, const N: usize>(
654    reader: &mut R,
655    checksum_bytes: &mut Vec<u8>,
656) -> io::Result<[u8; N]> {
657    let mut bytes = [0u8; N];
658    reader.read_exact(&mut bytes)?;
659    checksum_bytes.extend_from_slice(&bytes);
660    Ok(bytes)
661}
662
663fn read_u32_untracked<R: Read>(reader: &mut R) -> io::Result<u32> {
664    let mut bytes = [0u8; 4];
665    reader.read_exact(&mut bytes)?;
666    Ok(u32::from_le_bytes(bytes))
667}
668
669#[cfg(test)]
670mod tests {
671    use super::*;
672    use std::io::Cursor;
673
674    #[test]
675    fn main_wal_record_types_are_stable() {
676        assert_eq!(
677            MainWalRecordType::from_u8(1),
678            Some(MainWalRecordType::Begin)
679        );
680        assert_eq!(
681            MainWalRecordType::from_u8(9),
682            Some(MainWalRecordType::VectorInsert)
683        );
684        assert_eq!(
685            MainWalRecordType::from_u8(10),
686            Some(MainWalRecordType::ProbabilisticDelta)
687        );
688        assert_eq!(MainWalRecordType::from_u8(11), None);
689    }
690
691    #[test]
692    fn main_wal_records_round_trip_current_format() {
693        let frames = vec![
694            MainWalRecordFrame::Begin { tx_id: 1 },
695            MainWalRecordFrame::Commit { tx_id: 2 },
696            MainWalRecordFrame::Rollback { tx_id: 3 },
697            MainWalRecordFrame::Checkpoint { lsn: 4 },
698            MainWalRecordFrame::PageWrite {
699                tx_id: 5,
700                page_id: 6,
701                data: vec![1, 2, 3],
702            },
703            MainWalRecordFrame::TxCommitBatch {
704                tx_id: 7,
705                actions: vec![b"insert".to_vec(), b"update".to_vec()],
706            },
707            MainWalRecordFrame::FullPageImage {
708                tx_id: 8,
709                page_id: 9,
710                ckpt_epoch: 10,
711                data: vec![0xAA; 128],
712            },
713            MainWalRecordFrame::VectorInsert {
714                collection: "vectors".into(),
715                entity_id: 11,
716                vector: vec![1.0, -0.5, 0.25],
717            },
718            MainWalRecordFrame::ProbabilisticDelta {
719                kind: 1,
720                operation: 1,
721                name: "visitors".into(),
722                operands: vec![b"alice".to_vec()],
723            },
724        ];
725
726        for frame in frames {
727            let encoded = encode_main_wal_record_frame(&frame, 42).unwrap();
728            let mut cursor = Cursor::new(encoded);
729            let (term, decoded) = decode_main_wal_record_frame(&mut cursor, WAL_FILE_VERSION, 0)
730                .unwrap()
731                .unwrap();
732            assert_eq!(term, 42);
733            assert_eq!(decoded, frame);
734        }
735    }
736
737    #[test]
738    fn main_wal_record_round_trip_current_format_authority_epoch() {
739        let frame = MainWalRecordFrame::Begin { tx_id: 42 };
740        let authority = MainWalRecordAuthority {
741            term: 7,
742            ownership_epoch: Some(11),
743        };
744        let encoded = encode_main_wal_record_frame_with_authority(&frame, authority).unwrap();
745
746        let mut cursor = Cursor::new(encoded);
747        let (decoded_authority, decoded) = decode_main_wal_record_frame_with_authority(
748            &mut cursor,
749            WAL_FILE_VERSION,
750            MainWalRecordAuthority {
751                term: 1,
752                ownership_epoch: None,
753            },
754        )
755        .unwrap()
756        .unwrap();
757
758        assert_eq!(decoded_authority, authority);
759        assert_eq!(decoded, frame);
760    }
761
762    #[test]
763    fn main_wal_record_accepts_legacy_v2_without_term() {
764        let mut encoded = Vec::new();
765        encoded.push(MainWalRecordType::Begin as u8);
766        encoded.extend_from_slice(&42u64.to_le_bytes());
767        let checksum = crc32(&encoded);
768        encoded.extend_from_slice(&checksum.to_le_bytes());
769
770        let mut cursor = Cursor::new(encoded);
771        let (term, frame) = decode_main_wal_record_frame(&mut cursor, WAL_FILE_VERSION_V2, 99)
772            .unwrap()
773            .unwrap();
774        assert_eq!(term, 99);
775        assert_eq!(frame, MainWalRecordFrame::Begin { tx_id: 42 });
776    }
777
778    #[test]
779    fn main_wal_record_detects_checksum_mismatch() {
780        let frame = MainWalRecordFrame::Begin { tx_id: 42 };
781        let mut encoded = encode_main_wal_record_frame(&frame, 1).unwrap();
782        let last = encoded.len() - 1;
783        encoded[last] ^= 0xFF;
784
785        let mut cursor = Cursor::new(encoded);
786        assert_eq!(
787            decode_main_wal_record_frame(&mut cursor, WAL_FILE_VERSION, 0)
788                .unwrap_err()
789                .to_string(),
790            "WAL record checksum mismatch"
791        );
792    }
793
794    #[test]
795    fn main_wal_record_rejects_implausible_payload_before_allocating() {
796        let mut encoded = Vec::new();
797        write_type_and_authority(
798            &mut encoded,
799            MainWalRecordType::PageWrite,
800            MainWalRecordAuthority {
801                term: 1,
802                ownership_epoch: None,
803            },
804        );
805        encoded.extend_from_slice(&7u64.to_le_bytes());
806        encoded.extend_from_slice(&3u32.to_le_bytes());
807        encoded.extend_from_slice(&u32::MAX.to_le_bytes());
808
809        let mut cursor = Cursor::new(encoded);
810        assert_eq!(
811            decode_main_wal_record_frame(&mut cursor, WAL_FILE_VERSION, 1)
812                .unwrap_err()
813                .to_string(),
814            "implausible main WAL payload length: 4294967295"
815        );
816    }
817
818    #[test]
819    fn main_wal_record_rejects_implausible_count_before_allocating() {
820        let mut encoded = Vec::new();
821        write_type_and_authority(
822            &mut encoded,
823            MainWalRecordType::TxCommitBatch,
824            MainWalRecordAuthority {
825                term: 1,
826                ownership_epoch: None,
827            },
828        );
829        encoded.extend_from_slice(&7u64.to_le_bytes());
830        encoded.extend_from_slice(&u32::MAX.to_le_bytes());
831
832        let mut cursor = Cursor::new(encoded);
833        assert_eq!(
834            decode_main_wal_record_frame(&mut cursor, WAL_FILE_VERSION, 1)
835                .unwrap_err()
836                .to_string(),
837            "implausible main WAL action count: 4294967295"
838        );
839    }
840
841    #[test]
842    fn main_wal_record_compresses_and_decompresses_page_writes() {
843        let frame = MainWalRecordFrame::PageWrite {
844            tx_id: 7,
845            page_id: 3,
846            data: vec![0xAB; 1024],
847        };
848        let encoded = encode_main_wal_record_frame(&frame, 1).unwrap();
849        assert_eq!(encoded[0], MainWalRecordType::PageWriteCompressed as u8);
850
851        let mut cursor = Cursor::new(encoded);
852        let (_, decoded) = decode_main_wal_record_frame(&mut cursor, WAL_FILE_VERSION, 0)
853            .unwrap()
854            .unwrap();
855        assert_eq!(decoded, frame);
856    }
857}
reddb_file/wal_record.rs

reddb_file/
wal_record.rs
