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 <noreply@anthropic.com>

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.

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<digits>`.
 

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<CultCacheEnvelope> = 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::<Result<Vec<_>>>()?;
-            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::<Result<Vec<_>>>()?;
-                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::<Vec<_>>())?;
         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<CultCache> {
+            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::<Settings>("added")?.theme, "other", "{path:?}");
+            assert_eq!(reader.get_required::<Note>("mine")?.title, "t", "{path:?}");
+        }
+        Ok(())
+    }
 }