From 80754cb25f7fe7b7bfe5d7b8dd2371f8bb11ee27 Mon Sep 17 00:00:00 2001 From: Libby Date: Fri, 13 Nov 2026 19:53:46 +0000 Subject: [PATCH 1/6] cultcache-rs: conditional update (read, commit_if, restamp, write_entry), blind update deleted Co-Authored-By: Claude Sonnet 5.5 --- packages/cultcache-rs/Cargo.lock | 72 ++++ packages/cultcache-rs/Cargo.toml | 1 + packages/cultcache-rs/src/lib.rs | 390 ++++++++++++++---- packages/cultcache-rs/tests/slot_readonly.rs | 2 +- packages/cultcache-rs/tests/union_parity.rs | 2 +- packages/cultcache-rs/tests/unique_index.rs | 4 +- packages/cultmesh-rs/Cargo.lock | 1 + packages/cultnet-rs/Cargo.lock | 1 + .../src/idunn_runtime_provider/tests.rs | 1 + packages/cultnet-rs/src/service_identity.rs | 4 +- 10 files changed, 400 insertions(+), 78 deletions(-) diff --git a/packages/cultcache-rs/Cargo.lock b/packages/cultcache-rs/Cargo.lock index 40d9e1e8..6fad548a 100644 --- a/packages/cultcache-rs/Cargo.lock +++ b/packages/cultcache-rs/Cargo.lock @@ -38,6 +38,15 @@ version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "bumpalo" version = "3.20.3" @@ -80,6 +89,25 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + [[package]] name = "cultcache-rs" version = "0.3.1" @@ -95,6 +123,7 @@ dependencies = [ "serde_bytes", "serde_json", "serde_stacker", + "sha2", "stacker", "tempfile", "uuid", @@ -116,6 +145,16 @@ version = "0.1.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "56254986775e3233ffa9c4d7d3faaf6d36a2c09d30b20687e9f88bc8bafc16c8" +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "errno" version = "0.3.15" @@ -172,6 +211,16 @@ dependencies = [ "slab", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "getrandom" version = "0.4.3" @@ -433,6 +482,17 @@ dependencies = [ "stacker", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "shlex" version = "2.0.1" @@ -493,6 +553,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "unicode-ident" version = "1.0.27" @@ -510,6 +576,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + [[package]] name = "wasm-bindgen" version = "0.2.130" diff --git a/packages/cultcache-rs/Cargo.toml b/packages/cultcache-rs/Cargo.toml index db8a46ac..1ed1ba3f 100644 --- a/packages/cultcache-rs/Cargo.toml +++ b/packages/cultcache-rs/Cargo.toml @@ -23,6 +23,7 @@ serde = { version = "1", features = ["derive"] } serde_bytes = "0.11" serde_json = "1" serde_stacker = "0.1.14" +sha2 = "0.10" stacker = "0.1.25" uuid = { version = "1", features = ["v4"] } diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index 8fccdf97..44b5da9f 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -1290,6 +1290,39 @@ pub struct PushAllOptions { pub soft: bool, } +/// A record read from a store with its content version, both from one read. The version is the lowercase hex SHA-256 of the +/// record as the store file encodes it (key, schema id, `stored_at`, payload), so a rewrite of the same bytes under a later +/// stamp is a different version. +#[derive(Clone, Debug, PartialEq)] +pub struct CultStoredRecord { + pub envelope: CultCacheEnvelope, + pub version: String, +} + +/// The verdict of [`CacheBackingStore::commit_if`]: the write landed and carries this stamp, or the record at the key was not at +/// the expected version and nothing was written. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum CultCommitIf { + Committed { stored_at: String }, + Mismatch, +} + +/// How [`CultCache::update`] ended. `Declined` wrote nothing because `apply` returned `None`; `LostRace` wrote nothing because +/// every attempt found the record changed under it. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum CultUpdateOutcome { + Updated, + Declined, + LostRace, +} + +/// The outcome of an update and the cache's document at the key after it (a read that decides nothing). +#[derive(Clone, Debug, PartialEq)] +pub struct CultUpdateResult { + pub outcome: CultUpdateOutcome, + pub document: Option, +} + /// One per-entry expectation for [`SingleFileMessagePackBackingStore::compare_exchange`]. /// `current` is the envelope the caller believes is stored, or `None` when it /// believes the identity is absent. @@ -1312,7 +1345,9 @@ pub enum TryCompareExchangeSnapshotOutcome { pub trait CacheBackingStore: Send { fn pull_all(&self) -> Result>; - fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()>; + /// Writes `entry` and returns the `stored_at` the store wrote: a store that restamps a replacing write reports the stamp it + /// minted, and the cache holds that one. + fn push(&mut self, entry: &CultCacheEnvelope) -> Result; fn delete(&mut self, entry: &CultCacheEnvelope) -> Result<()>; /// Writes `entries` onto the store as it is when the write lands, in one @@ -1322,7 +1357,9 @@ pub trait CacheBackingStore: Send { /// it held before. A cache batch (`put_prepared_batch`) is all-or-nothing /// only through this. There is no default, because a write assembled from /// `push` calls leaves the store partial when one of them fails; a store - /// without an atomic write does not compile: + /// without an atomic write does not compile. It returns one stamp per entry given, in order, the `stored_at` written for it + /// (an entry deduplicated away gets the stamp of the one that landed for its schema id and key); a soft write that wrote + /// nothing returns none: /// /// ```compile_fail,E0046 /// use cultcache_rs::{CacheBackingStore, CultCacheEnvelope}; @@ -1333,15 +1370,33 @@ pub trait CacheBackingStore: Send { /// fn pull_all(&self) -> anyhow::Result> { /// Ok(Vec::new()) /// } - /// fn push(&mut self, _entry: &CultCacheEnvelope) -> anyhow::Result<()> { - /// Ok(()) + /// fn push(&mut self, _entry: &CultCacheEnvelope) -> anyhow::Result { + /// Ok(String::new()) /// } /// fn delete(&mut self, _entry: &CultCacheEnvelope) -> anyhow::Result<()> { /// Ok(()) /// } /// } /// ``` - fn push_all(&mut self, entries: &[CultCacheEnvelope], options: PushAllOptions) -> Result<()>; + fn push_all(&mut self, entries: &[CultCacheEnvelope], options: PushAllOptions) -> Result>; + + /// The record at `key`, of any type, with its content version, from one read; `None` when the store holds nothing there. + /// [`CultCache::update`] needs this and [`CacheBackingStore::commit_if`]; a store without them refuses it. + fn read(&self, _key: &str) -> Result> { + Err(anyhow!( + "{} has no read and commit_if: CultCache::update needs read and commit_if", + std::any::type_name::() + )) + } + + /// Writes `entry` over the record at its key only if that record's version is `expected` (`None`: nothing is there), in + /// one step under the store's own exclusion. A mismatch writes nothing. + fn commit_if(&mut self, _entry: &CultCacheEnvelope, _expected: Option<&str>) -> Result { + Err(anyhow!( + "{} has no read and commit_if: CultCache::update needs read and commit_if", + std::any::type_name::() + )) + } } #[derive(Clone, Debug, PartialEq, Eq)] @@ -2026,14 +2081,45 @@ impl CacheBackingStore for SingleFileMessagePackBackingStore { self.with_shared_lock(|| self.read_all_unlocked()) } - fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + fn push(&mut self, entry: &CultCacheEnvelope) -> Result { self.with_exclusive_lock(|| { let mut entries = self.read_all_unlocked()?; + let entry = restamped_over(entry, &entries); // A write replaces the record of its own type under its key; a second type under the key reaches the commit guard. - entries.retain(|candidate| entry_id(candidate) != entry_id(entry)); + entries.retain(|candidate| entry_id(candidate) != entry_id(&entry)); entries.push(entry.clone()); entries.sort_by_key(entry_id); - self.write_all_unlocked(&entries) + self.write_all_unlocked(&entries)?; + Ok(entry.stored_at) + }) + } + + fn read(&self, key: &str) -> Result> { + if is_absent(&self.path)? && is_absent(&self.lock_path())? { + return Ok(None); + } + self.with_shared_lock(|| { + Ok(self + .read_all_unlocked()? + .into_iter() + .find(|entry| entry.key == key) + .map(|envelope| CultStoredRecord { version: record_version(&envelope), envelope })) + }) + } + + fn commit_if(&mut self, entry: &CultCacheEnvelope, expected: Option<&str>) -> Result { + self.with_exclusive_lock(|| { + let mut entries = self.read_all_unlocked()?; + let actual = entries.iter().find(|held| held.key == entry.key).map(record_version); + if actual.as_deref() != expected { + return Ok(CultCommitIf::Mismatch); + } + let entry = restamped_over(entry, &entries); + entries.retain(|held| held.key != entry.key); + entries.push(entry.clone()); + entries.sort_by_key(entry_id); + self.write_all_unlocked(&entries)?; + Ok(CultCommitIf::Committed { stored_at: entry.stored_at }) }) } @@ -2045,27 +2131,32 @@ impl CacheBackingStore for SingleFileMessagePackBackingStore { }) } - fn push_all(&mut self, entries: &[CultCacheEnvelope], options: PushAllOptions) -> Result<()> { + fn push_all(&mut self, entries: &[CultCacheEnvelope], options: PushAllOptions) -> Result> { self.with_exclusive_lock(|| { // A soft flush asks the one reader: nothing at the path is written, a store it reads is left alone, and anything // the reader refuses (an unreadable file, a dangling link) is refused here as it is for open. if options.soft && !self.read_store_unlocked()?.absent { - return Ok(()); + return Ok(Vec::new()); } // Of entries sharing a key and a schema id the last one given is the record, whatever their type names (a type // renamed under a stable id); it replaces the held record under that key and schema id. A second schema id // under a key reaches the commit guard, which judges the merged store. - let mut by_id = BTreeMap::<(String, String), &CultCacheEnvelope>::new(); - for entry in entries { - by_id.insert((schema_identity(entry).to_string(), entry.key.clone()), entry); - } let mut merged = self.read_all_unlocked()?; - let replaced: BTreeSet<_> = by_id.values().map(|entry| entry_id(entry)).collect(); + let mut by_id = BTreeMap::<(String, String), CultCacheEnvelope>::new(); + for entry in entries { + by_id.insert((schema_identity(entry).to_string(), entry.key.clone()), restamped_over(entry, &merged)); + } + let replaced: BTreeSet<_> = by_id.values().map(entry_id).collect(); merged.retain(|held| !replaced.contains(&entry_id(held))); - merged.extend(by_id.into_values().cloned()); + let stamps = entries + .iter() + .map(|entry| by_id[&(schema_identity(entry).to_string(), entry.key.clone())].stored_at.clone()) + .collect(); + merged.extend(by_id.into_values()); merged.sort_by_key(entry_id); - self.write_all_unlocked(&merged) + self.write_all_unlocked(&merged)?; + Ok(stamps) }) } } @@ -2827,14 +2918,14 @@ impl CacheBackingStore for OwnedRedbMessagePackBackingStore { read_all_redb(&table) } - fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + 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(()) + Ok(entry.stored_at.clone()) } fn delete(&mut self, entry: &CultCacheEnvelope) -> Result<()> { @@ -2848,7 +2939,7 @@ impl CacheBackingStore for OwnedRedbMessagePackBackingStore { Ok(()) } - fn push_all(&mut self, entries: &[CultCacheEnvelope], _options: PushAllOptions) -> Result<()> { + fn push_all(&mut self, entries: &[CultCacheEnvelope], _options: PushAllOptions) -> Result> { unique_batch_ids(entries, "push all")?; let write = self.inner.database.begin_write()?; { @@ -2858,7 +2949,7 @@ impl CacheBackingStore for OwnedRedbMessagePackBackingStore { } } write.commit()?; - Ok(()) + Ok(entries.iter().map(|entry| entry.stored_at.clone()).collect()) } } @@ -2957,7 +3048,7 @@ impl CacheBackingStore for RedbMessagePackBackingStore { }) } - fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + fn push(&mut self, entry: &CultCacheEnvelope) -> Result { self.with_database(|database| { let write = database.begin_write()?; { @@ -2965,7 +3056,7 @@ impl CacheBackingStore for RedbMessagePackBackingStore { insert_redb_entry(&mut table, entry)?; } write.commit()?; - Ok(()) + Ok(entry.stored_at.clone()) }) } @@ -2982,7 +3073,7 @@ impl CacheBackingStore for RedbMessagePackBackingStore { }) } - fn push_all(&mut self, entries: &[CultCacheEnvelope], _options: PushAllOptions) -> Result<()> { + 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()?; @@ -2993,7 +3084,7 @@ impl CacheBackingStore for RedbMessagePackBackingStore { } } write.commit()?; - Ok(()) + Ok(entries.iter().map(|entry| entry.stored_at.clone()).collect()) }) } } @@ -3413,19 +3504,40 @@ impl CultCache { /// carries past this build's, or because this build cannot read the payload at all. fn refuse_unread_slots(&self, type_id: &str, key: &str) -> Result<()> { match self.held_finding(type_id, key)? { + HeldFinding::Free => Ok(()), + finding => { + let held = &self.entries[&entry_id_parts(type_id, key)]; + self.refuse_finding(type_id, key, schema_identity(held), finding) + } + } + } + + /// The same refusal for a record read from a store: [`CultCache::update`] copies it, so it must not carry slots this build + /// would drop or bytes it cannot read. + fn refuse_record_slots(&self, type_id: &str, key: &str, schema_id: &str, payload: &[u8]) -> Result<()> { + let Some(definition) = self.definitions.get(type_id) else { + bail!("No entry type registered for CultCache envelope type {type_id:?}"); + }; + let finding = match (definition.unread_slots)(payload) { + Ok(None) => return Ok(()), + Ok(Some(unread)) => HeldFinding::Marked(unread), + Err(error) => HeldFinding::Unreadable(error), + }; + self.refuse_finding(type_id, key, schema_id, finding) + } + + fn refuse_finding(&self, type_id: &str, key: &str, schema_id: &str, finding: HeldFinding) -> Result<()> { + match finding { HeldFinding::Free => Ok(()), HeldFinding::Unreadable(error) => Err(error.context(format!( "CultCache entry {type_id:?} at key {key:?} is held read-only: this build cannot read its payload. A newer writer's record is read with that build; a corrupt one is restored from a backup or removed at the store itself, not through the cache" ))), - HeldFinding::Marked(unread) => { - let held = &self.entries[&entry_id_parts(type_id, key)]; - Err(anyhow::Error::new(SchemaConflictError { - schema_id: schema_identity(held).to_string(), - schema_names: vec![self.definitions[type_id].schema_name.to_string()], - record_key: key.to_string(), - unread_slots: Some(unread), - })) - } + HeldFinding::Marked(unread) => Err(anyhow::Error::new(SchemaConflictError { + schema_id: schema_id.to_string(), + schema_names: vec![self.definitions[type_id].schema_name.to_string()], + record_key: key.to_string(), + unread_slots: Some(unread), + })), } } @@ -3561,14 +3673,7 @@ impl CultCache { self.require_loaded()?; let (entry, parsed) = self.prepare_entry(key, value)?; self.refuse_unread_slots(&entry.r#type, &entry.key)?; - self.refuse_second_global(&self.entries, &entry.r#type, &entry.key)?; - self.refuse_retype(&self.entries, &self.foreign, &entry)?; - let change = self.judge_unique(&[&entry])?; - if let Some(home) = self.home_index(T::TYPE)? { - self.stores[home].store.push(&entry)?; - } - self.entries.insert(entry_id(&entry), entry); - change.apply(&mut self.index_holders); + self.write_entry(entry, None)?; Ok(parsed) } @@ -3687,10 +3792,22 @@ impl CultCache { if let Some(store_index) = store_index { // The home store is handed the staged records alone and applies them onto what it holds now. // A key staged twice lands as its last record, whatever the store. - let landing: BTreeMap<_, _> = staged.iter().map(|entry| (entry_id(entry), entry.clone())).collect(); - self.stores[store_index] - .store - .push_all(&landing.into_values().collect::>(), PushAllOptions::default())?; + let landing: Vec = staged + .iter() + .map(|entry| (entry_id(entry), entry.clone())) + .collect::>() + .into_values() + .collect(); + let stamps = self.stores[store_index].store.push_all(&landing, PushAllOptions::default())?; + if stamps.len() == landing.len() { + for (entry, stamp) in landing.iter().zip(stamps) { + if chrono::DateTime::parse_from_rfc3339(&stamp).is_ok() + && let Some(held) = next.get_mut(&entry_id(entry)) + { + held.stored_at = stamp; + } + } + } } self.entries = next; change.apply(&mut self.index_holders); @@ -3702,14 +3819,7 @@ impl CultCache { let parsed = self.validate_envelope::(&entry)?; self.refuse_unread_slots(&entry.r#type, &entry.key)?; self.refuse_claimed_id(&entry)?; - self.refuse_second_global(&self.entries, &entry.r#type, &entry.key)?; - self.refuse_retype(&self.entries, &self.foreign, &entry)?; - let change = self.judge_unique(&[&entry])?; - if let Some(home) = self.home_index(T::TYPE)? { - self.stores[home].store.push(&entry)?; - } - self.entries.insert(entry_id(&entry), entry); - change.apply(&mut self.index_holders); + self.write_entry(entry, None)?; Ok(parsed) } @@ -3741,15 +3851,33 @@ impl CultCache { require_record_payload(&entry)?; self.refuse_unread_slots(&entry.r#type, &entry.key)?; self.refuse_claimed_id(&entry)?; + self.write_entry(entry, None)?; + Ok(()) + } + + /// The one landing of a single record, for `put`, `put_envelope`, `put_raw_envelope` and `update`: the global, retype and + /// unique judges, then the home store's write, then the install of the stamp the store reports (kept only when it parses + /// as RFC 3339, else the entry's own). `expected` is `None` for a blind write; `Some(version)` writes only over the record at + /// that version (`Some(None)`: over nothing) and returns `false`, installing nothing, when the store found another. + fn write_entry(&mut self, mut entry: CultCacheEnvelope, expected: Option>) -> Result { self.refuse_second_global(&self.entries, &entry.r#type, &entry.key)?; self.refuse_retype(&self.entries, &self.foreign, &entry)?; let change = self.judge_unique(&[&entry])?; if let Some(home) = self.home_index(&entry.r#type)? { - self.stores[home].store.push(&entry)?; + let stored_at = match expected { + None => self.stores[home].store.push(&entry)?, + Some(version) => match self.stores[home].store.commit_if(&entry, version.as_deref())? { + CultCommitIf::Committed { stored_at } => stored_at, + CultCommitIf::Mismatch => return Ok(false), + }, + }; + if chrono::DateTime::parse_from_rfc3339(&stored_at).is_ok() { + entry.stored_at = stored_at; + } } self.entries.insert(entry_id(&entry), entry); change.apply(&mut self.index_holders); - Ok(()) + Ok(true) } fn validate_envelope(&self, entry: &CultCacheEnvelope) -> Result { @@ -3784,13 +3912,82 @@ impl CultCache { }) } - pub fn update(&mut self, key: &str, updater: F) -> Result + /// Read the record at `key` as its home store holds it now, apply `apply` to a copy, and write the result only if the store + /// still holds the record that was read; on a lost race run `apply` again on a fresh read, up to 5 attempts + /// ([`CultCache::update_with_attempts`] takes the bound). `apply` may run several times and must have no side effects. + /// `None` from `apply` writes nothing ([`CultUpdateOutcome::Declined`]); losing every attempt is a value + /// ([`CultUpdateOutcome::LostRace`]), not an error. On `Updated` the write is durable. [`CultCache::put`] is the blind write. + pub fn update(&mut self, key: &str, apply: F) -> Result> where T: DatabaseEntry, - F: FnOnce(Option) -> T, + F: FnMut(Option) -> Option, { - let current = self.get::(key)?; - self.put::(key.to_string(), &updater(current)) + self.update_with_attempts(key, 5, apply) + } + + /// [`CultCache::update`] with its attempt bound: `max_attempts` of 0 is refused before `apply` runs. A refusal, a store + /// error or a failed judge is returned at once and never retried; after the last lost attempt the cache pulls its stores + /// (display only) and reports `LostRace`. + pub fn update_with_attempts(&mut self, key: &str, max_attempts: usize, mut apply: F) -> Result> + where + T: DatabaseEntry, + F: FnMut(Option) -> Option, + { + if max_attempts == 0 { + bail!("update max_attempts must be at least 1"); + } + for attempt in 1..=max_attempts { + let (copy, expected) = self.read_for_update::(key)?; + let Some(next) = apply(copy) else { + return Ok(CultUpdateResult { outcome: CultUpdateOutcome::Declined, document: self.get::(key)? }); + }; + let (entry, _) = self.prepare_entry(key, &next)?; + if self.write_entry(entry, expected)? { + return Ok(CultUpdateResult { outcome: CultUpdateOutcome::Updated, document: self.get::(key)? }); + } + if attempt == max_attempts { + self.pull_all_backing_stores()?; + return Ok(CultUpdateResult { outcome: CultUpdateOutcome::LostRace, document: self.get::(key)? }); + } + } + unreachable!("the last attempt returns") + } + + /// One attempt's read: the refusals that precede `apply`, then the record at `key` as the home store holds it with the version + /// to commit against (`Some(None)` for nothing there; `None` for a cache with no store, which has no version). Installs + /// nothing in the cache. + #[allow(clippy::type_complexity)] + fn read_for_update(&self, key: &str) -> Result<(Option, Option>)> { + self.require_loaded()?; + self.require_entry_type::()?; + self.refuse_unread_slots(T::TYPE, key)?; + let probe = CultCacheEnvelope { + key: key.to_string(), + r#type: T::TYPE.to_string(), + payload: Vec::new(), + stored_at: String::new(), + schema_id: None, + }; + self.refuse_retype(&self.entries, &self.foreign, &probe)?; + let Some(home) = self.home_index(T::TYPE)? else { + return Ok((self.get::(key)?, None)); + }; + let Some(found) = self.stores[home].store.read(key)? else { + return Ok((None, Some(None))); + }; + let known_types: BTreeSet = self.definitions.keys().cloned().collect(); + let resolved = resolve_registered_type( + &known_types, + &self.schema_name_definitions, + found.envelope.schema_id.as_deref(), + &found.envelope.r#type, + ); + if resolved.as_deref() != Some(T::TYPE) { + bail!("Record {key} is not a {}; update cannot copy it.", T::TYPE); + } + self.refuse_record_slots(T::TYPE, key, schema_identity(&found.envelope), &found.envelope.payload)?; + let copy = decode_payload::(&found.envelope.payload)?; + Ok((Some(copy), Some(Some(found.version)))) } pub fn delete(&mut self, key: &str) -> Result { @@ -4076,6 +4273,52 @@ fn entry_id_parts(r#type: &str, key: &str) -> (String, String) { (r#type.to_string(), key.to_string()) } +/// The content version of a stored record: lowercase hex SHA-256 of the record exactly as the store file encodes it, so it +/// covers the key, schema id, `stored_at` and payload. The only hash in this crate. +fn record_version(entry: &CultCacheEnvelope) -> String { + use sha2::{Digest, Sha256}; + let record = PersistedRecord( + entry.key.clone(), + schema_identity(entry).to_string(), + entry.stored_at.clone(), + entry.payload.clone(), + ); + let bytes = rmp_serde::to_vec(&record).expect("a record of strings and bytes encodes"); + Sha256::digest(bytes).iter().map(|byte| format!("{byte:02x}")).collect() +} + +/// A stamp strictly later than `previous` (to the microsecond): `now` truncated to the microsecond when that is later, +/// otherwise `previous` truncated plus one microsecond. An absent or unparseable `previous` gives `now`. +fn mint_stored_at(previous: Option<&str>, now: chrono::DateTime) -> String { + use chrono::SubsecRound; + let now = now.trunc_subsecs(6); + let minted = match previous.and_then(|text| chrono::DateTime::parse_from_rfc3339(text).ok()) { + Some(previous) => { + let previous = previous.with_timezone(&chrono::Utc).trunc_subsecs(6); + if now <= previous { previous + chrono::Duration::microseconds(1) } else { now } + } + None => now, + }; + minted.to_rfc3339_opts(chrono::SecondsFormat::Micros, true) +} + +/// The stamp a replacing write carries: `staged` stands only when both stamps parse and it is strictly later than `replaced`. +fn stored_at_over(staged: &str, replaced: &str) -> String { + match (chrono::DateTime::parse_from_rfc3339(staged), chrono::DateTime::parse_from_rfc3339(replaced)) { + (Ok(staged_at), Ok(replaced_at)) if staged_at > replaced_at => staged.to_string(), + _ => mint_stored_at(Some(replaced), chrono::Utc::now()), + } +} + +/// `entry` stamped later than the record the store holds at its key, of any type; unchanged where nothing is held. +fn restamped_over(entry: &CultCacheEnvelope, held: &[CultCacheEnvelope]) -> CultCacheEnvelope { + let mut entry = entry.clone(); + if let Some(replaced) = held.iter().find(|candidate| candidate.key == entry.key) { + entry.stored_at = stored_at_over(&entry.stored_at, &replaced.stored_at); + } + entry +} + fn now_utc_second() -> String { chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true) } @@ -4924,7 +5167,7 @@ mod tests { fn pull_all(&self) -> Result> { self.inner.pull_all() } - fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + fn push(&mut self, entry: &CultCacheEnvelope) -> Result { self.inner.push(entry) } fn delete(&mut self, entry: &CultCacheEnvelope) -> Result<()> { @@ -4934,7 +5177,7 @@ mod tests { &mut self, entries: &[CultCacheEnvelope], options: PushAllOptions, - ) -> Result<()> { + ) -> Result> { if self.refuse_batch.load(Ordering::SeqCst) { return Err(anyhow!("injected batch refusal")); } @@ -5554,8 +5797,8 @@ mod tests { let absent_note = CultCacheExpectedEnvelope { key: "k".to_string(), r#type: Note::TYPE.to_string(), current: None }; type Writer = fn(&mut SingleFileMessagePackBackingStore, &[CultCacheEnvelope; 3], &CultCacheExpectedEnvelope) -> Result; let writers: Vec<(&str, Writer)> = vec![ - ("push", |s, e, _| s.push(&e[2]).map(|()| true)), - ("push_all", |s, e, _| s.push_all(&[e[0].clone(), e[2].clone()], PushAllOptions::default()).map(|()| true)), + ("push", |s, e, _| s.push(&e[2]).map(|_| true)), + ("push_all", |s, e, _| s.push_all(&[e[0].clone(), e[2].clone()], PushAllOptions::default()).map(|_| true)), ("insert_entry_if_absent", |s, e, _| s.insert_entry_if_absent(e[2].clone())), ("compare_and_swap_entry", |s, e, _| s.compare_and_swap_entry(&e[0], e[2].clone())), ("compare_exchange", |s, e, x| s.compare_exchange(std::slice::from_ref(x), &[e[2].clone()])), @@ -6633,9 +6876,10 @@ mod tests { retries: 0, }); current.retries += 1; - current + Some(current) })?; - assert_eq!(updated.retries, 1); + assert_eq!(updated.outcome, CultUpdateOutcome::Updated); + assert_eq!(updated.document.map(|settings| settings.retries), Some(1)); assert!(cache.delete::("app")?); assert!(cache.get::("app")?.is_none()); Ok(()) @@ -7351,10 +7595,10 @@ mod tests { let mut store = SingleFileMessagePackBackingStore::new(&path); let outcome = match operation { "open" => store.pull_all().map(|_| ()), - "soft_push_all" => store.push_all(&[snapshot_envelope("x", b"one")], PushAllOptions { soft: true }), - "push_all" => store.push_all(&[snapshot_envelope("x", b"one")], PushAllOptions::default()), + "soft_push_all" => store.push_all(&[snapshot_envelope("x", b"one")], PushAllOptions { soft: true }).map(|_| ()), + "push_all" => store.push_all(&[snapshot_envelope("x", b"one")], PushAllOptions::default()).map(|_| ()), "delete" => store.delete(&snapshot_envelope("x", b"one")), - _ => store.push(&snapshot_envelope("x", b"one")), + _ => store.push(&snapshot_envelope("x", b"one")).map(|_| ()), }; let what = format!("{vector} {operation}"); assert_eq!(outcome.is_ok(), reads, "{what}: {outcome:?}"); @@ -8016,7 +8260,7 @@ mod tests { let mut store = SingleFileMessagePackBackingStore::new(&path); let outcome = match operation { "open" => store.pull_all().map(|_| ()), - _ => store.push_all(&[snapshot_envelope("x", b"one")], PushAllOptions::default()), + _ => store.push_all(&[snapshot_envelope("x", b"one")], PushAllOptions::default()).map(|_| ()), }; let error = outcome.expect_err(&format!("{vector} {operation}")); assert!(unreadable_kind(&error).is_some(), "{vector} {operation}: {error:#}"); @@ -8168,7 +8412,7 @@ mod tests { let mut writer = SingleFileMessagePackBackingStore::new(&path); let delta = snapshot_envelope("delta", b"4"); let writes = [ - writer.push(&delta), + writer.push(&delta).map(|_| ()), writer.delete(&delta), writer.insert_entry_if_absent(delta.clone()).map(|_| ()), writer.compare_and_swap_batch(&[], vec![delta.clone()]).map(|_| ()), @@ -8339,8 +8583,8 @@ mod tests { let dangling = volume.join("missing.cc"); let path = temp.path().join("late.cc"); let writers: [(&str, fn(&mut SingleFileMessagePackBackingStore) -> Result<()>); 4] = [ - ("push", |s| s.push(&snapshot_envelope("b", b"2"))), - ("push_all", |s| s.push_all(&[snapshot_envelope("b", b"2")], PushAllOptions::default())), + ("push", |s| s.push(&snapshot_envelope("b", b"2")).map(|_| ())), + ("push_all", |s| s.push_all(&[snapshot_envelope("b", b"2")], PushAllOptions::default()).map(|_| ())), ("delete", |s| s.delete(&snapshot_envelope("a", b"1"))), ("compare_exchange_snapshot", |s| s.compare_exchange_snapshot(&[], &[snapshot_envelope("b", b"2")]).map(|_| ())), ]; diff --git a/packages/cultcache-rs/tests/slot_readonly.rs b/packages/cultcache-rs/tests/slot_readonly.rs index cea0853a..7851c818 100644 --- a/packages/cultcache-rs/tests/slot_readonly.rs +++ b/packages/cultcache-rs/tests/slot_readonly.rs @@ -270,7 +270,7 @@ fn every_write_entry_refuses_a_marked_key_before_any_store_is_touched() { assert_refused_with(&cache.put_prepared_batch(vec![replacing]).unwrap_err(), "marked", &expected); assert_refused_with(&cache.delete::("marked").unwrap_err(), "marked", &expected); assert_refused_with( - &cache.update::("marked", |current| current.unwrap()).unwrap_err(), + &cache.update::("marked", |current| current).unwrap_err(), "marked", &expected, ); diff --git a/packages/cultcache-rs/tests/union_parity.rs b/packages/cultcache-rs/tests/union_parity.rs index 5dbe0e45..81a0da7a 100644 --- a/packages/cultcache-rs/tests/union_parity.rs +++ b/packages/cultcache-rs/tests/union_parity.rs @@ -674,7 +674,7 @@ fn a_refused_load_refuses_every_write() { ("put_prepared_batch", cache.put_prepared_batch(vec![staged.clone()]).unwrap_err()), ("put_envelope", cache.put_envelope::(staged.clone()).unwrap_err()), ("put_raw_envelope", cache.put_raw_envelope(staged.clone()).unwrap_err()), - ("update", cache.update::("good", |current| current.unwrap_or_else(constructed)).unwrap_err()), + ("update", cache.update::("good", |current| Some(current.unwrap_or_else(constructed))).unwrap_err()), ("delete", cache.delete::("good").unwrap_err()), ("delete a key the image lacks", cache.delete::("absent").unwrap_err()), ]; diff --git a/packages/cultcache-rs/tests/unique_index.rs b/packages/cultcache-rs/tests/unique_index.rs index 58b97293..1e3ea6a2 100644 --- a/packages/cultcache-rs/tests/unique_index.rs +++ b/packages/cultcache-rs/tests/unique_index.rs @@ -221,7 +221,7 @@ fn a_record_rewriting_its_own_key_keeps_its_value_and_a_change_releases_the_old_ assert_eq!(cache.get_by_index::("Sku", "S").unwrap().unwrap().sku, "S"); assert_eq!(cache.get_by_index::("Sku", "T").unwrap().unwrap().sku, "T"); // Updating through `update` is the same rule. - assert!(cache.update::("k", |_| unique("S")).is_err()); + assert!(cache.update::("k", |_| Some(unique("S"))).is_err()); assert_eq!(cache.get::("k").unwrap().unwrap().sku, "T"); } @@ -326,7 +326,7 @@ fn every_write_path_refuses_and_changes_nothing() { assert_eq!(text, refusal(first, second, "Multi", alias)); }; pick(&error_text(cache.put("m:new", clash))); - pick(&error_text(cache.update::("m:new", |_| clash.clone()))); + pick(&error_text(cache.update::("m:new", |_| Some(clash.clone())))); let envelope = staged(&cache, &[("m:new", clash.clone())]).remove(0); pick(&error_text(cache.put_envelope::(envelope.clone()))); pick(&error_text(cache.put_raw_envelope(envelope.clone()))); diff --git a/packages/cultmesh-rs/Cargo.lock b/packages/cultmesh-rs/Cargo.lock index de7c506e..c45da112 100644 --- a/packages/cultmesh-rs/Cargo.lock +++ b/packages/cultmesh-rs/Cargo.lock @@ -213,6 +213,7 @@ dependencies = [ "serde_bytes", "serde_json", "serde_stacker", + "sha2", "stacker", "uuid", "windows-sys 0.59.0", diff --git a/packages/cultnet-rs/Cargo.lock b/packages/cultnet-rs/Cargo.lock index ed0aab43..e1af9bd6 100644 --- a/packages/cultnet-rs/Cargo.lock +++ b/packages/cultnet-rs/Cargo.lock @@ -195,6 +195,7 @@ dependencies = [ "serde_bytes", "serde_json", "serde_stacker", + "sha2", "stacker", "uuid", "windows-sys 0.59.0", diff --git a/packages/cultnet-rs/src/idunn_runtime_provider/tests.rs b/packages/cultnet-rs/src/idunn_runtime_provider/tests.rs index 9f2d53ec..31760f2c 100644 --- a/packages/cultnet-rs/src/idunn_runtime_provider/tests.rs +++ b/packages/cultnet-rs/src/idunn_runtime_provider/tests.rs @@ -51,6 +51,7 @@ fn write_record(path: &Path, key: &str, ty: &str, schema: &str, payload: Vec stored_at: "2026-10-29T00:00:00Z".into(), schema_id: Some(schema.into()), }) + .map(|_| ()) } // ------------------------------------------------------------------ fake host diff --git a/packages/cultnet-rs/src/service_identity.rs b/packages/cultnet-rs/src/service_identity.rs index 63ec29f6..8ea978c8 100644 --- a/packages/cultnet-rs/src/service_identity.rs +++ b/packages/cultnet-rs/src/service_identity.rs @@ -347,7 +347,9 @@ pub fn export_service_identity_trust_anchor( }; let mut backing = SingleFileMessagePackBackingStore::new(path); match backing.pull_all()?.as_slice() { - [] => backing.push(&envelope)?, + [] => { + backing.push(&envelope)?; + } [current] if current == &envelope => {} _ => bail!("service identity trust anchor output already contains different state"), } -- 2.47.3 From 6e6c12310071527d8b0ddf0bc07bb5b77c4ee4ea Mon Sep 17 00:00:00 2001 From: Libby Date: Fri, 13 Nov 2026 19:56:42 +0000 Subject: [PATCH 2/6] cultcache-rs: update tests, content-version vector, docs snippet, README and CHANGELOG Co-Authored-By: Claude Sonnet 5.5 --- docs/start/README.md | 32 +- packages/cultcache-rs/CHANGELOG.md | 8 + packages/cultcache-rs/README.md | 9 +- packages/cultcache-rs/src/lib.rs | 35 + .../tests/content_version_vector.rs | 26 + packages/cultcache-rs/tests/first_call.rs | 58 +- packages/cultcache-rs/tests/update.rs | 657 ++++++++++++++++++ 7 files changed, 816 insertions(+), 9 deletions(-) create mode 100644 packages/cultcache-rs/tests/content_version_vector.rs create mode 100644 packages/cultcache-rs/tests/update.rs diff --git a/docs/start/README.md b/docs/start/README.md index d877469c..a86f4732 100644 --- a/docs/start/README.md +++ b/docs/start/README.md @@ -18,7 +18,7 @@ this page. | --- | --- | --- | --- | --- | --- | | **C#** | [first call, tested](#c-first-call); [update, tested](#c-update-with-retry) | [package](../../packages/cultmath/README.md); snippet not yet | [package](../../src/GameCult.Geometry/README.md); snippet not yet | [first call, tested](#c-cultnet-first-call); [package](../../src/GameCult.Networking) | [quickstart](../../src/GameCult.Mesh/docs/durable-node-quickstart.md); run by `scripts/verify-eve-getting-started.ps1` | | **TypeScript** | [first call, tested](#typescript-first-call); [update, tested](#typescript-update-with-retry) | not yet | not yet | [first call, tested](#typescript-cultnet-first-call); [package](../../packages/cultnet-ts/README.md) | [package](../../packages/cultmesh-ts/README.md); snippet not yet; [recipe: two-client .io game](io-game.md), tested with two live clients, browser page not yet run | -| **Rust** | [first call, tested](#rust-first-call) | [native core](../../packages/cultmath/native/cultmath-core); snippet not yet | not yet | [first call, tested](#rust-cultnet-first-call); [package](../../packages/cultnet-rs/README.md), [examples](../../packages/cultnet-rs/examples) | [package](../../packages/cultmesh-rs); snippet not yet | +| **Rust** | [first call, tested](#rust-first-call); [update, tested](#rust-update-with-retry) | [native core](../../packages/cultmath/native/cultmath-core); snippet not yet | not yet | [first call, tested](#rust-cultnet-first-call); [package](../../packages/cultnet-rs/README.md), [examples](../../packages/cultnet-rs/examples) | [package](../../packages/cultmesh-rs); snippet not yet | | **Python** | [first call, tested](#python-first-call) | not yet | not yet | [first call, tested](#python-cultnet-first-call); [package](../../packages/cultnet-py/README.md) | [package](../../packages/cultmesh-py/README.md); snippet not yet | For the whole @@ -220,6 +220,36 @@ fn first_call() -> anyhow::Result<()> { } ``` +## Rust update with retry + +`update` changes a record from what the store holds now and commits only if it is still that record. The source is the +second region of +[`packages/cultcache-rs/tests/first_call.rs`](../../packages/cultcache-rs/tests/first_call.rs), which uses the `Item` of +the first call above; a test in that file fails if the block stops matching the file. + +```rust +fn restock(path: &Path) -> anyhow::Result> { + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path))?; + cache.pull_all_backing_stores()?; + cache.update::("item:potion", |item| { + // The store's copy of the record, or nothing if it has none; this may run again on a lost race. + let mut item = item.unwrap_or(Item { + name: "Potion".to_string(), + category: "Consumable".to_string(), + value: 0, + }); + item.value += 10; + Some(item) + }) +} +``` + +The write is durable when `update` returns. A lost race re-runs the closure, so keep it free of side effects; when the +attempts run out (five, or the bound `update_with_attempts` takes) the outcome says `LostRace` instead of failing, and +returning `None` writes nothing and says `Declined`. `put` is the blind opt-out: last writer wins. + ## Python first call A document is a definition with an encoder and decoder; `name=` registers the name lookup, `indexes=` unique lookups and diff --git a/packages/cultcache-rs/CHANGELOG.md b/packages/cultcache-rs/CHANGELOG.md index 327f42aa..20ea5146 100644 --- a/packages/cultcache-rs/CHANGELOG.md +++ b/packages/cultcache-rs/CHANGELOG.md @@ -6,8 +6,14 @@ only in git history. ## Unreleased +### Added + +- `CultCache::update_with_attempts` and the conditional `update`: each attempt reads the record from its home store with its content version in one read, applies a function to a copy and commits only over that version, rerunning on a lost race (default five attempts). The result is `CultUpdateResult`, outcome `Updated`, `Declined` (the function returned `None`) or `LostRace`, with the cache's document. Public types `CultStoredRecord`, `CultCommitIf`, `CultUpdateOutcome`, `CultUpdateResult`. +- `CacheBackingStore::read(key)` and `commit_if(entry, expected)`, provided methods whose defaults refuse with an error naming the store kind; `SingleFileMessagePackBackingStore` implements both. The content version is the lowercase hex SHA-256 of the record as the store file encodes it (key, schema id, `stored_at`, payload), equal to the C# reference's `tests/vectors/content-version`. New dependency `sha2` 0.10 (with `digest`, `block-buffer`, `crypto-common`, `generic-array`, `typenum`, `cpufeatures`). + ### Changed +- The single-file store's `push`, `push_all` and `commit_if` stamp a replacing write strictly later than the record it replaces at the same key (a staged stamp stands only when it parses and is strictly later; otherwise the stamp is minted as the replaced stamp plus one microsecond, or now if later), with no identical-content exception, as C# does. The cache holds the stamp the store wrote. The redb stores and the store-level exact-envelope writes (`compare_and_swap_entry` and its family) still write the caller's envelope unchanged. - The single-file store re-checks for a link once the store lock is held (every write and the try-exchange), so a link put at the path while a writer waited on the lock is refused with `StoreLinkedError` and the lock released, as C# does, instead of being replaced. Pinned by a test that puts the link while a second holder owns the lock. - `CultCache::is_writable` does not look at store paths: a link at a store path is not predicted, and the write refuses it with `StoreLinkedError`. Pinned by a test through a cache attached to a linked store. @@ -15,6 +21,8 @@ only in git history. Semver: this is a breaking change in the 0.x line (0.3.x to 0.4.0). +- `CultCache::update` is replaced, not extended: `update::(key, apply)` takes `FnMut(Option) -> Option` and returns `Result>`. The blind `FnOnce(Option) -> T` form that returned `T` is gone; use `put` for a blind write. It reads the home store, not the cache's image, so a cache whose store has no `read` and `commit_if` (the redb stores, a custom store) refuses it before `apply` runs. +- `CacheBackingStore::push` returns `Result`, the `stored_at` written, and `push_all` returns `Result>`, one stamp per entry given in order (a soft write that wrote nothing returns an empty vec). A custom store must return what it wrote; the redb stores return the entries' own stamps. Huginn implements `CacheBackingStore` in four test stores under a pinned git rev: they change at its next bump. - A write at a store path that is a symbolic link, live or dangling, is refused with the new public `StoreLinkedError` (`error.downcast_ref::()`) before the lock and the replace, and writes nothing: `push`, `push_all`, `delete`, the exchanges, and a soft `push_all`, which used to skip a link and now refuses. Before, a write at a live link replaced the link with a regular file and a dangling link failed with `io::ErrorKind::NotFound`. The redb stores (`RedbMessagePackBackingStore`, `OwnedRedbMessagePackBackingStore`) are exempt: they are ruled migrated and deleted (`paged-store-redb-migrate-and-delete`) and still open a live link's target. Configure the link's target as the store path. Reads through a live link work. - A record whose payload carries unread slots is read-only. A payload with a slot past its type's declared slots, at any depth (record, nested record, list element, union arm, unit arm), still decodes, and every write or removal of its key is now refused before any store is touched: `put`, `put_envelope`, `put_raw_envelope`, `put_prepared_batch` (the whole batch) and `delete`, so `update` too. The refusal is `SchemaConflictError` with the new `unread_slots`; its message names the type and both slot counts and never a slot value. Holders of such a record used to overwrite it and drop the extra slots silently. The payload nesting refusal still runs first. A held payload this build cannot read at all (an arm it lacks from a newer writer, or a corrupt record) is read-only too: every write or removal is refused with an error naming the type and key (a newer writer's record is read with that build; a corrupt one is restored from a backup or removed at the store itself, as no cache call removes it), and `is_writable` answers false. `load_soa` is a write of every held key it omits or changes, so it refuses to drop or replace a read-only record. `is_writable` also answers false for a key a store carries as a foreign record and for a type other than the one holding the key, as the writes do. Each write over a held key decodes the held payload once to ask (about 40% of a `put` of a 200,000-element record, measured in release); keys nothing holds pay nothing. - cultmesh-rs `pull_rudp_catalog_snapshot` applies a snapshot as one cache commit (`put_prepared_batch`), as C# does: a snapshot document for a key the node holds read-only, or any other refusal, applies nothing. It used to apply document by document, so a refusal ended the pull with the earlier documents kept. A snapshot whose types route to different backing stores is now refused whole, as a batch cannot span stores. C# `CultCache.Commit` refuses a cross-store commit the same way (`The batch spans A and B; a commit lands in one home store`), so the two agree; a test pins it. diff --git a/packages/cultcache-rs/README.md b/packages/cultcache-rs/README.md index 06c530b1..193851b2 100644 --- a/packages/cultcache-rs/README.md +++ b/packages/cultcache-rs/README.md @@ -186,7 +186,7 @@ without pretending Rust has runtime reflection hiding under the floorboards. - `soa::` - `put::` - `put_envelope::` -- `update::` +- `update::` and `update_with_attempts::`: the conditional golden-path write (read the stored record, apply, commit only over what was read, rerun on a lost race) - `delete::` - `snapshot` - `snapshot_soa` @@ -304,6 +304,13 @@ The contract, shared with every runtime, is one home store per document type ## Persistence Semantics +`update` is the golden path for changing a record another writer may also change: each attempt reads the record and its content +version from the home store, hands `apply` a copy, and commits only if the store still holds that version. `put` is the blind +write. The single-file store stamps a replacing write (`push`, `push_all` and `update`'s commit) strictly later than the record +it replaces, and the cache holds the stamp the store reports; the redb stores and the exact-envelope family +(`compare_and_swap_entry` and the other store-level conditional writes) write the caller's envelope byte for byte. A store +needs `read` and `commit_if` to serve `update`; the redb stores do not have them. + `put` persists before mutating the in-memory cache. If persistence fails, the cache does not pretend the write succeeded. diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index 44b5da9f..87d93d5b 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -5958,6 +5958,41 @@ mod tests { Ok(()) } + fn is_minted_text(text: &str) -> bool { + let bytes = text.as_bytes(); + bytes.len() == 27 + && bytes.ends_with(b"Z") + && bytes[19] == b'.' + && bytes[20..26].iter().all(u8::is_ascii_digit) + && chrono::DateTime::parse_from_rfc3339(text).is_ok() + } + + fn at(text: &str) -> chrono::DateTime { + chrono::DateTime::parse_from_rfc3339(text).unwrap() + } + + #[test] + fn stamps_stand_only_when_strictly_later_and_otherwise_mint_over_the_replaced() { + // A strictly later staged stamp stands as written. + assert_eq!(stored_at_over("2099-01-01T00:00:00Z", "2026-01-01T00:00:00Z"), "2099-01-01T00:00:00Z"); + // An equal or earlier one is replaced by a minted stamp later than the replaced, over C#'s seven digits too. + for replaced in ["2026-01-01T00:00:00.1234567+00:00", "2026-01-01T00:00:00Z", "2099-01-01T00:00:00Z"] { + for staged in [replaced, "2020-01-01T00:00:00Z"] { + let minted = stored_at_over(staged, replaced); + assert!(is_minted_text(&minted), "{minted}"); + assert!(at(&minted) > at(replaced), "{minted} over {replaced}"); + } + } + assert_eq!(stored_at_over("2020-01-01T00:00:00Z", "2099-01-01T00:00:00Z"), "2099-01-01T00:00:00.000001Z"); + // An unparseable stamp on either side is never kept: the minted stamp is the replaced's successor or now. + let minted = stored_at_over("not a stamp", "2099-01-01T00:00:00Z"); + assert_eq!(minted, "2099-01-01T00:00:00.000001Z"); + let minted = stored_at_over("2099-01-01T00:00:00Z", "not a stamp"); + assert!(is_minted_text(&minted)); + assert!((chrono::Utc::now() - at(&minted).with_timezone(&chrono::Utc)).num_seconds().abs() < 60, "{minted}"); + assert!(is_minted_text(&mint_stored_at(None, chrono::Utc::now()))); + } + #[test] fn failed_snapshot_publication_preserves_the_committed_destination() -> Result<()> { let temp = tempfile::tempdir()?; diff --git a/packages/cultcache-rs/tests/content_version_vector.rs b/packages/cultcache-rs/tests/content_version_vector.rs new file mode 100644 index 00000000..56304b6f --- /dev/null +++ b/packages/cultcache-rs/tests/content_version_vector.rs @@ -0,0 +1,26 @@ +//! `tests/vectors/content-version`, written only by the C# reference: the content version of a stored record is the lowercase +//! hex SHA-256 of its persisted encoding (`variants:ruling:write-set-content-version-declared-ownership`). The store reports +//! the version of every record of the committed `store.cc` as `versions.msgpack` says. + +use std::path::PathBuf; + +use cultcache_rs::{CacheBackingStore, SingleFileMessagePackBackingStore}; + +fn vectors() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../tests/vectors/content-version") +} + +#[test] +fn read_reports_every_version_the_reference_reads() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("store.cc"); + std::fs::copy(vectors().join("store.cc"), &path).unwrap(); + let rows: Vec<(String, String)> = rmp_serde::from_slice(&std::fs::read(vectors().join("versions.msgpack")).unwrap()).unwrap(); + assert!(rows.len() >= 2); + let store = SingleFileMessagePackBackingStore::new(&path); + for (key, version) in rows { + let record = store.read(&key).unwrap().unwrap_or_else(|| panic!("no record {key}")); + assert_eq!(record.version, version, "{key}"); + } + assert_eq!(store.read("absent").unwrap(), None); +} diff --git a/packages/cultcache-rs/tests/first_call.rs b/packages/cultcache-rs/tests/first_call.rs index 6859ae07..c6d8d7f5 100644 --- a/packages/cultcache-rs/tests/first_call.rs +++ b/packages/cultcache-rs/tests/first_call.rs @@ -1,7 +1,9 @@ -//! The Rust CultCache first call from docs/start/README.md. The docs show the region between the -//! `snippet` markers verbatim, and `docs_show_the_tested_snippet` fails when they differ. +//! The Rust CultCache first call and update from docs/start/README.md. The docs show each region between its +//! `snippet` markers verbatim, and `docs_show_the_tested_snippets` fails when they differ. -use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore}; +use std::path::Path; + +use cultcache_rs::{CultCache, CultUpdateResult, DatabaseEntry, SingleFileMessagePackBackingStore}; // snippet:begin #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] @@ -40,17 +42,59 @@ fn first_call() -> anyhow::Result<()> { } // snippet:end +// snippet:update:begin +fn restock(path: &Path) -> anyhow::Result> { + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path))?; + cache.pull_all_backing_stores()?; + cache.update::("item:potion", |item| { + // The store's copy of the record, or nothing if it has none; this may run again on a lost race. + let mut item = item.unwrap_or(Item { + name: "Potion".to_string(), + category: "Consumable".to_string(), + value: 0, + }); + item.value += 10; + Some(item) + }) +} +// snippet:update:end + #[test] fn first_call_round_trips_through_a_second_cache() { first_call().unwrap(); } #[test] -fn docs_show_the_tested_snippet() { +fn restock_twice_reads_twenty_through_a_second_cache() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("restock.cc"); + assert_eq!(restock(&path).unwrap().outcome, cultcache_rs::CultUpdateOutcome::Updated); + let second = restock(&path).unwrap(); + assert_eq!(second.outcome, cultcache_rs::CultUpdateOutcome::Updated); + assert_eq!(second.document.unwrap().value, 20); + + let mut reopened = CultCache::new(); + reopened.register_entry_type::().unwrap(); + reopened.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap(); + reopened.pull_all_backing_stores().unwrap(); + assert_eq!(reopened.get_required::("item:potion").unwrap().value, 20); +} + +/// The docs show the region between `// snippet:begin` and `end` of this file as one fenced block. +fn docs_show_region(region: &str) { let source = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/first_call.rs")).unwrap(); - let begin = source.find("// snippet:begin\n").unwrap() + "// snippet:begin\n".len(); - let end = source.find("// snippet:end\n").unwrap(); + let (begin_marker, end_marker) = (format!("// snippet:{region}begin\n"), format!("// snippet:{region}end\n")); + let begin = source.find(&begin_marker).unwrap() + begin_marker.len(); + let end = source.find(&end_marker).unwrap(); let docs = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/../../docs/start/README.md")).unwrap(); let block = format!("```rust\n{}```", &source[begin..end]); - assert!(docs.contains(&block), "docs/start/README.md no longer shows tests/first_call.rs verbatim"); + assert!(docs.contains(&block), "docs/start/README.md no longer shows the {region:?} region of tests/first_call.rs verbatim"); +} + +#[test] +fn docs_show_the_tested_snippets() { + docs_show_region(""); + docs_show_region("update:"); } diff --git a/packages/cultcache-rs/tests/update.rs b/packages/cultcache-rs/tests/update.rs new file mode 100644 index 00000000..7891a234 --- /dev/null +++ b/packages/cultcache-rs/tests/update.rs @@ -0,0 +1,657 @@ +//! `CultCache::update`: read the record as its home store holds it, apply, commit only over what was read, rerun on a lost race +//! (C# `CultCache.Update`, as amended by PR #170). A rival is a second `CultCache` on the same file. + +use std::cell::Cell; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Barrier}; + +use cultcache_rs::{ + CacheBackingStore, CultCache, CultCacheEnvelope, CultCommitIf, CultStoredRecord, CultUpdateOutcome, DatabaseEntry, + PushAllOptions, RedbMessagePackBackingStore, SingleFileMessagePackBackingStore, +}; + +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache(type = "counter")] +struct Counter { + #[cultcache(key = 0)] + n: u32, +} + +/// The same type with a second slot: a record it writes carries a slot `Counter` does not read. +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache(type = "counter")] +struct WideCounter { + #[cultcache(key = 0)] + n: u32, + #[cultcache(key = 1)] + extra: u32, +} + +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache(type = "note")] +struct Note { + #[cultcache(key = 0)] + text: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache(type = "unique_sku")] +struct Sku { + #[cultcache(key = 0, index = "Sku")] + sku: String, +} + +fn counter(n: u32) -> Counter { + Counter { n } +} + +fn open(path: &Path) -> CultCache { + let mut cache = CultCache::new(); + cache.register_entry_type::().unwrap(); + cache.register_entry_type::().unwrap(); + cache.register_entry_type::().unwrap(); + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path)).unwrap(); + cache.pull_all_backing_stores().unwrap(); + cache +} + +fn path_in(dir: &tempfile::TempDir) -> PathBuf { + dir.path().join("store.cc") +} + +/// What a fresh cache reads at `key`. +fn stored(path: &Path, key: &str) -> Option { + open(path).get::(key).unwrap().map(|counter| counter.n) +} + +fn file(path: &Path) -> Option> { + std::fs::read(path).ok() +} + +#[test] +fn two_writers_updating_lose_no_increment() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let calls = Arc::new(AtomicUsize::new(0)); + let start = Arc::new(Barrier::new(2)); + let threads: Vec<_> = (0..2) + .map(|_| { + let (path, calls, start) = (path.clone(), calls.clone(), start.clone()); + std::thread::spawn(move || { + let mut cache = open(&path); + start.wait(); + for _ in 0..50 { + let result = cache + .update_with_attempts::("c", 100, |current| { + calls.fetch_add(1, Ordering::SeqCst); + // The window a rival can land in. + std::thread::sleep(std::time::Duration::from_millis(1)); + Some(counter(current.map_or(0, |held| held.n) + 1)) + }) + .unwrap(); + assert_eq!(result.outcome, CultUpdateOutcome::Updated); + } + }) + }) + .collect(); + for thread in threads { + thread.join().unwrap(); + } + assert_eq!(stored(&path, "c"), Some(100)); + assert!(calls.load(Ordering::SeqCst) > 100, "no apply was rerun, so no race was exercised"); +} + +#[test] +fn a_lost_attempt_reruns_apply_on_the_fresh_value() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut cache = open(&path); + cache.put("c", &counter(1)).unwrap(); + let mut rival = open(&path); + let mut seen = Vec::new(); + let result = cache + .update::("c", |current| { + let current = current.unwrap().n; + if seen.is_empty() { + rival.put("c", &counter(100)).unwrap(); + } + seen.push(current); + Some(counter(current + 1)) + }) + .unwrap(); + assert_eq!(seen, [1, 100]); + assert_eq!(result.outcome, CultUpdateOutcome::Updated); + assert_eq!(result.document, Some(counter(101))); + assert_eq!(stored(&path, "c"), Some(101)); +} + +#[test] +fn exhausted_attempts_return_lost_race() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut cache = open(&path); + cache.put("c", &counter(1)).unwrap(); + let mut rival = open(&path); + let mut calls = 0; + let result = cache + .update_with_attempts::("c", 3, |current| { + calls += 1; + rival.put("c", &counter(100 * calls)).unwrap(); + current + }) + .unwrap(); + assert_eq!(calls, 3); + assert_eq!(result.outcome, CultUpdateOutcome::LostRace); + // The pull before LostRace shows the value that won. + assert_eq!(result.document, Some(counter(300))); + assert_eq!(stored(&path, "c"), Some(300)); +} + +#[test] +fn max_attempts_bounds() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut cache = open(&path); + cache.put("c", &counter(1)).unwrap(); + let mut rival = open(&path); + // Always loses; panics past 10 calls so a weakened bound fails here instead of hanging. + let mut losing = |calls: &Cell| { + let n = calls.get() + 1; + calls.set(n); + assert!(n <= 10, "apply ran past the bound"); + rival.put("c", &counter(1000 + n)).unwrap(); + Some(counter(n)) + }; + + let calls = Cell::new(0); + let one = cache.update_with_attempts::("c", 1, |_| losing(&calls)).unwrap(); + assert_eq!((calls.get(), one.outcome), (1, CultUpdateOutcome::LostRace)); + + let calls = Cell::new(0); + assert!(cache.update_with_attempts::("c", 0, |_| losing(&calls)).is_err()); + assert_eq!(calls.get(), 0); + + // The default bound is 5. + let calls = Cell::new(0); + let default = cache.update::("c", |_| losing(&calls)).unwrap(); + assert_eq!((calls.get(), default.outcome), (5, CultUpdateOutcome::LostRace)); +} + +#[test] +fn none_declines_and_writes_nothing() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut cache = open(&path); + let absent = cache.update::("c", |current| { + assert_eq!(current, None); + None + }); + let absent = absent.unwrap(); + assert_eq!((absent.outcome, absent.document), (CultUpdateOutcome::Declined, None)); + assert_eq!(file(&path), None, "a declined update created the store"); + + cache.put("c", &counter(1)).unwrap(); + let before = file(&path); + let held = cache.update::("c", |_| None).unwrap(); + assert_eq!((held.outcome, held.document), (CultUpdateOutcome::Declined, Some(counter(1)))); + assert_eq!(file(&path), before); +} + +#[test] +fn a_declined_update_installs_nothing() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut cache = open(&path); + cache.put("c", &counter(1)).unwrap(); + let mut rival = open(&path); + rival.put("c", &counter(7)).unwrap(); + let mut seen = None; + let result = cache + .update::("c", |current| { + seen = current; + None + }) + .unwrap(); + assert_eq!(seen, Some(counter(7)), "apply saw the held copy, not the store's"); + assert_eq!(result.document, Some(counter(1))); + assert_eq!(cache.get::("c").unwrap(), Some(counter(1))); + cache.pull_all_backing_stores().unwrap(); + assert_eq!(cache.get::("c").unwrap(), Some(counter(7))); +} + +#[test] +fn absent_key_rival_removal_and_a_store_holding_nothing() { + // Two writers creating one absent key. + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let start = Arc::new(Barrier::new(2)); + let threads: Vec<_> = (0..2) + .map(|_| { + let (path, start) = (path.clone(), start.clone()); + std::thread::spawn(move || { + let mut cache = open(&path); + start.wait(); + let result = cache + .update_with_attempts::("new", 20, |current| { + std::thread::sleep(std::time::Duration::from_millis(1)); + Some(counter(current.map_or(0, |held| held.n) + 1)) + }) + .unwrap(); + assert_eq!(result.outcome, CultUpdateOutcome::Updated); + }) + }) + .collect(); + for thread in threads { + thread.join().unwrap(); + } + assert_eq!(stored(&path, "new"), Some(2)); + + // A rival deletes the record inside apply: the retry sees nothing. + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut cache = open(&path); + cache.put("c", &counter(1)).unwrap(); + let mut rival = open(&path); + let mut seen = Vec::new(); + let result = cache + .update::("c", |current| { + if seen.is_empty() { + assert!(rival.delete::("c").unwrap()); + } + seen.push(current.clone()); + Some(counter(current.map_or(0, |held| held.n) + 1)) + }) + .unwrap(); + assert_eq!(seen, [Some(counter(1)), None]); + assert_eq!(result.outcome, CultUpdateOutcome::Updated); + assert_eq!(stored(&path, "c"), Some(1)); + + // The cache holds a record the store no longer has: apply is handed None. + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut cache = open(&path); + cache.put("c", &counter(5)).unwrap(); + assert!(open(&path).delete::("c").unwrap()); + let mut seen = Vec::new(); + cache + .update::("c", |current| { + seen.push(current); + Some(counter(1)) + }) + .unwrap(); + assert_eq!(seen, [None]); + assert_eq!(stored(&path, "c"), Some(1)); +} + +#[test] +fn a_rival_writing_and_restoring_the_value_still_reruns_apply() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut cache = open(&path); + cache.put("c", &counter(1)).unwrap(); + let mut rival = open(&path); + let mut seen = Vec::new(); + cache + .update::("c", |current| { + let current = current.unwrap().n; + if seen.is_empty() { + rival.put("c", &counter(5)).unwrap(); + rival.put("c", &counter(1)).unwrap(); + } + seen.push(current); + Some(counter(current + 1)) + }) + .unwrap(); + assert_eq!(seen, [1, 1]); + assert_eq!(stored(&path, "c"), Some(2)); +} + +#[test] +fn a_stale_caches_first_attempt_reads_the_store() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut a = open(&path); + a.put("c", &counter(1)).unwrap(); + let mut b = open(&path); + a.put("c", &counter(7)).unwrap(); + let mut seen = Vec::new(); + b.update::("c", |current| { + seen.push(current.clone()); + current.map(|held| counter(held.n + 1)) + }) + .unwrap(); + assert_eq!(seen, [Some(counter(7))]); + assert_eq!(stored(&path, "c"), Some(8)); +} + +/// A custom store over the single-file store that has no `read` or `commit_if`. +struct ReadlessStore(SingleFileMessagePackBackingStore); + +impl CacheBackingStore for ReadlessStore { + fn pull_all(&self) -> anyhow::Result> { + self.0.pull_all() + } + fn push(&mut self, entry: &CultCacheEnvelope) -> anyhow::Result { + self.0.push(entry) + } + fn delete(&mut self, entry: &CultCacheEnvelope) -> anyhow::Result<()> { + self.0.delete(entry) + } + fn push_all(&mut self, entries: &[CultCacheEnvelope], options: PushAllOptions) -> anyhow::Result> { + self.0.push_all(entries, options) + } +} + +/// Runs `update` with an apply that counts its calls; the refusal must come before the first. +fn refused(cache: &mut CultCache, path: &Path, key: &str) -> String { + let before = file(path); + let calls = Cell::new(0); + let error = cache + .update::(key, |current| { + calls.set(calls.get() + 1); + current + }) + .expect_err("the update must be refused"); + assert_eq!(calls.get(), 0, "apply ran before the refusal: {error:#}"); + assert_eq!(file(path), before, "a refused update touched the store"); + format!("{error:#}") +} + +#[test] +fn refusals_come_before_apply() { + // Another type's record written at the key since the pull: the read resolves it. + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut cache = open(&path); + open(&path).put("k", &Note { text: "n".into() }).unwrap(); + let text = refused(&mut cache, &path, "k"); + assert!(text.contains("update cannot copy it"), "{text}"); + + // Another type held at the key. + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut cache = open(&path); + cache.put("k", &Note { text: "n".into() }).unwrap(); + refused(&mut cache, &path, "k"); + + // A cache with no store holds another type at the key. + let mut memory = CultCache::new(); + memory.register_entry_type::().unwrap(); + memory.register_entry_type::().unwrap(); + memory.put("k", &Note { text: "n".into() }).unwrap(); + let calls = Cell::new(0); + assert!(memory + .update::("k", |current| { + calls.set(calls.get() + 1); + current + }) + .is_err()); + assert_eq!(calls.get(), 0); + + // A foreign record (no registered type claims it) held at the key, and one that appears after the pull. + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + open(&path).put("k", &Note { text: "n".into() }).unwrap(); + let mut only_counter = CultCache::new(); + only_counter.register_entry_type::().unwrap(); + only_counter.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap(); + only_counter.pull_all_backing_stores().unwrap(); + refused(&mut only_counter, &path, "k"); + let mut later = CultCache::new(); + later.register_entry_type::().unwrap(); + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + later.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap(); + later.pull_all_backing_stores().unwrap(); + open(&path).put("z", &Note { text: "n".into() }).unwrap(); + let text = refused(&mut later, &path, "z"); + assert!(text.contains("update cannot copy it"), "{text}"); + + // A redb home and a custom store without read. + let dir = tempfile::tempdir().unwrap(); + let redb_path = dir.path().join("store.redb"); + let mut redb = CultCache::new(); + redb.register_entry_type::().unwrap(); + redb.add_generic_backing_store(RedbMessagePackBackingStore::new(&redb_path).unwrap()).unwrap(); + redb.pull_all_backing_stores().unwrap(); + redb.put("c", &counter(1)).unwrap(); + let text = refused(&mut redb, &redb_path, "c"); + assert!(text.contains("read and commit_if"), "{text}"); + let path = path_in(&dir); + let mut custom = CultCache::new(); + custom.register_entry_type::().unwrap(); + custom.add_generic_backing_store(ReadlessStore(SingleFileMessagePackBackingStore::new(&path))).unwrap(); + custom.pull_all_backing_stores().unwrap(); + custom.put("c", &counter(1)).unwrap(); + let text = refused(&mut custom, &path, "c"); + assert!(text.contains("read and commit_if"), "{text}"); + + // A held read-only record, and a store record with slots the cache never loaded. + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut wide = CultCache::new(); + wide.register_entry_type::().unwrap(); + wide.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap(); + wide.pull_all_backing_stores().unwrap(); + wide.put("held", &WideCounter { n: 1, extra: 2 }).unwrap(); + let mut cache = open(&path); + let text = refused(&mut cache, &path, "held"); + assert!(text.contains("held"), "{text}"); + wide.put("unloaded", &WideCounter { n: 1, extra: 2 }).unwrap(); + refused(&mut cache, &path, "unloaded"); + + // An unpulled cache. + let mut unpulled = CultCache::new(); + unpulled.register_entry_type::().unwrap(); + unpulled.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap(); + refused(&mut unpulled, &path, "held"); +} + +#[test] +fn a_refusal_after_apply_is_not_retried() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut cache = open(&path); + cache.put("a", &Sku { sku: "S".into() }).unwrap(); + cache.put("b", &Sku { sku: "T".into() }).unwrap(); + let before = file(&path); + let calls = Cell::new(0); + let error = cache + .update::("b", |_| { + calls.set(calls.get() + 1); + Some(Sku { sku: "S".into() }) + }) + .expect_err("a unique-index clash"); + assert_eq!(calls.get(), 1, "{error:#}"); + assert_eq!(file(&path), before); +} + +#[test] +fn own_writes_and_the_store_less_cache() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut cache = open(&path); + cache.put("c", &counter(1)).unwrap(); + let calls = Cell::new(0); + for expected in [2, 3] { + let result = cache + .update::("c", |current| { + calls.set(calls.get() + 1); + current.map(|held| counter(held.n + 1)) + }) + .unwrap(); + assert_eq!(result.outcome, CultUpdateOutcome::Updated); + assert_eq!(result.document, Some(counter(expected))); + assert_eq!(result.document, cache.get::("c").unwrap()); + } + assert_eq!(calls.get(), 2); + + let mut memory = CultCache::new(); + memory.register_entry_type::().unwrap(); + let calls = Cell::new(0); + let result = memory + .update::("c", |current| { + calls.set(calls.get() + 1); + Some(counter(current.map_or(0, |held| held.n) + 1)) + }) + .unwrap(); + assert_eq!((calls.get(), result.outcome, result.document), (1, CultUpdateOutcome::Updated, Some(counter(1)))); +} + +#[test] +fn the_ledger_holds_under_contention() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let start = Arc::new(Barrier::new(3)); + let threads: Vec<_> = (0..3) + .map(|_| { + let (path, start) = (path.clone(), start.clone()); + std::thread::spawn(move || { + let mut cache = open(&path); + start.wait(); + (0..25) + .filter(|_| { + let result = cache + .update::("c", |current| Some(counter(current.map_or(0, |held| held.n) + 1))) + .unwrap(); + result.outcome == CultUpdateOutcome::Updated + }) + .count() as u32 + }) + }) + .collect(); + let updated: u32 = threads.into_iter().map(|thread| thread.join().unwrap()).sum(); + assert_eq!(stored(&path, "c"), Some(updated)); +} + +fn envelope_of(cache: &CultCache, key: &str, value: &T, stamp: &str) -> CultCacheEnvelope { + let (mut entry, _) = cache.prepare_entry(key, value).unwrap(); + entry.stored_at = stamp.to_string(); + entry +} + +const FUTURE: &str = "2099-01-01T00:00:00Z"; + +fn later_than_future(stamp: &str) -> bool { + chrono::DateTime::parse_from_rfc3339(stamp).unwrap() > chrono::DateTime::parse_from_rfc3339(FUTURE).unwrap() +} + +#[test] +fn commit_if_writes_only_over_the_version_it_was_given() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let cache = open(&path); + let mut store = SingleFileMessagePackBackingStore::new(&path); + assert_eq!(store.read("c").unwrap(), None); + let one = envelope_of(&cache, "c", &counter(1), FUTURE); + // Against an absent key, expecting a version mismatches; expecting nothing commits. + assert_eq!(store.commit_if(&one, Some("0000")).unwrap(), CultCommitIf::Mismatch); + assert_eq!(file(&path), None); + assert_eq!(store.commit_if(&one, None).unwrap(), CultCommitIf::Committed { stored_at: FUTURE.to_string() }); + let CultStoredRecord { envelope, version } = store.read("c").unwrap().unwrap(); + assert_eq!(envelope.stored_at, FUTURE); + + let before = file(&path); + let two = envelope_of(&cache, "c", &counter(2), FUTURE); + assert_eq!(store.commit_if(&two, None).unwrap(), CultCommitIf::Mismatch); + assert_eq!(store.commit_if(&two, Some("not-the-version")).unwrap(), CultCommitIf::Mismatch); + assert_eq!(file(&path), before, "a mismatch wrote"); + let CultCommitIf::Committed { stored_at } = store.commit_if(&two, Some(&version)).unwrap() else { panic!("the version matches") }; + assert!(later_than_future(&stored_at), "{stored_at}"); + assert_eq!(stored(&path, "c"), Some(2)); +} + +#[test] +fn the_cache_facing_writes_stamp_over_a_replaced_record() { + let seeded = |dir: &tempfile::TempDir| { + let path = path_in(dir); + let seed = open(&path); + let mut store = SingleFileMessagePackBackingStore::new(&path); + store.push(&envelope_of(&seed, "c", &counter(1), FUTURE)).unwrap(); + (path, seed) + }; + let on_disk = |path: &Path| { + let mut store = SingleFileMessagePackBackingStore::new(path); + let stamp = store.read("c").unwrap().unwrap().envelope.stored_at; + let _ = &mut store; + stamp + }; + + // The store: push, push_all and commit_if each return the stamp they wrote, strictly later than what they replaced. + let dir = tempfile::tempdir().unwrap(); + let (path, seed) = seeded(&dir); + let mut store = SingleFileMessagePackBackingStore::new(&path); + let stamp = store.push(&envelope_of(&seed, "c", &counter(2), FUTURE)).unwrap(); + assert!(later_than_future(&stamp) && stamp == on_disk(&path), "push: {stamp}"); + let stamps = store.push_all(&[envelope_of(&seed, "c", &counter(3), FUTURE)], PushAllOptions::default()).unwrap(); + assert!(later_than_future(&stamps[0]) && stamps[0] == on_disk(&path), "push_all: {stamps:?}"); + let version = store.read("c").unwrap().unwrap().version; + let CultCommitIf::Committed { stored_at } = store.commit_if(&envelope_of(&seed, "c", &counter(4), FUTURE), Some(&version)).unwrap() + else { + panic!("matching version") + }; + assert!(later_than_future(&stored_at) && stored_at == on_disk(&path), "commit_if: {stored_at}"); + // One stamp per input, in order, a duplicate getting the stamp of the entry that landed; a soft write that wrote nothing none. + let stamps = store + .push_all( + &[envelope_of(&seed, "c", &counter(5), FUTURE), envelope_of(&seed, "d", &counter(6), FUTURE), envelope_of(&seed, "c", &counter(7), FUTURE)], + PushAllOptions::default(), + ) + .unwrap(); + assert_eq!(stamps.len(), 3); + assert_eq!(stamps[0], stamps[2]); + assert_eq!(stamps[1], FUTURE, "a record replacing nothing keeps its stamp"); + assert_eq!(stamps[0], on_disk(&path)); + assert!(store.push_all(&[envelope_of(&seed, "e", &counter(1), FUTURE)], PushAllOptions { soft: true }).unwrap().is_empty()); + + // The cache holds the stamp the store wrote after every write entry. + type Write = fn(&mut CultCache) -> (); + let writes: [(&str, Write); 5] = [ + ("put", |cache| drop(cache.put("c", &counter(9)).unwrap())), + ("put_envelope", |cache| { + let entry = envelope_of(cache, "c", &counter(9), FUTURE); + drop(cache.put_envelope::(entry).unwrap()) + }), + ("put_raw_envelope", |cache| { + let entry = envelope_of(cache, "c", &counter(9), FUTURE); + cache.put_raw_envelope(entry).unwrap() + }), + ("put_prepared_batch", |cache| { + let entry = envelope_of(cache, "c", &counter(9), FUTURE); + cache.put_prepared_batch(vec![entry]).unwrap() + }), + ("update", |cache| drop(cache.update::("c", |_| Some(counter(9))).unwrap())), + ]; + for (name, write) in writes { + let dir = tempfile::tempdir().unwrap(); + let (path, _) = seeded(&dir); + let mut cache = open(&path); + write(&mut cache); + let held = cache.get_envelope::("c").unwrap().unwrap().stored_at; + assert!(later_than_future(&held), "{name}: {held}"); + assert_eq!(held, on_disk(&path), "{name}: the cache holds a stamp the store did not write"); + } +} + +#[test] +fn the_byte_exact_writers_keep_the_callers_stamp() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let cache = open(&path); + let mut store = SingleFileMessagePackBackingStore::new(&path); + let held = envelope_of(&cache, "c", &counter(1), FUTURE); + store.push(&held).unwrap(); + let swapped = envelope_of(&cache, "c", &counter(2), "2026-01-01T00:00:00Z"); + assert!(store.compare_and_swap_entry(&held, swapped.clone()).unwrap()); + assert_eq!(store.read("c").unwrap().unwrap().envelope, swapped, "compare_and_swap_entry restamped"); + + let redb_path = dir.path().join("store.redb"); + let mut redb = RedbMessagePackBackingStore::new(&redb_path).unwrap(); + redb.push(&held).unwrap(); + let older = envelope_of(&cache, "c", &counter(3), "2026-01-01T00:00:00Z"); + assert_eq!(redb.push(&older).unwrap(), older.stored_at); + assert_eq!(redb.pull_all().unwrap(), vec![older.clone()]); + assert_eq!(redb.push_all(std::slice::from_ref(&held), PushAllOptions::default()).unwrap(), vec![held.stored_at.clone()]); + assert_eq!(redb.pull_all().unwrap(), vec![held]); +} -- 2.47.3 From c00b4ad6ed87c2d572bd703b6c67d5ddf037166b Mon Sep 17 00:00:00 2001 From: Libby Date: Fri, 13 Nov 2026 19:59:56 +0000 Subject: [PATCH 3/6] cultcache-rs: pin the reported-stamp validation (RFC 3339 only) in update tests Co-Authored-By: Claude Sonnet 5.5 --- packages/cultcache-rs/tests/update.rs | 78 +++++++++++++++++++++++++++ 1 file changed, 78 insertions(+) diff --git a/packages/cultcache-rs/tests/update.rs b/packages/cultcache-rs/tests/update.rs index 7891a234..93dd06f7 100644 --- a/packages/cultcache-rs/tests/update.rs +++ b/packages/cultcache-rs/tests/update.rs @@ -655,3 +655,81 @@ fn the_byte_exact_writers_keep_the_callers_stamp() { assert_eq!(redb.push_all(std::slice::from_ref(&held), PushAllOptions::default()).unwrap(), vec![held.stored_at.clone()]); assert_eq!(redb.pull_all().unwrap(), vec![held]); } + +/// A single-file store that reports whatever stamp it is told to, as a custom store might. +struct ReportingStore { + inner: SingleFileMessagePackBackingStore, + report: String, +} + +impl CacheBackingStore for ReportingStore { + fn pull_all(&self) -> anyhow::Result> { + self.inner.pull_all() + } + fn push(&mut self, entry: &CultCacheEnvelope) -> anyhow::Result { + self.inner.push(entry)?; + Ok(self.report.clone()) + } + fn delete(&mut self, entry: &CultCacheEnvelope) -> anyhow::Result<()> { + self.inner.delete(entry) + } + fn push_all(&mut self, entries: &[CultCacheEnvelope], options: PushAllOptions) -> anyhow::Result> { + self.inner.push_all(entries, options)?; + Ok(entries.iter().map(|_| self.report.clone()).collect()) + } + fn read(&self, key: &str) -> anyhow::Result> { + self.inner.read(key) + } + fn commit_if(&mut self, entry: &CultCacheEnvelope, expected: Option<&str>) -> anyhow::Result { + Ok(match self.inner.commit_if(entry, expected)? { + CultCommitIf::Committed { .. } => CultCommitIf::Committed { stored_at: self.report.clone() }, + CultCommitIf::Mismatch => CultCommitIf::Mismatch, + }) + } +} + +#[test] +fn the_cache_holds_a_reported_stamp_only_when_it_parses_as_rfc3339() { + type Write = fn(&mut CultCache); + let writes: [(&str, Write); 4] = [ + ("put", |cache| drop(cache.put("c", &counter(1)).unwrap())), + ("put_raw_envelope", |cache| { + let entry = envelope_of(cache, "c", &counter(1), "2020-01-01T00:00:00Z"); + cache.put_raw_envelope(entry).unwrap() + }), + ("put_prepared_batch", |cache| { + let entry = envelope_of(cache, "c", &counter(1), "2020-01-01T00:00:00Z"); + cache.put_prepared_batch(vec![entry]).unwrap() + }), + ("update", |cache| drop(cache.update::("c", |_| Some(counter(1))).unwrap())), + ]; + for (report, parses) in [ + ("2030-01-01T00:00:00Z", true), + ("2030-01-01T00:00:00.123456+00:00", true), + ("not a stamp", false), + ("", false), + ("2030-01-01", false), + ("2030-01-01 00:00:00", false), + ("2030-01-01T00:00:00", false), + ("1893456000", false), + ] { + for (name, write) in writes { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let mut cache = CultCache::new(); + cache.register_entry_type::().unwrap(); + cache + .add_generic_backing_store(ReportingStore { inner: SingleFileMessagePackBackingStore::new(&path), report: report.to_string() }) + .unwrap(); + cache.pull_all_backing_stores().unwrap(); + write(&mut cache); + let held = cache.get_envelope::("c").unwrap().unwrap().stored_at; + if parses { + assert_eq!(held, report, "{name}"); + } else { + assert_ne!(held, report, "{name} held a stamp that is not RFC 3339"); + assert!(chrono::DateTime::parse_from_rfc3339(&held).is_ok(), "{name}: {held}"); + } + } + } +} -- 2.47.3 From 033a612a22351ef3d8c6dbdb64993b413734cc76 Mon Sep 17 00:00:00 2001 From: Libby Date: Fri, 13 Nov 2026 20:10:56 +0000 Subject: [PATCH 4/6] cultcache-rs: commit_if replaces only its own identity; a mint past 9999 refuses Co-Authored-By: Claude Sonnet 5.5 --- packages/cultcache-rs/src/lib.rs | 58 ++++++++++++++++++--------- packages/cultcache-rs/tests/update.rs | 35 ++++++++++++++++ 2 files changed, 74 insertions(+), 19 deletions(-) diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index 87d93d5b..cd18aea8 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -2084,7 +2084,7 @@ impl CacheBackingStore for SingleFileMessagePackBackingStore { fn push(&mut self, entry: &CultCacheEnvelope) -> Result { self.with_exclusive_lock(|| { let mut entries = self.read_all_unlocked()?; - let entry = restamped_over(entry, &entries); + let entry = restamped_over(entry, &entries)?; // A write replaces the record of its own type under its key; a second type under the key reaches the commit guard. entries.retain(|candidate| entry_id(candidate) != entry_id(&entry)); entries.push(entry.clone()); @@ -2114,8 +2114,9 @@ impl CacheBackingStore for SingleFileMessagePackBackingStore { if actual.as_deref() != expected { return Ok(CultCommitIf::Mismatch); } - let entry = restamped_over(entry, &entries); - entries.retain(|held| held.key != entry.key); + let entry = restamped_over(entry, &entries)?; + // Only the record of the entry's own identity is replaced; a second type under the key reaches the commit guard, as push does. + entries.retain(|held| entry_id(held) != entry_id(&entry)); entries.push(entry.clone()); entries.sort_by_key(entry_id); self.write_all_unlocked(&entries)?; @@ -2145,7 +2146,7 @@ impl CacheBackingStore for SingleFileMessagePackBackingStore { let mut merged = self.read_all_unlocked()?; let mut by_id = BTreeMap::<(String, String), CultCacheEnvelope>::new(); for entry in entries { - by_id.insert((schema_identity(entry).to_string(), entry.key.clone()), restamped_over(entry, &merged)); + by_id.insert((schema_identity(entry).to_string(), entry.key.clone()), restamped_over(entry, &merged)?); } let replaced: BTreeSet<_> = by_id.values().map(entry_id).collect(); merged.retain(|held| !replaced.contains(&entry_id(held))); @@ -4288,35 +4289,43 @@ fn record_version(entry: &CultCacheEnvelope) -> String { } /// A stamp strictly later than `previous` (to the microsecond): `now` truncated to the microsecond when that is later, -/// otherwise `previous` truncated plus one microsecond. An absent or unparseable `previous` gives `now`. -fn mint_stored_at(previous: Option<&str>, now: chrono::DateTime) -> String { +/// otherwise `previous` truncated plus one microsecond. An absent or unparseable `previous` gives `now`. A successor past +/// year 9999 does not exist as a stamp any runtime reads, so it is refused (C# `MintStoredAt` throws there). +fn mint_stored_at(previous: Option<&str>, now: chrono::DateTime) -> Result { use chrono::SubsecRound; let now = now.trunc_subsecs(6); let minted = match previous.and_then(|text| chrono::DateTime::parse_from_rfc3339(text).ok()) { Some(previous) => { let previous = previous.with_timezone(&chrono::Utc).trunc_subsecs(6); - if now <= previous { previous + chrono::Duration::microseconds(1) } else { now } + if now <= previous { + previous + .checked_add_signed(chrono::Duration::microseconds(1)) + .filter(|successor| chrono::Datelike::year(successor) <= 9999) + .ok_or_else(|| anyhow!("stored_at {previous} has no later stamp: the successor is past year 9999"))? + } else { + now + } } None => now, }; - minted.to_rfc3339_opts(chrono::SecondsFormat::Micros, true) + Ok(minted.to_rfc3339_opts(chrono::SecondsFormat::Micros, true)) } /// The stamp a replacing write carries: `staged` stands only when both stamps parse and it is strictly later than `replaced`. -fn stored_at_over(staged: &str, replaced: &str) -> String { +fn stored_at_over(staged: &str, replaced: &str) -> Result { match (chrono::DateTime::parse_from_rfc3339(staged), chrono::DateTime::parse_from_rfc3339(replaced)) { - (Ok(staged_at), Ok(replaced_at)) if staged_at > replaced_at => staged.to_string(), + (Ok(staged_at), Ok(replaced_at)) if staged_at > replaced_at => Ok(staged.to_string()), _ => mint_stored_at(Some(replaced), chrono::Utc::now()), } } /// `entry` stamped later than the record the store holds at its key, of any type; unchanged where nothing is held. -fn restamped_over(entry: &CultCacheEnvelope, held: &[CultCacheEnvelope]) -> CultCacheEnvelope { +fn restamped_over(entry: &CultCacheEnvelope, held: &[CultCacheEnvelope]) -> Result { let mut entry = entry.clone(); if let Some(replaced) = held.iter().find(|candidate| candidate.key == entry.key) { - entry.stored_at = stored_at_over(&entry.stored_at, &replaced.stored_at); + entry.stored_at = stored_at_over(&entry.stored_at, &replaced.stored_at)?; } - entry + Ok(entry) } fn now_utc_second() -> String { @@ -5974,23 +5983,34 @@ mod tests { #[test] fn stamps_stand_only_when_strictly_later_and_otherwise_mint_over_the_replaced() { // A strictly later staged stamp stands as written. - assert_eq!(stored_at_over("2099-01-01T00:00:00Z", "2026-01-01T00:00:00Z"), "2099-01-01T00:00:00Z"); + assert_eq!(stored_at_over("2099-01-01T00:00:00Z", "2026-01-01T00:00:00Z").unwrap(), "2099-01-01T00:00:00Z"); // An equal or earlier one is replaced by a minted stamp later than the replaced, over C#'s seven digits too. for replaced in ["2026-01-01T00:00:00.1234567+00:00", "2026-01-01T00:00:00Z", "2099-01-01T00:00:00Z"] { for staged in [replaced, "2020-01-01T00:00:00Z"] { - let minted = stored_at_over(staged, replaced); + let minted = stored_at_over(staged, replaced).unwrap(); assert!(is_minted_text(&minted), "{minted}"); assert!(at(&minted) > at(replaced), "{minted} over {replaced}"); } } - assert_eq!(stored_at_over("2020-01-01T00:00:00Z", "2099-01-01T00:00:00Z"), "2099-01-01T00:00:00.000001Z"); + assert_eq!(stored_at_over("2020-01-01T00:00:00Z", "2099-01-01T00:00:00Z").unwrap(), "2099-01-01T00:00:00.000001Z"); // An unparseable stamp on either side is never kept: the minted stamp is the replaced's successor or now. - let minted = stored_at_over("not a stamp", "2099-01-01T00:00:00Z"); + let minted = stored_at_over("not a stamp", "2099-01-01T00:00:00Z").unwrap(); assert_eq!(minted, "2099-01-01T00:00:00.000001Z"); - let minted = stored_at_over("2099-01-01T00:00:00Z", "not a stamp"); + let minted = stored_at_over("2099-01-01T00:00:00Z", "not a stamp").unwrap(); assert!(is_minted_text(&minted)); assert!((chrono::Utc::now() - at(&minted).with_timezone(&chrono::Utc)).num_seconds().abs() < 60, "{minted}"); - assert!(is_minted_text(&mint_stored_at(None, chrono::Utc::now()))); + assert!(is_minted_text(&mint_stored_at(None, chrono::Utc::now()).unwrap())); + } + + #[test] + fn a_stamp_with_no_successor_in_year_9999_is_refused() { + for replaced in ["9999-12-31T23:59:59.9999999+00:00", "9999-12-31T23:59:59.999999Z"] { + for staged in [replaced, "2020-01-01T00:00:00Z"] { + assert!(stored_at_over(staged, replaced).is_err(), "{staged} over {replaced}"); + } + } + // One microsecond earlier still has a successor. + assert_eq!(stored_at_over("2020-01-01T00:00:00Z", "9999-12-31T23:59:59.999998Z").unwrap(), "9999-12-31T23:59:59.999999Z"); } #[test] diff --git a/packages/cultcache-rs/tests/update.rs b/packages/cultcache-rs/tests/update.rs index 93dd06f7..0cc19d95 100644 --- a/packages/cultcache-rs/tests/update.rs +++ b/packages/cultcache-rs/tests/update.rs @@ -733,3 +733,38 @@ fn the_cache_holds_a_reported_stamp_only_when_it_parses_as_rfc3339() { } } } + +#[test] +fn a_stamp_with_no_successor_refuses_every_single_file_write_and_writes_nothing() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let cache = open(&path); + let mut store = SingleFileMessagePackBackingStore::new(&path); + let last = "9999-12-31T23:59:59.999999Z"; + store.push(&envelope_of(&cache, "c", &counter(1), last)).unwrap(); + let before = file(&path); + let version = store.read("c").unwrap().unwrap().version; + let next = envelope_of(&cache, "c", &counter(2), "2020-01-01T00:00:00Z"); + assert!(store.push(&next).is_err()); + assert!(store.push_all(std::slice::from_ref(&next), PushAllOptions::default()).is_err()); + assert!(store.commit_if(&next, Some(&version)).is_err()); + assert_eq!(file(&path), before, "a refused mint wrote"); +} + +#[test] +fn commit_if_refuses_a_retype_as_push_does() { + let dir = tempfile::tempdir().unwrap(); + let path = path_in(&dir); + let cache = open(&path); + let mut store = SingleFileMessagePackBackingStore::new(&path); + let note = envelope_of(&cache, "k", &Note { text: "n".into() }, "2020-01-01T00:00:00Z"); + store.push(¬e).unwrap(); + let version = store.read("k").unwrap().unwrap().version; + let before = file(&path); + let counter_at_k = envelope_of(&cache, "k", &counter(1), "2020-01-01T00:00:00Z"); + let error = store.commit_if(&counter_at_k, Some(&version)).expect_err("a retype"); + assert!(error.downcast_ref::().is_some(), "{error:#}"); + assert!(store.push(&counter_at_k).is_err()); + assert_eq!(file(&path), before, "a refused retype wrote"); + assert_eq!(store.read("k").unwrap().unwrap().envelope.r#type, "note"); +} -- 2.47.3 From 1af951f1608c6e15f5ed9c612a9196a853201fd8 Mon Sep 17 00:00:00 2001 From: Libby Date: Fri, 13 Nov 2026 20:10:56 +0000 Subject: [PATCH 5/6] cultcache-rs: state the version's minimal-encoding condition; correct Huginn store and lockfile counts Co-Authored-By: Claude Sonnet 5.5 --- packages/cultcache-rs/CHANGELOG.md | 6 +++--- packages/cultcache-rs/README.md | 1 + 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/packages/cultcache-rs/CHANGELOG.md b/packages/cultcache-rs/CHANGELOG.md index 20ea5146..0cbf6708 100644 --- a/packages/cultcache-rs/CHANGELOG.md +++ b/packages/cultcache-rs/CHANGELOG.md @@ -9,11 +9,11 @@ only in git history. ### Added - `CultCache::update_with_attempts` and the conditional `update`: each attempt reads the record from its home store with its content version in one read, applies a function to a copy and commits only over that version, rerunning on a lost race (default five attempts). The result is `CultUpdateResult`, outcome `Updated`, `Declined` (the function returned `None`) or `LostRace`, with the cache's document. Public types `CultStoredRecord`, `CultCommitIf`, `CultUpdateOutcome`, `CultUpdateResult`. -- `CacheBackingStore::read(key)` and `commit_if(entry, expected)`, provided methods whose defaults refuse with an error naming the store kind; `SingleFileMessagePackBackingStore` implements both. The content version is the lowercase hex SHA-256 of the record as the store file encodes it (key, schema id, `stored_at`, payload), equal to the C# reference's `tests/vectors/content-version`. New dependency `sha2` 0.10 (with `digest`, `block-buffer`, `crypto-common`, `generic-array`, `typenum`, `cpufeatures`). +- `CacheBackingStore::read(key)` and `commit_if(entry, expected)`, provided methods whose defaults refuse with an error naming the store kind; `SingleFileMessagePackBackingStore` implements both. The content version is the lowercase hex SHA-256 of the record re-encoded as the store file writes it (key, schema id, `stored_at`, payload). It equals the SHA-256 of the record's bytes in the file, and so the C# reference's `tests/vectors/content-version`, whenever the file's MessagePack encodings are minimal, which every runtime's writer produces; a hand-edited file with a non-minimal encoding (for example a `str8` key that fits a fixstr) hashes differently from its slice here. New dependency `sha2` 0.10 and eight new lockfile crates: `sha2`, `digest`, `block-buffer`, `crypto-common`, `generic-array`, `typenum`, `cpufeatures`, `version_check`. ### Changed -- The single-file store's `push`, `push_all` and `commit_if` stamp a replacing write strictly later than the record it replaces at the same key (a staged stamp stands only when it parses and is strictly later; otherwise the stamp is minted as the replaced stamp plus one microsecond, or now if later), with no identical-content exception, as C# does. The cache holds the stamp the store wrote. The redb stores and the store-level exact-envelope writes (`compare_and_swap_entry` and its family) still write the caller's envelope unchanged. +- The single-file store's `push`, `push_all` and `commit_if` stamp a replacing write strictly later than the record it replaces at the same key (a staged stamp stands only when it parses and is strictly later; otherwise the stamp is minted as the replaced stamp plus one microsecond, or now if later), with no identical-content exception, as C# does. A stamp with no successor (the replaced stamp is at year 9999's last microsecond) refuses the write and writes nothing, as C# `MintStoredAt` does. `commit_if` replaces only the record of the entry's own type under the key, so a retype is refused as `push` refuses it. The cache holds the stamp the store wrote. The redb stores and the store-level exact-envelope writes (`compare_and_swap_entry` and its family) still write the caller's envelope unchanged. - The single-file store re-checks for a link once the store lock is held (every write and the try-exchange), so a link put at the path while a writer waited on the lock is refused with `StoreLinkedError` and the lock released, as C# does, instead of being replaced. Pinned by a test that puts the link while a second holder owns the lock. - `CultCache::is_writable` does not look at store paths: a link at a store path is not predicted, and the write refuses it with `StoreLinkedError`. Pinned by a test through a cache attached to a linked store. @@ -22,7 +22,7 @@ only in git history. Semver: this is a breaking change in the 0.x line (0.3.x to 0.4.0). - `CultCache::update` is replaced, not extended: `update::(key, apply)` takes `FnMut(Option) -> Option` and returns `Result>`. The blind `FnOnce(Option) -> T` form that returned `T` is gone; use `put` for a blind write. It reads the home store, not the cache's image, so a cache whose store has no `read` and `commit_if` (the redb stores, a custom store) refuses it before `apply` runs. -- `CacheBackingStore::push` returns `Result`, the `stored_at` written, and `push_all` returns `Result>`, one stamp per entry given in order (a soft write that wrote nothing returns an empty vec). A custom store must return what it wrote; the redb stores return the entries' own stamps. Huginn implements `CacheBackingStore` in four test stores under a pinned git rev: they change at its next bump. +- `CacheBackingStore::push` returns `Result`, the `stored_at` written, and `push_all` returns `Result>`, one stamp per entry given in order (a soft write that wrote nothing returns an empty vec). A custom store must return what it wrote; the redb stores return the entries' own stamps. Huginn implements `CacheBackingStore` in five test stores under a pinned git rev: they change at its next bump. - A write at a store path that is a symbolic link, live or dangling, is refused with the new public `StoreLinkedError` (`error.downcast_ref::()`) before the lock and the replace, and writes nothing: `push`, `push_all`, `delete`, the exchanges, and a soft `push_all`, which used to skip a link and now refuses. Before, a write at a live link replaced the link with a regular file and a dangling link failed with `io::ErrorKind::NotFound`. The redb stores (`RedbMessagePackBackingStore`, `OwnedRedbMessagePackBackingStore`) are exempt: they are ruled migrated and deleted (`paged-store-redb-migrate-and-delete`) and still open a live link's target. Configure the link's target as the store path. Reads through a live link work. - A record whose payload carries unread slots is read-only. A payload with a slot past its type's declared slots, at any depth (record, nested record, list element, union arm, unit arm), still decodes, and every write or removal of its key is now refused before any store is touched: `put`, `put_envelope`, `put_raw_envelope`, `put_prepared_batch` (the whole batch) and `delete`, so `update` too. The refusal is `SchemaConflictError` with the new `unread_slots`; its message names the type and both slot counts and never a slot value. Holders of such a record used to overwrite it and drop the extra slots silently. The payload nesting refusal still runs first. A held payload this build cannot read at all (an arm it lacks from a newer writer, or a corrupt record) is read-only too: every write or removal is refused with an error naming the type and key (a newer writer's record is read with that build; a corrupt one is restored from a backup or removed at the store itself, as no cache call removes it), and `is_writable` answers false. `load_soa` is a write of every held key it omits or changes, so it refuses to drop or replace a read-only record. `is_writable` also answers false for a key a store carries as a foreign record and for a type other than the one holding the key, as the writes do. Each write over a held key decodes the held payload once to ask (about 40% of a `put` of a 200,000-element record, measured in release); keys nothing holds pay nothing. - cultmesh-rs `pull_rudp_catalog_snapshot` applies a snapshot as one cache commit (`put_prepared_batch`), as C# does: a snapshot document for a key the node holds read-only, or any other refusal, applies nothing. It used to apply document by document, so a refusal ended the pull with the earlier documents kept. A snapshot whose types route to different backing stores is now refused whole, as a batch cannot span stores. C# `CultCache.Commit` refuses a cross-store commit the same way (`The batch spans A and B; a commit lands in one home store`), so the two agree; a test pins it. diff --git a/packages/cultcache-rs/README.md b/packages/cultcache-rs/README.md index 193851b2..d52fdea2 100644 --- a/packages/cultcache-rs/README.md +++ b/packages/cultcache-rs/README.md @@ -310,6 +310,7 @@ write. The single-file store stamps a replacing write (`push`, `push_all` and `u it replaces, and the cache holds the stamp the store reports; the redb stores and the exact-envelope family (`compare_and_swap_entry` and the other store-level conditional writes) write the caller's envelope byte for byte. A store needs `read` and `commit_if` to serve `update`; the redb stores do not have them. +The version is the SHA-256 of the record re-encoded as the store file writes it; that equals the hash of the record's bytes in the file (and the C# reference's vector) whenever the file's MessagePack encodings are minimal, as every runtime's writer produces. `put` persists before mutating the in-memory cache. If persistence fails, the cache does not pretend the write succeeded. -- 2.47.3 From 84d64d339a02dede27f5037dc1dfb126dac15c67 Mon Sep 17 00:00:00 2001 From: Libby Date: Fri, 13 Nov 2026 20:52:36 +0000 Subject: [PATCH 6/6] ci: rerun after the runner cap cancelled run 902 at the 30-minute mark Co-Authored-By: Claude Sonnet 5.5 -- 2.47.3