From 0744d526398f1316b47ded132ff185ba96b271a2 Mon Sep 17 00:00:00 2001 From: osobh Date: Sat, 19 Sep 2026 06:08:53 -0700 Subject: [PATCH] fix(agent): provenance survives compaction; bound alert/session growth; snapshot the WAL MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ProvenanceStore::remap: compaction renumbers cache indices (which are the provenance record ids) but nothing renumbered the ledger, so after any compaction — including the automatic one in delete() — every surviving record's hash was filed under a different record and the next save_or_update raised a bogus High "integrity mismatch" alert. - Pending anomaly alerts are capped (newest 1024 kept). Alerts never block a save, and a session over its write limit alerts on every write, so a caller that didn't drain them grew the queue without bound. - WriteAnomalyDetector tracks at most 4096 sessions, forgetting the least-active half on overflow instead of leaking one entry per session id for the life of the process. - snapshot() copies the pending WAL next to the .h5 copy, so a snapshot is the store as it is now rather than as of the last checkpoint (it used to silently omit up to wal_max_entries recent saves). Co-Authored-By: Claude Fable 5.1 --- crates/clawhdf5-agent/src/anomaly.rs | 20 ++++++ crates/clawhdf5-agent/src/lib.rs | 95 +++++++++++++++++++++++-- crates/clawhdf5-agent/src/provenance.rs | 17 +++++ 3 files changed, 128 insertions(+), 4 deletions(-) diff --git a/crates/clawhdf5-agent/src/anomaly.rs b/crates/clawhdf5-agent/src/anomaly.rs index 9564341..bc8975c 100644 --- a/crates/clawhdf5-agent/src/anomaly.rs +++ b/crates/clawhdf5-agent/src/anomaly.rs @@ -161,6 +161,9 @@ pub struct WriteEvent { // WriteAnomalyDetector // --------------------------------------------------------------------------- +/// Upper bound on distinct session ids the detector tracks at once. +const MAX_TRACKED_SESSIONS: usize = 4096; + /// Tracks write events and raises alerts for suspicious behaviour. #[derive(Debug)] pub struct WriteAnomalyDetector { @@ -189,6 +192,23 @@ impl WriteAnomalyDetector { if event.timestamp > self.last_timestamp { self.last_timestamp = event.timestamp; } + // Bound the per-session map: a long-lived process sees an unbounded + // number of distinct session ids. When it overflows, forget the + // sessions with the fewest writes (they are furthest from the limit + // this map exists to enforce); the current one is re-added below. + if self.session_counts.len() >= MAX_TRACKED_SESSIONS + && !self.session_counts.contains_key(&event.session_id) + { + let mut counts: Vec = self.session_counts.values().copied().collect(); + let keep_from = counts.len() / 2; + counts.select_nth_unstable(keep_from); + let threshold = counts[keep_from]; + self.session_counts.retain(|_, c| *c >= threshold); + if self.session_counts.len() >= MAX_TRACKED_SESSIONS { + // Every session had the same count: drop them all. + self.session_counts.clear(); + } + } *self .session_counts .entry(event.session_id.clone()) diff --git a/crates/clawhdf5-agent/src/lib.rs b/crates/clawhdf5-agent/src/lib.rs index a155b54..f07c755 100644 --- a/crates/clawhdf5-agent/src/lib.rs +++ b/crates/clawhdf5-agent/src/lib.rs @@ -205,6 +205,9 @@ pub trait AgentMemory { fn get_session_summary(&self, session_id: &str) -> Result>; } +/// Most anomaly alerts kept between `take_anomaly_alerts` calls. +const MAX_PENDING_ALERTS: usize = 1024; + // --- HDF5Memory --- pub struct HDF5Memory { @@ -527,7 +530,7 @@ impl HDF5Memory { .into_iter() .flatten() { - self.anomaly_alerts.push(alert); + self.push_anomaly_alert(alert); } } @@ -548,7 +551,7 @@ impl HDF5Memory { .provenance .verify_integrity(record_id as u64, current_chunk) { - self.anomaly_alerts.push(anomaly::AnomalyAlert { + self.push_anomaly_alert(anomaly::AnomalyAlert { severity: anomaly::Severity::High, message: format!( "provenance integrity mismatch for record {record_id}: stored content no \ @@ -559,6 +562,18 @@ impl HDF5Memory { } } + /// Queue an alert, keeping only the most recent [`MAX_PENDING_ALERTS`]. + /// Alerts never block a save, so a caller that never drains them — or a + /// session stuck over its write limit, which alerts on every write — + /// must not be able to grow this without bound. + fn push_anomaly_alert(&mut self, alert: anomaly::AnomalyAlert) { + if self.anomaly_alerts.len() >= MAX_PENDING_ALERTS { + let excess = self.anomaly_alerts.len() + 1 - MAX_PENDING_ALERTS; + self.anomaly_alerts.drain(..excess); + } + self.anomaly_alerts.push(alert); + } + /// Alerts raised by anomaly detection / provenance checks since the last /// call, draining the internal queue. pub fn take_anomaly_alerts(&mut self) -> Vec { @@ -874,8 +889,10 @@ impl AgentMemory for HDF5Memory { } fn compact(&mut self) -> Result { - let (removed, _index_map) = self.cache.compact(); + let (removed, index_map) = self.cache.compact(); if removed > 0 { + // Record ids are cache indices, which compaction just renumbered. + self.provenance.remap(&index_map); // Compaction renumbers cache indices; rebuild the index to match. self.hnsw_mark_dirty(); self.flush()?; @@ -892,7 +909,18 @@ impl AgentMemory for HDF5Memory { } fn snapshot(&self, dest: &Path) -> Result { - storage::snapshot_file(&self.config.path, dest) + let snapshot = storage::snapshot_file(&self.config.path, dest)?; + // Entries saved since the last checkpoint live only in the WAL. Copy + // it alongside (where `open()` looks for it) so the snapshot is the + // store as it is now, not as of the last checkpoint. The .h5 is + // copied first: if a checkpoint lands in between, the WAL copy is + // empty or its prefix is skipped via the checkpoint mark — never + // applied twice. + let wal_path = self.config.path.with_extension("h5.wal"); + if self.wal.as_ref().is_some_and(|w| !w.is_empty()) && wal_path.exists() { + storage::snapshot_file(&wal_path, &snapshot.with_extension("h5.wal"))?; + } + Ok(snapshot) } fn add_session( @@ -1508,6 +1536,65 @@ mod tests { assert_eq!(mem.count(), 1); } + #[test] + fn compaction_does_not_cause_false_provenance_alerts() { + let dir = TempDir::new().unwrap(); + let mut mem = HDF5Memory::create(make_config(&dir)).unwrap(); + for name in ["a", "b", "c"] { + let mut e = make_entry(name, &[1.0, 0.0, 0.0, 0.0]); + e.tags = format!("tag-{name}"); + mem.save(e).unwrap(); + } + // 1 of 3 tombstoned exceeds compact_threshold, so delete() compacts. + mem.delete(0).unwrap(); + assert_eq!(mem.cache.chunks, ["b", "c"]); + mem.take_anomaly_alerts(); + + // "c" moved from id 2 to id 1. Its recorded hash must have moved too, + // or this update is checked against "b"'s hash and flagged. + let mut update = make_entry("c2", &[0.0, 1.0, 0.0, 0.0]); + update.tags = "tag-c".into(); + assert_eq!(mem.save_or_update(update).unwrap(), 1); + let alerts = mem.take_anomaly_alerts(); + assert!( + !alerts.iter().any(|a| a.message.contains("provenance")), + "{alerts:?}" + ); + } + + #[test] + fn pending_alerts_are_bounded() { + let dir = TempDir::new().unwrap(); + let mut mem = HDF5Memory::create(make_config(&dir)).unwrap(); + for i in 0..(MAX_PENDING_ALERTS + 50) { + mem.push_anomaly_alert(anomaly::AnomalyAlert { + severity: anomaly::Severity::Low, + message: format!("alert {i}"), + timestamp: i as f64, + }); + } + let alerts = mem.take_anomaly_alerts(); + assert_eq!(alerts.len(), MAX_PENDING_ALERTS); + assert_eq!(alerts[0].message, "alert 50", "oldest are dropped first"); + } + + #[test] + fn snapshot_includes_entries_still_in_the_wal() { + let dir = TempDir::new().unwrap(); + let mut config = make_config(&dir); + config.wal_enabled = true; + let mut mem = HDF5Memory::create(config).unwrap(); + mem.save(make_entry("checkpointed", &[1.0, 0.0, 0.0, 0.0])) + .unwrap(); + mem.flush_wal().unwrap(); + mem.save(make_entry("wal-only", &[0.0, 1.0, 0.0, 0.0])) + .unwrap(); + + let snap = mem.snapshot(&dir.path().join("snap.h5")).unwrap(); + let restored = HDF5Memory::open(&snap).unwrap(); + assert_eq!(restored.cache.chunks, ["checkpointed", "wal-only"]); + } + #[test] fn store_has_a_single_writer() { let dir = TempDir::new().unwrap(); diff --git a/crates/clawhdf5-agent/src/provenance.rs b/crates/clawhdf5-agent/src/provenance.rs index 946940d..bbed6cc 100644 --- a/crates/clawhdf5-agent/src/provenance.rs +++ b/crates/clawhdf5-agent/src/provenance.rs @@ -105,6 +105,23 @@ impl ProvenanceStore { self.records.insert(provenance.record_id, provenance); } + /// Renumber records after the store was compacted. `index_map[old]` is + /// the record's new id, or `None` if it was removed. Without this, every + /// surviving record's hash ends up filed under some other record's id and + /// the next integrity check reports a bogus mismatch. + pub fn remap(&mut self, index_map: &[Option]) { + let old = std::mem::take(&mut self.records); + for (old_id, mut prov) in old { + let new_id = usize::try_from(old_id) + .ok() + .and_then(|i| index_map.get(i).copied().flatten()); + if let Some(new_id) = new_id { + prov.record_id = new_id as u64; + self.records.insert(new_id as u64, prov); + } + } + } + /// Retrieve by record ID. pub fn get(&self, record_id: u64) -> Option<&MemoryProvenance> { self.records.get(&record_id)