//! Write-Ahead Log (WAL) for edgehdf5 agent memory. //! //! Binary WAL format alongside the main .h5 file enables fast append-only //! writes without rewriting the entire HDF5 file on every save. use std::fs::{File, OpenOptions}; use std::io::{Read, Seek, SeekFrom, Write}; use std::path::{Path, PathBuf}; use clawhdf5_format::checksum::crc32; use crate::MemoryError; const WAL_MAGIC: [u8; 4] = [0x45, 0x48, 0x57, 0x4C]; // "EHWL" /// Bytes before the first entry: [`WAL_MAGIC`] (4) + version (1) + entry /// count (4). Named so the offset arithmetic in `open()` — which decides /// where an append lands, and therefore whether it is replayable — reads as /// a header length rather than a bare 9. const WAL_HEADER_LEN: u64 = WAL_MAGIC.len() as u64 + 1 + 4; /// Current WAL format version: every entry's CRC32 trailer is computed over /// its own bytes *chained with the previous entry's stored CRC* /// (`crc32(entry_bytes ++ prev_crc.to_le_bytes())`, seeded with 0 for the /// first entry after a truncation). A per-entry CRC alone only detects a /// bit-flip within that entry; chaining additionally detects entries being /// reordered, duplicated, or spliced (e.g. a Tombstone moved before/after /// its target Save) — the moved/inserted entry's stored CRC was computed /// against a different predecessor than the one now in front of it on disk, /// so the chain breaks at that point and replay stops there. const WAL_VERSION: u8 = 3; /// The previous WAL format version: still a CRC32 per entry (so a bit-flip /// within one entry is caught), but not chained to the previous entry's CRC /// (so reordering/splicing whole entries is not detected). Written by /// versions of this crate before the chaining hardening. Fully supported for /// reading via [`WalFile::read_entries`] — not restricted like /// [`WAL_VERSION_LEGACY_NO_CRC`], since it still verifies each entry /// individually. `WalFile::open` migrates it to [`WAL_VERSION`] by /// recreating the file fresh, the same as the legacy-no-CRC migration below. const WAL_VERSION_CRC_UNCHAINED: u8 = 2; /// The oldest WAL version this crate still knows how to *read*: no /// per-entry CRC trailer at all, so a bit-flip anywhere is silently /// accepted. Written by versions of this crate before the CRC32 hardening. /// Because of that — unlike [`WAL_VERSION_CRC_UNCHAINED`] — this version is /// deliberately *not* reachable through the public [`WalFile::read_entries`] /// API; only [`WalFile::read_entries_for_migration`] (used exclusively by /// `HDF5Memory::open`'s one-time migration path) will parse it. Flipping a /// version byte from 2/3 down to 1 no longer silently downgrades a file to /// the fully-unverified parser for an arbitrary caller. /// /// `WalFile::open` migrates a legacy file to [`WAL_VERSION`] by recreating /// it fresh — safe because every real call site reads existing entries via /// [`WalFile::read_entries_for_migration`] before calling `open` (see /// `HDF5Memory::open`), so no data is lost. const WAL_VERSION_LEGACY_NO_CRC: u8 = 1; /// Upper bound on a single length-prefixed WAL field (string bytes, or /// embedding element count), to reject a corrupted/truncated WAL length /// claim before allocating a large buffer for it. const MAX_WAL_FIELD_LEN: usize = 64 * 1024 * 1024; #[repr(u8)] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum WalEntryType { Save = 0x01, Tombstone = 0x02, ActivationUpdate = 0x03, } impl WalEntryType { fn from_u8(v: u8) -> Option { match v { 0x01 => Some(Self::Save), 0x02 => Some(Self::Tombstone), 0x03 => Some(Self::ActivationUpdate), _ => None, } } } #[derive(Debug, Clone)] pub struct WalEntry { pub entry_type: WalEntryType, pub timestamp: f64, pub chunk: String, pub embedding: Vec, pub source_channel: String, pub session_id: String, pub tags: String, /// For tombstone entries: the index of the entry to delete. pub tombstone_index: Option, } /// How many entries to accumulate before updating the header entry_count. /// /// The header count is only needed for replay; `read_entries` already handles /// stale counts by reading until EOF. Updating every N entries rather than /// every entry eliminates 3 lseek() + 1 write() per entry — see arXiv:2507.13062. const GROUP_COMMIT_SIZE: u32 = 8; #[derive(Debug)] pub struct WalFile { path: PathBuf, file: Option, entry_count: u32, /// Entries written since the last header count update. pending_header_sync: u32, /// CRC32 chain state: the previous entry's stored CRC (0 if this file /// has no entries yet), folded into the next entry's CRC computation. /// Reset to 0 by `truncate()`/`create_fresh_wal_file`, and re-derived by /// scanning existing entries when `open()` attaches to a non-empty file. running_crc: u32, } impl WalFile { /// Open or create a WAL file. If it exists, read the header and entry count. /// /// A pre-chaining WAL file ([`WAL_VERSION_CRC_UNCHAINED`] or /// [`WAL_VERSION_LEGACY_NO_CRC`]) is migrated to the current format by /// recreating it fresh. Callers that need an existing file's entries must /// call [`WalFile::read_entries`] (or, for a legacy-no-CRC file, /// [`WalFile::read_entries_for_migration`]) first, before calling `open`. pub fn open(path: &Path) -> Result { if path.exists() { // Read existing header let mut f = OpenOptions::new() .read(true) .write(true) .append(false) .open(path)?; let mut magic = [0u8; 4]; f.read_exact(&mut magic)?; if magic != WAL_MAGIC { return Err(MemoryError::Schema("invalid WAL magic bytes".into())); } let mut ver = [0u8; 1]; f.read_exact(&mut ver)?; match ver[0] { WAL_VERSION => { let mut count_buf = [0u8; 4]; f.read_exact(&mut count_buf)?; let header_count = u32::from_le_bytes(count_buf); // Scan any existing entries to resume the CRC chain // correctly for further appends (the header's count may // be stale from deferred group-commit sync, same // tolerance `read_entries` already has, so the scanned // count is also the more accurate of the two). let (entries, running_crc, verified_bytes) = read_chained_entries(&mut f, 0); let entry_count = if entries.is_empty() { header_count } else { entries.len() as u32 }; // Position the append at the end of the VERIFIED prefix, // and drop anything after it. // // This used to `seek(End(0))`, which appends PAST a torn // tail — the ordinary outcome of a crash mid-append. The // new entry is then chained to the last good entry, but // sits on disk behind the garbage: // // [1..N verified][torn bytes][N+1 chained to N] // // Replay stops at the torn bytes, so N+1 is unreachable // FOREVER even though its `append` returned Ok and synced. // That is silent data loss in the one situation a WAL // exists for. Truncating to the verified end is the // standard recovery: the torn tail was never acknowledged // to any caller, so discarding it loses nothing, and the // chain then continues from a byte offset that matches // `running_crc`. let verified_end = WAL_HEADER_LEN + verified_bytes; let file_len = f.metadata()?.len(); if file_len > verified_end { eprintln!( "clawhdf5-agent: WAL {} has {} unverifiable byte(s) after entry {}; \ discarding them so appends stay replayable", path.display(), file_len - verified_end, entries.len() ); f.set_len(verified_end)?; } f.seek(SeekFrom::Start(verified_end))?; Ok(Self { path: path.to_path_buf(), file: Some(f), entry_count, pending_header_sync: 0, running_crc, }) } WAL_VERSION_CRC_UNCHAINED | WAL_VERSION_LEGACY_NO_CRC => { drop(f); let f = create_fresh_wal_file(path)?; Ok(Self { path: path.to_path_buf(), file: Some(f), entry_count: 0, pending_header_sync: 0, running_crc: 0, }) } v => Err(MemoryError::Schema(format!("unsupported WAL version {v}"))), } } else { let f = create_fresh_wal_file(path)?; Ok(Self { path: path.to_path_buf(), file: Some(f), entry_count: 0, pending_header_sync: 0, running_crc: 0, }) } } /// Append a save entry to the WAL. /// /// Serializes the entry into a single buffer before writing to minimize /// syscall count (1 write() vs ~8 previously). The header entry_count is /// updated every GROUP_COMMIT_SIZE entries rather than on every write, /// eliminating 3 lseek() + 1 write() per entry (arXiv:2507.13062). /// /// Crash safety: `read_entries` reads until EOF and handles stale header /// counts, so deferred header updates do not compromise recovery. pub fn append_save(&mut self, entry: &WalEntry) -> Result<(), MemoryError> { let emb_len = entry.embedding.len(); let mut buf = Vec::with_capacity( 1 + 8 + // type + timestamp 4 + entry.chunk.len() + 4 + emb_len * 4 + 4 + entry.source_channel.len() + 4 + entry.session_id.len() + 4 + entry.tags.len(), ); buf.push(WalEntryType::Save as u8); buf.extend_from_slice(&entry.timestamp.to_le_bytes()); serialize_str(&mut buf, &entry.chunk); buf.extend_from_slice(&(emb_len as u32).to_le_bytes()); for &val in &entry.embedding { buf.extend_from_slice(&val.to_le_bytes()); } serialize_str(&mut buf, &entry.source_channel); serialize_str(&mut buf, &entry.session_id); serialize_str(&mut buf, &entry.tags); // Chain this entry's CRC to the previous one's so reordering/ // splicing entries (not just flipping a bit within one) is detected // on replay — see WAL_VERSION's doc comment. let crc = chained_crc(&buf, self.running_crc); buf.extend_from_slice(&crc.to_le_bytes()); let f = self .file .as_mut() .ok_or_else(|| MemoryError::Io(std::io::Error::other("WAL file not open")))?; f.write_all(&buf)?; self.running_crc = crc; self.entry_count += 1; self.pending_header_sync += 1; if self.pending_header_sync >= GROUP_COMMIT_SIZE { self.write_entry_count()?; } Ok(()) } /// Append a tombstone entry (deletion). pub fn append_tombstone(&mut self, index: usize, timestamp: f64) -> Result<(), MemoryError> { let mut buf = [0u8; 1 + 8 + 4 + 4]; // type + timestamp + index + crc32 buf[0] = WalEntryType::Tombstone as u8; buf[1..9].copy_from_slice(×tamp.to_le_bytes()); buf[9..13].copy_from_slice(&(index as u32).to_le_bytes()); let crc = chained_crc(&buf[..13], self.running_crc); buf[13..17].copy_from_slice(&crc.to_le_bytes()); let f = self .file .as_mut() .ok_or_else(|| MemoryError::Io(std::io::Error::other("WAL file not open")))?; f.write_all(&buf)?; self.running_crc = crc; self.entry_count += 1; self.pending_header_sync += 1; if self.pending_header_sync >= GROUP_COMMIT_SIZE { self.write_entry_count()?; } Ok(()) } /// Read all entries from the WAL (for replay on open). /// /// Reads until EOF — the header `entry_count` is used only for pre-allocation /// (and may be stale if written with deferred group-commit updates). This /// tolerates both truncated files (crash mid-write) and stale header counts /// (crash before the next group-commit header sync). On a `WAL_VERSION` /// file, a broken CRC chain (bit-flip, or an entry reordered/duplicated/ /// spliced in) is treated the same way — replay stops there rather than /// accepting corrupted or tampered data. `WAL_VERSION_CRC_UNCHAINED` /// files are read the same way minus the chain check (each entry's own /// CRC is still verified). /// /// Does **not** read [`WAL_VERSION_LEGACY_NO_CRC`] files — that format has /// no integrity verification at all, so it's only reachable through /// [`WalFile::read_entries_for_migration`], used exclusively by /// `HDF5Memory::open`'s one-time migration path. Calling this on a /// legacy-no-CRC file returns a typed error instead of silently /// downgrading to the unverified parser. pub fn read_entries(path: &Path) -> Result, MemoryError> { Self::read_entries_impl(path, false) } /// Like [`WalFile::read_entries`], but also accepts /// [`WAL_VERSION_LEGACY_NO_CRC`] files (no per-entry integrity check at /// all). Restricted to `pub(crate)` and named accordingly: the only /// legitimate caller is `HDF5Memory::open`'s one-time migration of a /// pre-CRC WAL file, which immediately recreates it in the current /// format afterward. Do not use this for anything else. pub(crate) fn read_entries_for_migration(path: &Path) -> Result, MemoryError> { Self::read_entries_impl(path, true) } fn read_entries_impl( path: &Path, allow_legacy_no_crc: bool, ) -> Result, MemoryError> { if !path.exists() { return Ok(Vec::new()); } let mut f = File::open(path)?; // Read header let mut header = [0u8; 9]; f.read_exact(&mut header)?; if header[0..4] != WAL_MAGIC { return Err(MemoryError::Schema("invalid WAL magic bytes".into())); } // entry_count is a pre-allocation hint only — we read until EOF. let entry_count_hint = u32::from_le_bytes([header[5], header[6], header[7], header[8]]); match header[4] { WAL_VERSION => { let (entries, _final_crc, _verified_bytes) = read_chained_entries(&mut f, 0); Ok(entries) } WAL_VERSION_CRC_UNCHAINED => { let mut entries = Vec::with_capacity(entry_count_hint as usize); loop { let raw_and_result = { let mut tee = TeeReader::new(&mut f); let result = read_one_entry(&mut tee); (tee.into_buf(), result) }; let (raw, result) = raw_and_result; let entry_opt = match result { Err(()) => break, Ok(v) => v, }; let mut crc_buf = [0u8; 4]; if f.read_exact(&mut crc_buf).is_err() { break; } let stored_crc = u32::from_le_bytes(crc_buf); if crc32(&raw) != stored_crc { // Corruption detected — stop replay here, same as a // clean truncation/EOF, rather than accepting the bad // entry. break; } if let Some(entry) = entry_opt { entries.push(entry); } } Ok(entries) } WAL_VERSION_LEGACY_NO_CRC if allow_legacy_no_crc => { let mut entries = Vec::with_capacity(entry_count_hint as usize); loop { match read_one_entry(&mut f) { Err(()) => break, Ok(Some(entry)) => entries.push(entry), Ok(None) => {} } } Ok(entries) } WAL_VERSION_LEGACY_NO_CRC => Err(MemoryError::Schema( "WAL file is in the legacy no-CRC format (version 1), which read_entries() no \ longer accepts — it has no per-entry integrity verification. Only the one-time \ migration path (WalFile::open) can read and upgrade it." .into(), )), v => Err(MemoryError::Schema(format!("unsupported WAL version {v}"))), } } /// Truncate the WAL (after merge into .h5). pub fn truncate(&mut self) -> Result<(), MemoryError> { // Close existing handle and recreate self.file = None; let f = create_fresh_wal_file(&self.path)?; self.file = Some(f); self.entry_count = 0; self.pending_header_sync = 0; self.running_crc = 0; Ok(()) } /// Number of pending entries. pub fn pending_count(&self) -> u32 { self.entry_count } /// Is the WAL empty? pub fn is_empty(&self) -> bool { self.entry_count == 0 } /// Update the entry_count in the header (seek to offset 5, write u32 LE). fn write_entry_count(&mut self) -> Result<(), MemoryError> { let f = self .file .as_mut() .ok_or_else(|| MemoryError::Io(std::io::Error::other("WAL file not open")))?; let pos = f.stream_position()?; f.seek(SeekFrom::Start(5))?; f.write_all(&self.entry_count.to_le_bytes())?; f.seek(SeekFrom::Start(pos))?; self.pending_header_sync = 0; Ok(()) } } /// Replay WAL entries into a MemoryCache. pub fn replay_into_cache(entries: &[WalEntry], cache: &mut crate::cache::MemoryCache) { for entry in entries { match entry.entry_type { WalEntryType::Save => { cache.push( entry.chunk.clone(), entry.embedding.clone(), entry.source_channel.clone(), entry.timestamp, entry.session_id.clone(), entry.tags.clone(), ); } WalEntryType::Tombstone => { if let Some(idx) = entry.tombstone_index { cache.mark_deleted(idx); } } WalEntryType::ActivationUpdate => {} } } } // --- Binary helpers --- /// Serialize a length-prefixed string into an in-memory buffer (zero syscalls). fn serialize_str(buf: &mut Vec, s: &str) { let bytes = s.as_bytes(); buf.extend_from_slice(&(bytes.len() as u32).to_le_bytes()); buf.extend_from_slice(bytes); } fn read_len_prefixed_str(f: &mut R) -> Result { let mut len_buf = [0u8; 4]; f.read_exact(&mut len_buf)?; let len = u32::from_le_bytes(len_buf) as usize; if len > MAX_WAL_FIELD_LEN { return Err(MemoryError::Schema(format!( "WAL string field length {len} exceeds max {MAX_WAL_FIELD_LEN}" ))); } let mut buf = vec![0u8; len]; f.read_exact(&mut buf)?; String::from_utf8(buf).map_err(|e| MemoryError::Schema(format!("invalid UTF-8 in WAL: {e}"))) } fn read_embedding(f: &mut R) -> Result, MemoryError> { let mut len_buf = [0u8; 4]; f.read_exact(&mut len_buf)?; let count = u32::from_le_bytes(len_buf) as usize; if count > MAX_WAL_FIELD_LEN / 4 { return Err(MemoryError::Schema(format!( "WAL embedding element count {count} exceeds max {}", MAX_WAL_FIELD_LEN / 4 ))); } let mut vals = Vec::with_capacity(count); for _ in 0..count { let mut val_buf = [0u8; 4]; f.read_exact(&mut val_buf)?; vals.push(f32::from_le_bytes(val_buf)); } Ok(vals) } /// Compute the CRC32 trailer for a `WAL_VERSION` entry, chaining in the /// previous entry's stored CRC (0 for the first entry after a truncation). fn chained_crc(entry_bytes: &[u8], prev_crc: u32) -> u32 { let mut chained = Vec::with_capacity(entry_bytes.len() + 4); chained.extend_from_slice(entry_bytes); chained.extend_from_slice(&prev_crc.to_le_bytes()); crc32(&chained) } /// Read and verify all entries from a `WAL_VERSION` (chained-CRC) stream /// starting at the reader's current position, given the chain state to /// resume from (0 for a stream starting at the beginning of a fresh WAL). /// /// Returns the parsed entries, the final running CRC — the chain state to /// continue from for further appends — and the number of BYTES consumed by /// those verified entries. Stops (without erroring) at the first entry that /// fails to parse or whose stored CRC doesn't match the expected chain value /// — a bit-flip, truncation/EOF, or an entry having been /// reordered/duplicated/spliced all produce a chain mismatch at that point, /// and are all handled the same way: replay stops there. /// /// The byte count is what lets `open()` position an append at the end of the /// VERIFIED prefix rather than at end-of-file. Appending past a torn tail /// writes entries that replay can never reach — see `open`. fn read_chained_entries(f: &mut R, start_crc: u32) -> (Vec, u32, u64) { let mut entries = Vec::new(); let mut running_crc = start_crc; let mut verified_bytes: u64 = 0; loop { let raw_and_result = { let mut tee = TeeReader::new(f); let result = read_one_entry(&mut tee); (tee.into_buf(), result) }; let (raw, result) = raw_and_result; let entry_opt = match result { Err(()) => break, Ok(v) => v, }; let mut crc_buf = [0u8; 4]; if f.read_exact(&mut crc_buf).is_err() { break; } let stored_crc = u32::from_le_bytes(crc_buf); if chained_crc(&raw, running_crc) != stored_crc { break; } running_crc = stored_crc; // Only counted once the entry AND its CRC trailer verified, so the // offset always points just past a complete, checked entry. verified_bytes += raw.len() as u64 + crc_buf.len() as u64; if let Some(entry) = entry_opt { entries.push(entry); } } (entries, running_crc, verified_bytes) } /// Create a fresh WAL file at `path` with the current-version header, /// truncating/overwriting anything already there. fn create_fresh_wal_file(path: &Path) -> Result { let mut f = File::create(path)?; f.write_all(&WAL_MAGIC)?; f.write_all(&[WAL_VERSION])?; f.write_all(&0u32.to_le_bytes())?; f.flush()?; Ok(f) } /// Wraps a [`Read`]er, accumulating every byte actually consumed (including /// via `read_exact`, which is implemented in terms of `read`) into an /// internal buffer — used to capture a WAL entry's raw bytes for CRC32 /// verification without needing to know its length up front. struct TeeReader<'a, R: Read> { inner: &'a mut R, buf: Vec, } impl<'a, R: Read> TeeReader<'a, R> { fn new(inner: &'a mut R) -> Self { Self { inner, buf: Vec::new(), } } fn into_buf(self) -> Vec { self.buf } } impl Read for TeeReader<'_, R> { fn read(&mut self, out: &mut [u8]) -> std::io::Result { let n = self.inner.read(out)?; self.buf.extend_from_slice(&out[..n]); Ok(n) } } /// Read one WAL entry (type + timestamp + type-specific payload) from `r`. /// /// Returns `Ok(None)` for entry types with no representable `WalEntry` (only /// `ActivationUpdate`, reserved for future use). Returns `Err(())` on any /// read failure or unrecognized entry type — the caller treats this the same /// as a clean end-of-log (crash-mid-write tolerance). fn read_one_entry(r: &mut R) -> Result, ()> { let mut type_buf = [0u8; 1]; r.read_exact(&mut type_buf).map_err(|_| ())?; let entry_type = WalEntryType::from_u8(type_buf[0]).ok_or(())?; let mut ts_buf = [0u8; 8]; r.read_exact(&mut ts_buf).map_err(|_| ())?; let timestamp = f64::from_le_bytes(ts_buf); match entry_type { WalEntryType::Save => { let chunk = read_len_prefixed_str(r).map_err(|_| ())?; let embedding = read_embedding(r).map_err(|_| ())?; let source_channel = read_len_prefixed_str(r).map_err(|_| ())?; let session_id = read_len_prefixed_str(r).map_err(|_| ())?; let tags = read_len_prefixed_str(r).map_err(|_| ())?; Ok(Some(WalEntry { entry_type, timestamp, chunk, embedding, source_channel, session_id, tags, tombstone_index: None, })) } WalEntryType::Tombstone => { let mut idx_buf = [0u8; 4]; r.read_exact(&mut idx_buf).map_err(|_| ())?; let idx = u32::from_le_bytes(idx_buf) as usize; Ok(Some(WalEntry { entry_type, timestamp, chunk: String::new(), embedding: Vec::new(), source_channel: String::new(), session_id: String::new(), tags: String::new(), tombstone_index: Some(idx), })) } WalEntryType::ActivationUpdate => Ok(None), } } // --- Tests --- #[cfg(test)] mod tests { use super::*; use tempfile::TempDir; fn make_wal_entry(chunk: &str, embedding: &[f32]) -> WalEntry { WalEntry { entry_type: WalEntryType::Save, timestamp: 1234567.89, chunk: chunk.to_string(), embedding: embedding.to_vec(), source_channel: "test-channel".to_string(), session_id: "sess-001".to_string(), tags: "tag1,tag2".to_string(), tombstone_index: None, } } #[test] fn test_wal_create_and_header() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("test.h5.wal"); let wal = WalFile::open(&wal_path).unwrap(); assert_eq!(wal.pending_count(), 0); assert!(wal.is_empty()); drop(wal); // Verify raw bytes on disk let bytes = std::fs::read(&wal_path).unwrap(); assert_eq!(&bytes[0..4], &WAL_MAGIC); assert_eq!(bytes[4], WAL_VERSION); assert_eq!(&bytes[5..9], &0u32.to_le_bytes()); } #[test] fn test_wal_append_and_read() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("test.h5.wal"); { let mut wal = WalFile::open(&wal_path).unwrap(); wal.append_save(&make_wal_entry("first", &[1.0, 2.0])) .unwrap(); wal.append_save(&make_wal_entry("second", &[3.0, 4.0])) .unwrap(); wal.append_save(&make_wal_entry("third", &[5.0, 6.0])) .unwrap(); assert_eq!(wal.pending_count(), 3); } let entries = WalFile::read_entries(&wal_path).unwrap(); assert_eq!(entries.len(), 3); assert_eq!(entries[0].chunk, "first"); assert_eq!(entries[0].embedding, vec![1.0, 2.0]); assert_eq!(entries[1].chunk, "second"); assert_eq!(entries[2].chunk, "third"); assert_eq!(entries[2].embedding, vec![5.0, 6.0]); } #[test] fn read_len_prefixed_str_rejects_oversized_len_claim() { let dir = TempDir::new().unwrap(); let path = dir.path().join("oversized_str.bin"); { let mut f = File::create(&path).unwrap(); // Claim a length far beyond MAX_WAL_FIELD_LEN; no payload follows. f.write_all(&(u32::MAX).to_le_bytes()).unwrap(); } let mut f = File::open(&path).unwrap(); let result = read_len_prefixed_str(&mut f); assert!( matches!(result, Err(MemoryError::Schema(_))), "expected a clean Schema error, got {result:?}" ); } #[test] fn read_embedding_rejects_oversized_count_claim() { let dir = TempDir::new().unwrap(); let path = dir.path().join("oversized_embedding.bin"); { let mut f = File::create(&path).unwrap(); // Claim a count far beyond MAX_WAL_FIELD_LEN / 4; no payload follows. f.write_all(&(u32::MAX).to_le_bytes()).unwrap(); } let mut f = File::open(&path).unwrap(); let result = read_embedding(&mut f); assert!( matches!(result, Err(MemoryError::Schema(_))), "expected a clean Schema error, got {result:?}" ); } #[test] fn test_wal_truncate() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("test.h5.wal"); let mut wal = WalFile::open(&wal_path).unwrap(); for i in 0..5 { wal.append_save(&make_wal_entry(&format!("entry {i}"), &[i as f32])) .unwrap(); } assert_eq!(wal.pending_count(), 5); wal.truncate().unwrap(); assert_eq!(wal.pending_count(), 0); assert!(wal.is_empty()); let entries = WalFile::read_entries(&wal_path).unwrap(); assert!(entries.is_empty()); } #[test] fn test_wal_append_tombstone() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("test.h5.wal"); { let mut wal = WalFile::open(&wal_path).unwrap(); wal.append_tombstone(42, 9999.0).unwrap(); assert_eq!(wal.pending_count(), 1); } let entries = WalFile::read_entries(&wal_path).unwrap(); assert_eq!(entries.len(), 1); assert_eq!(entries[0].entry_type, WalEntryType::Tombstone); assert_eq!(entries[0].tombstone_index, Some(42)); assert!((entries[0].timestamp - 9999.0).abs() < 1e-6); } #[test] fn test_wal_binary_roundtrip() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("test.h5.wal"); let unicode_chunk = "Hello 世界! 🌍 émojis & ünïcödé"; let embedding = vec![0.1, -0.2, 3.4567, f32::MAX, f32::MIN_POSITIVE]; { let mut wal = WalFile::open(&wal_path).unwrap(); let entry = WalEntry { entry_type: WalEntryType::Save, timestamp: std::f64::consts::PI, chunk: unicode_chunk.to_string(), embedding: embedding.clone(), source_channel: "channel/with/slashes".to_string(), session_id: "sess-öö-123".to_string(), tags: "α,β,γ".to_string(), tombstone_index: None, }; wal.append_save(&entry).unwrap(); } let entries = WalFile::read_entries(&wal_path).unwrap(); assert_eq!(entries.len(), 1); let e = &entries[0]; assert_eq!(e.entry_type, WalEntryType::Save); assert!((e.timestamp - std::f64::consts::PI).abs() < 1e-15); assert_eq!(e.chunk, unicode_chunk); assert_eq!(e.embedding, embedding); assert_eq!(e.source_channel, "channel/with/slashes"); assert_eq!(e.session_id, "sess-öö-123"); assert_eq!(e.tags, "α,β,γ"); } #[test] fn test_wal_empty_on_create() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("test.h5.wal"); let wal = WalFile::open(&wal_path).unwrap(); assert_eq!(wal.pending_count(), 0); assert!(wal.is_empty()); } // --- Integration tests (WAL + HDF5Memory) --- use crate::{AgentMemory, HDF5Memory, MemoryConfig, MemoryEntry}; fn make_config(dir: &TempDir) -> MemoryConfig { let mut config = MemoryConfig::new(dir.path().join("test.h5"), "agent-test", 4); config.wal_enabled = true; config } fn make_entry(chunk: &str, embedding: &[f32]) -> MemoryEntry { MemoryEntry { chunk: chunk.to_string(), embedding: embedding.to_vec(), source_channel: "test".to_string(), timestamp: 1000000.0, session_id: "session-1".to_string(), tags: "tag1,tag2".to_string(), } } #[test] fn test_save_with_wal() { let dir = TempDir::new().unwrap(); let config = make_config(&dir); let h5_path = config.path.clone(); let mut mem = HDF5Memory::create(config).unwrap(); // Get initial .h5 size (empty file) let initial_size = std::fs::metadata(&h5_path).unwrap().len(); mem.save(make_entry("a", &[1.0, 0.0, 0.0, 0.0])).unwrap(); mem.save(make_entry("b", &[0.0, 1.0, 0.0, 0.0])).unwrap(); mem.save(make_entry("c", &[0.0, 0.0, 1.0, 0.0])).unwrap(); // Cache has 3 entries assert_eq!(mem.count(), 3); // .h5 file should NOT have been updated (still initial size) let after_size = std::fs::metadata(&h5_path).unwrap().len(); assert_eq!( initial_size, after_size, ".h5 should not grow with WAL enabled" ); // .wal file should exist let wal_path = h5_path.with_extension("h5.wal"); assert!(wal_path.exists(), ".wal file should exist"); assert_eq!(mem.wal_pending_count(), 3); } #[test] fn test_wal_auto_merge() { let dir = TempDir::new().unwrap(); let mut config = make_config(&dir); config.wal_max_entries = 5; let h5_path = config.path.clone(); let mut mem = HDF5Memory::create(config).unwrap(); // Save 5 entries (at threshold but not over) for i in 0..5 { mem.save(make_entry( &format!("entry {i}"), &[i as f32, 0.0, 0.0, 0.0], )) .unwrap(); } // WAL should still have 5 pending (not yet merged, threshold is >=) assert_eq!(mem.wal_pending_count(), 5); // 6th entry triggers auto-merge (pending > wal_max_entries) mem.save(make_entry("entry 5", &[5.0, 0.0, 0.0, 0.0])) .unwrap(); // After auto-merge: WAL should be empty, cache still has all entries assert_eq!(mem.wal_pending_count(), 0); assert_eq!(mem.count(), 6); // WAL file should be truncated (only header) let wal_path = h5_path.with_extension("h5.wal"); let entries = WalFile::read_entries(&wal_path).unwrap(); assert!(entries.is_empty(), "WAL should be empty after auto-merge"); } #[test] fn test_wal_flush_explicit() { let dir = TempDir::new().unwrap(); let config = make_config(&dir); let h5_path = config.path.clone(); let mut mem = HDF5Memory::create(config).unwrap(); mem.save(make_entry("a", &[1.0, 0.0, 0.0, 0.0])).unwrap(); mem.save(make_entry("b", &[0.0, 1.0, 0.0, 0.0])).unwrap(); mem.save(make_entry("c", &[0.0, 0.0, 1.0, 0.0])).unwrap(); assert_eq!(mem.wal_pending_count(), 3); mem.flush_wal().unwrap(); // WAL should be empty after explicit flush assert_eq!(mem.wal_pending_count(), 0); // Cache should still have 3 assert_eq!(mem.count(), 3); // WAL file on disk should be empty let wal_path = h5_path.with_extension("h5.wal"); let entries = WalFile::read_entries(&wal_path).unwrap(); assert!(entries.is_empty()); } #[test] fn test_wal_replay_on_open() { // Test WAL replay using read_entries + replay_into_cache directly, // since the HDF5 read path is independent of WAL functionality. let dir = TempDir::new().unwrap(); let config = make_config(&dir); let h5_path = config.path.clone(); { let mut mem = HDF5Memory::create(config).unwrap(); mem.save(make_entry("replay-a", &[1.0, 0.0, 0.0, 0.0])) .unwrap(); mem.save(make_entry("replay-b", &[0.0, 1.0, 0.0, 0.0])) .unwrap(); mem.save(make_entry("replay-c", &[0.0, 0.0, 1.0, 0.0])) .unwrap(); assert_eq!(mem.wal_pending_count(), 3); // Drop without flushing — WAL has 3 entries } // Verify WAL file has the entries let wal_path = h5_path.with_extension("h5.wal"); assert!(wal_path.exists()); let entries = WalFile::read_entries(&wal_path).unwrap(); assert_eq!(entries.len(), 3); assert_eq!(entries[0].chunk, "replay-a"); assert_eq!(entries[1].chunk, "replay-b"); assert_eq!(entries[2].chunk, "replay-c"); // Replay into a fresh cache (simulates what open() does) let mut cache = crate::cache::MemoryCache::new(4); super::replay_into_cache(&entries, &mut cache); assert_eq!(cache.len(), 3); assert_eq!(cache.chunks[0], "replay-a"); assert_eq!(cache.chunks[1], "replay-b"); assert_eq!(cache.chunks[2], "replay-c"); assert_eq!(cache.count_active(), 3); } #[test] fn test_tick_session_merges_wal() { let dir = TempDir::new().unwrap(); let config = make_config(&dir); let h5_path = config.path.clone(); let mut mem = HDF5Memory::create(config).unwrap(); mem.save(make_entry("tick-a", &[1.0, 0.0, 0.0, 0.0])) .unwrap(); mem.save(make_entry("tick-b", &[0.0, 1.0, 0.0, 0.0])) .unwrap(); mem.save(make_entry("tick-c", &[0.0, 0.0, 1.0, 0.0])) .unwrap(); assert_eq!(mem.wal_pending_count(), 3); mem.tick_session().unwrap(); // WAL should be empty after tick_session merges assert_eq!(mem.wal_pending_count(), 0); // Cache should still have 3 entries assert_eq!(mem.count(), 3); // WAL file on disk should be empty let wal_path = h5_path.with_extension("h5.wal"); let entries = WalFile::read_entries(&wal_path).unwrap(); assert!(entries.is_empty()); } #[test] fn test_wal_header_only_with_nonzero_count() { // Simulate crash: header says 5 entries but file is only 9 bytes (header only). // This is the exact scenario from the bug report — process exits before WAL // flushes, leaving a stale entry_count in the header. let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("corrupted.h5.wal"); { let mut f = File::create(&wal_path).unwrap(); f.write_all(&WAL_MAGIC).unwrap(); f.write_all(&[WAL_VERSION]).unwrap(); f.write_all(&5u32.to_le_bytes()).unwrap(); // claims 5 entries f.flush().unwrap(); } // Should NOT error — should return empty vec let entries = WalFile::read_entries(&wal_path).unwrap(); assert!(entries.is_empty()); } #[test] fn test_wal_partial_truncation() { // Write 2 valid entries, then corrupt the header to claim 5. // read_entries should return the 2 valid entries, not error. let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("partial.h5.wal"); { let mut wal = WalFile::open(&wal_path).unwrap(); wal.append_save(&make_wal_entry("first", &[1.0, 2.0])) .unwrap(); wal.append_save(&make_wal_entry("second", &[3.0, 4.0])) .unwrap(); assert_eq!(wal.pending_count(), 2); } // Corrupt the header: overwrite entry_count to 5 { let mut f = OpenOptions::new().write(true).open(&wal_path).unwrap(); f.seek(SeekFrom::Start(5)).unwrap(); f.write_all(&5u32.to_le_bytes()).unwrap(); f.flush().unwrap(); } // Should recover the 2 valid entries, not fail let entries = WalFile::read_entries(&wal_path).unwrap(); assert_eq!(entries.len(), 2); assert_eq!(entries[0].chunk, "first"); assert_eq!(entries[1].chunk, "second"); } #[test] fn test_wal_invalid_version_rejected() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("badversion.h5.wal"); { let mut f = File::create(&wal_path).unwrap(); f.write_all(&WAL_MAGIC).unwrap(); f.write_all(&[0xFF]).unwrap(); // bad version f.write_all(&0u32.to_le_bytes()).unwrap(); f.flush().unwrap(); } let result = WalFile::read_entries(&wal_path); assert!(result.is_err()); let err = result.unwrap_err().to_string(); assert!(err.contains("unsupported WAL version"), "got: {err}"); } #[test] fn test_wal_v2_detects_corrupted_payload_and_stops_replay() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("test.h5.wal"); let mut wal = WalFile::open(&wal_path).unwrap(); wal.append_save(&make_wal_entry("first", &[1.0, 2.0])) .unwrap(); let len_after_first = std::fs::metadata(&wal_path).unwrap().len(); wal.append_save(&make_wal_entry("second", &[3.0, 4.0])) .unwrap(); drop(wal); // Flip one byte inside the second entry's "second" chunk string // (well past the header and the first entry, and not touching any // length-prefix field) — this must be caught by the CRC32 trailer, // not by any length-cap guard. let mut bytes = std::fs::read(&wal_path).unwrap(); let corrupt_at = len_after_first as usize + 15; bytes[corrupt_at] ^= 0xFF; std::fs::write(&wal_path, &bytes).unwrap(); let entries = WalFile::read_entries(&wal_path).unwrap(); assert_eq!( entries.len(), 1, "the corrupted second entry must not be returned" ); assert_eq!(entries[0].chunk, "first"); } /// A crash mid-append leaves a torn final entry. Reopening the WAL must /// place the next append at the end of the VERIFIED prefix, not at /// end-of-file, or that append is written behind garbage the replay /// scanner stops at — unreachable forever despite having returned Ok. /// /// This is the ordinary crash case, so getting it wrong loses /// acknowledged writes in exactly the situation a WAL exists for. #[test] fn test_wal_append_after_torn_tail_stays_replayable() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("test.h5.wal"); let mut wal = WalFile::open(&wal_path).unwrap(); wal.append_save(&make_wal_entry("first", &[1.0, 2.0])) .unwrap(); drop(wal); // Simulate the crash: a partial entry appended after the good one. { use std::io::Write; let mut f = std::fs::OpenOptions::new() .append(true) .open(&wal_path) .unwrap(); f.write_all(&[0xAB, 0xCD, 0xEF, 0x01, 0x02]).unwrap(); f.flush().unwrap(); } // Reopen and append. The torn bytes must not survive between the // verified prefix and the new entry. let mut wal = WalFile::open(&wal_path).unwrap(); wal.append_save(&make_wal_entry("second", &[3.0, 4.0])) .unwrap(); drop(wal); let entries = WalFile::read_entries(&wal_path).unwrap(); assert_eq!( entries.len(), 2, "the append after a torn tail must be replayable; got {} entr(y/ies) — \ the post-crash write was silently lost", entries.len() ); } /// Reordering two entries on disk must break the CRC chain — the /// second entry's stored CRC was computed against the first entry's /// real CRC, not against the chain state a reader sees after swapping /// them, so replay stops immediately instead of accepting the tampered /// order (INT-09). #[test] fn test_wal_detects_reordered_entries() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("test.h5.wal"); let mut wal = WalFile::open(&wal_path).unwrap(); wal.append_save(&make_wal_entry("first", &[1.0, 2.0])) .unwrap(); let len_after_first = std::fs::metadata(&wal_path).unwrap().len() as usize; wal.append_save(&make_wal_entry("second", &[3.0, 4.0])) .unwrap(); let len_after_second = std::fs::metadata(&wal_path).unwrap().len() as usize; drop(wal); let bytes = std::fs::read(&wal_path).unwrap(); let header_len = 9usize; let entry1_bytes = bytes[header_len..len_after_first].to_vec(); let entry2_bytes = bytes[len_after_first..len_after_second].to_vec(); let mut spliced = bytes[..header_len].to_vec(); spliced.extend_from_slice(&entry2_bytes); spliced.extend_from_slice(&entry1_bytes); std::fs::write(&wal_path, &spliced).unwrap(); let entries = WalFile::read_entries(&wal_path).unwrap(); assert!( entries.is_empty(), "reordered entries must break the CRC chain and stop replay, got {} entries", entries.len() ); } /// Splicing a third-party entry in between two legitimate entries (e.g. /// moving a Tombstone in front of the Save it's meant to follow) must /// also break the chain for everything after the splice point. #[test] fn test_wal_detects_spliced_entry() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("test.h5.wal"); let mut wal = WalFile::open(&wal_path).unwrap(); wal.append_save(&make_wal_entry("first", &[1.0])).unwrap(); let len_after_first = std::fs::metadata(&wal_path).unwrap().len() as usize; wal.append_save(&make_wal_entry("second", &[2.0])).unwrap(); let len_after_second = std::fs::metadata(&wal_path).unwrap().len() as usize; wal.append_save(&make_wal_entry("third", &[3.0])).unwrap(); drop(wal); let bytes = std::fs::read(&wal_path).unwrap(); let entry2_bytes = bytes[len_after_first..len_after_second].to_vec(); // Duplicate "second" right after itself: [first][second][second][third] let mut spliced = bytes[..len_after_second].to_vec(); spliced.extend_from_slice(&entry2_bytes); spliced.extend_from_slice(&bytes[len_after_second..]); std::fs::write(&wal_path, &spliced).unwrap(); let entries = WalFile::read_entries(&wal_path).unwrap(); assert_eq!( entries.len(), 2, "replay must stop at the spliced duplicate, keeping only the entries before it" ); assert_eq!(entries[0].chunk, "first"); assert_eq!(entries[1].chunk, "second"); } /// A WAL closed (without truncating) and reopened must continue the CRC /// chain correctly for newly appended entries — this is the normal /// crash-restart-without-flush scenario (`HDF5Memory::open` replays /// existing entries, then reopens the same file for further appends /// without clearing it), and must not produce a false "reordering" /// detection for its own legitimately-appended entries. #[test] fn test_wal_chain_continues_across_reopen() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("test.h5.wal"); let mut wal = WalFile::open(&wal_path).unwrap(); wal.append_save(&make_wal_entry("first", &[1.0])).unwrap(); drop(wal); // simulate a restart without ever truncating the WAL let mut wal2 = WalFile::open(&wal_path).unwrap(); wal2.append_save(&make_wal_entry("second", &[2.0])).unwrap(); drop(wal2); let entries = WalFile::read_entries(&wal_path).unwrap(); assert_eq!( entries.len(), 2, "both pre- and post-reopen entries must replay cleanly" ); assert_eq!(entries[0].chunk, "first"); assert_eq!(entries[1].chunk, "second"); } /// Build a legacy (WAL_VERSION_LEGACY_NO_CRC) WAL file containing one /// Save entry, with no trailing CRC32. fn build_legacy_v1_wal_bytes() -> Vec { let mut buf = Vec::new(); buf.extend_from_slice(&WAL_MAGIC); buf.push(WAL_VERSION_LEGACY_NO_CRC); buf.extend_from_slice(&1u32.to_le_bytes()); buf.push(WalEntryType::Save as u8); buf.extend_from_slice(&42.0f64.to_le_bytes()); serialize_str(&mut buf, "legacy-chunk"); let embedding = [1.0f32, 2.0]; buf.extend_from_slice(&(embedding.len() as u32).to_le_bytes()); for v in embedding { buf.extend_from_slice(&v.to_le_bytes()); } serialize_str(&mut buf, "chan"); serialize_str(&mut buf, "sess"); serialize_str(&mut buf, "tags"); buf } #[test] fn test_wal_reads_legacy_v1_format_without_crc() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("legacy.h5.wal"); std::fs::write(&wal_path, build_legacy_v1_wal_bytes()).unwrap(); // Only the migration-only reader may read a legacy no-CRC file. let entries = WalFile::read_entries_for_migration(&wal_path).unwrap(); assert_eq!(entries.len(), 1); assert_eq!(entries[0].chunk, "legacy-chunk"); assert_eq!(entries[0].embedding, vec![1.0, 2.0]); } /// The public `read_entries` must reject a legacy no-CRC file instead of /// silently downgrading to the fully-unverified parser (INT-09) — flipping /// a version byte from 2/3 down to 1 must not be a way to bypass every /// integrity check for an arbitrary caller of the public API. #[test] fn test_wal_read_entries_rejects_legacy_v1_format() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("legacy.h5.wal"); std::fs::write(&wal_path, build_legacy_v1_wal_bytes()).unwrap(); let result = WalFile::read_entries(&wal_path); assert!( result.is_err(), "read_entries() must reject a legacy no-CRC WAL file, not silently parse it" ); } #[test] fn test_wal_open_migrates_legacy_v1_to_current_version() { let dir = TempDir::new().unwrap(); let wal_path = dir.path().join("legacy.h5.wal"); let mut buf = Vec::new(); buf.extend_from_slice(&WAL_MAGIC); buf.push(WAL_VERSION_LEGACY_NO_CRC); buf.extend_from_slice(&0u32.to_le_bytes()); std::fs::write(&wal_path, &buf).unwrap(); let wal = WalFile::open(&wal_path).unwrap(); assert!(wal.is_empty()); drop(wal); let bytes = std::fs::read(&wal_path).unwrap(); assert_eq!( bytes[4], WAL_VERSION, "legacy file must be migrated to the current version" ); } #[test] fn test_wal_disabled() { let dir = TempDir::new().unwrap(); let mut config = make_config(&dir); config.wal_enabled = false; let h5_path = config.path.clone(); let mut mem = HDF5Memory::create(config).unwrap(); mem.save(make_entry("no-wal", &[1.0, 0.0, 0.0, 0.0])) .unwrap(); // With WAL disabled, save goes through flush() directly (old behavior) assert_eq!(mem.count(), 1); // No WAL file should exist let wal_path = h5_path.with_extension("h5.wal"); assert!(!wal_path.exists(), "no .wal file when WAL disabled"); assert_eq!(mem.wal_pending_count(), 0); } }