diff --git a/crates/clawhdf5-agent/src/async_memory.rs b/crates/clawhdf5-agent/src/async_memory.rs index e6200e0..1f665ad 100644 --- a/crates/clawhdf5-agent/src/async_memory.rs +++ b/crates/clawhdf5-agent/src/async_memory.rs @@ -408,6 +408,10 @@ impl AsyncHDF5Memory { let (tx, rx) = oneshot::channel(); let _ = self.write_tx.send(WriteCmd::Shutdown(tx)).await; let _ = rx.await; + // The writer task has stopped, so nothing can write through this + // handle any more: release the single-writer lock now rather than at + // drop, so the store can be reopened while `self` is still in scope. + self.inner.lock().await.release_store_lock(); Ok(()) } } diff --git a/crates/clawhdf5-agent/src/lib.rs b/crates/clawhdf5-agent/src/lib.rs index 151e784..a155b54 100644 --- a/crates/clawhdf5-agent/src/lib.rs +++ b/crates/clawhdf5-agent/src/lib.rs @@ -37,6 +37,7 @@ pub mod schema; pub mod search; pub mod session; pub mod storage; +mod store_lock; pub mod temporal; pub mod wal; @@ -86,6 +87,8 @@ pub enum MemoryError { Hdf5(String), Schema(String), NotFound(String), + /// Another `HDF5Memory` (in this or another process) has the store open. + Locked(String), } impl std::fmt::Display for MemoryError { @@ -95,6 +98,7 @@ impl std::fmt::Display for MemoryError { MemoryError::Hdf5(e) => write!(f, "HDF5 error: {e}"), MemoryError::Schema(e) => write!(f, "schema error: {e}"), MemoryError::NotFound(e) => write!(f, "not found: {e}"), + MemoryError::Locked(e) => write!(f, "store is locked: {e}"), } } } @@ -240,6 +244,15 @@ pub struct HDF5Memory { /// via [`HDF5Memory::take_anomaly_alerts`]. Saves are never blocked on /// these — surfacing is opt-in for callers that want to act on them. anomaly_alerts: Vec, + /// Opened with [`HDF5Memory::open_read_only`]: nothing may reach the disk. + read_only: bool, + /// A WAL that `open()` could not read and moved aside; see + /// [`HDF5Memory::quarantined_wal`]. + quarantined_wal: Option, + /// Single-writer guard. Declared last so it is released only after the + /// WAL and everything else has been dropped. `None` once a wrapper that + /// has stopped all writes released it early (see `release_store_lock`). + _lock: Option, } impl std::fmt::Debug for HDF5Memory { @@ -251,6 +264,7 @@ impl std::fmt::Debug for HDF5Memory { impl HDF5Memory { /// Create a new HDF5 memory file with the given configuration. pub fn create(config: MemoryConfig) -> Result { + let lock = store_lock::StoreLock::acquire(&config.path)?; let cache = MemoryCache::new(config.embedding_dim); let sessions = SessionCache::new(); let knowledge = KnowledgeCache::new(); @@ -282,17 +296,103 @@ impl HDF5Memory { provenance: provenance::ProvenanceStore::new(), anomaly: anomaly::WriteAnomalyDetector::new(anomaly::AnomalyConfig::default()), anomaly_alerts: Vec::new(), + read_only: false, + quarantined_wal: None, + _lock: Some(lock), }) } /// Open an existing HDF5 memory file. + /// If the WAL at `wal_path` can't possibly be replayed — its header is + /// torn (crash while the file was being created) or isn't a WAL header at + /// all — move it aside so a healthy `.h5` still opens, and return where it + /// went. A well-formed header with an *unknown version* is left alone and + /// still fails `open()`: that WAL was most likely written by a newer + /// build, and discarding it would lose data this binary merely can't read. + fn quarantine_unreadable_wal(wal_path: &Path) -> Result> { + if !wal_path.exists() { + return Ok(None); + } + let reason = match wal::wal_header_status(wal_path)? { + wal::WalHeaderStatus::Readable | wal::WalHeaderStatus::UnknownVersion(_) => { + return Ok(None); + } + wal::WalHeaderStatus::Torn => "truncated header", + wal::WalHeaderStatus::BadMagic => "bad magic bytes", + }; + let ts = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + let dest = wal_path.with_extension(format!("wal.corrupt-{ts}")); + std::fs::rename(wal_path, &dest)?; + eprintln!( + "clawhdf5-agent: WAL {} is unreadable ({reason}); moved to {} and continuing \ + from the last checkpoint", + wal_path.display(), + dest.display() + ); + Ok(Some(dest)) + } + + /// Give up the single-writer lock before this value is dropped. Only for + /// wrappers that have already stopped every write path but keep the handle + /// alive (`AsyncHDF5Memory::shutdown`), so the store can be reopened. + #[cfg_attr(not(feature = "async"), allow(dead_code))] + pub(crate) fn release_store_lock(&mut self) { + self._lock = None; + } + + /// Where `open()` moved an unreadable WAL, if it had to. Entries that were + /// only in that WAL are not in this store; the file is kept for forensics. + pub fn quarantined_wal(&self) -> Option<&Path> { + self.quarantined_wal.as_deref() + } + pub fn open(path: &Path) -> Result { + Self::open_impl(path, false) + } + + /// Open a store for reading only, without taking the single-writer lock — + /// so it works while another `HDF5Memory` (in this or another process) + /// has the store open for writing, e.g. to inspect what is on disk. + /// + /// It loads the last checkpoint plus whatever the WAL held at that + /// moment; it is a point-in-time view and does not follow later writes. + /// Nothing is written: the WAL file is not repaired, upgraded or moved, + /// and every operation that would persist state returns an error. + pub fn open_read_only(path: &Path) -> Result { + Self::open_impl(path, true) + } + + fn open_impl(path: &Path, read_only: bool) -> Result { + let lock = if read_only { + None + } else { + Some(store_lock::StoreLock::acquire(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"); - let wal = if wal_path.exists() { + let quarantined_wal = if read_only { + None + } else { + Self::quarantine_unreadable_wal(&wal_path)? + }; + let wal = if read_only { + // Replay in memory only. `WalFile::open` would truncate a torn + // tail and may rewrite the header — both belong to the writer. An + // unreadable WAL is simply skipped: the writer will deal with it. + if wal_path.exists() + && let Ok(entries) = + wal::WalFile::read_entries_for_migration(&wal_path, wal_applied) + { + wal::replay_into_cache(&entries, &mut cache); + } + None + } else if wal_path.exists() { // 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. @@ -331,6 +431,9 @@ impl HDF5Memory { provenance: provenance::ProvenanceStore::new(), anomaly: anomaly::WriteAnomalyDetector::new(anomaly::AnomalyConfig::default()), anomaly_alerts: Vec::new(), + read_only, + quarantined_wal, + _lock: lock, }) } @@ -340,6 +443,12 @@ 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<()> { + if self.read_only { + return Err(MemoryError::Io(std::io::Error::new( + std::io::ErrorKind::PermissionDenied, + "store was opened read-only", + ))); + } // 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()); @@ -1399,6 +1508,90 @@ mod tests { assert_eq!(mem.count(), 1); } + #[test] + fn store_has_a_single_writer() { + let dir = TempDir::new().unwrap(); + let config = make_config(&dir); + let path = config.path.clone(); + let mem = HDF5Memory::create(config).unwrap(); + assert!(matches!( + HDF5Memory::open(&path), + Err(MemoryError::Locked(_)) + )); + drop(mem); + HDF5Memory::open(&path).unwrap(); + } + + #[test] + fn read_only_open_coexists_with_a_writer_and_never_writes() { + let dir = TempDir::new().unwrap(); + let mut config = make_config(&dir); + config.wal_enabled = true; + let path = config.path.clone(); + let wal_path = path.with_extension("h5.wal"); + let mut writer = HDF5Memory::create(config).unwrap(); + writer + .save(make_entry("pending", &[1.0, 0.0, 0.0, 0.0])) + .unwrap(); + let wal_before = std::fs::read(&wal_path).unwrap(); + let h5_before = std::fs::read(&path).unwrap(); + + // Sees the checkpoint plus the writer's un-checkpointed WAL entry. + let mut reader = HDF5Memory::open_read_only(&path).unwrap(); + assert_eq!(reader.cache.chunks, ["pending"]); + assert!(reader.save(make_entry("nope", &[0.0; 4])).is_err()); + assert!(reader.flush_wal().is_err()); + drop(reader); + + assert_eq!(std::fs::read(&wal_path).unwrap(), wal_before); + assert_eq!(std::fs::read(&path).unwrap(), h5_before); + // The writer is unaffected. + writer + .save(make_entry("more", &[0.0, 1.0, 0.0, 0.0])) + .unwrap(); + } + + #[test] + fn unreadable_wal_is_quarantined_not_fatal() { + let dir = TempDir::new().unwrap(); + let config = make_config(&dir); + let path = config.path.clone(); + let wal_path = path.with_extension("h5.wal"); + { + let mut mem = HDF5Memory::create(config).unwrap(); + mem.save(make_entry("kept", &[1.0, 0.0, 0.0, 0.0])).unwrap(); + mem.flush_wal().unwrap(); + } + std::fs::write(&wal_path, b"not a wal at all").unwrap(); + + let mem = HDF5Memory::open(&path).unwrap(); + assert_eq!(mem.cache.chunks, ["kept"]); + let moved = mem.quarantined_wal().expect("WAL should be quarantined"); + assert_eq!(std::fs::read(moved).unwrap(), b"not a wal at all"); + // A fresh, valid WAL took its place. + assert!(wal::WalFile::read_entries(&wal_path).unwrap().is_empty()); + } + + #[test] + fn wal_from_a_newer_build_is_refused_not_discarded() { + let dir = TempDir::new().unwrap(); + let mut config = make_config(&dir); + config.wal_enabled = true; + let path = config.path.clone(); + let wal_path = path.with_extension("h5.wal"); + drop(HDF5Memory::create(config).unwrap()); + let mut bytes = std::fs::read(&wal_path).unwrap(); + bytes[4] = 200; // a version this build has never heard of + std::fs::write(&wal_path, &bytes).unwrap(); + + assert!(HDF5Memory::open(&path).is_err()); + assert_eq!( + std::fs::read(&wal_path).unwrap(), + bytes, + "WAL left untouched" + ); + } + #[test] fn empty_file_operations() { let dir = TempDir::new().unwrap(); @@ -1407,6 +1600,7 @@ mod tests { let mem = HDF5Memory::create(config).unwrap(); assert_eq!(mem.count(), 0); assert_eq!(mem.count_active(), 0); + drop(mem); // a store has a single writer; release it before reopening let mem2 = HDF5Memory::open(&path).unwrap(); assert_eq!(mem2.count(), 0); diff --git a/crates/clawhdf5-agent/src/store_lock.rs b/crates/clawhdf5-agent/src/store_lock.rs new file mode 100644 index 0000000..2813f5b --- /dev/null +++ b/crates/clawhdf5-agent/src/store_lock.rs @@ -0,0 +1,79 @@ +//! Single-writer guard for a memory store. +//! +//! `HDF5Memory` keeps the whole store in memory and rewrites the `.h5` file at +//! every checkpoint, so two handles on one store (two processes, or two opens +//! in one process) silently destroy each other's data: whoever checkpoints +//! last wins, and both append to the same WAL with independent CRC chains. +//! The lock turns that into an immediate, explicit error. + +use std::fs::{File, OpenOptions, TryLockError}; +use std::path::{Path, PathBuf}; + +use crate::MemoryError; + +const LOCK_RETRIES: u32 = 25; +const LOCK_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(10); + +/// An exclusive advisory lock on `.h5.lock`, held for the lifetime of +/// the owning `HDF5Memory` and released when it is dropped (or when the +/// process dies — the OS drops the lock with the file descriptor, so a crash +/// never leaves a stale lock behind; the empty lock file itself is harmless). +#[derive(Debug)] +pub(crate) struct StoreLock { + _file: File, +} + +impl StoreLock { + pub(crate) fn lock_path(store: &Path) -> PathBuf { + store.with_extension("h5.lock") + } + + pub(crate) fn acquire(store: &Path) -> Result { + let path = Self::lock_path(store); + let file = OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(&path)?; + // A previous owner may be mid-teardown (e.g. an `AsyncHDF5Memory` + // dropped without `shutdown()`: its background task releases the + // store a moment later), so give the lock a short, bounded grace + // period before reporting a genuine second writer. + let mut attempts_left = LOCK_RETRIES; + loop { + match file.try_lock() { + Ok(()) => return Ok(Self { _file: file }), + Err(TryLockError::WouldBlock) if attempts_left > 0 => { + attempts_left -= 1; + std::thread::sleep(LOCK_RETRY_DELAY); + } + Err(TryLockError::WouldBlock) => { + return Err(MemoryError::Locked(format!( + "{} is already open in this or another process (lock file {})", + store.display(), + path.display() + ))); + } + Err(TryLockError::Error(e)) => return Err(MemoryError::Io(e)), + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn second_acquire_fails_until_first_is_dropped() { + let dir = tempfile::TempDir::new().unwrap(); + let store = dir.path().join("s.h5"); + let first = StoreLock::acquire(&store).unwrap(); + assert!(matches!( + StoreLock::acquire(&store), + Err(MemoryError::Locked(_)) + )); + drop(first); + StoreLock::acquire(&store).unwrap(); + } +} diff --git a/crates/clawhdf5-agent/src/wal.rs b/crates/clawhdf5-agent/src/wal.rs index dcf2c3b..43bcd2f 100644 --- a/crates/clawhdf5-agent/src/wal.rs +++ b/crates/clawhdf5-agent/src/wal.rs @@ -135,6 +135,45 @@ pub struct WalFile { chain_len: u64, } +/// What a WAL file's 9-byte header looks like, without reading any entries. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WalHeaderStatus { + /// A version this build can read (current or legacy). + Readable, + /// Shorter than a header — e.g. a crash while the file was being created. + /// It cannot contain entries. + Torn, + /// Not a WAL file at all. + BadMagic, + /// Well-formed header from a version this build doesn't know — most + /// likely written by a *newer* build. Never discard this: the entries are + /// probably fine, this binary just can't read them. + UnknownVersion(u8), +} + +/// Classify the header of the WAL at `path`. +pub fn wal_header_status(path: &Path) -> std::io::Result { + let mut header = [0u8; WAL_HEADER_LEN as usize]; + let mut f = File::open(path)?; + let mut filled = 0; + while filled < header.len() { + match f.read(&mut header[filled..])? { + 0 => return Ok(WalHeaderStatus::Torn), + n => filled += n, + } + } + if header[0..4] != WAL_MAGIC { + return Ok(WalHeaderStatus::BadMagic); + } + Ok(match header[4] { + WAL_VERSION + | WAL_VERSION_CHAINED_NO_UPDATE + | WAL_VERSION_CRC_UNCHAINED + | WAL_VERSION_LEGACY_NO_CRC => WalHeaderStatus::Readable, + v => WalHeaderStatus::UnknownVersion(v), + }) +} + /// A position in a WAL's CRC chain: `len` bytes of entries after the header, /// whose chained CRC is `crc`. /// diff --git a/crates/clawhdf5-agent/tests/e2e_tests.rs b/crates/clawhdf5-agent/tests/e2e_tests.rs index d151ece..3321d8b 100644 --- a/crates/clawhdf5-agent/tests/e2e_tests.rs +++ b/crates/clawhdf5-agent/tests/e2e_tests.rs @@ -196,7 +196,7 @@ fn test_migration_round_trip() { mem.add_relation(e1, e2, "discusses", 0.8).unwrap(); // Verify all data transferred by reopening - let reopened = HDF5Memory::open(&path).unwrap(); + let reopened = HDF5Memory::open_read_only(&path).unwrap(); assert_eq!(reopened.count(), 500); // Verify sessions @@ -266,7 +266,7 @@ fn test_knowledge_graph_workflow() { assert_eq!(entity.entity_type, "library"); // Persistence - let reopened = HDF5Memory::open(&path).unwrap(); + let reopened = HDF5Memory::open_read_only(&path).unwrap(); assert_eq!(reopened.knowledge().entities.len(), 4); assert_eq!(reopened.knowledge().relations.len(), 4); @@ -316,7 +316,7 @@ fn test_multi_session_workflow() { assert_eq!(mem.count(), 100); // 5 sessions * 20 entries // Reopen and verify sessions - let reopened = HDF5Memory::open(&path).unwrap(); + let reopened = HDF5Memory::open_read_only(&path).unwrap(); for sess in 0..5 { let summary = reopened .get_session_summary(&format!("sess_{sess}")) @@ -460,7 +460,7 @@ fn test_snapshot_and_continue() { assert_eq!(snap_mem.count(), 50); // Original should have 100 - let orig_mem = HDF5Memory::open(&path).unwrap(); + let orig_mem = HDF5Memory::open_read_only(&path).unwrap(); assert_eq!(orig_mem.count(), 100); } @@ -483,7 +483,7 @@ fn test_config_persistence_across_ops() { mem.add_session("s1", 0, 0, "ch", "summary").unwrap(); mem.add_entity("Entity", "type", -1).unwrap(); - let reopened = HDF5Memory::open(&path).unwrap(); + let reopened = HDF5Memory::open_read_only(&path).unwrap(); assert_eq!(reopened.config().embedding_dim, 128); assert_eq!(reopened.config().embedder, "custom:my-embedder-v2"); assert_eq!(reopened.config().chunk_size, 2048); @@ -695,7 +695,7 @@ fn test_large_text_chunks() { mem.save_batch(entries).unwrap(); // Reopen and verify - let reopened = HDF5Memory::open(&path).unwrap(); + let reopened = HDF5Memory::open_read_only(&path).unwrap(); assert_eq!(reopened.count(), 10); let (_, cache, _, _) = read_cache(&path); @@ -752,7 +752,7 @@ fn test_interleaved_sessions_entries() { mem.flush_wal().unwrap(); // Verify - let reopened = HDF5Memory::open(&path).unwrap(); + let reopened = HDF5Memory::open_read_only(&path).unwrap(); assert_eq!(reopened.count(), 6); assert_eq!( reopened.get_session_summary("s1").unwrap().as_deref(), @@ -806,7 +806,7 @@ fn test_knowledge_graph_with_embeddings() { mem.add_relation(e_python, e_hdf5, "reads", 0.9).unwrap(); // Verify entity-embedding linkage persists - let reopened = HDF5Memory::open(&path).unwrap(); + let reopened = HDF5Memory::open_read_only(&path).unwrap(); let rust_entity = reopened.knowledge().get_entity(e_rust).unwrap(); assert_eq!(rust_entity.embedding_idx, idx0 as i64); diff --git a/crates/clawhdf5-agent/tests/stress_tests.rs b/crates/clawhdf5-agent/tests/stress_tests.rs index eb09d84..572da6f 100644 --- a/crates/clawhdf5-agent/tests/stress_tests.rs +++ b/crates/clawhdf5-agent/tests/stress_tests.rs @@ -105,7 +105,7 @@ fn test_heavy_tombstoning() { assert_eq!(mem.count_active(), 5000); // Verify persistence - let reopened = HDF5Memory::open(&path).unwrap(); + let reopened = HDF5Memory::open_read_only(&path).unwrap(); assert_eq!(reopened.count(), 5000); } @@ -163,7 +163,7 @@ fn test_large_embeddings_1536() { assert_eq!(mem.count(), 10_000); // Verify persistence - let reopened = HDF5Memory::open(&path).unwrap(); + let reopened = HDF5Memory::open_read_only(&path).unwrap(); assert_eq!(reopened.count(), 10_000); // Verify search works on large dims @@ -545,7 +545,7 @@ fn test_delete_all_entries() { assert_eq!(mem.count(), 0); // Verify persistence - let reopened = HDF5Memory::open(&path).unwrap(); + let reopened = HDF5Memory::open_read_only(&path).unwrap(); assert_eq!(reopened.count(), 0); } @@ -639,7 +639,7 @@ fn test_unicode_content() { ]; mem.save_batch(entries).unwrap(); - let reopened = HDF5Memory::open(&path).unwrap(); + let reopened = HDF5Memory::open_read_only(&path).unwrap(); assert_eq!(reopened.count(), 3); let (_, cache, _, _) = clawhdf5_agent::storage::read_from_disk(&path).unwrap(); @@ -685,6 +685,6 @@ fn test_rapid_save_delete_cycles() { assert_eq!(removed, 250); assert_eq!(mem.count(), 250); - let reopened = HDF5Memory::open(&path).unwrap(); + let reopened = HDF5Memory::open_read_only(&path).unwrap(); assert_eq!(reopened.count(), 250); } diff --git a/crates/clawhdf5-cli/src/main.rs b/crates/clawhdf5-cli/src/main.rs index e6053b9..03076a2 100644 --- a/crates/clawhdf5-cli/src/main.rs +++ b/crates/clawhdf5-cli/src/main.rs @@ -146,7 +146,7 @@ fn run(cli: Cli) -> Result<(), Box> { } Commands::Recall { index } => { - let mem = HDF5Memory::open(&cli.path)?; + let mem = HDF5Memory::open_read_only(&cli.path)?; match mem.get_chunk(index) { Some(content) => { let j = serde_json::json!({ "index": index, "chunk": content }); @@ -160,7 +160,7 @@ fn run(cli: Cli) -> Result<(), Box> { } Commands::Stats => { - let mem = HDF5Memory::open(&cli.path)?; + let mem = HDF5Memory::open_read_only(&cli.path)?; let cfg = mem.config(); let j = serde_json::json!({ "path": cli.path.display().to_string(), @@ -187,7 +187,7 @@ fn run(cli: Cli) -> Result<(), Box> { } Commands::AgentsMd { output } => { - let mem = HDF5Memory::open(&cli.path)?; + let mem = HDF5Memory::open_read_only(&cli.path)?; let md = mem.generate_agents_md(); match output { Some(p) => { @@ -199,7 +199,7 @@ fn run(cli: Cli) -> Result<(), Box> { } Commands::Export => { - let mem = HDF5Memory::open(&cli.path)?; + let mem = HDF5Memory::open_read_only(&cli.path)?; for i in 0..mem.count() { if let Some(chunk) = mem.get_chunk(i) { let j = serde_json::json!({ "index": i, "chunk": chunk });