From 743eaa550779d0058e8910b8ed6ecfbcd3235d95 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 19 Jul 2026 00:49:19 +0100 Subject: [PATCH] Harden Rust CultCache snapshot replacement --- packages/cultcache-rs/Cargo.toml | 3 + packages/cultcache-rs/README.md | 37 ++- packages/cultcache-rs/src/lib.rs | 546 ++++++++++++++++++++++++++++++- 3 files changed, 569 insertions(+), 17 deletions(-) diff --git a/packages/cultcache-rs/Cargo.toml b/packages/cultcache-rs/Cargo.toml index b11a909e..ef06727d 100644 --- a/packages/cultcache-rs/Cargo.toml +++ b/packages/cultcache-rs/Cargo.toml @@ -22,6 +22,9 @@ serde_bytes = "0.11" serde_json = "1" uuid = { version = "1", features = ["v4"] } +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.59", features = ["Win32_Storage_FileSystem"] } + [dev-dependencies] pretty_assertions = "1.4" tempfile = "3" diff --git a/packages/cultcache-rs/README.md b/packages/cultcache-rs/README.md index 2d75d94e..74b20dd3 100644 --- a/packages/cultcache-rs/README.md +++ b/packages/cultcache-rs/README.md @@ -250,8 +250,38 @@ marked `default` use `Default::default()` when older payloads are read. Deleted field slots should stay reserved until an explicit store migration rewrites the data. -The single-file MessagePack store rewrites an atomic snapshot. That is a sane -starting point for small typed state surfaces, settings, Epiphany agent memory, +The single-file MessagePack store rewrites one complete snapshot. Under the +exclusive sidecar lock it removes abandoned candidates without following them, +creates a unique same-directory candidate with create-new semantics, writes and +flushes it, and then replaces the committed pathname without deleting that +pathname first. A process interruption before replacement leaves the previous +snapshot authoritative. Once replacement succeeds, readers observe the complete +new snapshot rather than a partial file. + +Unix requests a parent-directory flush after replacement. Windows uses +`MoveFileExW` with replace-existing and write-through flags. If the final Unix +directory flush fails, the method returns a downcastable +`CommittedSnapshotDurabilityUncertain`: the complete new snapshot may already be +visible, so the caller must fail-stop or reopen and reconcile before issuing +higher-level work again. Treating that error as "not committed" and blindly +retrying can duplicate commands or receipts. + +Abandoned `..tmp` candidates are never read as committed state and are +removed by the next exclusive writer. Cleanup recognizes only the exact +canonical UUID form emitted for that target, including targets whose native +filename is not UTF-8; similar `.tmp` names remain untouched. Candidate creation +uses create-new semantics, so an unexpected file or symlink collision fails +instead of being opened or truncated. The canonical candidate filename namespace +is reserved to the store. + +These guarantees assume a local filesystem that honors the platform rename and +flush primitives, and writers that honor the sidecar lock. Network filesystems, +faulty storage hardware, and unrelated writers may provide weaker behavior. +Fault-injection tests prove the process-visible old-or-new outcomes around the +replacement boundary; they do not simulate sudden power loss or prove that a +device honored a successful flush. + +This is a sane starting point for small typed state surfaces, settings, Epiphany agent memory, heartbeat state, and other compact control-plane data. Large corpora should use a sharded store or a real database instead of asking one file to become a warehouse and then acting wounded when physics invoices us. @@ -260,6 +290,9 @@ The single-file store uses a sidecar lock file for shared reads and exclusive writes. That protects ordinary multi-process access to the same file. It is still not a multi-master replication protocol, and it is not a substitute for a coordinator when higher-level write ordering matters. +An explicit unlock error cannot replace the write outcome: dropping the lock-file +handle is the final OS lock-release path, so a committed-durability-uncertain +result remains inspectable rather than being masked by cleanup noise. ## Near-Term Ergonomic Improvements diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index 67e88cdf..b85b763a 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -5,9 +5,12 @@ use serde::Serialize; use serde::de::DeserializeOwned; use std::collections::BTreeMap; use std::collections::BTreeSet; +use std::ffi::OsStr; +use std::fmt; use std::fs; use std::fs::File; use std::fs::OpenOptions; +use std::io::Write; use std::path::Path; use std::path::PathBuf; @@ -97,6 +100,45 @@ pub struct PushAllOptions { pub soft: bool, } +/// A complete new snapshot replaced the committed pathname, but its final +/// durability barrier failed. Callers must fail-stop or reopen and reconcile; +/// retrying as though the write never committed can duplicate higher-level work. +#[derive(Debug)] +pub struct CommittedSnapshotDurabilityUncertain { + path: PathBuf, + detail: String, +} + +impl CommittedSnapshotDurabilityUncertain { + fn new(path: PathBuf, error: impl fmt::Display) -> Self { + Self { + path, + detail: error.to_string(), + } + } + + pub fn path(&self) -> &Path { + &self.path + } + + pub fn detail(&self) -> &str { + &self.detail + } +} + +impl fmt::Display for CommittedSnapshotDurabilityUncertain { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + formatter, + "snapshot {} was committed but its durability is uncertain: {}", + self.path.display(), + self.detail + ) + } +} + +impl std::error::Error for CommittedSnapshotDurabilityUncertain {} + pub trait CacheBackingStore: Send { fn pull_all(&self) -> Result>; fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()>; @@ -144,26 +186,50 @@ impl SingleFileMessagePackBackingStore { } fn write_all_unlocked(&self, entries: &[CultCacheEnvelope]) -> Result<()> { + self.write_all_unlocked_with_observer(entries, |_| Ok(())) + } + + fn write_all_unlocked_with_observer( + &self, + entries: &[CultCacheEnvelope], + mut observe: impl FnMut(SnapshotReplacementStage) -> Result<()>, + ) -> Result<()> { if let Some(parent) = self.path.parent() { fs::create_dir_all(parent) .with_context(|| format!("failed to create {}", parent.display()))?; } + remove_stale_candidates(&self.path)?; let bytes = rmp_serde::to_vec(&encode_store_snapshot(entries)) .context("failed to encode MessagePack")?; let tmp_path = temporary_path_for(&self.path); - fs::write(&tmp_path, bytes) + let mut candidate = OpenOptions::new() + .create_new(true) + .write(true) + .open(&tmp_path) + .with_context(|| format!("failed to open {}", tmp_path.display()))?; + candidate + .write_all(&bytes) .with_context(|| format!("failed to write {}", tmp_path.display()))?; - if self.path.exists() { - fs::remove_file(&self.path) - .with_context(|| format!("failed to replace {}", self.path.display()))?; + candidate + .sync_all() + .with_context(|| format!("failed to flush {}", tmp_path.display()))?; + drop(candidate); + + observe(SnapshotReplacementStage::CandidateSynced)?; + replace_file(&tmp_path, &self.path)?; + if let Err(error) = observe(SnapshotReplacementStage::TargetReplaced) { + return Err(anyhow::Error::new( + CommittedSnapshotDurabilityUncertain::new(self.path.clone(), error), + )); + } + + if let Some(parent) = self.path.parent() { + if let Err(error) = sync_parent_directory(parent) { + return Err(anyhow::Error::new( + CommittedSnapshotDurabilityUncertain::new(self.path.clone(), error), + )); + } } - fs::rename(&tmp_path, &self.path).with_context(|| { - format!( - "failed to move {} to {}", - tmp_path.display(), - self.path.display() - ) - })?; Ok(()) } @@ -178,13 +244,26 @@ impl SingleFileMessagePackBackingStore { } fn with_exclusive_lock(&self, action: impl FnOnce() -> Result) -> Result { + self.with_exclusive_lock_using_unlock(action, |lock| fs2::FileExt::unlock(lock)) + } + + fn with_exclusive_lock_using_unlock( + &self, + action: impl FnOnce() -> Result, + unlock: impl FnOnce(&File) -> std::io::Result<()>, + ) -> Result { let lock = self.open_lock_file()?; fs2::FileExt::lock_exclusive(&lock) .with_context(|| format!("failed to lock {}", self.lock_path().display()))?; - let result = action(); - fs2::FileExt::unlock(&lock) - .with_context(|| format!("failed to unlock {}", self.lock_path().display()))?; - result + let action_result = action(); + + // Closing the file handle releases the OS lock even when an explicit + // unlock reports an error. The action outcome remains authoritative: + // in particular, never replace a committed-durability-uncertain error + // with secondary lock-cleanup noise. + let _unlock_result = unlock(&lock); + drop(lock); + action_result } fn open_lock_file(&self) -> Result { @@ -742,6 +821,186 @@ fn temporary_path_for(path: &Path) -> PathBuf { path.with_file_name(file_name) } +fn is_canonical_candidate_uuid(value: &[u8]) -> bool { + if value.len() != 36 || !value.is_ascii() { + return false; + } + let Ok(value) = std::str::from_utf8(value) else { + return false; + }; + let Ok(uuid) = uuid::Uuid::parse_str(value) else { + return false; + }; + uuid.hyphenated().to_string() == value +} + +#[cfg(unix)] +fn is_owned_candidate_name(target_name: &OsStr, candidate_name: &OsStr) -> bool { + use std::os::unix::ffi::OsStrExt; + + let target = target_name.as_bytes(); + let candidate = candidate_name.as_bytes(); + let Some(remainder) = candidate.strip_prefix(target) else { + return false; + }; + let Some(remainder) = remainder.strip_prefix(b".") else { + return false; + }; + let Some(uuid) = remainder.strip_suffix(b".tmp") else { + return false; + }; + is_canonical_candidate_uuid(uuid) +} + +#[cfg(windows)] +fn is_owned_candidate_name(target_name: &OsStr, candidate_name: &OsStr) -> bool { + use std::os::windows::ffi::OsStrExt; + + let target = target_name.encode_wide().collect::>(); + let candidate = candidate_name.encode_wide().collect::>(); + let Some(remainder) = candidate.strip_prefix(target.as_slice()) else { + return false; + }; + let Some(remainder) = remainder.strip_prefix(&[b'.' as u16]) else { + return false; + }; + let suffix = [b'.' as u16, b't' as u16, b'm' as u16, b'p' as u16]; + let Some(uuid_wide) = remainder.strip_suffix(&suffix) else { + return false; + }; + let Ok(uuid) = uuid_wide + .iter() + .map(|value| u8::try_from(*value)) + .collect::, _>>() + else { + return false; + }; + is_canonical_candidate_uuid(&uuid) +} + +#[cfg(not(any(unix, windows)))] +fn is_owned_candidate_name(target_name: &OsStr, candidate_name: &OsStr) -> bool { + let (Some(target), Some(candidate)) = (target_name.to_str(), candidate_name.to_str()) else { + return false; + }; + let Some(remainder) = candidate.strip_prefix(target) else { + return false; + }; + let Some(remainder) = remainder.strip_prefix('.') else { + return false; + }; + let Some(uuid) = remainder.strip_suffix(".tmp") else { + return false; + }; + is_canonical_candidate_uuid(uuid.as_bytes()) +} + +fn stale_candidate_paths(path: &Path) -> Result> { + let parent = path.parent().unwrap_or_else(|| Path::new(".")); + if !parent.exists() { + return Ok(Vec::new()); + } + let target_name = path + .file_name() + .unwrap_or_else(|| OsStr::new("cultcache.msgpack")); + let mut candidates = Vec::new(); + for entry in + fs::read_dir(parent).with_context(|| format!("failed to inspect {}", parent.display()))? + { + let entry = entry.with_context(|| format!("failed to inspect {}", parent.display()))?; + let name = entry.file_name(); + if is_owned_candidate_name(target_name, &name) { + candidates.push(entry.path()); + } + } + Ok(candidates) +} + +fn remove_stale_candidates(path: &Path) -> Result<()> { + for candidate in stale_candidate_paths(path)? { + fs::remove_file(&candidate) + .with_context(|| format!("failed to remove stale candidate {}", candidate.display()))?; + } + Ok(()) +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum SnapshotReplacementStage { + CandidateSynced, + TargetReplaced, +} + +#[cfg(unix)] +fn replace_file(candidate: &Path, target: &Path) -> Result<()> { + fs::rename(candidate, target).with_context(|| { + format!( + "failed to atomically replace {} with {}", + target.display(), + candidate.display() + ) + }) +} + +#[cfg(windows)] +fn replace_file(candidate: &Path, target: &Path) -> Result<()> { + use std::os::windows::ffi::OsStrExt; + use windows_sys::Win32::Storage::FileSystem::{ + MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, MoveFileExW, + }; + + let candidate_wide = candidate + .as_os_str() + .encode_wide() + .chain(std::iter::once(0)) + .collect::>(); + let target_wide = target + .as_os_str() + .encode_wide() + .chain(std::iter::once(0)) + .collect::>(); + let replaced = unsafe { + MoveFileExW( + candidate_wide.as_ptr(), + target_wide.as_ptr(), + MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH, + ) + }; + if replaced == 0 { + return Err(std::io::Error::last_os_error()).with_context(|| { + format!( + "failed to atomically replace {} with {}", + target.display(), + candidate.display() + ) + }); + } + Ok(()) +} + +#[cfg(not(any(unix, windows)))] +fn replace_file(candidate: &Path, target: &Path) -> Result<()> { + fs::rename(candidate, target).with_context(|| { + format!( + "failed to replace {} with {}", + target.display(), + candidate.display() + ) + }) +} + +#[cfg(unix)] +fn sync_parent_directory(parent: &Path) -> Result<()> { + File::open(parent) + .with_context(|| format!("failed to open directory {}", parent.display()))? + .sync_all() + .with_context(|| format!("failed to flush directory {}", parent.display())) +} + +#[cfg(not(unix))] +fn sync_parent_directory(_parent: &Path) -> Result<()> { + Ok(()) +} + #[cfg(test)] mod tests { use super::*; @@ -778,6 +1037,263 @@ mod tests { cultcache_registry!(TestEntries { Settings, Note }); + fn test_envelope(value: &str) -> CultCacheEnvelope { + CultCacheEnvelope { + key: "state".to_string(), + r#type: "test-state".to_string(), + payload: value.as_bytes().to_vec(), + stored_at: "2026-07-15T00:00:00Z".to_string(), + schema_id: Some("tests.test_state.v1".to_string()), + } + } + + #[test] + fn interruption_before_replacement_preserves_committed_snapshot() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("cache.msgpack"); + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + let old = test_envelope("old"); + let new = test_envelope("new"); + store.push_all(std::slice::from_ref(&old), PushAllOptions::default())?; + + let result = store.write_all_unlocked_with_observer(std::slice::from_ref(&new), |stage| { + assert_eq!(stage, SnapshotReplacementStage::CandidateSynced); + assert!( + store_path.exists(), + "the committed pathname must never be removed" + ); + Err(anyhow!("simulated interruption before replacement")) + }); + + assert!(result.is_err()); + let recovered = store.pull_all()?; + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].payload, old.payload); + assert_eq!(stale_candidate_paths(&store_path)?.len(), 1); + + store.push_all(std::slice::from_ref(&new), PushAllOptions::default())?; + let recovered = store.pull_all()?; + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].payload, new.payload); + assert!(stale_candidate_paths(&store_path)?.is_empty()); + Ok(()) + } + + #[test] + fn interruption_after_replacement_recovers_new_snapshot() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("cache.msgpack"); + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + let old = test_envelope("old"); + let new = test_envelope("new"); + store.push_all(std::slice::from_ref(&old), PushAllOptions::default())?; + + let result = store.write_all_unlocked_with_observer(std::slice::from_ref(&new), |stage| { + if stage == SnapshotReplacementStage::TargetReplaced { + return Err(anyhow!("simulated interruption after replacement")); + } + Ok(()) + }); + + let error = result.expect_err("the injected post-replacement failure must surface"); + let uncertain = error + .downcast_ref::() + .expect("post-replacement failures must be typed as committed but uncertain"); + assert_eq!(uncertain.path(), store_path.as_path()); + assert!(uncertain.detail().contains("simulated interruption")); + let recovered = store.pull_all()?; + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].payload, new.payload); + assert!(stale_candidate_paths(&store_path)?.is_empty()); + Ok(()) + } + + #[test] + fn first_write_interruption_before_replacement_leaves_store_uncommitted() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("cache.msgpack"); + let store = SingleFileMessagePackBackingStore::new(&store_path); + let new = test_envelope("new"); + + let result = store.write_all_unlocked_with_observer(std::slice::from_ref(&new), |stage| { + assert_eq!(stage, SnapshotReplacementStage::CandidateSynced); + Err(anyhow!("simulated interruption before first replacement")) + }); + + assert!(result.is_err()); + assert!(!store_path.exists()); + assert!(store.pull_all()?.is_empty()); + assert_eq!(stale_candidate_paths(&store_path)?.len(), 1); + Ok(()) + } + + #[test] + fn first_write_interruption_after_replacement_exposes_complete_snapshot() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("cache.msgpack"); + let store = SingleFileMessagePackBackingStore::new(&store_path); + let new = test_envelope("new"); + + let result = store.write_all_unlocked_with_observer(std::slice::from_ref(&new), |stage| { + if stage == SnapshotReplacementStage::TargetReplaced { + return Err(anyhow!("simulated interruption after first replacement")); + } + Ok(()) + }); + + let error = result.expect_err("the injected post-replacement failure must surface"); + assert!( + error + .downcast_ref::() + .is_some() + ); + let recovered = store.pull_all()?; + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].payload, new.payload); + assert!(stale_candidate_paths(&store_path)?.is_empty()); + Ok(()) + } + + #[test] + fn unlock_failure_never_masks_the_action_outcome() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("cache.msgpack"); + let store = SingleFileMessagePackBackingStore::new(&store_path); + + let completed = store.with_exclusive_lock_using_unlock( + || Ok(42_u32), + |_| Err(std::io::Error::other("simulated unlock failure")), + )?; + assert_eq!(completed, 42); + + let uncertain: Result<()> = store.with_exclusive_lock_using_unlock( + || { + Err(anyhow::Error::new( + CommittedSnapshotDurabilityUncertain::new( + store_path.clone(), + "primary post-commit failure", + ), + )) + }, + |_| Err(std::io::Error::other("secondary unlock failure")), + ); + let error = uncertain.expect_err("the action error must survive unlock cleanup"); + let error = error + .downcast_ref::() + .expect("the primary typed outcome must not be masked"); + assert!(error.detail().contains("primary post-commit failure")); + assert!(!error.detail().contains("secondary unlock failure")); + + // Both injected unlock failures still close their lock handles. A new + // exclusive operation must therefore acquire the lock normally. + store.with_exclusive_lock(|| Ok(()))?; + Ok(()) + } + + #[test] + fn stale_cleanup_preserves_non_uuid_lookalikes() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("cache.msgpack"); + let lookalike = temp.path().join("cache.msgpack.not-a-uuid.tmp"); + fs::write(&lookalike, b"belongs to somebody else")?; + + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + store.push_all(&[test_envelope("new")], PushAllOptions::default())?; + + assert_eq!(fs::read(&lookalike)?, b"belongs to somebody else"); + assert!(stale_candidate_paths(&store_path)?.is_empty()); + Ok(()) + } + + #[cfg(unix)] + #[test] + fn stale_candidate_symlink_is_removed_without_clobbering_its_target() -> Result<()> { + use std::os::unix::fs::symlink; + + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("cache.msgpack"); + let victim_path = temp.path().join("victim.txt"); + fs::write(&victim_path, b"untouched")?; + let stale_path = temporary_path_for(&store_path); + symlink(&victim_path, &stale_path)?; + + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + store.push_all(&[test_envelope("new")], PushAllOptions::default())?; + + assert_eq!(fs::read(&victim_path)?, b"untouched"); + assert!(!stale_path.exists()); + assert!(stale_candidate_paths(&store_path)?.is_empty()); + Ok(()) + } + + #[cfg(unix)] + #[test] + fn non_utf8_target_names_recover_their_own_candidates() -> Result<()> { + use std::ffi::OsString; + use std::os::unix::ffi::OsStringExt; + + let temp = tempfile::tempdir()?; + let store_name = OsString::from_vec(b"cache-\xff.msgpack".to_vec()); + let store_path = temp.path().join(store_name); + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + let new = test_envelope("new"); + + let interrupted = store + .write_all_unlocked_with_observer(std::slice::from_ref(&new), |_| { + Err(anyhow!("simulated interruption")) + }); + assert!(interrupted.is_err()); + assert_eq!(stale_candidate_paths(&store_path)?.len(), 1); + + store.push_all(std::slice::from_ref(&new), PushAllOptions::default())?; + assert!(stale_candidate_paths(&store_path)?.is_empty()); + assert_eq!(store.pull_all()?[0].payload, new.payload); + Ok(()) + } + + #[test] + fn two_contending_writers_leave_one_complete_snapshot() -> Result<()> { + use std::sync::{Arc, Barrier}; + + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("cache.msgpack"); + let barrier = Arc::new(Barrier::new(3)); + let spawn_writer = |marker: u8| { + let store_path = store_path.clone(); + let barrier = Arc::clone(&barrier); + std::thread::spawn(move || -> Result<()> { + let mut entries = Vec::new(); + for index in 0..128 { + let mut entry = test_envelope(std::str::from_utf8(&[marker]).unwrap()); + entry.key = format!("state-{index:03}"); + entry.payload = vec![marker; 4096]; + entries.push(entry); + } + let mut store = SingleFileMessagePackBackingStore::new(store_path); + barrier.wait(); + store.push_all(&entries, PushAllOptions::default()) + }) + }; + + let writer_a = spawn_writer(b'a'); + let writer_b = spawn_writer(b'b'); + barrier.wait(); + writer_a.join().expect("writer A panicked")?; + writer_b.join().expect("writer B panicked")?; + + let recovered = SingleFileMessagePackBackingStore::new(&store_path).pull_all()?; + assert_eq!(recovered.len(), 128); + let winner = recovered[0].payload[0]; + assert!(winner == b'a' || winner == b'b'); + assert!( + recovered + .iter() + .all(|entry| entry.payload == vec![winner; 4096]) + ); + assert!(stale_candidate_paths(&store_path)?.is_empty()); + Ok(()) + } + #[test] fn familiar_cultcache_flow_persists_and_reloads_typed_documents() -> Result<()> { let temp = tempfile::tempdir()?;