fix(agent): crash between checkpoint and WAL truncate no longer duplicates entries

flush() writes the new .h5 and only then truncates the WAL. A crash in that
window left a .h5 that already contained the pending entries AND a WAL that
still listed them, and open() replayed the WAL unconditionally — every pending
entry came back twice.

A checkpoint now records a WalMark in /meta (wal_applied_len/wal_applied_crc):
the byte length and chained CRC of the WAL prefix it folded in. On open, if
the WAL's v3 CRC chain passes through exactly that position, the entries up to
it are skipped; otherwise (the normal case: the WAL was truncated) everything
is replayed. No WAL format change; files without the attributes behave as
before. WalFile tracks its chain length alongside running_crc and resumes both
on reopen.

Also make the checkpoint and snapshot durable as a unit: sync the temp file
before the rename and the parent directory after it, so a power loss can't
leave an empty or partial .h5 under the final name. This is per-checkpoint
cost only; individual WAL appends remain unsynced by design.

Co-Authored-By: Claude Fable 5.1 <[email protected]>

crates/clawhdf5-agent/src/lib.rs

@@ -287,7 +287,8 @@ impl HDF5Memory {
 
     /// Open an existing HDF5 memory file.
     pub fn open(path: &Path) -> Result<Self> {
-        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()?;

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<Vec<u8>, 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<WalMark>,
 ) -> Result<Vec<u8>, 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<WalMark> {
+    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> {

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<WalMark>,
+) -> 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)?;
+    write_synced(&tmp_path, &bytes)?;
-    std::fs::rename(&tmp_path, path).map_err(MemoryError::Io)?;
+    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<WalMark>), 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<std::path::PathBuf, Memo
     // Atomic copy: write to temp, then rename
     let tmp_path = dest_file.with_extension("h5.tmp");
     std::fs::copy(src, &tmp_path).map_err(MemoryError::Io)?;
-    std::fs::rename(&tmp_path, &dest_file).map_err(MemoryError::Io)?;
+    std::fs::File::open(&tmp_path)
+        .and_then(|f| f.sync_all())
+        .map_err(MemoryError::Io)?;
+    rename_synced(&tmp_path, &dest_file)?;
 
     Ok(dest_file)
 }

crates/clawhdf5-agent/src/wal.rs

@@ -112,6 +112,23 @@ pub struct WalFile {
     /// 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,
+    /// Bytes of verified entries after the header (the length of the chain
+    /// `running_crc` covers). Together they form the [`WalMark`].
+    chain_len: u64,
+}
+
+/// A position in a WAL's CRC chain: `len` bytes of entries after the header,
+/// whose chained CRC is `crc`.
+///
+/// A checkpoint stores the mark of the WAL prefix it folded into the `.h5`
+/// file. If the process dies after the new `.h5` is in place but before the
+/// WAL is truncated, the next `open()` finds that exact prefix still in the
+/// WAL and skips it instead of replaying it on top of data that already
+/// contains it (which used to duplicate every pending entry).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct WalMark {
+    pub len: u64,
+    pub crc: u32,
 }
 
 impl WalFile {
@@ -147,7 +164,8 @@ impl WalFile {
                     // 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 (entries, running_crc, verified_bytes) =
+                        read_chained_entries(&mut f, 0, None);
                     let entry_count = if entries.is_empty() {
                         header_count
                     } else {
@@ -190,6 +208,7 @@ impl WalFile {
                         entry_count,
                         pending_header_sync: 0,
                         running_crc,
+                        chain_len: verified_bytes,
                     })
                 }
                 WAL_VERSION_CRC_UNCHAINED | WAL_VERSION_LEGACY_NO_CRC => {
@@ -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<Vec<WalEntry>, 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<Vec<WalEntry>, 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<WalMark>,
+    ) -> Result<Vec<WalEntry>, MemoryError> {
+        Self::read_entries_impl(path, true, applied)
     }
 
     fn read_entries_impl(
         path: &Path,
         allow_legacy_no_crc: bool,
+        applied: Option<WalMark>,
     ) -> Result<Vec<WalEntry>, 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<R: Read>(f: &mut R, start_crc: u32) -> (Vec<WalEntry>, 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<R: Read>(
+    f: &mut R,
+    start_crc: u32,
+    applied: Option<WalMark>,
+) -> (Vec<WalEntry>, 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<R: Read>(f: &mut R, start_crc: u32) -> (Vec<WalEntry>, 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<String> {
+        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::<Vec<_>>()
+        };
+        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]);