Bytes after the value: the payload check runs at load, for every record (Rust)

Single-file, redb and load_soa refuse a tail at load through check_record_payload,
indexed or not; depth beats a tail is pinned; the dead end guard is gone.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

packages/cultcache-rs/CHANGELOG.md

@@ -8,7 +8,7 @@ only in git history.
 
 ### Breaking
 
-- `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 loads, and its read refuses, as for a payload past the nesting bound. Behaviour change in a public API: the next release is a minor bump (0.x breaking).
+- `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. Behaviour change in a public API: the next release is a minor bump (0.x breaking).
 - 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.
 - A write over a file the reader refuses (a zero-byte file, a truncated store, a stamp this runtime does not read) refuses with `CultCacheStoreUnreadable` and leaves the file as it is. A whole-store write used to replace such a file. This reverses 0.3.1, where `push_all` replaced a file with no readable stamp (healing it); the stamp rule now has one owner, the reader every write asks under the lock, and it accepts v1 and v3 (0.3.1 refused v3).
 - `#[cultcache(index)]` means unique, as in C# `RefuseIndexSharing`: within one type and one alias no two records hold one non-blank value. A put, batch, push or load that would make two records hold a value is refused, naming both keys and the alias, and a store that already holds a duplicate does not open. A grouping index is no longer expressible; a consumer that relied on shared values must drop the `index` and look up by another way.

packages/cultcache-rs/src/lib.rs

@@ -607,7 +607,8 @@ pub fn check_payload_depth(bytes: &[u8]) -> std::result::Result<(), PayloadTooDe
 }
 
 /// Where the first MessagePack value of `bytes` ends, walking only that value (its tail is never
-/// read). `None` when the payload ends before the value does: the decoder's to refuse.
+/// read). `None` when the payload ends inside a header; a final scalar cut short ends past the
+/// payload's length, which is not bytes after the value either. Both are the decoder's to refuse.
 fn first_value_end(bytes: &[u8]) -> std::result::Result<Option<usize>, PayloadTooDeep> {
     let mut remaining = [0usize; MAX_PAYLOAD_DEPTH];
     let mut depth = 0usize;
@@ -636,7 +637,7 @@ fn first_value_end(bytes: &[u8]) -> std::result::Result<Option<usize>, PayloadTo
             depth -= 1;
         }
         if depth == 0 {
-            return Ok((position <= bytes.len()).then_some(position));
+            return Ok(Some(position));
         }
     }
     Ok(None)
@@ -2976,6 +2977,7 @@ impl CultCache {
         let schema_name_definitions = self.schema_name_definitions.clone();
         for (index, registration) in self.stores.iter().enumerate() {
             for mut entry in registration.store.pull_all()? {
+                require_record_payload(&entry)?;
                 let Some(canonical_type) = resolve_registered_type(
                     &known_types,
                     &schema_name_definitions,
@@ -3455,6 +3457,7 @@ impl CultCache {
                     entry.r#type
                 ));
             }
+            require_record_payload(&entry)?;
             self.home_index(&entry.r#type)?;
             self.refuse_claimed_id(&entry)?;
             self.refuse_retype(&loaded, &self.foreign, &entry)?;

packages/cultcache-rs/tests/union_parity.rs

@@ -6,6 +6,7 @@ use std::path::PathBuf;
 use cultcache_rs::CultCache;
 use cultcache_rs::CacheBackingStore;
 use cultcache_rs::CultCacheEnvelope;
+use cultcache_rs::CultCacheSoaTable;
 use cultcache_rs::MAX_PAYLOAD_DEPTH;
 use cultcache_rs::PayloadTooDeep;
 use cultcache_rs::check_payload_depth;
@@ -694,7 +695,7 @@ fn envelope(key: &str, payload: &[u8]) -> CultCacheEnvelope {
 }
 
 // A payload past the bound cannot be admitted through the cache, so a store file written behind its
-// back stands in for one a peer or an older runtime left.
+// back stands in for one a peer or an older runtime left; load or read refuses it.
 fn through_cache(payload: &[u8]) -> Result<UnionSurface, String> {
     let temp = tempfile::tempdir().unwrap();
     let path = temp.path().join("foreign.cc");
@@ -702,7 +703,10 @@ fn through_cache(payload: &[u8]) -> Result<UnionSurface, String> {
     let mut cache = CultCache::new();
     cache.register_entry_type::<UnionSurface>().unwrap();
     cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
-    cache.pull_all_backing_stores().unwrap();
+    // The payload check runs at load, so a payload past the bound is refused there, before any read.
+    if let Err(error) = cache.pull_all_backing_stores() {
+        return Err(format!("{error:#}"));
+    }
     match cache.get::<UnionSurface>("k") {
         Ok(value) => Ok(value.expect("the envelope was stored")),
         Err(error) => Err(format!("{error:#}")),
@@ -782,25 +786,30 @@ fn bytes_after_vectors() {
         let error = decode_payload::<UnionSurface>(&bytes).unwrap_err();
         assert!(error.downcast_ref::<PayloadTooDeep>().is_none(), "{file}");
         check("decode_payload", &format!("{error:#}"));
-        // A store written behind the cache's back loads (the bound is the read's, as for depth) and
-        // its read refuses, naming the key and the type.
+        // A store written behind the cache's back is refused at load, naming the type, the key and the count.
         let temp = tempfile::tempdir().unwrap();
         let path = temp.path().join("foreign.cc");
         SingleFileMessagePackBackingStore::new(&path).push(&envelope("k", &bytes)).unwrap();
         let mut single = CultCache::new();
         single.register_entry_type::<UnionSurface>().unwrap();
         single.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
-        single.pull_all_backing_stores().unwrap();
-        let message = format!("{:#}", single.get::<UnionSurface>("k").unwrap_err());
-        check("single-file read", &message);
+        let message = format!("{:#}", single.pull_all_backing_stores().unwrap_err());
+        check("single-file load", &message);
         assert!(message.contains("\"k\"") && message.contains(UnionSurface::TYPE), "{file}: {message}");
         let redb_path = temp.path().join("foreign.redb");
         RedbMessagePackBackingStore::new(&redb_path).unwrap().push(&envelope("k", &bytes)).unwrap();
         let mut redb = CultCache::new();
         redb.register_entry_type::<UnionSurface>().unwrap();
         redb.add_generic_backing_store(RedbMessagePackBackingStore::new(&redb_path).unwrap()).unwrap();
-        redb.pull_all_backing_stores().unwrap();
-        check("redb read", &format!("{:#}", redb.get::<UnionSurface>("k").unwrap_err()));
+        let message = format!("{:#}", redb.pull_all_backing_stores().unwrap_err());
+        check("redb load", &message);
+        assert!(message.contains("\"k\"") && message.contains(UnionSurface::TYPE), "{file}: {message}");
+        let mut soa = CultCache::new();
+        soa.register_entry_type::<UnionSurface>().unwrap();
+        let message = format!("{:#}", soa.load_soa(CultCacheSoaTable::from_envelopes(vec![envelope("k", &bytes)])).unwrap_err());
+        check("load_soa", &message);
+        assert!(message.contains("\"k\"") && message.contains(UnionSurface::TYPE), "{file}: {message}");
+        assert!(soa.snapshot().is_empty(), "{file}");
         // Raw admission refuses and stages nothing.
         let mut raw = CultCache::new();
         raw.register_entry_type::<UnionSurface>().unwrap();
@@ -810,6 +819,17 @@ fn bytes_after_vectors() {
     }
 }
 
+// Depth is found inside the first value, so a tail never hides it: a payload nested past the bound
+// with a tail after it is the depth refusal, not bytes after.
+#[test]
+fn depth_beats_a_tail() {
+    let mut deep = vec![0x91u8; MAX_PAYLOAD_DEPTH + 1];
+    deep.push(0xc0);
+    deep.push(0xc0);
+    let error = decode_payload::<UnionSurface>(&deep).unwrap_err();
+    assert_eq!(error.downcast_ref::<PayloadTooDeep>(), Some(&TOO_DEEP), "{error:#}");
+}
+
 // A payload that ends early is not bytes after the value: the decoder refuses it, not the check.
 #[test]
 fn a_truncated_payload_is_the_decoders_not_bytes_after() {