From dc3fca627b9bb792464b2658d40b99ecc4badfd7 Mon Sep 17 00:00:00 2001 From: Libby Date: Thu, 12 Nov 2026 08:04:13 +0000 Subject: [PATCH] Rust: batch writes merge onto the store (rs-batch-merges-onto-store) push_all applies the given entries onto the store as it is under its lock or transaction instead of replacing it, so a cache with a stale image cannot drop a record another writer added. put_prepared_batch sends only its staged records. Co-Authored-By: Claude Sonnet 5.5 --- packages/cultcache-rs/CHANGELOG.md | 1 + packages/cultcache-rs/README.md | 7 +-- packages/cultcache-rs/src/lib.rs | 88 +++++++++++++++++------------- 3 files changed, 55 insertions(+), 41 deletions(-) diff --git a/packages/cultcache-rs/CHANGELOG.md b/packages/cultcache-rs/CHANGELOG.md index c6d72492..b43c13c6 100644 --- a/packages/cultcache-rs/CHANGELOG.md +++ b/packages/cultcache-rs/CHANGELOG.md @@ -16,6 +16,7 @@ Semver: this is a breaking change in the 0.x line (0.3.x to 0.4.0). - A unit arm with slots (`[N, [extra...]]`) decodes as the unit arm and marks the record, where the derive refused it. - The registration map's value changes from the schema name to a private struct holding the schema name and the unread-slots check. - A cache whose last `pull_all_backing_stores` was refused (a tail, a payload past the nesting bound, a duplicate unique-index value, a retype conflict, a record from the wrong store) refuses every `put`, `put_envelope`, `put_raw_envelope`, `put_prepared_batch`, `update` and `delete` with `CultCacheLoadRefused` until a pull succeeds. Before, it stayed writable over an empty image, and the next batch write replaced the store with only the new records. +- `CacheBackingStore::push_all` writes the given entries onto the store as it is when the write lands, under the store's lock or transaction: a record under the same identity is replaced and no other record is removed. It used to replace everything the store held, so a cache whose image was stale (another writer had added a record since its pull) wiped that record with `put_prepared_batch`. `put_prepared_batch` now hands the store its staged records alone, as C# `CommitBatch` and Python `push_all` do; the lay-back of foreign records is gone because they are never rewritten. A custom `CacheBackingStore` must give `push_all` merge semantics. `push_all` reads the store, so through a dangling link it is refused like `push` and no longer replaces the link; a soft flush is unchanged. - A cache with a backing store also refuses every write and removal, with `CultCacheLoadRefused`, until it has pulled: a cache that never pulled, or that gained a store since its last pull, holds no image of its stores, and a batch from it replaced a populated store with only the new records. `is_writable` answers false meanwhile. A cache with no store needs no pull. Callers that attach a store and write (cultmesh nodes created with `pull_on_start: false`) must call `pull_all_backing_stores` first. - `decode_payload`, `put_raw_envelope` and prepared batches refuse a payload with bytes after its first MessagePack value (`payload has {n} bytes after its first MessagePack value`), by count; the tail is never walked or quoted. They used to read the first value and keep or drop the tail, so a record with a stray tail was served and rewritten without it. A store holding such a record (single-file, redb, `load_soa`) is refused at load, naming the type, the key and the count, and stays unreadable until the record is repaired; the same check now runs on every record at load, as the C# and Python runtimes do at open. The nesting bound is part of that check, so a stored payload nested past it is refused at load too, where it used to load and refuse on read. - A single-file store written as `cultcache.store.v3` is read and written: a store the C# runtime marked v3 (it holds a record with an element id: an owned list element with a `[CultElementId]` member) keeps its header here, and is carried through every rewrite with its ids; this runtime never drops the mark. A runtime from before this change refuses a v3 store by name. Rolling back to an older release after a store has been written as v3 is one-way: there is no carry-back, so restore from a copy taken before upgrading. diff --git a/packages/cultcache-rs/README.md b/packages/cultcache-rs/README.md index ed9701d5..da4185e2 100644 --- a/packages/cultcache-rs/README.md +++ b/packages/cultcache-rs/README.md @@ -384,10 +384,9 @@ an empty store. Any other failure to reach it, such as a symlink loop, a file where a parent directory should be, or a missing permission, is an I/O error. A store file that exists and holds zero bytes is `Undecodable`: no CultCache writer leaves one. A symbolic link whose target is gone is not a missing store: -reading it is an I/O error, so a read-then-write (`push`, `delete`, the exchanges) -fails before writing. Writes do not yet follow links: a write through a live link, -and a `push_all` through a dangling one, replace the link with a regular file and -leave its target alone. A refusal names no +reading it is an I/O error, so a read-then-write (`push`, `push_all`, `delete`, the exchanges) +fails before writing. Writes do not yet follow links: a write through a live link +replaces the link with a regular file and leaves its target alone. A refusal names no value read from the file; it names a record only by its key and schema id, and a header only when it has the shape `cultcache.store.v`. diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index 47a3dd18..9085f21d 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -1253,12 +1253,14 @@ pub trait CacheBackingStore: Send { fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()>; fn delete(&mut self, entry: &CultCacheEnvelope) -> Result<()>; - /// Replaces everything the store holds with `entries` in one atomic step: - /// on `Err` the store still holds exactly what it held before. A cache - /// batch (`put_prepared_batch`) is all-or-nothing only through this. There - /// is no default, because a replace assembled from `delete` and `push` - /// calls leaves the store emptied or partial when one of them fails; a - /// store without an atomic replace does not compile: + /// Writes `entries` onto the store as it is when the write lands, in one + /// atomic step: a record under the same identity is replaced and no other + /// record is removed or changed, so a writer whose image is stale cannot + /// drop a record it never saw. On `Err` the store still holds exactly what + /// 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: /// /// ```compile_fail,E0046 /// use cultcache_rs::{CacheBackingStore, CultCacheEnvelope}; @@ -1962,14 +1964,17 @@ impl CacheBackingStore for SingleFileMessagePackBackingStore { } // 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); a second schema id under a key reaches the commit guard. + // 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 entries: Vec = by_id.into_values().cloned().collect(); - entries.sort_by_key(entry_id); - self.write_all_unlocked(&entries) + let mut merged = self.read_all_unlocked()?; + merged.retain(|held| !by_id.contains_key(&(schema_identity(held).to_string(), held.key.clone()))); + merged.extend(by_id.into_values().cloned()); + merged.sort_by_key(entry_id); + self.write_all_unlocked(&merged) }) } } @@ -2757,13 +2762,6 @@ impl CacheBackingStore for OwnedRedbMessagePackBackingStore { let write = self.inner.database.begin_write()?; { let mut table = write.open_table(REDB_ENVELOPES)?; - let keys = table - .iter()? - .map(|row| Ok(row?.0.value().to_vec())) - .collect::>>()?; - for key in keys { - table.remove(key.as_slice())?; - } for entry in entries { insert_redb_entry(&mut table, entry)?; } @@ -2899,13 +2897,6 @@ impl CacheBackingStore for RedbMessagePackBackingStore { let write = database.begin_write()?; { let mut table = write.open_table(REDB_ENVELOPES)?; - let keys = table - .iter()? - .map(|row| Ok(row?.0.value().to_vec())) - .collect::>>()?; - for key in keys { - table.remove(key.as_slice())?; - } for entry in entries { insert_redb_entry(&mut table, entry)?; } @@ -3602,18 +3593,10 @@ impl CultCache { } let change = self.judge_unique(&staged.iter().collect::>())?; if let Some(store_index) = store_index { - // The home store receives its own records only: those it already holds in this cache plus the staged ones, and - // the records it carries that no registered type claims: push_all replaces the store's records, so a foreign - // record is laid back as the store read it. - let snapshot: Vec<_> = next - .values() - .filter(|entry| self.resolve_route_index(&entry.r#type) == Some(store_index)) - .chain(self.foreign.iter().filter(|(home, _)| *home == store_index).map(|(_, entry)| entry)) - .cloned() - .collect(); + // The home store is handed the staged records alone and applies them onto what it holds now. self.stores[store_index] .store - .push_all(&snapshot, PushAllOptions::default())?; + .push_all(&staged, PushAllOptions::default())?; } self.entries = next; change.apply(&mut self.index_holders); @@ -7980,15 +7963,17 @@ mod tests { #[cfg(unix)] #[test] - fn r3_decides_resolve_or_refuse_a_push_all_through_a_dangling_link_replaces_the_link() -> Result<()> { + fn a_push_all_through_a_dangling_link_is_refused_like_push_and_writes_nothing() -> Result<()> { let temp = tempfile::tempdir()?; let volume = temp.path().join("volume"); fs::create_dir(&volume)?; let link = temp.path().join("store.cc"); std::os::unix::fs::symlink(volume.join("store.cc"), &link)?; - SingleFileMessagePackBackingStore::new(&link).push_all(&[snapshot_envelope("a", b"1")], PushAllOptions::default())?; - assert!(fs::symlink_metadata(&link)?.file_type().is_file(), "the link was replaced by a regular file"); + let mut store = SingleFileMessagePackBackingStore::new(&link); + assert!(store.push_all(&[snapshot_envelope("a", b"1")], PushAllOptions::default()).is_err()); + assert!(store.push(&snapshot_envelope("a", b"1")).is_err()); + assert!(fs::symlink_metadata(&link)?.file_type().is_symlink(), "the link is left as it was"); assert_eq!(fs::read_dir(&volume)?.count(), 0, "nothing reached the link's volume"); Ok(()) } @@ -8191,4 +8176,33 @@ mod tests { routed.put("again", &settings)?; Ok(()) } + + #[test] + fn a_stale_cache_batch_keeps_the_records_another_writer_added() -> Result<()> { + let temp = tempfile::tempdir()?; + let single = temp.path().join("single.cc"); + let redb = temp.path().join("redb.db"); + let open = |path: &Path| -> Result { + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + if path.extension().is_some_and(|extension| extension == "db") { + cache.add_generic_backing_store(RedbMessagePackBackingStore::new(path)?)?; + } else { + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path))?; + } + cache.pull_all_backing_stores()?; + Ok(cache) + }; + for path in [&single, &redb] { + let mut stale = open(path)?; + let mut other = open(path)?; + other.put("added", &Settings { theme: "other".to_string(), retries: 2 })?; + let (staged, _) = stale.prepare_entry("mine", &Note { title: "t".to_string(), body: "b".to_string() })?; + stale.put_prepared_batch(vec![staged])?; + let reader = open(path)?; + assert_eq!(reader.get_required::("added")?.theme, "other", "{path:?}"); + assert_eq!(reader.get_required::("mine")?.title, "t", "{path:?}"); + } + Ok(()) + } }