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"), }