horon/
wal.rs

1//! Write-Ahead Log — append-only mutation log with per-entry CRC.
2//!
3//! Every Store mutation (insert, update, delete, set_meta, set_semantic)
4//! is serialized as a WAL entry. On recovery, entries are replayed in
5//! sequence order. The first entry with a bad CRC is treated as a partial
6//! write — the WAL is truncated there.
7
8use std::io::{Read, Write};
9
10use crate::error::{HoronError, HoronResult};
11use crate::format::*;
12use crate::quant::SemLayout;
13use crate::snapshot::NodeEntry;
14
15/// A single WAL entry.
16#[derive(Debug, Clone)]
17pub struct WalEntry {
18    /// Monotonically increasing sequence number assigned at append time.
19    pub seq: u32,
20    /// Operation code (one of the `format::OP_*` constants).
21    pub op: u8,
22    /// Key of the node the operation targets.
23    pub key: String,
24    /// Operation-specific payload.
25    pub payload: WalPayload,
26}
27
28/// Op-specific payload.
29#[derive(Debug, Clone)]
30pub enum WalPayload {
31    /// INSERT: full node entry.
32    Insert(NodeEntry),
33    /// UPDATE: replacement data + metadata.
34    Update {
35        /// Replacement payload bytes for the node.
36        data: Vec<u8>,
37        /// Replacement metadata pairs (replaces all existing metadata).
38        metadata: Vec<(String, String)>,
39    },
40    /// DELETE: no additional data.
41    Delete,
42    /// SET_META: single key/value.
43    SetMeta {
44        /// Metadata key to set.
45        meta_key: String,
46        /// Value to associate with `meta_key`.
47        meta_value: String,
48    },
49    /// SET_SEMANTIC: full semantic coordinate replacement.
50    SetSemantic {
51        /// New semantic coordinates as raw Q64.64 bytes (16 bytes per dimension).
52        coords: Vec<u8>,
53    },
54    /// EPOCH: seal marker (temporal epochs). No node state change; replay advances the
55    /// in-memory epoch counter. `key` is empty — epochs are file-scoped.
56    Epoch {
57        /// Monotonically increasing logical epoch counter (not wall-clock).
58        epoch_id: u64,
59        /// `EPOCH_FLAG_*` bits (bit 0: speculative).
60        flags: u8,
61    },
62}
63
64impl WalEntry {
65    /// Serialize this entry to bytes (excluding the trailing CRC).
66    ///
67    /// Semantic coordinates are held full-width (16 bytes/dim) in memory;
68    /// `layout` decides the on-disk tail encoding (quantization).
69    fn write_body<W: Write>(&self, w: &mut W, layout: &SemLayout) -> HoronResult<()> {
70        // seq (u32)
71        w.write_all(&self.seq.to_le_bytes())?;
72        // op (u8)
73        w.write_all(&[self.op])?;
74        // key
75        let key_bytes = self.key.as_bytes();
76        w.write_all(&(key_bytes.len() as u16).to_le_bytes())?;
77        w.write_all(key_bytes)?;
78
79        match &self.payload {
80            WalPayload::Insert(entry) => {
81                // data
82                w.write_all(&(entry.data.len() as u32).to_le_bytes())?;
83                w.write_all(&entry.data)?;
84                // metadata
85                w.write_all(&(entry.metadata.len() as u16).to_le_bytes())?;
86                for (mk, mv) in &entry.metadata {
87                    let mk_b = mk.as_bytes();
88                    let mv_b = mv.as_bytes();
89                    w.write_all(&(mk_b.len() as u16).to_le_bytes())?;
90                    w.write_all(mk_b)?;
91                    w.write_all(&(mv_b.len() as u16).to_le_bytes())?;
92                    w.write_all(mv_b)?;
93                }
94                // semantic coords
95                if layout.quantized {
96                    if layout.disk_bytes() > 0 {
97                        w.write_all(&layout.encode_tail(&entry.semantic_coords)?)?;
98                    }
99                } else if !entry.semantic_coords.is_empty() {
100                    w.write_all(&entry.semantic_coords)?;
101                }
102            }
103            WalPayload::Update { data, metadata } => {
104                w.write_all(&(data.len() as u32).to_le_bytes())?;
105                w.write_all(data)?;
106                w.write_all(&(metadata.len() as u16).to_le_bytes())?;
107                for (mk, mv) in metadata {
108                    let mk_b = mk.as_bytes();
109                    let mv_b = mv.as_bytes();
110                    w.write_all(&(mk_b.len() as u16).to_le_bytes())?;
111                    w.write_all(mk_b)?;
112                    w.write_all(&(mv_b.len() as u16).to_le_bytes())?;
113                    w.write_all(mv_b)?;
114                }
115            }
116            WalPayload::Delete => {
117                // No additional fields
118            }
119            WalPayload::SetMeta { meta_key, meta_value } => {
120                let mk_b = meta_key.as_bytes();
121                let mv_b = meta_value.as_bytes();
122                w.write_all(&(mk_b.len() as u16).to_le_bytes())?;
123                w.write_all(mk_b)?;
124                w.write_all(&(mv_b.len() as u16).to_le_bytes())?;
125                w.write_all(mv_b)?;
126            }
127            WalPayload::SetSemantic { coords } => {
128                if layout.quantized {
129                    w.write_all(&layout.encode_tail(coords)?)?;
130                } else {
131                    w.write_all(coords)?;
132                }
133            }
134            WalPayload::Epoch { epoch_id, flags } => {
135                w.write_all(&epoch_id.to_le_bytes())?;
136                w.write_all(&[*flags])?;
137            }
138        }
139
140        Ok(())
141    }
142
143    /// Serialize entry with trailing CRC32.
144    pub fn write_to<W: Write>(&self, w: &mut W, layout: &SemLayout) -> HoronResult<()> {
145        let mut body = Vec::new();
146        self.write_body(&mut body, layout)?;
147
148        let crc = crc32fast::hash(&body);
149        w.write_all(&body)?;
150        w.write_all(&crc.to_le_bytes())?;
151
152        Ok(())
153    }
154
155    /// Deserialize an entry from bytes. Returns None if CRC check fails
156    /// (indicating a partial write / truncation point). The payload's
157    /// semantic coordinates come back full-width regardless of the disk
158    /// encoding (`layout` — quantization).
159    pub fn read_from<R: Read>(
160        r: &mut R,
161        layout: &SemLayout,
162    ) -> HoronResult<Option<Self>> {
163        let mut buf2 = [0u8; 2];
164        let mut buf4 = [0u8; 4];
165
166        // We need to buffer the body for CRC verification
167        // Read seq
168        if r.read_exact(&mut buf4).is_err() {
169            return Ok(None); // EOF — no more entries
170        }
171        let mut body = Vec::new();
172        body.extend_from_slice(&buf4);
173        let seq = u32::from_le_bytes(buf4);
174
175        // op
176        let mut op_buf = [0u8; 1];
177        r.read_exact(&mut op_buf)?;
178        body.push(op_buf[0]);
179        let op = op_buf[0];
180
181        // key
182        r.read_exact(&mut buf2)?;
183        body.extend_from_slice(&buf2);
184        let key_len = u16::from_le_bytes(buf2) as usize;
185        let mut key_buf = vec![0u8; key_len];
186        r.read_exact(&mut key_buf)?;
187        body.extend_from_slice(&key_buf);
188        let key = String::from_utf8(key_buf)
189            .map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8 key: {}", e)))?;
190
191        // Op-specific body reading (accumulates into body for CRC)
192        let payload = match op {
193            OP_INSERT => {
194                let entry = read_insert_body(r, &mut body, layout)?;
195                WalPayload::Insert(entry)
196            }
197            OP_UPDATE => {
198                let (data, metadata) = read_update_body(r, &mut body)?;
199                WalPayload::Update { data, metadata }
200            }
201            OP_DELETE => {
202                WalPayload::Delete
203            }
204            OP_SET_META => {
205                let (mk, mv) = read_meta_body(r, &mut body)?;
206                WalPayload::SetMeta { meta_key: mk, meta_value: mv }
207            }
208            OP_SET_SEMANTIC => {
209                // The CRC covers the on-disk bytes; the payload is decoded
210                // to full width afterwards.
211                let mut disk = vec![0u8; layout.disk_bytes()];
212                r.read_exact(&mut disk)?;
213                body.extend_from_slice(&disk);
214                let coords = if layout.quantized {
215                    layout.decode_tail(&disk)?
216                } else {
217                    disk
218                };
219                WalPayload::SetSemantic { coords }
220            }
221            OP_EPOCH => {
222                let mut buf8 = [0u8; 8];
223                r.read_exact(&mut buf8)?;
224                body.extend_from_slice(&buf8);
225                let epoch_id = u64::from_le_bytes(buf8);
226                let mut flag_buf = [0u8; 1];
227                r.read_exact(&mut flag_buf)?;
228                body.push(flag_buf[0]);
229                WalPayload::Epoch { epoch_id, flags: flag_buf[0] }
230            }
231            _ => {
232                return Err(HoronError::InvalidFormat(
233                    format!("unknown WAL op code: 0x{:02X}", op)
234                ));
235            }
236        };
237
238        // Read and verify CRC
239        r.read_exact(&mut buf4)?;
240        let stored_crc = u32::from_le_bytes(buf4);
241        let computed_crc = crc32fast::hash(&body);
242
243        if stored_crc != computed_crc {
244            // Partial write detected — truncation point
245            return Ok(None);
246        }
247
248        Ok(Some(WalEntry { seq, op, key, payload }))
249    }
250}
251
252/// Read INSERT body fields, appending raw bytes to `body` for CRC.
253fn read_insert_body<R: Read>(
254    r: &mut R,
255    body: &mut Vec<u8>,
256    layout: &SemLayout,
257) -> HoronResult<NodeEntry> {
258    let mut buf2 = [0u8; 2];
259    let mut buf4 = [0u8; 4];
260
261    // data
262    r.read_exact(&mut buf4)?;
263    body.extend_from_slice(&buf4);
264    let data_len = u32::from_le_bytes(buf4) as usize;
265    if data_len > MAX_ENTRY_DATA {
266        return Err(HoronError::InvalidFormat(format!(
267            "WAL entry data length {} exceeds maximum {} — corrupt length field",
268            data_len, MAX_ENTRY_DATA
269        )));
270    }
271    let data = crate::format::read_bounded_vec(r, data_len, "WAL entry data")?;
272    body.extend_from_slice(&data);
273
274    // metadata
275    r.read_exact(&mut buf2)?;
276    body.extend_from_slice(&buf2);
277    let meta_count = u16::from_le_bytes(buf2) as usize;
278    let mut metadata = Vec::with_capacity(meta_count);
279    for _ in 0..meta_count {
280        r.read_exact(&mut buf2)?;
281        body.extend_from_slice(&buf2);
282        let mk_len = u16::from_le_bytes(buf2) as usize;
283        let mut mk_buf = vec![0u8; mk_len];
284        r.read_exact(&mut mk_buf)?;
285        body.extend_from_slice(&mk_buf);
286
287        r.read_exact(&mut buf2)?;
288        body.extend_from_slice(&buf2);
289        let mv_len = u16::from_le_bytes(buf2) as usize;
290        let mut mv_buf = vec![0u8; mv_len];
291        r.read_exact(&mut mv_buf)?;
292        body.extend_from_slice(&mv_buf);
293
294        let mk = String::from_utf8(mk_buf)
295            .map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8: {}", e)))?;
296        let mv = String::from_utf8(mv_buf)
297            .map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8: {}", e)))?;
298        metadata.push((mk, mv));
299    }
300
301    // semantic coords (CRC over disk bytes; payload decoded to full width)
302    let disk_bytes = layout.disk_bytes();
303    let mut semantic_coords = vec![0u8; disk_bytes];
304    if disk_bytes > 0 {
305        r.read_exact(&mut semantic_coords)?;
306        body.extend_from_slice(&semantic_coords);
307        if layout.quantized {
308            semantic_coords = layout.decode_tail(&semantic_coords)?;
309        }
310    }
311
312    Ok(NodeEntry {
313        key: String::new(), // key already parsed by caller
314        data,
315        metadata,
316        semantic_coords,
317    })
318}
319
320/// Read UPDATE body fields.
321fn read_update_body<R: Read>(
322    r: &mut R,
323    body: &mut Vec<u8>,
324) -> HoronResult<(Vec<u8>, Vec<(String, String)>)> {
325    let mut buf2 = [0u8; 2];
326    let mut buf4 = [0u8; 4];
327
328    // data
329    r.read_exact(&mut buf4)?;
330    body.extend_from_slice(&buf4);
331    let data_len = u32::from_le_bytes(buf4) as usize;
332    if data_len > MAX_ENTRY_DATA {
333        return Err(HoronError::InvalidFormat(format!(
334            "WAL entry data length {} exceeds maximum {} — corrupt length field",
335            data_len, MAX_ENTRY_DATA
336        )));
337    }
338    let data = crate::format::read_bounded_vec(r, data_len, "WAL entry data")?;
339    body.extend_from_slice(&data);
340
341    // metadata
342    r.read_exact(&mut buf2)?;
343    body.extend_from_slice(&buf2);
344    let meta_count = u16::from_le_bytes(buf2) as usize;
345    let mut metadata = Vec::with_capacity(meta_count);
346    for _ in 0..meta_count {
347        r.read_exact(&mut buf2)?;
348        body.extend_from_slice(&buf2);
349        let mk_len = u16::from_le_bytes(buf2) as usize;
350        let mut mk_buf = vec![0u8; mk_len];
351        r.read_exact(&mut mk_buf)?;
352        body.extend_from_slice(&mk_buf);
353
354        r.read_exact(&mut buf2)?;
355        body.extend_from_slice(&buf2);
356        let mv_len = u16::from_le_bytes(buf2) as usize;
357        let mut mv_buf = vec![0u8; mv_len];
358        r.read_exact(&mut mv_buf)?;
359        body.extend_from_slice(&mv_buf);
360
361        let mk = String::from_utf8(mk_buf)
362            .map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8: {}", e)))?;
363        let mv = String::from_utf8(mv_buf)
364            .map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8: {}", e)))?;
365        metadata.push((mk, mv));
366    }
367
368    Ok((data, metadata))
369}
370
371/// Read SET_META body fields.
372fn read_meta_body<R: Read>(
373    r: &mut R,
374    body: &mut Vec<u8>,
375) -> HoronResult<(String, String)> {
376    let mut buf2 = [0u8; 2];
377
378    r.read_exact(&mut buf2)?;
379    body.extend_from_slice(&buf2);
380    let mk_len = u16::from_le_bytes(buf2) as usize;
381    let mut mk_buf = vec![0u8; mk_len];
382    r.read_exact(&mut mk_buf)?;
383    body.extend_from_slice(&mk_buf);
384
385    r.read_exact(&mut buf2)?;
386    body.extend_from_slice(&buf2);
387    let mv_len = u16::from_le_bytes(buf2) as usize;
388    let mut mv_buf = vec![0u8; mv_len];
389    r.read_exact(&mut mv_buf)?;
390    body.extend_from_slice(&mv_buf);
391
392    let mk = String::from_utf8(mk_buf)
393        .map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8: {}", e)))?;
394    let mv = String::from_utf8(mv_buf)
395        .map_err(|e| HoronError::InvalidFormat(format!("invalid UTF-8: {}", e)))?;
396
397    Ok((mk, mv))
398}
399
400/// Maximum decompressed WAL block size (256 MB safety cap).
401const MAX_BLOCK_DECOMPRESSED: usize = 256 * 1024 * 1024;
402
403/// Write a compressed WAL block.
404///
405/// Block format: `entry_count (u16 LE) + compressed_len (u32 LE) + compressed_data`.
406pub fn write_wal_block<W: Write>(
407    w: &mut W,
408    entries_data: &[u8],
409    entry_count: u16,
410    algo: u8,
411) -> HoronResult<()> {
412    let compressed = crate::compression::compress(entries_data, algo)?;
413    w.write_all(&entry_count.to_le_bytes())?;
414    w.write_all(&(compressed.len() as u32).to_le_bytes())?;
415    w.write_all(&compressed)?;
416    Ok(())
417}
418
419/// Read a single compressed WAL block. Returns `(decompressed_bytes, entry_count)`
420/// or `None` on clean EOF.
421pub fn read_wal_block<R: Read>(
422    r: &mut R,
423    algo: u8,
424) -> HoronResult<Option<(Vec<u8>, u16)>> {
425    // Read block_entry_count (u16)
426    let mut buf2 = [0u8; 2];
427    if r.read_exact(&mut buf2).is_err() {
428        return Ok(None); // clean EOF
429    }
430    let entry_count = u16::from_le_bytes(buf2);
431
432    if entry_count == 0 || entry_count > WAL_BLOCK_SIZE as u16 {
433        return Ok(None); // invalid block header — treat as truncation
434    }
435
436    // Read compressed_len (u32)
437    let mut buf4 = [0u8; 4];
438    if r.read_exact(&mut buf4).is_err() {
439        return Ok(None); // truncated block header
440    }
441    let compressed_len = u32::from_le_bytes(buf4) as usize;
442
443    if compressed_len == 0 || compressed_len > MAX_BLOCK_DECOMPRESSED {
444        return Ok(None); // invalid or absurd size
445    }
446
447    // Read compressed data (allocation bounded by bytes actually present)
448    let compressed = match crate::format::read_bounded_vec(r, compressed_len, "WAL block") {
449        Ok(c) => c,
450        Err(_) => return Ok(None), // truncated compressed data
451    };
452
453    // Decompress
454    let decompressed = crate::compression::decompress(&compressed, algo, MAX_BLOCK_DECOMPRESSED)?;
455    Ok(Some((decompressed, entry_count)))
456}
457
458/// Write WAL section header.
459pub fn write_wal_header<W: Write>(w: &mut W, entry_count: u32, base_seq: u32) -> HoronResult<()> {
460    w.write_all(&entry_count.to_le_bytes())?;
461    w.write_all(&base_seq.to_le_bytes())?;
462    Ok(())
463}
464
465/// Read WAL section header. Returns (entry_count, base_seq).
466pub fn read_wal_header<R: Read>(r: &mut R) -> HoronResult<(u32, u32)> {
467    let mut buf4 = [0u8; 4];
468    r.read_exact(&mut buf4)?;
469    let entry_count = u32::from_le_bytes(buf4);
470    r.read_exact(&mut buf4)?;
471    let base_seq = u32::from_le_bytes(buf4);
472    Ok((entry_count, base_seq))
473}
474
475#[cfg(test)]
476mod tests {
477    use super::*;
478    use std::io::Cursor;
479
480    /// Zero-dim plain layout — the tail-less common case for these tests.
481    fn l0() -> SemLayout {
482        SemLayout::plain(0)
483    }
484
485    fn make_insert_entry(seq: u32, key: &str, data: &[u8]) -> WalEntry {
486        WalEntry {
487            seq,
488            op: OP_INSERT,
489            key: key.to_string(),
490            payload: WalPayload::Insert(NodeEntry {
491                key: key.to_string(),
492                data: data.to_vec(),
493                metadata: vec![],
494                semantic_coords: vec![],
495            }),
496        }
497    }
498
499    #[test]
500    fn test_wal_entry_insert_roundtrip() {
501        let entry = make_insert_entry(1, "/test", b"hello");
502
503        let mut buf = Vec::new();
504        entry.write_to(&mut buf, &l0()).unwrap();
505
506        let mut cursor = Cursor::new(&buf);
507        let parsed = WalEntry::read_from(&mut cursor, &l0()).unwrap().unwrap();
508
509        assert_eq!(parsed.seq, 1);
510        assert_eq!(parsed.op, OP_INSERT);
511        assert_eq!(parsed.key, "/test");
512        match parsed.payload {
513            WalPayload::Insert(e) => assert_eq!(e.data, b"hello"),
514            _ => panic!("expected Insert"),
515        }
516    }
517
518    #[test]
519    fn test_wal_entry_delete_roundtrip() {
520        let entry = WalEntry {
521            seq: 5,
522            op: OP_DELETE,
523            key: "/gone".to_string(),
524            payload: WalPayload::Delete,
525        };
526
527        let mut buf = Vec::new();
528        entry.write_to(&mut buf, &l0()).unwrap();
529
530        let mut cursor = Cursor::new(&buf);
531        let parsed = WalEntry::read_from(&mut cursor, &l0()).unwrap().unwrap();
532
533        assert_eq!(parsed.seq, 5);
534        assert_eq!(parsed.op, OP_DELETE);
535        assert!(matches!(parsed.payload, WalPayload::Delete));
536    }
537
538    #[test]
539    fn test_wal_entry_set_meta_roundtrip() {
540        let entry = WalEntry {
541            seq: 10,
542            op: OP_SET_META,
543            key: "/doc".to_string(),
544            payload: WalPayload::SetMeta {
545                meta_key: "author".to_string(),
546                meta_value: "alice".to_string(),
547            },
548        };
549
550        let mut buf = Vec::new();
551        entry.write_to(&mut buf, &l0()).unwrap();
552
553        let mut cursor = Cursor::new(&buf);
554        let parsed = WalEntry::read_from(&mut cursor, &l0()).unwrap().unwrap();
555
556        assert_eq!(parsed.seq, 10);
557        match parsed.payload {
558            WalPayload::SetMeta { meta_key, meta_value } => {
559                assert_eq!(meta_key, "author");
560                assert_eq!(meta_value, "alice");
561            }
562            _ => panic!("expected SetMeta"),
563        }
564    }
565
566    #[test]
567    fn test_wal_corrupted_crc_returns_none() {
568        let entry = make_insert_entry(1, "/test", b"data");
569        let mut buf = Vec::new();
570        entry.write_to(&mut buf, &l0()).unwrap();
571
572        // Corrupt last byte (part of CRC)
573        let len = buf.len();
574        buf[len - 1] ^= 0xFF;
575
576        let mut cursor = Cursor::new(&buf);
577        let result = WalEntry::read_from(&mut cursor, &l0()).unwrap();
578        assert!(result.is_none(), "corrupted CRC should return None");
579    }
580
581    #[test]
582    fn test_wal_multiple_entries() {
583        let entries = vec![
584            make_insert_entry(1, "/a", b"aaa"),
585            make_insert_entry(2, "/b", b"bbb"),
586            WalEntry {
587                seq: 3,
588                op: OP_DELETE,
589                key: "/a".to_string(),
590                payload: WalPayload::Delete,
591            },
592        ];
593
594        let mut buf = Vec::new();
595        for e in &entries {
596            e.write_to(&mut buf, &l0()).unwrap();
597        }
598
599        let mut cursor = Cursor::new(&buf);
600        let e1 = WalEntry::read_from(&mut cursor, &l0()).unwrap().unwrap();
601        let e2 = WalEntry::read_from(&mut cursor, &l0()).unwrap().unwrap();
602        let e3 = WalEntry::read_from(&mut cursor, &l0()).unwrap().unwrap();
603
604        assert_eq!(e1.seq, 1);
605        assert_eq!(e2.seq, 2);
606        assert_eq!(e3.seq, 3);
607        assert_eq!(e3.op, OP_DELETE);
608
609        // No more entries
610        let e4 = WalEntry::read_from(&mut cursor, &l0()).unwrap();
611        assert!(e4.is_none());
612    }
613
614    #[test]
615    fn test_wal_header_roundtrip() {
616        let mut buf = Vec::new();
617        write_wal_header(&mut buf, 42, 100).unwrap();
618
619        let mut cursor = Cursor::new(&buf);
620        let (count, base) = read_wal_header(&mut cursor).unwrap();
621        assert_eq!(count, 42);
622        assert_eq!(base, 100);
623    }
624
625    #[test]
626    fn test_wal_block_roundtrip() {
627        let entries: Vec<WalEntry> = (1..=10)
628            .map(|i| make_insert_entry(i, &format!("/n{}", i), b"data"))
629            .collect();
630
631        // Serialize entries into raw bytes
632        let mut raw = Vec::new();
633        for e in &entries {
634            e.write_to(&mut raw, &l0()).unwrap();
635        }
636
637        // Write as compressed block
638        let mut block_buf = Vec::new();
639        write_wal_block(&mut block_buf, &raw, entries.len() as u16, ALGO_ZSTD).unwrap();
640
641        // Read back
642        let mut cursor = Cursor::new(&block_buf);
643        let (decompressed, count) = read_wal_block(&mut cursor, ALGO_ZSTD).unwrap().unwrap();
644        assert_eq!(count, 10);
645        assert_eq!(decompressed, raw);
646
647        // Parse individual entries from decompressed block
648        let mut inner = Cursor::new(&decompressed);
649        for i in 1..=10u32 {
650            let parsed = WalEntry::read_from(&mut inner, &l0()).unwrap().unwrap();
651            assert_eq!(parsed.seq, i);
652        }
653        assert!(WalEntry::read_from(&mut inner, &l0()).unwrap().is_none());
654    }
655
656    #[test]
657    fn test_wal_block_single_entry() {
658        let entry = make_insert_entry(1, "/single", b"one");
659        let mut raw = Vec::new();
660        entry.write_to(&mut raw, &l0()).unwrap();
661
662        let mut block_buf = Vec::new();
663        write_wal_block(&mut block_buf, &raw, 1, ALGO_ZSTD).unwrap();
664
665        let mut cursor = Cursor::new(&block_buf);
666        let (decompressed, count) = read_wal_block(&mut cursor, ALGO_ZSTD).unwrap().unwrap();
667        assert_eq!(count, 1);
668        assert_eq!(decompressed, raw);
669    }
670
671    #[test]
672    fn test_wal_block_full_64() {
673        let entries: Vec<WalEntry> = (1..=64)
674            .map(|i| make_insert_entry(i, &format!("/node_{}", i), b"payload"))
675            .collect();
676
677        let mut raw = Vec::new();
678        for e in &entries {
679            e.write_to(&mut raw, &l0()).unwrap();
680        }
681
682        let mut block_buf = Vec::new();
683        write_wal_block(&mut block_buf, &raw, 64, ALGO_ZSTD).unwrap();
684
685        let mut cursor = Cursor::new(&block_buf);
686        let (decompressed, count) = read_wal_block(&mut cursor, ALGO_ZSTD).unwrap().unwrap();
687        assert_eq!(count, 64);
688        assert_eq!(decompressed, raw);
689    }
690
691    #[test]
692    fn test_wal_block_eof_returns_none() {
693        let cursor = Cursor::new(Vec::<u8>::new());
694        let result = read_wal_block(&mut cursor.clone(), ALGO_ZSTD).unwrap();
695        assert!(result.is_none());
696    }
697
698    #[test]
699    fn test_wal_block_truncated_header() {
700        // Only 1 byte — partial block_entry_count
701        let mut cursor = Cursor::new(vec![0x01]);
702        let result = read_wal_block(&mut cursor, ALGO_ZSTD).unwrap();
703        assert!(result.is_none());
704    }
705
706    #[test]
707    fn test_wal_block_multiple_blocks() {
708        let mut all_blocks = Vec::new();
709        let mut expected_entries = Vec::new();
710
711        // Write 3 blocks: 64 + 64 + 12
712        for block_idx in 0..3u32 {
713            let count = if block_idx < 2 { 64 } else { 12 };
714            let entries: Vec<WalEntry> = (0..count)
715                .map(|i| {
716                    let seq = block_idx * 64 + i + 1;
717                    make_insert_entry(seq, &format!("/b{}/n{}", block_idx, i), b"x")
718                })
719                .collect();
720
721            let mut raw = Vec::new();
722            for e in &entries {
723                e.write_to(&mut raw, &l0()).unwrap();
724            }
725
726            write_wal_block(&mut all_blocks, &raw, count as u16, ALGO_ZSTD).unwrap();
727            expected_entries.extend(entries);
728        }
729
730        // Read all 3 blocks back
731        let mut cursor = Cursor::new(&all_blocks);
732        let mut read_count = 0u32;
733        for block_idx in 0..3 {
734            let (decompressed, count) = read_wal_block(&mut cursor, ALGO_ZSTD).unwrap().unwrap();
735            let expected_count: u16 = if block_idx < 2 { 64 } else { 12 };
736            assert_eq!(count, expected_count);
737
738            let mut inner = Cursor::new(&decompressed);
739            for _ in 0..count {
740                let parsed = WalEntry::read_from(&mut inner, &l0()).unwrap().unwrap();
741                assert_eq!(parsed.seq, read_count + 1);
742                read_count += 1;
743            }
744        }
745        assert_eq!(read_count, 140);
746
747        // No more blocks
748        assert!(read_wal_block(&mut cursor, ALGO_ZSTD).unwrap().is_none());
749    }
750}
horon/wal.rs

horon/
wal.rs
