diff --git a/crates/clawhdf5-agent/src/lib.rs b/crates/clawhdf5-agent/src/lib.rs index d2be326..b1a5253 100644 --- a/crates/clawhdf5-agent/src/lib.rs +++ b/crates/clawhdf5-agent/src/lib.rs @@ -287,7 +287,8 @@ impl HDF5Memory { /// Open an existing HDF5 memory file. pub fn open(path: &Path) -> Result { - let (config, mut cache, sessions, knowledge) = storage::read_from_disk(path)?; + let ((config, mut cache, sessions, knowledge), wal_applied) = + storage::read_from_disk_with_mark(path)?; // Replay WAL if present let wal_path = path.with_extension("h5.wal"); @@ -295,7 +296,10 @@ impl HDF5Memory { // Uses the migration-only reader since this is the one legitimate // path that may need to read a legacy (pre-CRC) WAL file — see // WalFile::read_entries_for_migration. - let entries = wal::WalFile::read_entries_for_migration(&wal_path)?; + // `wal_applied` drops the prefix a checkpoint already folded in, + // in case the process died between writing the .h5 and + // truncating the WAL. + let entries = wal::WalFile::read_entries_for_migration(&wal_path, wal_applied)?; wal::replay_into_cache(&entries, &mut cache); Some(wal::WalFile::open(&wal_path)?) } else if config.wal_enabled { @@ -336,12 +340,16 @@ impl HDF5Memory { /// also clear the WAL, otherwise `open()` will replay stale entries /// on top of the already-persisted data, duplicating them. fn flush(&mut self) -> Result<()> { - storage::write_to_disk( + // Record which WAL prefix this checkpoint contains, so a crash before + // the truncate below can't replay those entries a second time. + let wal_applied = self.wal.as_ref().map(|w| w.mark()); + storage::write_to_disk_with_mark( &self.config.path, &self.config, &self.cache, &self.sessions, &self.knowledge, + wal_applied, )?; if let Some(ref mut w) = self.wal { w.truncate()?; diff --git a/crates/clawhdf5-agent/src/schema.rs b/crates/clawhdf5-agent/src/schema.rs index a366fe3..8c234eb 100644 --- a/crates/clawhdf5-agent/src/schema.rs +++ b/crates/clawhdf5-agent/src/schema.rs @@ -12,16 +12,36 @@ use crate::MemoryError; use crate::cache::MemoryCache; use crate::knowledge::KnowledgeCache; use crate::session::SessionCache; +use crate::wal::WalMark; pub const SCHEMA_VERSION: &str = "1.0"; pub const ZEROCLAW_VERSION: &str = "0.8.0"; +/// `/meta` attributes holding the [`WalMark`] of the WAL prefix already folded +/// into this file. Absent on files written before the mark existed, and when +/// the checkpoint was taken with an empty WAL. +const WAL_APPLIED_LEN_ATTR: &str = "wal_applied_len"; +const WAL_APPLIED_CRC_ATTR: &str = "wal_applied_crc"; + /// Build a complete HDF5 file from the in-memory state. pub fn build_hdf5_file( config: &MemoryConfig, cache: &MemoryCache, sessions: &SessionCache, knowledge: &KnowledgeCache, +) -> Result, MemoryError> { + build_hdf5_file_with_mark(config, cache, sessions, knowledge, None) +} + +/// [`build_hdf5_file`], recording which WAL prefix this state already +/// contains (see [`WalMark`]) so a crash before the WAL is truncated doesn't +/// replay those entries a second time. +pub fn build_hdf5_file_with_mark( + config: &MemoryConfig, + cache: &MemoryCache, + sessions: &SessionCache, + knowledge: &KnowledgeCache, + wal_applied: Option, ) -> Result, MemoryError> { let mut builder = clawhdf5::FileBuilder::new(); @@ -38,6 +58,10 @@ pub fn build_hdf5_file( "edgehdf5_version", AttrValue::String(ZEROCLAW_VERSION.into()), ); + if let Some(mark) = wal_applied.filter(|m| m.len > 0) { + meta.set_attr(WAL_APPLIED_LEN_ATTR, AttrValue::I64(mark.len as i64)); + meta.set_attr(WAL_APPLIED_CRC_ATTR, AttrValue::I64(i64::from(mark.crc))); + } // Need at least one dataset in the group for it to be a proper group meta.create_dataset("_marker").with_u8_data(&[1]).compact(); let finished_meta = meta.finish(); @@ -309,6 +333,20 @@ fn write_string_dataset( } /// Validate an HDF5 file has the correct schema and load all data. +/// Read the checkpoint's [`WalMark`] from `/meta`, if it has one. +pub fn read_wal_mark(file: &clawhdf5::File) -> Option { + let attrs = file.group("meta").ok()?.attrs().ok()?; + let len = match attrs.get(WAL_APPLIED_LEN_ATTR)? { + AttrValue::I64(v) => u64::try_from(*v).ok()?, + _ => return None, + }; + let crc = match attrs.get(WAL_APPLIED_CRC_ATTR)? { + AttrValue::I64(v) => u32::try_from(*v).ok()?, + _ => return None, + }; + Some(WalMark { len, crc }) +} + pub fn validate_and_load( file: &clawhdf5::File, ) -> Result<(MemoryConfig, MemoryCache, SessionCache, KnowledgeCache), MemoryError> { diff --git a/crates/clawhdf5-agent/src/storage.rs b/crates/clawhdf5-agent/src/storage.rs index 8d8c1b0..c9000a1 100644 --- a/crates/clawhdf5-agent/src/storage.rs +++ b/crates/clawhdf5-agent/src/storage.rs @@ -11,6 +11,7 @@ use crate::cache::MemoryCache; use crate::knowledge::KnowledgeCache; use crate::schema; use crate::session::SessionCache; +use crate::wal::WalMark; /// Write all in-memory state to an HDF5 file on disk. pub fn write_to_disk( @@ -20,7 +21,20 @@ pub fn write_to_disk( sessions: &SessionCache, knowledge: &KnowledgeCache, ) -> Result<(), MemoryError> { - let bytes = schema::build_hdf5_file(config, cache, sessions, knowledge)?; + write_to_disk_with_mark(path, config, cache, sessions, knowledge, None) +} + +/// [`write_to_disk`] for a checkpoint: `wal_applied` is the mark of the WAL +/// prefix whose entries `cache` already contains. +pub fn write_to_disk_with_mark( + path: &Path, + config: &MemoryConfig, + cache: &MemoryCache, + sessions: &SessionCache, + knowledge: &KnowledgeCache, + wal_applied: Option, +) -> Result<(), MemoryError> { + let bytes = schema::build_hdf5_file_with_mark(config, cache, sessions, knowledge, wal_applied)?; if bytes.is_empty() { return Err(MemoryError::Hdf5("build_hdf5_file produced 0 bytes".into())); @@ -28,9 +42,41 @@ pub fn write_to_disk( // Write to a temp file first, then rename for atomicity let tmp_path = path.with_extension("h5.tmp"); - std::fs::write(&tmp_path, &bytes).map_err(MemoryError::Io)?; - std::fs::rename(&tmp_path, path).map_err(MemoryError::Io)?; + write_synced(&tmp_path, &bytes)?; + rename_synced(&tmp_path, path) +} +/// Write `bytes` to `path` and flush them to stable storage. +fn write_synced(path: &Path, bytes: &[u8]) -> Result<(), MemoryError> { + use std::io::Write; + let mut f = std::fs::File::create(path).map_err(MemoryError::Io)?; + f.write_all(bytes).map_err(MemoryError::Io)?; + f.sync_all().map_err(MemoryError::Io) +} + +/// Rename `from` over `to`, then sync the parent directory so the rename +/// itself survives a power loss. `from` must already be synced: without that, +/// the rename can reach disk before the data and leave an empty or partial +/// file under the final name. +/// +/// This is per-checkpoint/snapshot cost only (each is already a full file +/// write). Individual WAL appends are deliberately not synced — see the +/// durability notes in the crate docs. +fn rename_synced(from: &Path, to: &Path) -> Result<(), MemoryError> { + std::fs::rename(from, to).map_err(MemoryError::Io)?; + #[cfg(unix)] + if let Some(dir) = to.parent() { + let dir = if dir.as_os_str().is_empty() { + Path::new(".") + } else { + dir + }; + // Directory fsync is best-effort: some filesystems refuse it, and the + // rename has already happened. + if let Ok(d) = std::fs::File::open(dir) { + let _ = d.sync_all(); + } + } Ok(()) } @@ -42,6 +88,15 @@ pub fn write_to_disk( pub fn read_from_disk( path: &Path, ) -> Result<(MemoryConfig, MemoryCache, SessionCache, KnowledgeCache), MemoryError> { + read_from_disk_with_mark(path).map(|(state, _mark)| state) +} + +/// Everything [`read_from_disk`] returns. +pub type StoreState = (MemoryConfig, MemoryCache, SessionCache, KnowledgeCache); + +/// [`read_from_disk`], plus the checkpoint's [`WalMark`] (if any) so the +/// caller can skip WAL entries this file already contains. +pub fn read_from_disk_with_mark(path: &Path) -> Result<(StoreState, Option), MemoryError> { let mmap = clawhdf5_io::MmapReader::open(path).map_err(MemoryError::Io)?; // Advise the OS we'll need the whole file for parsing @@ -53,8 +108,9 @@ pub fn read_from_disk( let (mut config, cache, sessions, knowledge) = schema::validate_and_load(&file)?; config.path = path.to_path_buf(); + let wal_applied = schema::read_wal_mark(&file); - Ok((config, cache, sessions, knowledge)) + Ok(((config, cache, sessions, knowledge), wal_applied)) } /// Copy an HDF5 file atomically to a destination. @@ -78,7 +134,10 @@ pub fn snapshot_file(src: &Path, dest: &Path) -> Result { @@ -201,6 +220,7 @@ impl WalFile { entry_count: 0, pending_header_sync: 0, running_crc: 0, + chain_len: 0, }) } v => Err(MemoryError::Schema(format!("unsupported WAL version {v}"))), @@ -213,6 +233,7 @@ impl WalFile { entry_count: 0, pending_header_sync: 0, running_crc: 0, + chain_len: 0, }) } } @@ -258,6 +279,7 @@ impl WalFile { .as_mut() .ok_or_else(|| MemoryError::Io(std::io::Error::other("WAL file not open")))?; f.write_all(&buf)?; + self.chain_len += buf.len() as u64; self.running_crc = crc; self.entry_count += 1; @@ -282,6 +304,7 @@ impl WalFile { .as_mut() .ok_or_else(|| MemoryError::Io(std::io::Error::other("WAL file not open")))?; f.write_all(&buf)?; + self.chain_len += buf.len() as u64; self.running_crc = crc; self.entry_count += 1; @@ -311,7 +334,7 @@ impl WalFile { /// 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) + Self::read_entries_impl(path, false, None) } /// Like [`WalFile::read_entries`], but also accepts @@ -320,13 +343,23 @@ impl WalFile { /// 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) + /// + /// `applied` is the checkpoint mark read from the `.h5` file, if any: if + /// the WAL's chain passes through it (same byte length, same chained + /// CRC), everything up to that point is already in the `.h5` and is + /// dropped. If it never does — the normal case, because the WAL was + /// truncated after the checkpoint — every entry is returned. + pub(crate) fn read_entries_for_migration( + path: &Path, + applied: Option, + ) -> Result, MemoryError> { + Self::read_entries_impl(path, true, applied) } fn read_entries_impl( path: &Path, allow_legacy_no_crc: bool, + applied: Option, ) -> Result, MemoryError> { if !path.exists() { return Ok(Vec::new()); @@ -343,7 +376,8 @@ impl WalFile { match header[4] { WAL_VERSION => { - let (entries, _final_crc, _verified_bytes) = read_chained_entries(&mut f, 0); + let (entries, _final_crc, _verified_bytes) = + read_chained_entries(&mut f, 0, applied); Ok(entries) } WAL_VERSION_CRC_UNCHAINED => { @@ -406,9 +440,19 @@ impl WalFile { self.entry_count = 0; self.pending_header_sync = 0; self.running_crc = 0; + self.chain_len = 0; Ok(()) } + /// The mark covering every entry currently in this WAL. Store it with a + /// checkpoint taken from the state those entries produced. + pub fn mark(&self) -> WalMark { + WalMark { + len: self.chain_len, + crc: self.running_crc, + } + } + /// Number of pending entries. pub fn pending_count(&self) -> u32 { self.entry_count @@ -524,7 +568,16 @@ fn chained_crc(entry_bytes: &[u8], prev_crc: u32) -> u32 { /// 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) { +/// +/// `applied`, when given, is a checkpoint mark: once the chain reaches exactly +/// that position, the entries collected so far are discarded (they are +/// already in the `.h5` file). A zero-length mark matches nothing. +fn read_chained_entries( + f: &mut R, + start_crc: u32, + applied: Option, +) -> (Vec, u32, u64) { + let applied = applied.filter(|m| m.len > 0); let mut entries = Vec::new(); let mut running_crc = start_crc; let mut verified_bytes: u64 = 0; @@ -554,6 +607,14 @@ fn read_chained_entries(f: &mut R, start_crc: u32) -> (Vec, u if let Some(entry) = entry_opt { entries.push(entry); } + if applied + == Some(WalMark { + len: verified_bytes, + crc: running_crc, + }) + { + entries.clear(); + } } (entries, running_crc, verified_bytes) } @@ -932,6 +993,96 @@ mod tests { assert!(entries.is_empty()); } + /// Reopen `path` and return the stored chunks in order. + fn reopen_chunks(path: &std::path::Path) -> Vec { + let mem = HDF5Memory::open(path).unwrap(); + mem.cache.chunks.clone() + } + + #[test] + fn crash_between_checkpoint_and_wal_truncate_does_not_duplicate() { + // flush() writes the new .h5 and only then truncates the WAL. Dying in + // between leaves BOTH a .h5 that contains the pending entries and a + // WAL that still lists them; replaying blindly used to double them. + let dir = TempDir::new().unwrap(); + let config = make_config(&dir); + let h5_path = config.path.clone(); + let wal_path = h5_path.with_extension("h5.wal"); + let stale_wal = dir.path().join("stale.wal"); + + { + let mut mem = HDF5Memory::create(config).unwrap(); + for name in ["a", "b", "c"] { + mem.save(make_entry(name, &[1.0, 0.0, 0.0, 0.0])).unwrap(); + } + assert_eq!(mem.wal_pending_count(), 3); + std::fs::copy(&wal_path, &stale_wal).unwrap(); + mem.flush_wal().unwrap(); + } + // Undo the truncate: this is the on-disk state right after the crash. + std::fs::copy(&stale_wal, &wal_path).unwrap(); + assert_eq!(WalFile::read_entries(&wal_path).unwrap().len(), 3); + + assert_eq!(reopen_chunks(&h5_path), ["a", "b", "c"]); + + // Entries appended to that same WAL after recovery are still replayed. + { + let mut mem = HDF5Memory::open(&h5_path).unwrap(); + mem.save(make_entry("d", &[0.0, 1.0, 0.0, 0.0])).unwrap(); + } + assert_eq!(reopen_chunks(&h5_path), ["a", "b", "c", "d"]); + } + + #[test] + fn entries_written_after_a_completed_checkpoint_are_all_replayed() { + // Normal case: the checkpoint's mark refers to a WAL that has since + // been truncated, so it must not suppress anything in the new one — + // including when the new WAL grows past the old mark's length. + 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.flush_wal().unwrap(); + for name in ["b", "c", "d"] { + mem.save(make_entry(name, &[1.0, 0.0, 0.0, 0.0])).unwrap(); + } + } + assert_eq!(reopen_chunks(&h5_path), ["a", "b", "c", "d"]); + } + + #[test] + fn mark_matching_is_exact() { + let dir = TempDir::new().unwrap(); + let wal_path = dir.path().join("m.wal"); + let mut wal = WalFile::open(&wal_path).unwrap(); + wal.append_save(&make_wal_entry("one", &[1.0])).unwrap(); + let after_one = wal.mark(); + wal.append_save(&make_wal_entry("two", &[2.0])).unwrap(); + let after_two = wal.mark(); + drop(wal); + + let read = |m| { + WalFile::read_entries_for_migration(&wal_path, m) + .unwrap() + .into_iter() + .map(|e| e.chunk) + .collect::>() + }; + assert_eq!(read(None), ["one", "two"]); + assert_eq!(read(Some(after_one)), ["two"]); + assert!(read(Some(after_two)).is_empty()); + // Right length, wrong CRC (a different WAL generation): skip nothing. + let foreign = WalMark { + crc: after_one.crc ^ 1, + ..after_one + }; + assert_eq!(read(Some(foreign)), ["one", "two"]); + // Reopening resumes the same mark. + assert_eq!(WalFile::open(&wal_path).unwrap().mark(), after_two); + } + #[test] fn test_wal_replay_on_open() { // Test WAL replay using read_entries + replay_into_cache directly, @@ -1269,7 +1420,7 @@ mod tests { 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(); + let entries = WalFile::read_entries_for_migration(&wal_path, None).unwrap(); assert_eq!(entries.len(), 1); assert_eq!(entries[0].chunk, "legacy-chunk"); assert_eq!(entries[0].embedding, vec![1.0, 2.0]);