diff --git a/packages/cultcache-rs/Cargo.toml b/packages/cultcache-rs/Cargo.toml index ef06727d..e6ec5e59 100644 --- a/packages/cultcache-rs/Cargo.toml +++ b/packages/cultcache-rs/Cargo.toml @@ -17,14 +17,15 @@ chrono = { version = "0.4", features = ["serde"] } cultcache-rs-derive = { path = "cultcache-rs-derive" } fs2 = "0.4" rmp-serde = "1.3" +redb = "4.1.0" serde = { version = "1", features = ["derive"] } 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" + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.59", features = ["Win32_Foundation", "Win32_Storage_FileSystem"] } diff --git a/packages/cultcache-rs/README.md b/packages/cultcache-rs/README.md index 74b20dd3..0176f474 100644 --- a/packages/cultcache-rs/README.md +++ b/packages/cultcache-rs/README.md @@ -53,7 +53,7 @@ struct Settings { let mut cache = CultCache::new(); cache.register_entry_type::()?; -cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new("cache.msgpack")); +cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new("cache.cc")); cache.pull_all_backing_stores()?; cache.put("app", &Settings { @@ -136,7 +136,7 @@ pub struct PlayerData { let mut cache = CultCache::builder() .with_generated_entries(game_entries()) - .with_generic_store(SingleFileMessagePackBackingStore::new("cache.msgpack")) + .with_generic_store(SingleFileMessagePackBackingStore::new("cache.cc")) .build()?; cache.pull_all_backing_stores()?; @@ -173,26 +173,59 @@ without pretending Rust has runtime reflection hiding under the floorboards. - `get_envelope::` - `get_required_envelope::` - `get_all::` +- `get_all_with_keys::` +- `soa::` - `put::` - `put_envelope::` - `update::` - `delete::` - `snapshot` +- `snapshot_soa` +- `load_soa` - `SingleFileMessagePackBackingStore` +## Structure Of Arrays + +The C# CultCache hot path exposes CPU-local SoA scans as: + +```csharp +Span x = cache.Soa() + .Column(nameof(PlayerTransform.PositionX)) + .Span; +``` + +Rust cannot discover those columns through runtime reflection, so each domain +type declares its column extractors explicitly with `SoaDocument` or the +`cultcache_soa!` helper: + +```rust +cultcache_soa!(PlayerTransform { + "position_x" => |row: &PlayerTransform| row.position_x, + "health" => |row: &PlayerTransform| row.health, +}); + +let table = cache.soa::()?; +let health = table.column::("health")?; +``` + +This is the same public shape as the C# rite: callers ask the cache for a typed +table, then pull contiguous typed columns by name. The Rust version is explicit +about the extractor table because the closed-world schema boundary is a virtue, +not a missing reflection party trick. + ## Backing Store Routing Generic store: ```rust -cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new("cache.msgpack")); +cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new("cache.cc")); ``` Type-specific store: ```rust cache.add_backing_store( - SingleFileMessagePackBackingStore::new("players.msgpack"), + SingleFileMessagePackBackingStore::new("players.cc"), ["player"], ); ``` @@ -250,38 +283,8 @@ 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 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, +The single-file MessagePack store rewrites an atomic snapshot. That 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. @@ -290,9 +293,6 @@ 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/cultcache-rs-derive/src/lib.rs b/packages/cultcache-rs/cultcache-rs-derive/src/lib.rs index f05a23f3..7d613194 100644 --- a/packages/cultcache-rs/cultcache-rs-derive/src/lib.rs +++ b/packages/cultcache-rs/cultcache-rs-derive/src/lib.rs @@ -75,6 +75,7 @@ pub fn derive_database_entry(input: TokenStream) -> TokenStream { }; let mut key: Option = None; let mut default = false; + let mut bytes = false; for attribute in field .attrs .iter() @@ -88,6 +89,9 @@ pub fn derive_database_entry(input: TokenStream) -> TokenStream { } else if meta.path.is_ident("default") { default = true; Ok(()) + } else if meta.path.is_ident("bytes") { + bytes = true; + Ok(()) } else { Err(meta.error("unsupported field cultcache attribute")) } @@ -117,6 +121,7 @@ pub fn derive_database_entry(input: TokenStream) -> TokenStream { ident: field_ident, ty: field.ty, default, + bytes, }); } slots.sort_by_key(|slot| slot.key); @@ -125,8 +130,14 @@ pub fn derive_database_entry(input: TokenStream) -> TokenStream { let serialize_elements = (0..=max_slot).map(|slot| { if let Some(field) = slots.iter().find(|field| field.key == slot) { let field_ident = &field.ident; - quote! { - tuple.serialize_element(&self.#field_ident)?; + if field.bytes { + quote! { + tuple.serialize_element(::serde_bytes::Bytes::new(&self.#field_ident))?; + } + } else { + quote! { + tuple.serialize_element(&self.#field_ident)?; + } } } else { quote! { @@ -138,8 +149,14 @@ pub fn derive_database_entry(input: TokenStream) -> TokenStream { let option_declarations = slots.iter().map(|field| { let variable = format_ident!("slot_{}", field.key); let ty = &field.ty; - quote! { - let mut #variable: ::std::option::Option<#ty> = ::std::option::Option::None; + if field.bytes { + quote! { + let mut #variable: ::std::option::Option<::serde_bytes::ByteBuf> = ::std::option::Option::None; + } + } else { + quote! { + let mut #variable: ::std::option::Option<#ty> = ::std::option::Option::None; + } } }); @@ -147,9 +164,17 @@ pub fn derive_database_entry(input: TokenStream) -> TokenStream { if let Some(field) = slots.iter().find(|field| field.key == slot) { let variable = format_ident!("slot_{}", field.key); let ty = &field.ty; - quote! { - #slot => { - #variable = seq.next_element::<#ty>()?; + if field.bytes { + quote! { + #slot => { + #variable = seq.next_element::<::serde_bytes::ByteBuf>()?; + } + } + } else { + quote! { + #slot => { + #variable = seq.next_element::<#ty>()?; + } } } } else { @@ -164,7 +189,16 @@ pub fn derive_database_entry(input: TokenStream) -> TokenStream { let build_fields = slots.iter().map(|field| { let field_ident = &field.ident; let variable = format_ident!("slot_{}", field.key); - if field.default { + if field.bytes && field.default { + quote! { + #field_ident: #variable.unwrap_or_default().into_vec() + } + } else if field.bytes { + let field_name = field_ident.to_string(); + quote! { + #field_ident: #variable.ok_or_else(|| ::serde::de::Error::missing_field(#field_name))?.into_vec() + } + } else if field.default { quote! { #field_ident: #variable.unwrap_or_default() } @@ -240,4 +274,5 @@ struct CultCacheSlot { ident: syn::Ident, ty: Type, default: bool, + bytes: bool, } diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index b85b763a..b8d6d8f9 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -1,18 +1,22 @@ use anyhow::Context; use anyhow::Result; use anyhow::anyhow; +use anyhow::bail; +use anyhow::ensure; use serde::Serialize; use serde::de::DeserializeOwned; +use std::any::Any; 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::marker::PhantomData; use std::path::Path; use std::path::PathBuf; +use std::sync::Arc; + +use redb::{Database, ReadableDatabase, ReadableTable, TableDefinition}; extern crate self as cultcache_rs; @@ -27,6 +31,83 @@ pub trait CultCacheRegistry { fn register_entries(&self, cache: &mut CultCache) -> Result<()>; } +pub trait SoaDocument: DatabaseEntry { + fn soa_columns(rows: &[Self]) -> BTreeMap<&'static str, CultSoaColumnValues>; +} + +pub struct CultSoaColumnValues { + type_name: &'static str, + values: Box, +} + +impl CultSoaColumnValues { + pub fn new(values: Vec) -> Self { + Self { + type_name: std::any::type_name::(), + values: Box::new(values), + } + } +} + +#[derive(Clone, Debug, PartialEq)] +pub struct CultSoaColumn { + name: String, + values: Vec, +} + +impl CultSoaColumn { + pub fn name(&self) -> &str { + &self.name + } + + pub fn values(&self) -> &[TValue] { + &self.values + } +} + +pub struct CultSoaTable { + keys: Vec, + documents: Vec, + columns: BTreeMap, + _document: PhantomData, +} + +impl CultSoaTable { + pub fn keys(&self) -> &[String] { + &self.keys + } + + pub fn documents(&self) -> &[TDocument] { + &self.documents + } + + pub fn len(&self) -> usize { + self.keys.len() + } + + pub fn is_empty(&self) -> bool { + self.keys.is_empty() + } + + pub fn column(&self, name: &str) -> Result> { + let Some(column) = self.columns.get(name) else { + return Err(anyhow!("SoA table has no column {name:?}")); + }; + let Some(values) = column.values.downcast_ref::>() else { + return Err(anyhow!( + "SoA table column {:?} stores {}, not {}", + name, + column.type_name, + std::any::type_name::() + )); + }; + Ok(CultSoaColumn { + name: name.to_string(), + values: values.clone(), + }) + } +} + #[macro_export] macro_rules! cultcache_registry { ($name:ident { $($entry:ty),* $(,)? }) => { @@ -44,6 +125,22 @@ macro_rules! cultcache_registry { }; } +#[macro_export] +macro_rules! cultcache_soa { + ($entry:ty { $($column:literal => $extractor:expr),* $(,)? }) => { + impl $crate::SoaDocument for $entry { + fn soa_columns(rows: &[Self]) -> ::std::collections::BTreeMap<&'static str, $crate::CultSoaColumnValues> { + let mut columns = ::std::collections::BTreeMap::new(); + $( + let values = rows.iter().map($extractor).collect::>(); + columns.insert($column, $crate::CultSoaColumnValues::new(values)); + )* + columns + } + } + }; +} + #[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "camelCase")] pub struct CultCacheEnvelope { @@ -57,6 +154,93 @@ pub struct CultCacheEnvelope { pub schema_id: Option, } +#[derive(Clone, Debug, Default, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CultCacheSoaTable { + pub keys: Vec, + pub types: Vec, + pub payloads: Vec>, + pub stored_ats: Vec, + pub schema_ids: Vec>, +} + +impl CultCacheSoaTable { + pub fn from_envelopes(entries: impl IntoIterator) -> Self { + let mut table = Self::default(); + for entry in entries { + table.keys.push(entry.key); + table.types.push(entry.r#type); + table.payloads.push(entry.payload); + table.stored_ats.push(entry.stored_at); + table.schema_ids.push(entry.schema_id); + } + table + } + + pub fn len(&self) -> usize { + self.keys.len() + } + + pub fn is_empty(&self) -> bool { + self.keys.is_empty() + } + + pub fn validate(&self) -> Result<()> { + let len = self.keys.len(); + for (name, candidate) in [ + ("types", self.types.len()), + ("payloads", self.payloads.len()), + ("storedAts", self.stored_ats.len()), + ("schemaIds", self.schema_ids.len()), + ] { + if candidate != len { + return Err(anyhow!( + "CultCache SoA column {name} has length {candidate}, expected {len}" + )); + } + } + for (index, key) in self.keys.iter().enumerate() { + if key.trim().is_empty() { + return Err(anyhow!("CultCache SoA row {index} has an empty key")); + } + if self.types[index].trim().is_empty() { + return Err(anyhow!("CultCache SoA row {index} has an empty type")); + } + if self.stored_ats[index].trim().is_empty() { + return Err(anyhow!("CultCache SoA row {index} has an empty stored_at")); + } + } + Ok(()) + } + + pub fn into_envelopes(self) -> Result> { + self.validate()?; + let Self { + keys, + types, + payloads, + stored_ats, + schema_ids, + } = self; + Ok(keys + .into_iter() + .zip(types) + .zip(payloads) + .zip(stored_ats) + .zip(schema_ids) + .map( + |((((key, r#type), payload), stored_at), schema_id)| CultCacheEnvelope { + key, + r#type, + payload, + stored_at, + schema_id, + }, + ) + .collect()) + } +} + #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] struct PersistedStoreSnapshot( String, @@ -100,45 +284,6 @@ 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<()>; @@ -170,6 +315,305 @@ impl SingleFileMessagePackBackingStore { &self.path } + /// Holds the pre-created sibling lock shared while a read-only consumer + /// evaluates and acts on one snapshot. The reader never creates or writes + /// either file. Writers using the sibling lock cannot replace the snapshot + /// until `action` returns. + /// + /// Releasing the lock is RAII cleanup. A release failure therefore cannot + /// rewrite the completed action's result; consumers that need to observe + /// that cleanup failure can use + /// [`Self::with_read_only_shared_snapshot_and_unlock_diagnostic`]. + pub fn with_read_only_shared_snapshot( + &self, + action: impl FnOnce(Vec) -> Result, + ) -> Result { + self.with_read_only_shared_snapshot_and_unlock_diagnostic(action, |_| {}) + } + + /// The diagnostic-bearing form of + /// [`Self::with_read_only_shared_snapshot`]. `on_unlock_failure` observes a + /// cleanup failure separately and never changes the action result. + pub fn with_read_only_shared_snapshot_and_unlock_diagnostic( + &self, + action: impl FnOnce(Vec) -> Result, + on_unlock_failure: impl FnOnce(anyhow::Error), + ) -> Result { + self.with_read_only_shared_snapshot_using( + action, + |lock| { + fs2::FileExt::unlock(lock) + .with_context(|| format!("failed to unlock {}", self.lock_path().display())) + }, + on_unlock_failure, + ) + } + + fn with_read_only_shared_snapshot_using( + &self, + action: impl FnOnce(Vec) -> Result, + unlock: U, + on_unlock_failure: D, + ) -> Result + where + U: FnOnce(&File) -> Result<()>, + D: FnOnce(anyhow::Error), + { + let lock_path = self.lock_path(); + let lock = OpenOptions::new() + .read(true) + .open(&lock_path) + .with_context(|| format!("failed to open pre-created {}", lock_path.display()))?; + fs2::FileExt::lock_shared(&lock) + .with_context(|| format!("failed to lock {}", lock_path.display()))?; + let guard = ReadOnlySharedLockGuard::new(lock, unlock, on_unlock_failure); + let result = self.read_all_unlocked().and_then(action); + drop(guard); + result + } + + /// Replaces one exact envelope under the store's cross-process exclusive + /// lock. A changed or missing expected envelope returns false without a write. + pub fn compare_and_swap_entry( + &self, + expected: &CultCacheEnvelope, + replacement: CultCacheEnvelope, + ) -> Result { + self.with_exclusive_lock(|| { + let mut entries = self.read_all_unlocked()?; + let Some(index) = entries + .iter() + .position(|candidate| entry_id(candidate) == entry_id(expected)) + else { + return Ok(false); + }; + if entries[index] != *expected { + return Ok(false); + } + entries[index] = replacement; + entries.sort_by_key(entry_id); + self.write_all_unlocked(&entries)?; + Ok(true) + }) + } + + /// Inserts one envelope only when its polymorphic type/key identity is + /// absent, under the same cross-process lock used by all single-file writes. + pub fn insert_entry_if_absent(&self, entry: CultCacheEnvelope) -> Result { + self.with_exclusive_lock(|| { + let mut entries = self.read_all_unlocked()?; + if entries + .iter() + .any(|candidate| entry_id(candidate) == entry_id(&entry)) + { + return Ok(false); + } + entries.push(entry); + entries.sort_by_key(entry_id); + self.write_all_unlocked(&entries)?; + Ok(true) + }) + } + + /// Atomically replaces an exact set of existing envelopes and inserts any + /// additional companion envelopes. Every precondition is checked beneath + /// the same cross-process lock as the single snapshot write. + pub fn compare_and_swap_batch( + &self, + expected: &[CultCacheEnvelope], + replacements: Vec, + ) -> Result { + if replacements.is_empty() { + return Err(anyhow!( + "conditional batch requires a non-empty replacement set" + )); + } + let expected_ids = unique_batch_ids(expected, "expected")?; + let replacement_ids = unique_batch_ids(&replacements, "replacement")?; + if !expected_ids.is_subset(&replacement_ids) { + return Err(anyhow!( + "conditional batch must replace every expected identity" + )); + } + + self.with_exclusive_lock(|| { + let mut entries = self.read_all_unlocked()?; + for expected_entry in expected { + let Some(current) = entries + .iter() + .find(|candidate| entry_id(candidate) == entry_id(expected_entry)) + else { + return Ok(false); + }; + if current != expected_entry { + return Ok(false); + } + } + for replacement in &replacements { + let identity = entry_id(replacement); + if !expected_ids.contains(&identity) + && entries + .iter() + .any(|candidate| entry_id(candidate) == identity) + { + return Ok(false); + } + } + + entries.retain(|candidate| !expected_ids.contains(&entry_id(candidate))); + entries.extend(replacements); + entries.sort_by_key(entry_id); + self.write_all_unlocked(&entries)?; + Ok(true) + }) + } + + /// Atomically appends companions only when the complete backing-store + /// snapshot is byte-for-byte unchanged. Unlike identity-scoped CAS, this + /// fences predicate cardinality: a concurrently inserted unknown identity + /// makes the expected snapshot stale. + pub fn append_if_snapshot_unchanged( + &self, + expected_snapshot: &[CultCacheEnvelope], + additions: Vec, + ) -> Result { + if additions.is_empty() { + return Err(anyhow!("conditional snapshot append requires additions")); + } + unique_batch_ids(expected_snapshot, "expected snapshot")?; + let addition_ids = unique_batch_ids(&additions, "snapshot additions")?; + self.with_exclusive_lock(|| { + let mut current = self.read_all_unlocked()?; + let mut expected = expected_snapshot.to_vec(); + current.sort_by_key(entry_id); + expected.sort_by_key(entry_id); + if current != expected { + return Ok(false); + } + if current + .iter() + .any(|row| addition_ids.contains(&entry_id(row))) + { + return Ok(false); + } + current.extend(additions); + current.sort_by_key(entry_id); + self.write_all_unlocked(¤t)?; + Ok(true) + }) + } + + /// Atomically deletes an exact set of envelopes. Any changed or missing + /// member refuses the entire deletion without a partial write. + pub fn delete_batch_if_unchanged(&self, expected: &[CultCacheEnvelope]) -> Result { + if expected.is_empty() { + return Err(anyhow!( + "conditional delete requires a non-empty expected set" + )); + } + let expected_ids = unique_batch_ids(expected, "delete expected")?; + self.with_exclusive_lock(|| { + let mut entries = self.read_all_unlocked()?; + for expected_entry in expected { + let Some(current) = entries + .iter() + .find(|candidate| entry_id(candidate) == entry_id(expected_entry)) + else { + return Ok(false); + }; + if current != expected_entry { + return Ok(false); + } + } + entries.retain(|candidate| !expected_ids.contains(&entry_id(candidate))); + self.write_all_unlocked(&entries)?; + Ok(true) + }) + } + + /// Atomically replaces existing rows and appends companion rows only while + /// the complete backing-store snapshot remains byte-exact. + pub fn replace_and_append_if_snapshot_unchanged( + &self, + expected_snapshot: &[CultCacheEnvelope], + replacements: Vec, + ) -> Result { + if replacements.is_empty() { + return Err(anyhow!( + "conditional snapshot replacement requires replacements" + )); + } + unique_batch_ids(expected_snapshot, "expected snapshot")?; + let replacement_ids = unique_batch_ids(&replacements, "snapshot replacements")?; + self.with_exclusive_lock(|| { + let mut current = self.read_all_unlocked()?; + let mut expected = expected_snapshot.to_vec(); + current.sort_by_key(entry_id); + expected.sort_by_key(entry_id); + if current != expected { + return Ok(false); + } + current.retain(|row| !replacement_ids.contains(&entry_id(row))); + current.extend(replacements); + current.sort_by_key(entry_id); + self.write_all_unlocked(¤t)?; + Ok(true) + }) + } + + /// Atomically replaces and deletes rows only while the complete store + /// snapshot remains byte-exact. This is the single-file compaction + /// primitive: summary publication and history deletion share one write. + pub fn replace_and_delete_if_snapshot_unchanged( + &self, + expected_snapshot: &[CultCacheEnvelope], + replacements: Vec, + deletions: &[CultCacheEnvelope], + ) -> Result { + if replacements.is_empty() || deletions.is_empty() { + return Err(anyhow!( + "conditional compaction requires replacements and deletions" + )); + } + let expected_ids = unique_batch_ids(expected_snapshot, "expected snapshot")?; + let replacement_ids = unique_batch_ids(&replacements, "replacement")?; + let deletion_ids = unique_batch_ids(deletions, "deletion")?; + if !deletion_ids.is_subset(&expected_ids) { + return Err(anyhow!( + "compaction deletion is absent from expected snapshot" + )); + } + if !deletion_ids.is_disjoint(&replacement_ids) { + return Err(anyhow!("compaction cannot replace and delete one identity")); + } + self.with_exclusive_lock(|| { + let mut current = self.read_all_unlocked()?; + let mut expected = expected_snapshot.to_vec(); + current.sort_by_key(entry_id); + expected.sort_by_key(entry_id); + if current != expected { + return Ok(false); + } + for row in &replacements { + if !expected_ids.contains(&entry_id(row)) + && current + .iter() + .any(|candidate| entry_id(candidate) == entry_id(row)) + { + return Ok(false); + } + } + current.retain(|row| !deletion_ids.contains(&entry_id(row))); + for row in replacements { + current.retain(|candidate| entry_id(candidate) != entry_id(&row)); + current.push(row); + } + current.sort_by_key(entry_id); + self.write_all_unlocked(¤t)?; + Ok(true) + }) + } + fn read_all_unlocked(&self) -> Result> { if !self.path.exists() { return Ok(Vec::new()); @@ -179,57 +623,47 @@ impl SingleFileMessagePackBackingStore { if bytes.is_empty() { return Ok(Vec::new()); } - decode_store_snapshot(&bytes).or_else(|_| { - rmp_serde::from_slice(&bytes) - .with_context(|| format!("failed to decode MessagePack {}", self.path.display())) - }) + decode_store_snapshot(&bytes) + .or_else(|_| rmp_serde::from_slice(&bytes)) + .with_context(|| format!("failed to decode MessagePack {}", self.path.display())) + } + + /// Reads one filesystem snapshot without creating or opening the sibling + /// CultCache lock file. + /// + /// This is deliberately narrower than `pull_all`: use it only at a + /// provider-owned, read-only crossing where the producer replaces the + /// complete file atomically and the consumer has no authority to mutate + /// the provider directory. An absent file is an empty snapshot. Writers + /// and locally owned stores must continue to use the locked APIs. + pub fn pull_all_read_only_snapshot(&self) -> Result> { + self.read_all_unlocked() } 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)) + remove_abandoned_staging_files(&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); - let mut candidate = OpenOptions::new() + let mut staged = OpenOptions::new() .create_new(true) .write(true) .open(&tmp_path) - .with_context(|| format!("failed to open {}", tmp_path.display()))?; - candidate + .with_context(|| format!("failed to create {}", tmp_path.display()))?; + use std::io::Write; + staged .write_all(&bytes) .with_context(|| format!("failed to write {}", tmp_path.display()))?; - candidate + staged .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), - )); - } - } + .with_context(|| format!("failed to sync {}", tmp_path.display()))?; + drop(staged); + replace_file_atomically(&tmp_path, &self.path)?; + sync_parent_directory(&self.path)?; Ok(()) } @@ -244,26 +678,13 @@ 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 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 + let result = action(); + fs2::FileExt::unlock(&lock) + .with_context(|| format!("failed to unlock {}", self.lock_path().display()))?; + result } fn open_lock_file(&self) -> Result { @@ -286,14 +707,71 @@ impl SingleFileMessagePackBackingStore { .path .file_name() .map(|value| value.to_os_string()) - .unwrap_or_else(|| "cultcache.msgpack".into()); + .unwrap_or_else(|| "cultcache.cc".into()); lock_name.push(".lock"); self.path.with_file_name(lock_name) } } +struct ReadOnlySharedLockGuard +where + U: FnOnce(&File) -> Result<()>, + D: FnOnce(anyhow::Error), +{ + lock: Option, + unlock: Option, + on_unlock_failure: Option, +} + +impl ReadOnlySharedLockGuard +where + U: FnOnce(&File) -> Result<()>, + D: FnOnce(anyhow::Error), +{ + fn new(lock: File, unlock: U, on_unlock_failure: D) -> Self { + Self { + lock: Some(lock), + unlock: Some(unlock), + on_unlock_failure: Some(on_unlock_failure), + } + } +} + +impl Drop for ReadOnlySharedLockGuard +where + U: FnOnce(&File) -> Result<()>, + D: FnOnce(anyhow::Error), +{ + fn drop(&mut self) { + let (Some(lock), Some(unlock)) = (self.lock.as_ref(), self.unlock.take()) else { + return; + }; + if let Err(error) = unlock(lock) + && let Some(on_unlock_failure) = self.on_unlock_failure.take() + { + on_unlock_failure(error); + } + } +} + +fn unique_batch_ids( + entries: &[CultCacheEnvelope], + label: &str, +) -> Result> { + let ids = entries.iter().map(entry_id).collect::>(); + if ids.len() != entries.len() { + return Err(anyhow!( + "conditional batch {label} set contains duplicate identities" + )); + } + Ok(ids) +} + impl CacheBackingStore for SingleFileMessagePackBackingStore { fn pull_all(&self) -> Result> { + if !self.path.exists() && !self.lock_path().exists() { + return Ok(Vec::new()); + } self.with_shared_lock(|| self.read_all_unlocked()) } @@ -324,6 +802,947 @@ impl CacheBackingStore for SingleFileMessagePackBackingStore { } } +const REDB_ENVELOPES: TableDefinition<&[u8], &[u8]> = TableDefinition::new("cultcache_envelopes"); + +/// Transactional keyed CultCache storage. Each polymorphic `(type, key)` +/// identity is an independent redb row whose value is the exact MessagePack +/// serialization of its `CultCacheEnvelope`. +#[derive(Clone)] +pub struct RedbMessagePackBackingStore { + path: PathBuf, +} + +impl std::fmt::Debug for RedbMessagePackBackingStore { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("RedbMessagePackBackingStore") + .field("path", &self.path) + .finish_non_exhaustive() + } +} + +impl RedbMessagePackBackingStore { + pub fn new(path: impl Into) -> Result { + let path = path.into(); + if let Some(parent) = path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + { + fs::create_dir_all(parent) + .with_context(|| format!("failed to create {}", parent.display()))?; + } + let store = Self { path }; + store.with_database(|database| { + let write = database.begin_write()?; + { + write.open_table(REDB_ENVELOPES)?; + } + write.commit()?; + Ok(()) + })?; + Ok(store) + } + + pub fn path(&self) -> &Path { + &self.path + } + + fn open_database(&self) -> Result { + Database::create(&self.path) + .with_context(|| format!("failed to open redb CultCache {}", self.path.display())) + } + + /// redb permits one open writable database handle per path. The external + /// CultCache lock therefore owns the complete open/transaction/close + /// interval, preserving the fresh-handle and cross-process construction + /// model used by the other backing store. + fn with_database(&self, action: impl FnOnce(&Database) -> Result) -> Result { + let lock_path = self.lock_path(); + if let Some(parent) = lock_path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + { + fs::create_dir_all(parent) + .with_context(|| format!("failed to create {}", parent.display()))?; + } + let lock = OpenOptions::new() + .create(true) + .read(true) + .truncate(false) + .write(true) + .open(&lock_path) + .with_context(|| format!("failed to open {}", lock_path.display()))?; + fs2::FileExt::lock_exclusive(&lock) + .with_context(|| format!("failed to lock {}", lock_path.display()))?; + let result = self.open_database().and_then(|database| action(&database)); + fs2::FileExt::unlock(&lock) + .with_context(|| format!("failed to unlock {}", lock_path.display()))?; + result + } + + fn lock_path(&self) -> PathBuf { + let mut lock_name = self + .path + .file_name() + .map(|value| value.to_os_string()) + .unwrap_or_else(|| "cultcache.cc".into()); + lock_name.push(".lock"); + self.path.with_file_name(lock_name) + } + + pub fn compare_and_swap_entry( + &self, + expected: &CultCacheEnvelope, + replacement: CultCacheEnvelope, + ) -> Result { + if entry_id(expected) != entry_id(&replacement) { + return Err(anyhow!("entry CAS replacement must preserve identity")); + } + self.with_database(|database| { + let write = database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + if read_redb_entry(&table, expected)? != Some(expected.clone()) { + false + } else { + insert_redb_entry(&mut table, &replacement)?; + true + } + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + }) + } + + pub fn insert_entry_if_absent(&self, entry: CultCacheEnvelope) -> Result { + self.with_database(|database| { + let write = database.begin_write()?; + let inserted = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let key = redb_identity(&entry)?; + if table.get(key.as_slice())?.is_some() { + false + } else { + insert_redb_entry(&mut table, &entry)?; + true + } + }; + if inserted { + write.commit()?; + } else { + write.abort()?; + } + Ok(inserted) + }) + } + + pub fn compare_and_swap_batch( + &self, + expected: &[CultCacheEnvelope], + replacements: Vec, + ) -> Result { + if replacements.is_empty() { + return Err(anyhow!( + "conditional batch requires a non-empty replacement set" + )); + } + let expected_ids = unique_batch_ids(expected, "expected")?; + let replacement_ids = unique_batch_ids(&replacements, "replacement")?; + if !expected_ids.is_subset(&replacement_ids) { + return Err(anyhow!( + "conditional batch must replace every expected identity" + )); + } + self.with_database(|database| { + let write = database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let mut valid = true; + for row in expected { + if read_redb_entry(&table, row)? != Some(row.clone()) { + valid = false; + break; + } + } + if valid { + for row in &replacements { + if !expected_ids.contains(&entry_id(row)) + && read_redb_entry(&table, row)?.is_some() + { + valid = false; + break; + } + } + } + if valid { + for row in &replacements { + insert_redb_entry(&mut table, row)?; + } + } + valid + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + }) + } + + pub fn append_if_snapshot_unchanged( + &self, + expected_snapshot: &[CultCacheEnvelope], + additions: Vec, + ) -> Result { + if additions.is_empty() { + return Err(anyhow!("conditional snapshot append requires additions")); + } + unique_batch_ids(expected_snapshot, "expected snapshot")?; + let addition_ids = unique_batch_ids(&additions, "snapshot additions")?; + self.with_database(|database| { + let write = database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let mut current = read_all_redb(&table)?; + let mut expected = expected_snapshot.to_vec(); + current.sort_by_key(entry_id); + expected.sort_by_key(entry_id); + let valid = current == expected + && !current + .iter() + .any(|row| addition_ids.contains(&entry_id(row))); + if valid { + for row in &additions { + insert_redb_entry(&mut table, row)?; + } + } + valid + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + }) + } + + pub fn replace_and_append_if_snapshot_unchanged( + &self, + expected_snapshot: &[CultCacheEnvelope], + replacements: Vec, + ) -> Result { + if replacements.is_empty() { + return Err(anyhow!( + "conditional snapshot replacement requires replacements" + )); + } + unique_batch_ids(expected_snapshot, "expected snapshot")?; + unique_batch_ids(&replacements, "snapshot replacements")?; + self.with_database(|database| { + let write = database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let mut current = read_all_redb(&table)?; + let mut expected = expected_snapshot.to_vec(); + current.sort_by_key(entry_id); + expected.sort_by_key(entry_id); + let valid = current == expected; + if valid { + for row in &replacements { + insert_redb_entry(&mut table, row)?; + } + } + valid + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + }) + } + + /// Atomically replaces and deletes rows only while the complete store + /// snapshot remains byte-exact. This is the compaction primitive for a + /// non-owning keyed handle: a stale reader cannot summarize one history + /// while deleting another. + pub fn replace_and_delete_if_snapshot_unchanged( + &self, + expected_snapshot: &[CultCacheEnvelope], + replacements: Vec, + deletions: &[CultCacheEnvelope], + ) -> Result { + if replacements.is_empty() || deletions.is_empty() { + return Err(anyhow!( + "conditional compaction requires replacements and deletions" + )); + } + let expected_ids = unique_batch_ids(expected_snapshot, "expected snapshot")?; + let replacement_ids = unique_batch_ids(&replacements, "replacement")?; + let deletion_ids = unique_batch_ids(deletions, "deletion")?; + if !deletion_ids.is_subset(&expected_ids) { + return Err(anyhow!( + "compaction deletion is absent from expected snapshot" + )); + } + if !deletion_ids.is_disjoint(&replacement_ids) { + return Err(anyhow!("compaction cannot replace and delete one identity")); + } + self.with_database(|database| { + let write = database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let mut current = read_all_redb(&table)?; + let mut expected = expected_snapshot.to_vec(); + current.sort_by_key(entry_id); + expected.sort_by_key(entry_id); + let mut valid = current == expected; + if valid { + for row in &replacements { + if !expected_ids.contains(&entry_id(row)) + && read_redb_entry(&table, row)?.is_some() + { + valid = false; + break; + } + } + } + if valid { + for identity in deletion_ids { + let key = rmp_serde::to_vec(&identity) + .context("failed to encode CultCache identity")?; + table.remove(key.as_slice())?; + } + for row in &replacements { + insert_redb_entry(&mut table, row)?; + } + } + valid + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + }) + } + + pub fn delete_batch_if_unchanged(&self, expected: &[CultCacheEnvelope]) -> Result { + if expected.is_empty() { + return Err(anyhow!( + "conditional delete requires a non-empty expected set" + )); + } + unique_batch_ids(expected, "delete expected")?; + self.with_database(|database| { + let write = database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let mut valid = true; + for row in expected { + if read_redb_entry(&table, row)? != Some(row.clone()) { + valid = false; + break; + } + } + if valid { + for row in expected { + let key = redb_identity(row)?; + table.remove(key.as_slice())?; + } + } + valid + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + }) + } +} + +struct OwnedRedbInner { + database: Database, + _owner_lock: File, + file_identity: String, +} + +/// A pinned redb store for one long-lived service owner. Unlike the transient +/// redb store, this handle holds the CultCache external lock, an open file +/// handle, and the redb database for its entire lifetime. Clones share that +/// exact ownership authority. +#[derive(Clone)] +pub struct OwnedRedbMessagePackBackingStore { + path: PathBuf, + inner: Arc, +} + +impl std::fmt::Debug for OwnedRedbMessagePackBackingStore { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("OwnedRedbMessagePackBackingStore") + .field("path", &self.path) + .field("file_identity", &self.inner.file_identity) + .finish_non_exhaustive() + } +} + +impl OwnedRedbMessagePackBackingStore { + pub fn new(path: impl Into) -> Result { + let path = path.into(); + if let Some(parent) = path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + { + fs::create_dir_all(parent) + .with_context(|| format!("failed to create {}", parent.display()))?; + } + let lock_path = redb_lock_path(&path); + let owner_lock = OpenOptions::new() + .create(true) + .read(true) + .truncate(false) + .write(true) + .open(&lock_path) + .with_context(|| format!("failed to open {}", lock_path.display()))?; + fs2::FileExt::try_lock_exclusive(&owner_lock).with_context(|| { + format!( + "redb CultCache {} already has an active owner", + path.display() + ) + })?; + + // Give redb ownership of the already pinned file, structurally closing + // path substitution during database creation. The post-open identity + // comparison separately verifies that the pathname still names it. + let database_file = OpenOptions::new() + .create(true) + .read(true) + .truncate(false) + .write(true) + .open(&path) + .with_context(|| format!("failed to pin redb CultCache {}", path.display()))?; + let held_identity = file_identity_from_file(&database_file)?; + let database = Database::builder() + .create_file(database_file) + .with_context(|| format!("failed to open redb CultCache {}", path.display()))?; + let path_identity_file = File::open(&path)?; + let path_identity = file_identity_from_file(&path_identity_file)?; + if held_identity != path_identity { + bail!( + "redb CultCache path identity changed while opening: held {held_identity}, path {path_identity}" + ); + } + let write = database.begin_write()?; + { + write.open_table(REDB_ENVELOPES)?; + } + write.commit()?; + Ok(Self { + path, + inner: Arc::new(OwnedRedbInner { + database, + _owner_lock: owner_lock, + file_identity: held_identity, + }), + }) + } + + pub fn path(&self) -> &Path { + &self.path + } + + pub fn file_identity(&self) -> &str { + &self.inner.file_identity + } + + pub fn require_file_identity(&self, expected: &str) -> Result<()> { + if expected != self.file_identity() { + bail!( + "redb CultCache file identity mismatch: expected {expected}, owned {}", + self.file_identity() + ); + } + Ok(()) + } + + pub fn validate_path_identity(&self) -> Result<()> { + let current_file = File::open(&self.path) + .with_context(|| format!("owned redb path {} is missing", self.path.display()))?; + let current = file_identity_from_file(¤t_file)?; + self.require_file_identity(¤t).with_context(|| { + format!( + "owned redb path {} no longer names the pinned file", + self.path.display() + ) + }) + } + + pub fn compare_and_swap_entry( + &self, + expected: &CultCacheEnvelope, + replacement: CultCacheEnvelope, + ) -> Result { + if entry_id(expected) != entry_id(&replacement) { + return Err(anyhow!("entry CAS replacement must preserve identity")); + } + let write = self.inner.database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + if read_redb_entry(&table, expected)? != Some(expected.clone()) { + false + } else { + insert_redb_entry(&mut table, &replacement)?; + true + } + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + } + + pub fn insert_entry_if_absent(&self, entry: CultCacheEnvelope) -> Result { + let write = self.inner.database.begin_write()?; + let inserted = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let key = redb_identity(&entry)?; + if table.get(key.as_slice())?.is_some() { + false + } else { + insert_redb_entry(&mut table, &entry)?; + true + } + }; + if inserted { + write.commit()?; + } else { + write.abort()?; + } + Ok(inserted) + } + + pub fn compare_and_swap_batch( + &self, + expected: &[CultCacheEnvelope], + replacements: Vec, + ) -> Result { + if replacements.is_empty() { + return Err(anyhow!( + "conditional batch requires a non-empty replacement set" + )); + } + let expected_ids = unique_batch_ids(expected, "expected")?; + let replacement_ids = unique_batch_ids(&replacements, "replacement")?; + if !expected_ids.is_subset(&replacement_ids) { + return Err(anyhow!( + "conditional batch must replace every expected identity" + )); + } + let write = self.inner.database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let mut valid = true; + for row in expected { + if read_redb_entry(&table, row)? != Some(row.clone()) { + valid = false; + break; + } + } + if valid { + for row in &replacements { + if !expected_ids.contains(&entry_id(row)) + && read_redb_entry(&table, row)?.is_some() + { + valid = false; + break; + } + } + } + if valid { + for row in &replacements { + insert_redb_entry(&mut table, row)?; + } + } + valid + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + } + + pub fn append_if_snapshot_unchanged( + &self, + expected_snapshot: &[CultCacheEnvelope], + additions: Vec, + ) -> Result { + if additions.is_empty() { + return Err(anyhow!("conditional snapshot append requires additions")); + } + unique_batch_ids(expected_snapshot, "expected snapshot")?; + let addition_ids = unique_batch_ids(&additions, "snapshot additions")?; + let write = self.inner.database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let mut current = read_all_redb(&table)?; + let mut expected = expected_snapshot.to_vec(); + current.sort_by_key(entry_id); + expected.sort_by_key(entry_id); + let valid = current == expected + && !current + .iter() + .any(|row| addition_ids.contains(&entry_id(row))); + if valid { + for row in &additions { + insert_redb_entry(&mut table, row)?; + } + } + valid + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + } + + pub fn replace_and_append_if_snapshot_unchanged( + &self, + expected_snapshot: &[CultCacheEnvelope], + replacements: Vec, + ) -> Result { + if replacements.is_empty() { + return Err(anyhow!( + "conditional snapshot replacement requires replacements" + )); + } + unique_batch_ids(expected_snapshot, "expected snapshot")?; + unique_batch_ids(&replacements, "snapshot replacements")?; + let write = self.inner.database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let mut current = read_all_redb(&table)?; + let mut expected = expected_snapshot.to_vec(); + current.sort_by_key(entry_id); + expected.sort_by_key(entry_id); + let valid = current == expected; + if valid { + for row in &replacements { + insert_redb_entry(&mut table, row)?; + } + } + valid + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + } + + pub fn delete_batch_if_unchanged(&self, expected: &[CultCacheEnvelope]) -> Result { + if expected.is_empty() { + return Err(anyhow!( + "conditional delete requires a non-empty expected set" + )); + } + let expected_ids = unique_batch_ids(expected, "delete expected")?; + let write = self.inner.database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let mut valid = true; + for row in expected { + if read_redb_entry(&table, row)? != Some(row.clone()) { + valid = false; + break; + } + } + if valid { + for identity in expected_ids { + let key = rmp_serde::to_vec(&identity) + .context("failed to encode CultCache identity")?; + table.remove(key.as_slice())?; + } + } + valid + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + } + + /// Atomically replaces and deletes rows only while the complete store + /// snapshot remains byte-exact. This is the compaction primitive: a stale + /// reader cannot summarize one history while deleting another. + pub fn replace_and_delete_if_snapshot_unchanged( + &self, + expected_snapshot: &[CultCacheEnvelope], + replacements: Vec, + deletions: &[CultCacheEnvelope], + ) -> Result { + if replacements.is_empty() || deletions.is_empty() { + return Err(anyhow!( + "conditional compaction requires replacements and deletions" + )); + } + let expected_ids = unique_batch_ids(expected_snapshot, "expected snapshot")?; + let replacement_ids = unique_batch_ids(&replacements, "replacement")?; + let deletion_ids = unique_batch_ids(deletions, "deletion")?; + if !deletion_ids.is_subset(&expected_ids) { + return Err(anyhow!( + "compaction deletion is absent from expected snapshot" + )); + } + if !deletion_ids.is_disjoint(&replacement_ids) { + return Err(anyhow!("compaction cannot replace and delete one identity")); + } + let write = self.inner.database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let mut current = read_all_redb(&table)?; + let mut expected = expected_snapshot.to_vec(); + current.sort_by_key(entry_id); + expected.sort_by_key(entry_id); + let mut valid = current == expected; + if valid { + for row in &replacements { + if !expected_ids.contains(&entry_id(row)) + && read_redb_entry(&table, row)?.is_some() + { + valid = false; + break; + } + } + } + if valid { + for identity in deletion_ids { + let key = rmp_serde::to_vec(&identity) + .context("failed to encode CultCache identity")?; + table.remove(key.as_slice())?; + } + for row in &replacements { + insert_redb_entry(&mut table, row)?; + } + } + valid + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + } +} + +impl CacheBackingStore for OwnedRedbMessagePackBackingStore { + fn pull_all(&self) -> Result> { + let read = self.inner.database.begin_read()?; + let table = read.open_table(REDB_ENVELOPES)?; + read_all_redb(&table) + } + + fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + let write = self.inner.database.begin_write()?; + { + let mut table = write.open_table(REDB_ENVELOPES)?; + insert_redb_entry(&mut table, entry)?; + } + write.commit()?; + Ok(()) + } + + fn delete(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + let write = self.inner.database.begin_write()?; + { + let mut table = write.open_table(REDB_ENVELOPES)?; + let key = redb_identity(entry)?; + table.remove(key.as_slice())?; + } + write.commit()?; + Ok(()) + } + + fn push_all(&mut self, entries: &[CultCacheEnvelope], _options: PushAllOptions) -> Result<()> { + unique_batch_ids(entries, "push all")?; + let write = self.inner.database.begin_write()?; + { + let mut table = write.open_table(REDB_ENVELOPES)?; + let keys = table + .iter()? + .map(|row| Ok(row?.0.value().to_vec())) + .collect::>>()?; + for key in keys { + table.remove(key.as_slice())?; + } + for entry in entries { + insert_redb_entry(&mut table, entry)?; + } + } + write.commit()?; + Ok(()) + } +} + +fn redb_lock_path(path: &Path) -> PathBuf { + let mut lock_name = path + .file_name() + .map(|value| value.to_os_string()) + .unwrap_or_else(|| "cultcache.cc".into()); + lock_name.push(".lock"); + path.with_file_name(lock_name) +} + +#[cfg(unix)] +fn file_identity_from_file(file: &File) -> Result { + use std::os::unix::fs::MetadataExt; + let metadata = file.metadata()?; + Ok(format!( + "unix:{:016x}:{:016x}", + metadata.dev(), + metadata.ino() + )) +} + +#[cfg(windows)] +fn file_identity_from_file(file: &File) -> Result { + use std::mem::zeroed; + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::Storage::FileSystem::{ + BY_HANDLE_FILE_INFORMATION, GetFileInformationByHandle, + }; + let mut information: BY_HANDLE_FILE_INFORMATION = unsafe { zeroed() }; + let success = + unsafe { GetFileInformationByHandle(file.as_raw_handle() as _, &mut information) }; + if success == 0 { + return Err(std::io::Error::last_os_error()) + .context("failed to read Windows file identity"); + } + let index = ((information.nFileIndexHigh as u64) << 32) | information.nFileIndexLow as u64; + Ok(format!( + "windows:{:08x}:{index:016x}", + information.dwVolumeSerialNumber + )) +} + +fn redb_identity(entry: &CultCacheEnvelope) -> Result> { + rmp_serde::to_vec(&entry_id(entry)).context("failed to encode CultCache identity") +} + +fn read_redb_entry( + table: &impl ReadableTable<&'static [u8], &'static [u8]>, + identity: &CultCacheEnvelope, +) -> Result> { + let key = redb_identity(identity)?; + table + .get(key.as_slice())? + .map(|value| { + rmp_serde::from_slice(value.value()).context("failed to decode redb CultCache envelope") + }) + .transpose() +} + +fn insert_redb_entry( + table: &mut redb::Table<&[u8], &[u8]>, + entry: &CultCacheEnvelope, +) -> Result<()> { + let key = redb_identity(entry)?; + let value = rmp_serde::to_vec(entry).context("failed to encode redb CultCache envelope")?; + table.insert(key.as_slice(), value.as_slice())?; + Ok(()) +} + +fn read_all_redb( + table: &impl ReadableTable<&'static [u8], &'static [u8]>, +) -> Result> { + let mut entries = table + .iter()? + .map(|row| { + let (_, value) = row?; + rmp_serde::from_slice(value.value()).context("failed to decode redb CultCache envelope") + }) + .collect::>>()?; + entries.sort_by_key(entry_id); + Ok(entries) +} + +impl CacheBackingStore for RedbMessagePackBackingStore { + fn pull_all(&self) -> Result> { + self.with_database(|database| { + let read = database.begin_read()?; + let table = read.open_table(REDB_ENVELOPES)?; + read_all_redb(&table) + }) + } + + fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + self.with_database(|database| { + let write = database.begin_write()?; + { + let mut table = write.open_table(REDB_ENVELOPES)?; + insert_redb_entry(&mut table, entry)?; + } + write.commit()?; + Ok(()) + }) + } + + fn delete(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + self.with_database(|database| { + let write = database.begin_write()?; + { + let mut table = write.open_table(REDB_ENVELOPES)?; + let key = redb_identity(entry)?; + table.remove(key.as_slice())?; + } + write.commit()?; + Ok(()) + }) + } + + fn push_all(&mut self, entries: &[CultCacheEnvelope], _options: PushAllOptions) -> Result<()> { + unique_batch_ids(entries, "push all")?; + self.with_database(|database| { + let write = database.begin_write()?; + { + let mut table = write.open_table(REDB_ENVELOPES)?; + let keys = table + .iter()? + .map(|row| Ok(row?.0.value().to_vec())) + .collect::>>()?; + for key in keys { + table.remove(key.as_slice())?; + } + for entry in entries { + insert_redb_entry(&mut table, entry)?; + } + } + write.commit()?; + Ok(()) + }) + } +} + struct CultCacheStoreRegistration { store: Box, types: BTreeSet, @@ -331,8 +1750,7 @@ struct CultCacheStoreRegistration { pub struct CultCache { definitions: BTreeMap, - schema_name_definitions: BTreeMap, - entries: BTreeMap, + entries: BTreeMap<(String, String), CultCacheEnvelope>, stores: Vec, } @@ -340,7 +1758,6 @@ impl CultCache { pub fn new() -> Self { Self { definitions: BTreeMap::new(), - schema_name_definitions: BTreeMap::new(), entries: BTreeMap::new(), stores: Vec::new(), } @@ -360,18 +1777,7 @@ impl CultCache { T::TYPE )); } - if let Some(existing_type) = self.schema_name_definitions.get(T::SCHEMA_NAME) - && existing_type != T::TYPE - { - return Err(anyhow!( - "CultCache schema name {:?} is already registered for type {:?}", - T::SCHEMA_NAME, - existing_type - )); - } self.definitions.insert(T::TYPE.to_string(), T::SCHEMA_NAME); - self.schema_name_definitions - .insert(T::SCHEMA_NAME.to_string(), T::TYPE.to_string()); Ok(()) } @@ -379,6 +1785,10 @@ impl CultCache { self.register_entry_type::() } + pub fn registered_entry_types(&self) -> Vec { + self.definitions.keys().cloned().collect() + } + pub fn register_registry(&mut self, registry: R) -> Result<&mut Self> { registry.register_entries(self)?; Ok(self) @@ -402,26 +1812,27 @@ impl CultCache { pub fn pull_all_backing_stores(&mut self) -> Result<()> { self.entries.clear(); let known_types: BTreeSet = self.definitions.keys().cloned().collect(); - let schema_name_definitions = self.schema_name_definitions.clone(); for registration in &mut self.stores { - for mut entry in registration.store.pull_all()? { - let Some(canonical_type) = resolve_registered_type( - &known_types, - &schema_name_definitions, - &entry.r#type, - ) else { + for entry in registration.store.pull_all()? { + if !known_types.contains(&entry.r#type) { return Err(anyhow!( "No schema is registered for persisted entry type {:?}", entry.r#type )); - }; - entry.r#type = canonical_type; + } self.entries.insert(entry_id(&entry), entry); } } Ok(()) } + /// Returns the exact envelopes in the currently loaded cache image. + /// Callers that need optimistic concurrency can validate typed documents + /// and capture CAS expectations from one coherent read. + pub fn snapshot_envelopes(&self) -> Vec { + self.entries.values().cloned().collect() + } + pub fn get(&self, key: &str) -> Result> { self.require_entry_type::()?; let Some(entry) = self.entries.get(&entry_id_parts(T::TYPE, key)) else { @@ -472,32 +1883,45 @@ impl CultCache { Ok(values) } - pub fn put(&mut self, key: impl Into, value: &T) -> Result { + pub fn get_all_with_keys(&self) -> Result> { self.require_entry_type::()?; - let key = key.into(); - let payload = rmp_serde::to_vec(value).with_context(|| { - format!( - "failed to encode CultCache entry {:?} at key {:?} as {}", - T::TYPE, - key, - T::SCHEMA_NAME - ) - })?; - let parsed: T = rmp_serde::from_slice(&payload).with_context(|| { - format!( - "failed to validate CultCache entry {:?} at key {:?} as {}", - T::TYPE, - key, - T::SCHEMA_NAME - ) - })?; - let entry = CultCacheEnvelope { - key: key.clone(), - r#type: T::TYPE.to_string(), - payload, - stored_at: now_utc_second(), - schema_id: Some(T::TYPE.to_string()), - }; + let mut values = Vec::new(); + for entry in self.entries.values() { + if entry.r#type != T::TYPE { + continue; + } + values.push(( + entry.key.clone(), + rmp_serde::from_slice(&entry.payload).with_context(|| { + format!( + "failed to decode CultCache entry {:?} at key {:?} as {}", + T::TYPE, + entry.key, + T::SCHEMA_NAME + ) + })?, + )); + } + Ok(values) + } + + pub fn soa(&self) -> Result> { + let keyed_rows = self.get_all_with_keys::()?; + let (keys, documents): (Vec<_>, Vec<_>) = keyed_rows.into_iter().unzip(); + let columns = T::soa_columns(&documents) + .into_iter() + .map(|(name, values)| (name.to_string(), values)) + .collect(); + Ok(CultSoaTable { + keys, + documents, + columns, + _document: PhantomData, + }) + } + + pub fn put(&mut self, key: impl Into, value: &T) -> Result { + let (entry, parsed) = self.prepare_entry(key, value)?; let route = self.resolve_route_indices(T::TYPE); let Some(primary_index) = route.first().copied() else { return Err(anyhow!( @@ -513,7 +1937,176 @@ impl CultCache { Ok(parsed) } + pub fn prepare_entry( + &self, + key: impl Into, + value: &T, + ) -> Result<(CultCacheEnvelope, T)> { + self.require_entry_type::()?; + let key = key.into(); + let payload = rmp_serde::to_vec(value).with_context(|| { + format!( + "failed to encode CultCache entry {:?} at key {:?} as {}", + T::TYPE, + key, + T::SCHEMA_NAME + ) + })?; + self.finish_prepared_entry::(key, payload) + } + + /// Prepares a typed entry while encoding ordinary nested structs as + /// MessagePack maps. DatabaseEntry itself remains the stable numeric-slot + /// tuple; named nested fields prevent serde's omitted optional fields from + /// shifting later values into the wrong positions. + pub fn prepare_entry_named( + &self, + key: impl Into, + value: &T, + ) -> Result<(CultCacheEnvelope, T)> { + self.require_entry_type::()?; + let key = key.into(); + let payload = rmp_serde::to_vec_named(value).with_context(|| { + format!( + "failed to encode named CultCache entry {:?} at key {:?} as {}", + T::TYPE, + key, + T::SCHEMA_NAME + ) + })?; + self.finish_prepared_entry::(key, payload) + } + + fn finish_prepared_entry( + &self, + key: String, + payload: Vec, + ) -> Result<(CultCacheEnvelope, T)> { + let parsed: T = rmp_serde::from_slice(&payload).with_context(|| { + format!( + "failed to validate CultCache entry {:?} at key {:?} as {}", + T::TYPE, + key, + T::SCHEMA_NAME + ) + })?; + let entry = CultCacheEnvelope { + key: key.clone(), + r#type: T::TYPE.to_string(), + payload, + stored_at: now_utc_second(), + schema_id: Some(T::TYPE.to_string()), + }; + Ok((entry, parsed)) + } + + pub fn put_prepared_batch(&mut self, staged: Vec) -> Result<()> { + if staged.is_empty() { + return Err(anyhow!("CultCache batch must contain at least one entry")); + } + let mut store_index = None; + for entry in staged.iter().chain(self.entries.values()) { + if !self.definitions.contains_key(&entry.r#type) { + return Err(anyhow!( + "CultCache batch entry type {:?} is not registered", + entry.r#type + )); + } + let route = self.resolve_route_indices(&entry.r#type); + if route.len() != 1 { + return Err(anyhow!( + "CultCache atomic batch requires exactly one backing-store route for type {:?}, found {}", + entry.r#type, + route.len() + )); + } + match store_index { + Some(expected) if expected != route[0] => { + return Err(anyhow!("CultCache atomic batch cannot span backing stores")); + } + None => store_index = Some(route[0]), + _ => {} + } + } + let store_index = store_index.expect("non-empty batch establishes a store route"); + let mut next = self.entries.clone(); + for entry in staged { + next.insert(entry_id(&entry), entry); + } + let snapshot: Vec<_> = next.values().cloned().collect(); + self.stores[store_index] + .store + .push_all(&snapshot, PushAllOptions::default())?; + self.entries = next; + Ok(()) + } + pub fn put_envelope(&mut self, entry: CultCacheEnvelope) -> Result { + let parsed = self.validate_envelope::(&entry)?; + let route = self.resolve_route_indices(T::TYPE); + let Some(primary_index) = route.first().copied() else { + return Err(anyhow!( + "No backing store is registered for entry type {:?}", + T::TYPE + )); + }; + self.stores[primary_index].store.push(&entry)?; + for mirror_index in route.iter().skip(1).copied() { + self.stores[mirror_index].store.push(&entry)?; + } + self.entries.insert(entry_id(&entry), entry); + Ok(parsed) + } + + /// Publishes an envelope for a registered dynamic document type without + /// claiming compile-time knowledge of its payload schema. This is the + /// CultMesh/schema-registry boundary: callers own payload validation before + /// admission; CultCache still enforces registered type, identity, timestamp, + /// routing, and persistence. + pub fn put_raw_envelope(&mut self, entry: CultCacheEnvelope) -> Result<()> { + if !self.definitions.contains_key(&entry.r#type) { + return Err(anyhow!( + "No entry type registered for CultCache envelope type {:?}", + entry.r#type + )); + } + if entry.key.trim().is_empty() { + return Err(anyhow!( + "CultCache envelope keys for type {:?} must be non-empty", + entry.r#type + )); + } + if entry.stored_at.trim().is_empty() { + return Err(anyhow!( + "CultCache envelope stored_at for type {:?} must be non-empty", + entry.r#type + )); + } + let route = self.resolve_route_indices(&entry.r#type); + let Some(primary_index) = route.first().copied() else { + return Err(anyhow!( + "No backing store is registered for entry type {:?}", + entry.r#type + )); + }; + self.stores[primary_index].store.push(&entry)?; + for mirror_index in route.iter().skip(1).copied() { + self.stores[mirror_index].store.push(&entry)?; + } + self.entries.insert(entry_id(&entry), entry); + Ok(()) + } + + /// Validates and admits an existing envelope into this cache image without + /// publishing it to any backing store. This is the typed read primitive for + /// owner-filtered views over a shared polymorphic store. + pub fn load_envelope(&mut self, entry: CultCacheEnvelope) -> Result { + let parsed = self.validate_envelope::(&entry)?; + self.entries.insert(entry_id(&entry), entry); + Ok(parsed) + } + + fn validate_envelope(&self, entry: &CultCacheEnvelope) -> Result { self.require_entry_type::()?; if entry.r#type != T::TYPE { return Err(anyhow!( @@ -535,27 +2128,14 @@ impl CultCache { )); } - let parsed: T = rmp_serde::from_slice(&entry.payload).with_context(|| { + rmp_serde::from_slice(&entry.payload).with_context(|| { format!( "failed to validate CultCache envelope {:?} at key {:?} as {}", T::TYPE, entry.key, T::SCHEMA_NAME ) - })?; - let route = self.resolve_route_indices(T::TYPE); - let Some(primary_index) = route.first().copied() else { - return Err(anyhow!( - "No backing store is registered for entry type {:?}", - T::TYPE - )); - }; - self.stores[primary_index].store.push(&entry)?; - for mirror_index in route.iter().skip(1).copied() { - self.stores[mirror_index].store.push(&entry)?; - } - self.entries.insert(entry_id(&entry), entry); - Ok(parsed) + }) } pub fn update(&mut self, key: &str, updater: F) -> Result @@ -592,6 +2172,25 @@ impl CultCache { self.entries.values().cloned().collect() } + pub fn snapshot_soa(&self) -> CultCacheSoaTable { + CultCacheSoaTable::from_envelopes(self.snapshot()) + } + + pub fn load_soa(&mut self, table: CultCacheSoaTable) -> Result<()> { + let known_types: BTreeSet = self.definitions.keys().cloned().collect(); + self.entries.clear(); + for entry in table.into_envelopes()? { + if !known_types.contains(&entry.r#type) { + return Err(anyhow!( + "No schema is registered for SoA entry type {:?}", + entry.r#type + )); + } + self.entries.insert(entry_id(&entry), entry); + } + Ok(()) + } + fn require_entry_type(&self) -> Result<()> { match self.definitions.get(T::TYPE) { Some(schema_name) if *schema_name == T::SCHEMA_NAME => Ok(()), @@ -620,7 +2219,6 @@ impl CultCache { .filter_map(|(index, registration)| registration.types.is_empty().then_some(index)) .collect() } - } impl Default for CultCache { @@ -629,47 +2227,40 @@ impl Default for CultCache { } } -fn entry_id(entry: &CultCacheEnvelope) -> String { +fn entry_id(entry: &CultCacheEnvelope) -> (String, String) { entry_id_parts(&entry.r#type, &entry.key) } -fn entry_id_parts(r#type: &str, key: &str) -> String { - format!("{type}::{key}", type = r#type) -} - -fn resolve_registered_type( - known_types: &BTreeSet, - schema_name_definitions: &BTreeMap, - persisted_type: &str, -) -> Option { - if known_types.contains(persisted_type) { - return Some(persisted_type.to_string()); - } - - schema_name_definitions.get(persisted_type).cloned() +fn entry_id_parts(r#type: &str, key: &str) -> (String, String) { + (r#type.to_string(), key.to_string()) } fn now_utc_second() -> String { chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true) } -fn encode_store_snapshot(entries: &[CultCacheEnvelope]) -> PersistedStoreSnapshot { - let mut schema_names = BTreeSet::::new(); +fn encode_store_snapshot(entries: &[CultCacheEnvelope]) -> Result { + let mut schema_types = BTreeMap::::new(); for entry in entries { - schema_names.insert( - entry - .schema_id - .clone() - .unwrap_or_else(|| entry.r#type.clone()), - ); + let schema_id = entry + .schema_id + .clone() + .unwrap_or_else(|| entry.r#type.clone()); + if let Some(existing_type) = schema_types.insert(schema_id.clone(), entry.r#type.clone()) { + ensure!( + existing_type == entry.r#type, + "CultCache schema {schema_id:?} cannot identify both {existing_type:?} and {:?}", + entry.r#type + ); + } } - let catalog = schema_names + let catalog = schema_types .into_iter() - .map(|schema_id| { + .map(|(schema_id, document_type)| { PersistedSchemaCatalogEntry( schema_id.clone(), - schema_id.clone(), + document_type, format!("{schema_id}.v1"), schema_id.clone(), format!( @@ -697,7 +2288,11 @@ fn encode_store_snapshot(entries: &[CultCacheEnvelope]) -> PersistedStoreSnapsho }) .collect(); - PersistedStoreSnapshot("cultcache.store.v1".to_string(), catalog, records) + Ok(PersistedStoreSnapshot( + "cultcache.store.v1".to_string(), + catalog, + records, + )) } fn decode_store_snapshot(bytes: &[u8]) -> Result> { @@ -716,25 +2311,13 @@ fn decode_store_snapshot(bytes: &[u8]) -> Result> { .2 .into_iter() .map(|record| { - let r#type = match catalog.get(&record.1) { - Some(schema_name) => schema_name.clone(), - None => { - let schema_version = infer_schema_version_from_payload(&record.3).ok_or_else(|| { - anyhow!( - "CultCache record {:?} references missing schema {:?}", - record.0, - record.1 - ) - })?; - infer_schema_name(&schema_version).ok_or_else(|| { - anyhow!( - "CultCache record {:?} references missing schema {:?}", - record.0, - record.1 - ) - })? - } - }; + let r#type = catalog.get(&record.1).cloned().ok_or_else(|| { + anyhow!( + "CultCache record {:?} references missing schema {:?}", + record.0, + record.1 + ) + })?; Ok(CultCacheEnvelope { key: record.0, r#type, @@ -746,68 +2329,6 @@ fn decode_store_snapshot(bytes: &[u8]) -> Result> { .collect() } -fn infer_schema_version_from_payload(payload: &[u8]) -> Option { - let mut offset = 0usize; - read_array_header(payload, &mut offset)?; - read_string(payload, &mut offset) -} - -fn read_array_header(payload: &[u8], offset: &mut usize) -> Option { - let marker = *payload.get(*offset)?; - *offset += 1; - match marker { - 0x90..=0x9f => Some((marker & 0x0f) as u32), - 0xdc => { - let bytes = payload.get(*offset..(*offset + 2))?; - *offset += 2; - Some(u16::from_be_bytes([bytes[0], bytes[1]]) as u32) - } - 0xdd => { - let bytes = payload.get(*offset..(*offset + 4))?; - *offset += 4; - Some(u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]])) - } - _ => None, - } -} - -fn read_string(payload: &[u8], offset: &mut usize) -> Option { - let marker = *payload.get(*offset)?; - *offset += 1; - let length = match marker { - 0xa0..=0xbf => (marker & 0x1f) as usize, - 0xd9 => { - let length = *payload.get(*offset)? as usize; - *offset += 1; - length - } - 0xda => { - let bytes = payload.get(*offset..(*offset + 2))?; - *offset += 2; - u16::from_be_bytes([bytes[0], bytes[1]]) as usize - } - 0xdb => { - let bytes = payload.get(*offset..(*offset + 4))?; - *offset += 4; - u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as usize - } - _ => return None, - }; - - let bytes = payload.get(*offset..(*offset + length))?; - std::str::from_utf8(bytes).ok().map(str::to_string) -} - -fn infer_schema_name(schema_version: &str) -> Option { - let marker = schema_version.rfind(".v")?; - let version = schema_version.get((marker + 2)..)?; - if marker == 0 || version.is_empty() || !version.bytes().all(|value| value.is_ascii_digit()) { - return None; - } - - Some(schema_version[..marker].to_string()) -} - fn escape_json_string(value: &str) -> String { value.replace('\\', "\\\\").replace('"', "\\\"") } @@ -816,188 +2337,108 @@ fn temporary_path_for(path: &Path) -> PathBuf { let mut file_name = path .file_name() .map(|value| value.to_os_string()) - .unwrap_or_else(|| "cultcache.msgpack".into()); + .unwrap_or_else(|| "cultcache.cc".into()); file_name.push(format!(".{}.tmp", uuid::Uuid::new_v4())); 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; +fn remove_abandoned_staging_files(destination: &Path) -> Result<()> { + let Some(parent) = destination.parent() else { + return Ok(()); }; - let Ok(uuid) = uuid::Uuid::parse_str(value) else { - return false; + let Some(destination_name) = destination.file_name().and_then(|value| value.to_str()) else { + return Ok(()); }; - 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()); + let prefix = format!("{destination_name}."); + for entry in fs::read_dir(parent).with_context(|| { + format!( + "failed to inspect {} for abandoned staging files", + parent.display() + ) + })? { + let entry = entry?; + let Some(name) = entry.file_name().to_str().map(str::to_owned) else { + continue; + }; + let Some(candidate) = name + .strip_prefix(&prefix) + .and_then(|value| value.strip_suffix(".tmp")) + else { + continue; + }; + if uuid::Uuid::parse_str(candidate).is_err() || !entry.file_type()?.is_file() { + continue; } - } - 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()))?; + fs::remove_file(entry.path()).with_context(|| { + format!( + "failed to remove abandoned staging file {}", + entry.path().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(|| { +fn replace_file_atomically(staged: &Path, destination: &Path) -> Result<()> { + fs::rename(staged, destination).with_context(|| { format!( "failed to atomically replace {} with {}", - target.display(), - candidate.display() + destination.display(), + staged.display() ) }) } #[cfg(windows)] -fn replace_file(candidate: &Path, target: &Path) -> Result<()> { +fn replace_file_atomically(staged: &Path, destination: &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 + let staged_wide = staged .as_os_str() .encode_wide() .chain(std::iter::once(0)) .collect::>(); - let target_wide = target + let destination_wide = destination .as_os_str() .encode_wide() .chain(std::iter::once(0)) .collect::>(); - let replaced = unsafe { + let success = unsafe { MoveFileExW( - candidate_wide.as_ptr(), - target_wide.as_ptr(), + staged_wide.as_ptr(), + destination_wide.as_ptr(), MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH, ) }; - if replaced == 0 { + if success == 0 { return Err(std::io::Error::last_os_error()).with_context(|| { format!( "failed to atomically replace {} with {}", - target.display(), - candidate.display() + destination.display(), + staged.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<()> { +fn sync_parent_directory(path: &Path) -> Result<()> { + let Some(parent) = path.parent() else { + return Ok(()); + }; File::open(parent) - .with_context(|| format!("failed to open directory {}", parent.display()))? - .sync_all() - .with_context(|| format!("failed to flush directory {}", parent.display())) + .and_then(|directory| directory.sync_all()) + .with_context(|| format!("failed to sync {}", parent.display())) } -#[cfg(not(unix))] -fn sync_parent_directory(_parent: &Path) -> Result<()> { +#[cfg(windows)] +fn sync_parent_directory(_path: &Path) -> Result<()> { + // MoveFileExW with MOVEFILE_WRITE_THROUGH owns publication durability. Ok(()) } @@ -1005,6 +2446,37 @@ fn sync_parent_directory(_parent: &Path) -> Result<()> { mod tests { use super::*; use pretty_assertions::assert_eq; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::mpsc; + use std::time::Duration; + + struct RefusingBatchStore { + inner: SingleFileMessagePackBackingStore, + refuse_batch: Arc, + } + + impl CacheBackingStore for RefusingBatchStore { + fn pull_all(&self) -> Result> { + self.inner.pull_all() + } + fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + self.inner.push(entry) + } + fn delete(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + self.inner.delete(entry) + } + fn push_all( + &mut self, + entries: &[CultCacheEnvelope], + options: PushAllOptions, + ) -> Result<()> { + if self.refuse_batch.load(Ordering::SeqCst) { + return Err(anyhow!("injected batch refusal")); + } + self.inner.push_all(entries, options) + } + } #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] #[cultcache(type = "settings")] @@ -1025,279 +2497,111 @@ mod tests { } #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] - #[cultcache(type = "runtime-policy", schema = "tests.schema_stamped_entry")] - struct SchemaStamped { - #[cultcache(key = 0)] - schema_version: String, - #[cultcache(key = 1)] - name: String, - #[cultcache(key = 2)] - value: String, + #[cultcache(type = "binary-record")] + struct BinaryRecord { + #[cultcache(key = 0, bytes)] + required: Vec, + #[cultcache(key = 1, bytes, default)] + optional: Vec, } + #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] + struct NestedOptionalValue { + label: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + count: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + symbol: Option, + } + + #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] + #[cultcache(type = "nested-note")] + struct NestedNote { + #[cultcache(key = 0)] + value: NestedOptionalValue, + } + + cultcache_soa!(Settings { + "theme" => |row: &Settings| row.theme.clone(), + "retries" => |row: &Settings| row.retries, + }); + 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<()> { + fn captured_snapshot_cannot_outlive_a_shared_consequence_gate() -> Result<()> { let temp = tempfile::tempdir()?; - let store_path = temp.path().join("cache.msgpack"); + let store_path = temp.path().join("authority.cc"); 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 released = CultCacheEnvelope { + key: "brake".into(), + r#type: "brake".into(), + payload: b"released".to_vec(), + stored_at: "2026-07-20T00:00:00Z".into(), + schema_id: Some("brake".into()), }; + store.push(&released)?; - 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 writer_path = store_path.clone(); + let (started_tx, started_rx) = mpsc::channel(); + let (finished_tx, finished_rx) = mpsc::channel(); + store.with_read_only_shared_snapshot(|captured| { + assert_eq!(captured, vec![released.clone()]); + std::thread::spawn(move || { + started_tx.send(()).unwrap(); + let engaged = CultCacheEnvelope { + payload: b"engaged".to_vec(), + ..released + }; + let mut writer = SingleFileMessagePackBackingStore::new(writer_path); + writer.push(&engaged).unwrap(); + finished_tx.send(()).unwrap(); + }); + started_rx.recv_timeout(Duration::from_secs(1))?; + assert!( + finished_rx + .recv_timeout(Duration::from_millis(100)) + .is_err() + ); + Ok(()) + })?; + finished_rx.recv_timeout(Duration::from_secs(1))?; + assert_eq!(store.pull_all()?[0].payload, b"engaged"); + Ok(()) + } - 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]) + #[test] + fn post_action_unlock_failure_is_diagnostic_not_action_result() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("authority.cc"); + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + store.push(&CultCacheEnvelope { + key: "brake".into(), + r#type: "brake".into(), + payload: b"released".to_vec(), + stored_at: "2026-07-20T00:00:00Z".into(), + schema_id: Some("brake".into()), + })?; + + let mut diagnostic = None; + let result = store.with_read_only_shared_snapshot_using( + |_| Ok(42), + |_| Err(anyhow!("injected post-action unlock failure")), + |error| diagnostic = Some(error.to_string()), + ); + + assert_eq!(result?, 42); + assert_eq!( + diagnostic.as_deref(), + Some("injected post-action unlock failure") ); - assert!(stale_candidate_paths(&store_path)?.is_empty()); Ok(()) } #[test] fn familiar_cultcache_flow_persists_and_reloads_typed_documents() -> Result<()> { let temp = tempfile::tempdir()?; - let store_path = temp.path().join("cache.msgpack"); + let store_path = temp.path().join("cache.cc"); let settings = Settings { theme: "ash".to_string(), retries: 3, @@ -1321,7 +2625,7 @@ mod tests { #[test] fn entry_identity_is_polymorphic_by_type_and_key() -> Result<()> { let temp = tempfile::tempdir()?; - let store_path = temp.path().join("cache.msgpack"); + let store_path = temp.path().join("cache.cc"); let mut cache = CultCache::new(); cache.register_registry(TestEntries)?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); @@ -1347,11 +2651,751 @@ mod tests { Ok(()) } + #[test] + fn prepared_batch_publishes_heterogeneous_entries_in_one_snapshot() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("batch.cc"); + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); + + let (settings, _) = cache.prepare_entry( + "app", + &Settings { + theme: "iron".to_string(), + retries: 4, + }, + )?; + let (note, _) = cache.prepare_entry( + "receipt", + &Note { + title: "committed".to_string(), + body: "same snapshot".to_string(), + }, + )?; + cache.put_prepared_batch(vec![settings, note])?; + + let mut reloaded = CultCache::new(); + reloaded.register_registry(TestEntries)?; + reloaded.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); + reloaded.pull_all_backing_stores()?; + assert_eq!(reloaded.get_required::("app")?.theme, "iron"); + assert_eq!(reloaded.get_required::("receipt")?.title, "committed"); + Ok(()) + } + + #[test] + fn named_preparation_preserves_omitted_nested_optional_slots() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("named.cc"); + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); + cache.pull_all_backing_stores()?; + let note = NestedNote { + value: NestedOptionalValue { + label: "nested".into(), + count: None, + symbol: Some("kept-in-its-own-field".into()), + }, + }; + assert!(cache.prepare_entry("compact", ¬e).is_err()); + let (entry, parsed) = cache.prepare_entry_named("named", ¬e)?; + assert_eq!(parsed, note); + cache.put_prepared_batch(vec![entry])?; + assert_eq!(cache.get_required::("named")?, note); + Ok(()) + } + + #[test] + fn pulling_an_absent_store_does_not_create_its_parent_or_lock() -> Result<()> { + let temp = tempfile::tempdir()?; + let missing_parent = temp.path().join("missing-body"); + let store_path = missing_parent.join("cache.cc"); + let store = SingleFileMessagePackBackingStore::new(&store_path); + + assert!(store.pull_all()?.is_empty()); + assert!(!missing_parent.exists()); + assert!(!store_path.with_file_name("cache.cc.lock").exists()); + Ok(()) + } + + #[test] + fn provider_snapshot_read_does_not_create_or_open_a_lock() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("provider.cc"); + let lock_path = temp.path().join("provider.cc.lock"); + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + let expected = test_envelope("provider.delivery", "one", b"payload"); + store.push(&expected)?; + std::fs::remove_file(&lock_path)?; + + assert_eq!(store.pull_all_read_only_snapshot()?, vec![expected]); + assert!(!lock_path.exists()); + Ok(()) + } + + #[test] + fn next_exclusive_write_removes_only_exact_abandoned_staging_files() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("cache.cc"); + let abandoned = temporary_path_for(&store_path); + let unknown = temp.path().join("cache.cc.not-a-uuid.tmp"); + let foreign = temp + .path() + .join("other.cc.00000000-0000-0000-0000-000000000000.tmp"); + fs::write(&abandoned, b"partial staging bytes")?; + fs::write(&unknown, b"operator-owned unknown file")?; + fs::write(&foreign, b"foreign store staging")?; + + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + let expected = test_envelope("model", "current", b"one"); + store.push(&expected)?; + + assert!(!abandoned.exists()); + assert_eq!(fs::read(&unknown)?, b"operator-owned unknown file"); + assert_eq!(fs::read(&foreign)?, b"foreign store staging"); + assert_eq!(store.pull_all()?, vec![expected]); + Ok(()) + } + + #[test] + fn failed_snapshot_publication_preserves_the_committed_destination() -> Result<()> { + let temp = tempfile::tempdir()?; + let destination = temp.path().join("cache.cc"); + let absent_staging = temp.path().join("absent.tmp"); + let committed = b"committed snapshot"; + fs::write(&destination, committed)?; + + assert!(replace_file_atomically(&absent_staging, &destination).is_err()); + assert_eq!(fs::read(&destination)?, committed); + Ok(()) + } + + #[test] + fn snapshot_publication_atomically_replaces_an_existing_destination() -> Result<()> { + let temp = tempfile::tempdir()?; + let destination = temp.path().join("cache.cc"); + let staged = temp.path().join("cache.cc.staged"); + fs::write(&destination, b"old snapshot")?; + fs::write(&staged, b"new snapshot")?; + + replace_file_atomically(&staged, &destination)?; + + assert_eq!(fs::read(&destination)?, b"new snapshot"); + assert!(!staged.exists()); + Ok(()) + } + + #[test] + #[ignore = "operator SIGKILL probe; requires CULTCACHE_INTERRUPTION_STORE"] + fn snapshot_publication_interruption_probe_writer() -> Result<()> { + let store_path = std::env::var_os("CULTCACHE_INTERRUPTION_STORE") + .map(PathBuf::from) + .ok_or_else(|| anyhow!("CULTCACHE_INTERRUPTION_STORE is required"))?; + let marker_path = std::env::var_os("CULTCACHE_INTERRUPTION_MARKER") + .map(PathBuf::from) + .ok_or_else(|| anyhow!("CULTCACHE_INTERRUPTION_MARKER is required"))?; + let mut store = SingleFileMessagePackBackingStore::new(store_path); + for sequence in 0_u64.. { + store.push(&test_envelope( + "interruption.probe", + "current", + &rmp_serde::to_vec(&sequence)?, + ))?; + fs::write(&marker_path, sequence.to_string())?; + } + unreachable!("the interruption writer is stopped by the operator") + } + + #[test] + #[ignore = "operator SIGKILL probe; requires CULTCACHE_INTERRUPTION_STORE"] + fn snapshot_publication_interruption_probe_reader() -> Result<()> { + let store_path = std::env::var_os("CULTCACHE_INTERRUPTION_STORE") + .map(PathBuf::from) + .ok_or_else(|| anyhow!("CULTCACHE_INTERRUPTION_STORE is required"))?; + let rows = SingleFileMessagePackBackingStore::new(store_path).pull_all()?; + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].r#type, "interruption.probe"); + assert_eq!(rows[0].key, "current"); + let _: u64 = rmp_serde::from_slice(&rows[0].payload)?; + Ok(()) + } + + fn test_envelope(r#type: &str, key: &str, payload: &[u8]) -> CultCacheEnvelope { + CultCacheEnvelope { + key: key.to_string(), + r#type: r#type.to_string(), + payload: payload.to_vec(), + stored_at: "2026-07-13T00:00:00Z".to_string(), + schema_id: Some(r#type.to_string()), + } + } + + #[test] + fn redb_store_crud_and_reopen_preserve_polymorphic_rows() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("keyed.cc"); + let model = test_envelope("model", "shared", b"one"); + let receipt = test_envelope("receipt", "shared", b"two"); + { + let mut store = RedbMessagePackBackingStore::new(&path)?; + store.push(&model)?; + store.push(&receipt)?; + assert_eq!(store.pull_all()?.len(), 2); + store.delete(&receipt)?; + } + let reopened = RedbMessagePackBackingStore::new(&path)?; + assert_eq!(reopened.pull_all()?, vec![model]); + Ok(()) + } + + #[test] + fn redb_exact_cas_refuses_stale_value_without_mutation() -> Result<()> { + let temp = tempfile::tempdir()?; + let mut store = RedbMessagePackBackingStore::new(temp.path().join("cas.cc"))?; + let current = test_envelope("model", "current", b"one"); + store.push(¤t)?; + let stale = test_envelope("model", "current", b"stale"); + let replacement = test_envelope("model", "current", b"two"); + assert!(!store.compare_and_swap_entry(&stale, replacement.clone())?); + assert_eq!(store.pull_all()?, vec![current.clone()]); + assert!(store.compare_and_swap_entry(¤t, replacement.clone())?); + assert_eq!(store.pull_all()?, vec![replacement]); + Ok(()) + } + + #[test] + fn redb_batch_is_atomic_and_stale_member_refuses_every_write() -> Result<()> { + let temp = tempfile::tempdir()?; + let mut store = RedbMessagePackBackingStore::new(temp.path().join("batch.cc"))?; + let first = test_envelope("model", "first", b"one"); + let second = test_envelope("model", "second", b"one"); + store.push(&first)?; + store.push(&second)?; + let stale_second = test_envelope("model", "second", b"stale"); + let first_two = test_envelope("model", "first", b"two"); + let second_two = test_envelope("model", "second", b"two"); + let receipt = test_envelope("receipt", "batch", b"committed"); + assert!(!store.compare_and_swap_batch( + &[first.clone(), stale_second], + vec![first_two.clone(), second_two.clone(), receipt.clone()], + )?); + assert_eq!(store.pull_all()?, vec![first.clone(), second.clone()]); + assert!(store.compare_and_swap_batch( + &[first, second], + vec![first_two.clone(), second_two.clone(), receipt.clone()], + )?); + assert_eq!(store.pull_all()?, vec![first_two, second_two, receipt]); + Ok(()) + } + + #[test] + fn redb_compaction_refuses_stale_snapshot_and_atomically_replaces_and_deletes() -> Result<()> { + let temp = tempfile::tempdir()?; + let mut store = RedbMessagePackBackingStore::new(temp.path().join("compact.cc"))?; + let old_head = test_envelope("retention", "current", b"one"); + let retired = test_envelope("receipt", "old", b"terminal"); + store.push(&old_head)?; + store.push(&retired)?; + let stale = store.pull_all()?; + let concurrent = test_envelope("receipt", "running", b"active"); + assert!(store.insert_entry_if_absent(concurrent.clone())?); + let new_head = test_envelope("retention", "current", b"two"); + assert!(!store.replace_and_delete_if_snapshot_unchanged( + &stale, + vec![new_head.clone()], + std::slice::from_ref(&retired), + )?); + assert!(store.pull_all()?.contains(&retired)); + let exact = store.pull_all()?; + assert!(store.replace_and_delete_if_snapshot_unchanged( + &exact, + vec![new_head.clone()], + &[retired], + )?); + assert_eq!(store.pull_all()?, vec![concurrent, new_head]); + Ok(()) + } + + #[test] + fn redb_entry_cas_does_not_depend_on_unrelated_snapshot_changes() -> Result<()> { + let temp = tempfile::tempdir()?; + let store = RedbMessagePackBackingStore::new(temp.path().join("independent.cc"))?; + let current = test_envelope("model", "current", b"one"); + assert!(store.insert_entry_if_absent(current.clone())?); + let unrelated = test_envelope("event", "later", b"noise"); + assert!(store.insert_entry_if_absent(unrelated.clone())?); + let replacement = test_envelope("model", "current", b"two"); + assert!(store.compare_and_swap_entry(¤t, replacement.clone())?); + assert_eq!(store.pull_all()?, vec![unrelated, replacement]); + Ok(()) + } + + #[test] + fn redb_fresh_handles_share_one_database_for_the_same_cultcache_path() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("shared.cc"); + let mut first = RedbMessagePackBackingStore::new(&path)?; + let second = RedbMessagePackBackingStore::new(&path)?; + let initial = test_envelope("model", "current", b"one"); + first.push(&initial)?; + assert_eq!(second.pull_all()?, vec![initial.clone()]); + let replacement = test_envelope("model", "current", b"two"); + assert!(second.compare_and_swap_entry(&initial, replacement.clone())?); + assert_eq!(first.pull_all()?, vec![replacement]); + Ok(()) + } + + #[test] + fn owned_redb_clones_share_authority_and_fresh_owner_is_refused() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("owned.cc"); + let owner = OwnedRedbMessagePackBackingStore::new(&path)?; + let clone = owner.clone(); + assert_eq!(owner.file_identity(), clone.file_identity()); + assert!(OwnedRedbMessagePackBackingStore::new(&path).is_err()); + drop(owner); + assert!(OwnedRedbMessagePackBackingStore::new(&path).is_err()); + drop(clone); + assert!(OwnedRedbMessagePackBackingStore::new(&path).is_ok()); + Ok(()) + } + + #[test] + fn owned_redb_file_identity_is_stable_across_writes_and_reopen() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("identity.cc"); + let identity = { + let mut owner = OwnedRedbMessagePackBackingStore::new(&path)?; + let identity = owner.file_identity().to_string(); + owner.push(&test_envelope("model", "current", b"one"))?; + owner.validate_path_identity()?; + owner.require_file_identity(&identity)?; + identity + }; + let reopened = OwnedRedbMessagePackBackingStore::new(&path)?; + assert_eq!(reopened.file_identity(), identity); + assert_eq!(reopened.pull_all()?.len(), 1); + Ok(()) + } + + #[test] + fn owned_redb_identity_helper_refuses_mismatch() -> Result<()> { + let temp = tempfile::tempdir()?; + let owner = OwnedRedbMessagePackBackingStore::new(temp.path().join("mismatch.cc"))?; + assert!(owner.require_file_identity("not-the-owned-file").is_err()); + owner.require_file_identity(owner.file_identity())?; + Ok(()) + } + + #[test] + fn owned_redb_batch_refuses_stale_member_and_commits_success_atomically() -> Result<()> { + let temp = tempfile::tempdir()?; + let mut owner = OwnedRedbMessagePackBackingStore::new(temp.path().join("batch-owned.cc"))?; + let first = test_envelope("model", "first", b"one"); + let second = test_envelope("model", "second", b"one"); + owner.push(&first)?; + owner.push(&second)?; + let first_two = test_envelope("model", "first", b"two"); + let second_two = test_envelope("model", "second", b"two"); + let companion = test_envelope("receipt", "batch", b"committed"); + let stale_second = test_envelope("model", "second", b"stale"); + assert!(!owner.compare_and_swap_batch( + &[first.clone(), stale_second], + vec![first_two.clone(), second_two.clone(), companion.clone()], + )?); + assert_eq!(owner.pull_all()?, vec![first.clone(), second.clone()]); + assert!(owner.compare_and_swap_batch( + &[first, second], + vec![first_two.clone(), second_two.clone(), companion.clone()], + )?); + assert_eq!(owner.pull_all()?, vec![first_two, second_two, companion]); + Ok(()) + } + + #[test] + fn owned_redb_compaction_refuses_stale_snapshot_and_atomically_replaces_and_deletes() + -> Result<()> { + let temp = tempfile::tempdir()?; + let mut owner = + OwnedRedbMessagePackBackingStore::new(temp.path().join("compact-owned.cc"))?; + let old_head = test_envelope("retention", "current", b"one"); + let retired_a = test_envelope("claim", "a", b"terminal"); + let retired_b = test_envelope("receipt", "a", b"terminal"); + owner.push(&old_head)?; + owner.push(&retired_a)?; + owner.push(&retired_b)?; + let stale = owner.pull_all()?; + let concurrent = test_envelope("claim", "concurrent", b"running"); + assert!(owner.insert_entry_if_absent(concurrent.clone())?); + let new_head = test_envelope("retention", "current", b"two"); + assert!(!owner.replace_and_delete_if_snapshot_unchanged( + &stale, + vec![new_head.clone()], + &[retired_a.clone(), retired_b.clone()], + )?); + assert!(owner.pull_all()?.contains(&retired_a)); + let exact = owner.pull_all()?; + assert!(owner.replace_and_delete_if_snapshot_unchanged( + &exact, + vec![new_head.clone()], + &[retired_a, retired_b], + )?); + assert_eq!(owner.pull_all()?, vec![concurrent, new_head]); + Ok(()) + } + + #[test] + fn owned_redb_path_replacement_cannot_redirect_pinned_authority() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("pinned.cc"); + let displaced = temp.path().join("displaced.cc"); + let mut owner = OwnedRedbMessagePackBackingStore::new(&path)?; + let identity = owner.file_identity().to_string(); + owner.push(&test_envelope("model", "before", b"one"))?; + fs::rename(&path, &displaced)?; + File::create(&path)?; + assert!(owner.validate_path_identity().is_err()); + assert_eq!(owner.file_identity(), identity); + owner.push(&test_envelope("model", "after", b"two"))?; + assert_eq!(owner.pull_all()?.len(), 2); + assert_ne!( + file_identity_from_file(&File::open(&path)?)?, + owner.file_identity() + ); + Ok(()) + } + + #[test] + fn redb_overlapping_fresh_handles_serialize_the_database_open_interval() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("contended.cc"); + let first = RedbMessagePackBackingStore::new(&path)?; + let second = RedbMessagePackBackingStore::new(&path)?; + let barrier = std::sync::Arc::new(std::sync::Barrier::new(3)); + let first_barrier = barrier.clone(); + let first_thread = std::thread::spawn(move || -> Result<()> { + first_barrier.wait(); + for index in 0..32 { + first.insert_entry_if_absent(test_envelope( + "event", + &format!("first-{index}"), + b"first", + ))?; + } + Ok(()) + }); + let second_barrier = barrier.clone(); + let second_thread = std::thread::spawn(move || -> Result<()> { + second_barrier.wait(); + for index in 0..32 { + second.insert_entry_if_absent(test_envelope( + "event", + &format!("second-{index}"), + b"second", + ))?; + } + Ok(()) + }); + barrier.wait(); + first_thread.join().expect("first redb writer")?; + second_thread.join().expect("second redb writer")?; + assert_eq!( + RedbMessagePackBackingStore::new(path)?.pull_all()?.len(), + 64 + ); + Ok(()) + } + + #[test] + fn redb_accepts_a_bare_relative_cultcache_path() -> Result<()> { + let name = format!("cultcache-redb-relative-{}.cc", uuid::Uuid::new_v4()); + let path = PathBuf::from(&name); + let lock_path = PathBuf::from(format!("{name}.lock")); + let store = RedbMessagePackBackingStore::new(&path)?; + assert!(store.insert_entry_if_absent(test_envelope("model", "current", b"one"))?); + assert_eq!(store.pull_all()?.len(), 1); + fs::remove_file(path)?; + fs::remove_file(lock_path)?; + Ok(()) + } + + #[test] + fn conditional_batch_replaces_expected_and_inserts_companion_atomically() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("conditional-batch.cc"); + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + let current = test_envelope("model", "current", b"revision-1"); + store.push(¤t)?; + let replacement = test_envelope("model", "current", b"revision-2"); + let companion = test_envelope("receipt", "migration-1", b"applied"); + + assert!(store.compare_and_swap_batch( + std::slice::from_ref(¤t), + vec![replacement.clone(), companion.clone()], + )?); + assert_eq!(store.pull_all()?, vec![replacement, companion]); + Ok(()) + } + + #[test] + fn conditional_batch_allows_atomic_first_import_when_identities_are_absent() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("conditional-import.cc"); + let store = SingleFileMessagePackBackingStore::new(&store_path); + let model = test_envelope("model", "current", b"revision-1"); + let receipt = test_envelope("receipt", "migration-1", b"applied"); + + assert!(store.compare_and_swap_batch(&[], vec![model.clone(), receipt.clone()])?); + assert_eq!(store.pull_all()?, vec![model, receipt]); + Ok(()) + } + + #[test] + fn conditional_batch_stale_expected_and_companion_collision_preserve_exact_bytes() -> Result<()> + { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("conditional-refusal.cc"); + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + let current = test_envelope("model", "current", b"revision-1"); + let occupied = test_envelope("receipt", "migration-1", b"someone-else"); + store.push(¤t)?; + store.push(&occupied)?; + let before = fs::read(&store_path)?; + + let stale = test_envelope("model", "current", b"stale-copy"); + let replacement = test_envelope("model", "current", b"revision-2"); + let fresh_companion = test_envelope("receipt", "migration-2", b"applied"); + assert!( + !store.compare_and_swap_batch(&[stale], vec![replacement.clone(), fresh_companion],)? + ); + assert_eq!(fs::read(&store_path)?, before); + + assert!(!store.compare_and_swap_batch( + &[current], + vec![ + replacement, + test_envelope("receipt", "migration-1", b"collision") + ], + )?); + assert_eq!(fs::read(&store_path)?, before); + Ok(()) + } + + #[test] + fn conditional_batch_rejects_empty_replacements_and_duplicate_identities() -> Result<()> { + let temp = tempfile::tempdir()?; + let store = SingleFileMessagePackBackingStore::new(temp.path().join("invalid-batch.cc")); + let entry = test_envelope("model", "current", b"one"); + + assert!( + store + .compare_and_swap_batch(&[entry.clone()], Vec::new()) + .is_err() + ); + assert!( + store + .compare_and_swap_batch(&[], vec![entry.clone(), entry.clone()]) + .is_err() + ); + assert!( + store + .compare_and_swap_batch( + &[entry.clone(), entry.clone()], + vec![test_envelope("model", "current", b"two")], + ) + .is_err() + ); + assert!(!store.path().exists()); + Ok(()) + } + + #[test] + fn full_snapshot_replacement_and_append_refuses_concurrent_rows() -> Result<()> { + let temp = tempfile::tempdir()?; + let snapshot_path = temp.path().join("snapshot-replace-append.cc"); + let mut snapshot_store = SingleFileMessagePackBackingStore::new(&snapshot_path); + let current = test_envelope("session", "current", b"active"); + snapshot_store.push(¤t)?; + let stale_snapshot = snapshot_store.pull_all()?; + let concurrent = test_envelope("event", "concurrent", b"live"); + snapshot_store.push(&concurrent)?; + let completed = test_envelope("session", "current", b"completed"); + let receipt = test_envelope("receipt", "terminal", b"done"); + assert!(!snapshot_store.replace_and_append_if_snapshot_unchanged( + &stale_snapshot, + vec![completed.clone(), receipt.clone()], + )?); + assert_eq!( + snapshot_store.pull_all()?, + vec![concurrent.clone(), current.clone()] + ); + let exact_snapshot = snapshot_store.pull_all()?; + assert!(snapshot_store.replace_and_append_if_snapshot_unchanged( + &exact_snapshot, + vec![completed.clone(), receipt.clone()], + )?); + assert_eq!( + snapshot_store.pull_all()?, + vec![concurrent.clone(), receipt.clone(), completed.clone()] + ); + + let keyed_path = temp.path().join("keyed-replace-append.redb"); + let mut keyed_store = RedbMessagePackBackingStore::new(&keyed_path)?; + keyed_store.push(¤t)?; + let stale_snapshot = keyed_store.pull_all()?; + keyed_store.push(&concurrent)?; + assert!(!keyed_store.replace_and_append_if_snapshot_unchanged( + &stale_snapshot, + vec![completed.clone(), receipt.clone()], + )?); + let exact_snapshot = keyed_store.pull_all()?; + assert!(keyed_store.replace_and_append_if_snapshot_unchanged( + &exact_snapshot, + vec![completed.clone(), receipt.clone()], + )?); + assert_eq!( + keyed_store.pull_all()?, + vec![concurrent, receipt, completed] + ); + Ok(()) + } + + #[test] + fn single_file_compaction_atomically_replaces_summary_and_deletes_history() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("conditional-compaction.cc"); + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + let history = test_envelope("history", "old", b"closed"); + let head = test_envelope("retention", "head", b"revision-1"); + store.push(&history)?; + store.push(&head)?; + let snapshot = store.pull_all()?; + let next_head = test_envelope("retention", "head", b"revision-2"); + + assert!(store.replace_and_delete_if_snapshot_unchanged( + &snapshot, + vec![next_head.clone()], + std::slice::from_ref(&history), + )?); + assert_eq!(store.pull_all()?, vec![next_head]); + + let before = fs::read(&store_path)?; + assert!(!store.replace_and_delete_if_snapshot_unchanged( + &snapshot, + vec![test_envelope("retention", "head", b"revision-3")], + std::slice::from_ref(&history), + )?); + assert_eq!(fs::read(&store_path)?, before); + Ok(()) + } + + #[test] + fn snapshot_append_fences_concurrent_unknown_identity() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("snapshot-append.cc"); + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + let model = test_envelope("model", "current", b"revision-1"); + store.push(&model)?; + let expected = store.pull_all()?; + let duplicate_admission = test_envelope("admission", "unexpected", b"also-current"); + store.push(&duplicate_admission)?; + let before = fs::read(&store_path)?; + + assert!(!store.append_if_snapshot_unchanged( + &expected, + vec![test_envelope("readiness", "proof", b"ready")], + )?); + assert_eq!(fs::read(&store_path)?, before); + Ok(()) + } + + #[test] + fn snapshot_append_is_order_insensitive_and_refuses_collisions() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("snapshot-order.cc"); + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + let a = test_envelope("a", "one", b"1"); + let b = test_envelope("b", "two", b"2"); + store.push(&a)?; + store.push(&b)?; + assert!(store.append_if_snapshot_unchanged( + &[b.clone(), a.clone()], + vec![test_envelope("proof", "three", b"3")], + )?); + let current = store.pull_all()?; + assert!(!store.append_if_snapshot_unchanged( + ¤t, + vec![test_envelope("a", "one", b"collision")], + )?); + assert!( + store + .append_if_snapshot_unchanged(¤t, Vec::new()) + .is_err() + ); + Ok(()) + } + + #[test] + fn type_and_key_identity_is_not_delimiter_ambiguous() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("delimiter-identity.cc"); + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + let left = test_envelope("a::b", "c", b"left"); + let right = test_envelope("a", "b::c", b"right"); + store.push(&left)?; + store.push(&right)?; + let rows = store.pull_all()?; + assert_eq!(rows.len(), 2); + assert!(rows.contains(&left)); + assert!(rows.contains(&right)); + Ok(()) + } + + #[test] + fn refused_batch_preserves_prior_snapshot() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("refused-batch.cc"); + let refusal = Arc::new(AtomicBool::new(false)); + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + cache.add_generic_backing_store(RefusingBatchStore { + inner: SingleFileMessagePackBackingStore::new(&store_path), + refuse_batch: refusal.clone(), + }); + cache.put( + "app", + &Settings { + theme: "before".to_string(), + retries: 1, + }, + )?; + let (note, _) = cache.prepare_entry( + "receipt", + &Note { + title: "after".to_string(), + body: "must not land".to_string(), + }, + )?; + refusal.store(true, Ordering::SeqCst); + assert!(cache.put_prepared_batch(vec![note]).is_err()); + assert!(cache.get::("receipt")?.is_none()); + + let mut reloaded = CultCache::new(); + reloaded.register_registry(TestEntries)?; + reloaded.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); + reloaded.pull_all_backing_stores()?; + assert_eq!(reloaded.get_required::("app")?.theme, "before"); + assert!(reloaded.get::("receipt")?.is_none()); + Ok(()) + } + #[test] fn type_specific_store_routes_before_generic_store() -> Result<()> { let temp = tempfile::tempdir()?; - let generic_path = temp.path().join("generic.msgpack"); - let settings_path = temp.path().join("settings.msgpack"); + let generic_path = temp.path().join("generic.cc"); + let settings_path = temp.path().join("settings.cc"); let mut cache = CultCache::new(); cache.register_entry_type::()?; cache.register_entry_type::()?; @@ -1386,7 +3430,7 @@ mod tests { #[test] fn update_and_delete_follow_the_cache_api() -> Result<()> { let temp = tempfile::tempdir()?; - let store_path = temp.path().join("cache.msgpack"); + let store_path = temp.path().join("cache.cc"); let mut cache = CultCache::new(); cache.register_entry_type::()?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); @@ -1408,7 +3452,7 @@ mod tests { #[test] fn pull_rejects_unregistered_persisted_entry_type() -> Result<()> { let temp = tempfile::tempdir()?; - let store_path = temp.path().join("cache.msgpack"); + let store_path = temp.path().join("cache.cc"); let mut store = SingleFileMessagePackBackingStore::new(&store_path); store.push(&CultCacheEnvelope { key: "unknown".to_string(), @@ -1432,7 +3476,7 @@ mod tests { #[test] fn payload_is_binary_messagepack_not_json_value() -> Result<()> { let temp = tempfile::tempdir()?; - let store_path = temp.path().join("cache.msgpack"); + let store_path = temp.path().join("cache.cc"); let mut cache = CultCache::new(); cache.register_entry_type::()?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); @@ -1451,10 +3495,23 @@ mod tests { Ok(()) } + #[test] + fn byte_slots_use_messagepack_binary_and_round_trip() -> Result<()> { + let record = BinaryRecord { + required: vec![1, 2, 3], + optional: Vec::new(), + }; + + let payload = rmp_serde::to_vec(&record)?; + assert_eq!(payload, vec![0x92, 0xc4, 3, 1, 2, 3, 0xc4, 0]); + assert_eq!(rmp_serde::from_slice::(&payload)?, record); + Ok(()) + } + #[test] fn corrupted_payload_fails_during_typed_retrieval() -> Result<()> { let temp = tempfile::tempdir()?; - let store_path = temp.path().join("cache.msgpack"); + let store_path = temp.path().join("cache.cc"); let mut store = SingleFileMessagePackBackingStore::new(&store_path); store.push(&CultCacheEnvelope { key: "app".to_string(), @@ -1480,8 +3537,8 @@ mod tests { #[test] fn put_envelope_reuses_existing_messagepack_payload() -> Result<()> { let temp = tempfile::tempdir()?; - let origin_store = temp.path().join("origin.msgpack"); - let target_store = temp.path().join("target.msgpack"); + let origin_store = temp.path().join("origin.cc"); + let target_store = temp.path().join("target.cc"); let mut origin = CultCache::new(); origin.register_entry_type::()?; @@ -1517,38 +3574,141 @@ mod tests { } #[test] - fn messagepack_store_recovers_schema_stamped_records_missing_catalog_entries() -> Result<()> { + fn raw_envelope_path_preserves_a_registered_dynamic_payload() -> Result<()> { let temp = tempfile::tempdir()?; - let store_path = temp.path().join("missing-catalog.msgpack"); - let expected = SchemaStamped { - schema_version: "tests.schema_stamped_entry.v1".to_string(), - name: "schema-stamped".to_string(), - value: "still readable".to_string(), + let store_path = temp.path().join("dynamic.cc"); + let envelope = CultCacheEnvelope { + key: "app".to_string(), + r#type: Settings::TYPE.to_string(), + payload: vec![0xc1], + stored_at: now_utc_second(), + schema_id: Some("dynamic.settings.v1".to_string()), }; - let snapshot = PersistedStoreSnapshot( - "cultcache.store.v1".to_string(), - Vec::new(), - vec![PersistedRecord( - "record-1".to_string(), - "sha256:stale-schema-id-from-cold-record".to_string(), - "2026-06-25T12:00:00Z".to_string(), - rmp_serde::to_vec(&expected)?, - )], - ); - fs::write(&store_path, rmp_serde::to_vec(&snapshot)?)?; let mut cache = CultCache::new(); - cache.register_entry_type::()?; + cache.register_entry_type::()?; + cache.register_entry_type::()?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); - cache.pull_all_backing_stores()?; + cache.put_raw_envelope(envelope.clone())?; - assert_eq!(cache.get_required::("record-1")?, expected); + assert_eq!(cache.snapshot(), vec![envelope.clone()]); assert_eq!( + SingleFileMessagePackBackingStore::new(&store_path).pull_all()?, + vec![envelope] + ); + assert!(cache.get_required::("app").is_err()); + let before_collision = fs::read(&store_path)?; + assert!( cache - .get_required_envelope::("record-1")? - .schema_id, - Some("sha256:stale-schema-id-from-cold-record".to_string()) + .put_raw_envelope(CultCacheEnvelope { + key: "note".to_string(), + r#type: Note::TYPE.to_string(), + payload: Vec::new(), + stored_at: now_utc_second(), + schema_id: Some("dynamic.settings.v1".to_string()), + }) + .is_err() + ); + assert_eq!(fs::read(&store_path)?, before_collision); + assert!( + cache + .put_raw_envelope(CultCacheEnvelope { + key: "foreign".to_string(), + r#type: "unregistered".to_string(), + payload: Vec::new(), + stored_at: now_utc_second(), + schema_id: None, + }) + .is_err() ); Ok(()) } + + #[test] + fn soa_snapshot_round_trips_registered_entries() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("cache.cc"); + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); + cache.put( + "app", + &Settings { + theme: "ash".to_string(), + retries: 3, + }, + )?; + cache.put( + "memo", + &Note { + title: "machine".to_string(), + body: "awake".to_string(), + }, + )?; + + let soa = cache.snapshot_soa(); + assert_eq!(soa.len(), 2); + assert_eq!(soa.keys.len(), soa.payloads.len()); + + let mut restored = CultCache::new(); + restored.register_registry(TestEntries)?; + restored.add_generic_backing_store(SingleFileMessagePackBackingStore::new( + temp.path().join("restored.cc"), + )); + restored.load_soa(soa)?; + assert_eq!(restored.get_required::("app")?.theme, "ash"); + assert_eq!(restored.get_required::("memo")?.body, "awake"); + Ok(()) + } + + #[test] + fn typed_soa_columns_match_csharp_cache_ergonomics() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("cache.cc"); + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); + cache.put( + "settings:a", + &Settings { + theme: "ash".to_string(), + retries: 3, + }, + )?; + cache.put( + "settings:b", + &Settings { + theme: "bone".to_string(), + retries: 5, + }, + )?; + + let table = cache.soa::()?; + assert_eq!( + table.keys(), + &["settings:a".to_string(), "settings:b".to_string()] + ); + assert_eq!( + table.column::("theme")?.values(), + &["ash".to_string(), "bone".to_string()] + ); + assert_eq!(table.column::("retries")?.values(), &[3, 5]); + + let error = table.column::("retries").unwrap_err(); + assert!(error.to_string().contains("not alloc::string::String")); + Ok(()) + } + + #[test] + fn soa_rejects_column_length_drift() { + let table = CultCacheSoaTable { + keys: vec!["app".to_string()], + types: Vec::new(), + payloads: Vec::new(), + stored_ats: Vec::new(), + schema_ids: Vec::new(), + }; + let error = table.validate().unwrap_err(); + assert!(error.to_string().contains("column types has length 0")); + } }