.github/workflows/cultnet-interop.yml

@@ -29,6 +29,12 @@ jobs:
     steps:
       - name: Check out CultLib
         uses: actions/checkout@v4
+        with:
+          # packages/cultnet-py/tests/test_declared_dependency_ranges.py reads the cultcache-py
+          # release tags and their file contents with `git show`; a shallow checkout has no tags
+          # and the test skips itself.
+          fetch-depth: 0
+          fetch-tags: true
 
       - name: Set up .NET
         uses: actions/setup-dotnet@v4

packages/cultcache-rs/tests/union_parity.rs

@@ -619,6 +619,23 @@ fn depth_is_refused_at_load() {
     assert!(cache.snapshot().is_empty());
 }
 
+// load_soa is a load too: a table holding a payload nested past the bound is refused with the typed depth refusal, the
+// prior view is kept, and a tail behind the depth is still the depth refusal.
+#[test]
+fn load_soa_refuses_a_payload_nested_past_the_bound() {
+    let mut deep = vec![0x91u8; MAX_PAYLOAD_DEPTH + 1];
+    deep.push(0xc0);
+    let mut cache = CultCache::new();
+    cache.register_entry_type::<UnionSurface>().unwrap();
+    cache.load_soa(CultCacheSoaTable::from_envelopes(vec![envelope("kept", &vector("union-surface.msgpack"))])).unwrap();
+    let before = cache.snapshot();
+    for payload in [deep.clone(), [deep.clone(), vec![0xc0]].concat()] {
+        let error = cache.load_soa(CultCacheSoaTable::from_envelopes(vec![envelope("k", &payload)])).unwrap_err();
+        assert_eq!(error.downcast_ref::<PayloadTooDeep>(), Some(&TOO_DEEP), "{error:#}");
+        assert_eq!(cache.snapshot(), before, "a refused load keeps the prior view");
+    }
+}
+
 // A cache whose load was refused refuses every write and removal and leaves the store as it was: a
 // later write would otherwise persist the empty image over the records it never read.
 #[test]

packages/cultcache-rs/tests/unique_index.rs

@@ -6,7 +6,7 @@
 use std::path::{Path, PathBuf};
 
 use cultcache_rs::{
-    CultCache, CultCacheEnvelope, CultCacheSoaTable, DatabaseEntry, SingleFileMessagePackBackingStore,
+    CultCache, CultCacheEnvelope, CultCacheLoadRefused, CultCacheSoaTable, DatabaseEntry, SingleFileMessagePackBackingStore,
 };
 use serde::Deserialize;
 
@@ -419,6 +419,54 @@ fn a_load_of_a_duplicate_image_is_refused_and_keeps_the_prior_view() {
     soa.put("item:y", &unique("K")).unwrap();
 }
 
+// A pull refused for a duplicate unique-index value, or for one key held under two types in two stores, closes the same
+// write gate a payload refusal does: the cache holds no image of its stores, so a write would lay the empty image over them.
+#[test]
+fn a_pull_refused_for_a_duplicate_index_or_a_retype_closes_every_write() {
+    fn assert_gated(cache: &mut CultCache, path: &Path, door_label: &str) {
+        let before = std::fs::read(path).unwrap();
+        let errors = [
+            ("put", cache.put("item:new", &unique("N")).unwrap_err()),
+            ("put_raw_envelope", cache.put_raw_envelope(cache.prepare_entry("item:new", &unique("N")).unwrap().0).unwrap_err()),
+            ("delete", cache.delete::<Unique>("item:keep").unwrap_err()),
+        ];
+        for (door, error) in errors {
+            assert!(error.downcast_ref::<CultCacheLoadRefused>().is_some(), "{door_label} {door}: {error:#}");
+        }
+        assert_eq!(std::fs::read(path).unwrap(), before, "{door_label}: the store changed");
+    }
+
+    // Duplicate index value found in the stored image.
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("dup.cc");
+    let mut cache = cache_at(&path);
+    cache.put("item:keep", &unique("K")).unwrap();
+    let mut twin = twin_cache_at(&path);
+    twin.put("item:b", &Twin { sku: "S".into(), category: "c".into() }).unwrap();
+    twin.put("item:a", &Twin { sku: "S".into(), category: "c".into() }).unwrap();
+    assert!(error_text(cache.pull_all_backing_stores()).contains("both hold unique index 'Sku'"));
+    assert_gated(&mut cache, &path, "duplicate index");
+
+    // One key held under two types by two stores. The cache pulled cleanly first, so only the refused pull can close the gate.
+    let (first, second) = (dir.path().join("first.cc"), dir.path().join("second.cc"));
+    let mut cache = CultCache::new();
+    cache.register_entry_type::<Plain>().unwrap();
+    cache.register_entry_type::<Other>().unwrap();
+    cache.register_entry_type::<Unique>().unwrap();
+    cache.add_backing_store(SingleFileMessagePackBackingStore::new(&first), [Plain::TYPE]).unwrap();
+    cache.add_backing_store(SingleFileMessagePackBackingStore::new(&second), [Other::TYPE]).unwrap();
+    cache.pull_all_backing_stores().unwrap();
+    cache.put("k", &Plain { value: "p".into() }).unwrap();
+    let mut seed = CultCache::new();
+    seed.register_entry_type::<Other>().unwrap();
+    seed.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&second)).unwrap();
+    seed.pull_all_backing_stores().unwrap();
+    seed.put("k", &Other { sku: "o".into() }).unwrap();
+    assert!(error_text(cache.pull_all_backing_stores()).contains("\"k\""));
+    assert_gated(&mut cache, &first, "retype");
+    assert_eq!(cache.get::<Plain>("k").unwrap().unwrap().value, "p", "a refused pull keeps the prior view");
+}
+
 #[test]
 fn a_multi_index_store_with_a_duplicate_on_the_second_alias_refuses_to_open() {
     let dir = tempfile::tempdir().unwrap();

packages/cultnet-py/CHANGELOG.md

@@ -2,7 +2,7 @@
 
 ## Unreleased (breaking; version 0.x, so the next minor)
 
-- `decode_witness_artifact_bundle_payload` (and `CultNetWitnessArtifactBundle.from_payload`) read the payload through `cultcache_py.read_payload_tree`: bytes after its first MessagePack value raise `ValueError` `cultnet.witness_artifact_bundle payload has N bytes after its first MessagePack value (refused on read).` instead of `msgpack.ExtraData`, and a payload nested past 256 raises `CultCachePayloadDepthError`. Needs `cultcache-py` 0.4.0, the release that exports `read_payload_tree`: the dependency is now `cultcache-py>=0.4.0` (it was `>=0.2.0`, which admits releases whose `import cultnet_py` fails), and `tests/test_declared_dependency_ranges.py` fails when a declared bound admits a tagged release lacking a name the package imports. The same check found `cultmesh-py` importing `cache.index_text`, which `cultcache-py` 0.2.0 lacks; its bound is `cultcache-py>=0.4.0` too (it has no changelog of its own). A raw record with bytes after its first value gives no schema hint to `resolve_document_for_raw_record`, so an unknown schema id stays `KeyError` rather than being routed; a registered one is refused by count when applied.
+- `decode_witness_artifact_bundle_payload` (and `CultNetWitnessArtifactBundle.from_payload`) read the payload through `cultcache_py.read_payload_tree`: bytes after its first MessagePack value raise `ValueError` `cultnet.witness_artifact_bundle payload has N bytes after its first MessagePack value (refused on read).` instead of `msgpack.ExtraData`, and a payload nested past 256 raises `CultCachePayloadDepthError`. Needs `cultcache-py` 0.4.0, the release that exports `read_payload_tree`: the dependency is now `cultcache-py>=0.4.0` (it was `>=0.2.0`, which admits releases whose `import cultnet_py` fails), and `tests/test_declared_dependency_ranges.py` fails when a declared bound admits a tagged release lacking a name the package imports. The same check found `cultmesh-py` importing `cache.index_text`, which `cultcache-py` 0.2.0 lacks; its bound is `cultcache-py>=0.4.0` too (it has no changelog of its own). `resolve_document_for_raw_record` reads the schema hint from a raw record's first value whether or not bytes follow it, so a record routed by its stamp is refused by count when applied (`ValueError`, in a snapshot, a shard log and a direct apply) like one routed by a registered schema id; `apply_shard_log_response` used to skip it silently as an unknown schema. A tail on a payload whose first value names no registered schema is still `KeyError`, which a shard log skips.
 - `apply_raw_document_record` and `apply_raw_snapshot` raise `SchemaConflictError` when a remote record is of another type than the one the cache holds at its key (`cultcache-py` holds one record per key whatever its type); the cache is unchanged, and the caller decides whether to drop the record or the peer. `cultmesh-py`'s server answers such a put with `message_invalid`; its client's `read_next_change` lets the error out.
 - `CultNetDatabaseChange` carries `commit_id` and `commit_count` (`commitId`/`commitCount`; None outside a multi-record commit, as C# writes nil): `from_wire` refuses an empty or non-string id and a count that is not an integer of at least 2, `to_wire` writes them when set, and the published `database_change_raw` schema lists both. Python applies no batch. The `simulation_fact` document's `shard`, `frame`, `subject` and `claim_kind` aliases are `roles` (many facts share each); the catalog bytes do not move.
 - Decodes the write receipt frames of `contracts/cultnet/interop/write-receipt.md`: `parse_message` validates `cultnet.write_admitted.v0` and `cultnet.error.v1` against their closed schemas (unknown keys, empty or non-string ids, an unknown operation or code are refused; `error.v1` requires `messageId`, string or null; `limit_reached` is `error.v1` only). Added `CULTNET_ERROR_V1_CODES`, `CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS`, `write_content_hash` and `hello_advertises_write_receipts`; `hello` takes `supported_message_versions`. `tests/test_write_receipt.py` replays the vectors' `contentHash`, `wire` and `schemaInvalid` sections. Not implemented here: the client write tracker (`scenarios`) and the server rules (`dedupe`, `blankFields`); this package has no receipt server and no write client.

packages/cultnet-py/src/cultnet_py/replication.py

@@ -174,10 +174,11 @@ def _infer_schema_version_from_payload(payload: bytes) -> str | None:
     import msgpack  # type: ignore
     try:
         decoded = msgpack.unpackb(payload, raw=False)
-    except msgpack.ExtraData:
-        # A payload with bytes after its value gives no routing hint: the record is unroutable by inference, and a
-        # registered schema id routes it to a definition whose decode refuses the tail by count.
-        return None
+    except msgpack.ExtraData as extra:
+        # The hint is read from the first value; the tail is not this function's to judge. The routed definition refuses
+        # it by count when the record is applied, as it does for a record routed by a registered schema id. Treating a
+        # tail as "no hint" made a stamp-routed record look like an unknown schema, which a shard log skips silently.
+        decoded = extra.unpacked
     except Exception:
         return None
     if isinstance(decoded, list) and decoded and isinstance(decoded[0], str):

packages/cultnet-py/tests/test_cultnet.py

@@ -1971,7 +1971,7 @@ class CultNetTests(unittest.TestCase):
         apply_raw_document_record(cache, schema_document_map([first, second]), raw(first))  # the same type still replaces
         self.assertEqual(cache.get_required(first, "shared").name, "remote")
 
-    def test_a_raw_record_with_a_tail_is_unroutable_by_inference_and_refused_by_count_when_routed(self) -> None:
+    def test_a_raw_record_with_a_tail_is_routed_by_its_stamp_and_refused_by_count_on_every_apply_path(self) -> None:
         import msgpack  # type: ignore
         from cultcache_py import define_document_type
         from cultnet_py import resolve_document_for_raw_record
@@ -1986,18 +1986,33 @@ class CultNetTests(unittest.TestCase):
                     "payloadEncoding": "messagepack", "payload": payload}
 
         documents = schema_document_map([document])
-        # Routing by inference reads a hint, not a payload: a payload with a tail gives no hint, so an unknown schema id
-        # stays the unknown-schema error (KeyError) even when the first value names a registered schema.
-        self.assertIs(resolve_document_for_raw_record(documents, "unknown", raw("unknown", head)), document)
+        # The stamp is read from the first value whether or not a tail follows, so an unknown schema id whose payload names
+        # a registered schema routes to it; the tail is then refused by count where the record is applied.
+        self.assertIs(resolve_document_for_raw_record(documents, "unknown", raw("unknown", head + b"\xc0")), document)
+        for schema_id in (known, "unknown"):
+            cache = CultCache()
+            cache.register_document_type(document)
+            with self.assertRaises(ValueError, msg=schema_id) as caught:
+                apply_raw_document_record(cache, documents, raw(schema_id, head + b"\xc0"))
+            self.assertIn("1 bytes after its first MessagePack value", str(caught.exception))
+            self.assertIsNone(cache.get(document, "k"))
+            snapshot = {"schemaVersion": "cultnet.snapshot_response_raw.v0", "messageId": "s",
+                        "documents": [raw(schema_id, head + b"\xc0")]}
+            with self.assertRaises(ValueError, msg=schema_id):
+                apply_raw_snapshot(cache, [document], snapshot)
+            log = {"schemaVersion": "cultnet.shard_log_response.v0", "messageId": "l", "shardId": "s", "shardEpoch": 1,
+                   "resyncRequired": False,
+                   "entries": [{"sequence": 0, "changeKind": "updated",
+                                "put": {"schemaVersion": "cultnet.document_put_raw.v0", "messageId": "p",
+                                        "document": raw(schema_id, head + b"\xc0"), "shardId": "s", "shardEpoch": 1}}]}
+            with self.assertRaises(ValueError, msg=schema_id) as caught:
+                apply_shard_log_response(cache, [document], log)
+            self.assertIn("1 bytes after its first MessagePack value", str(caught.exception))
+            self.assertIsNone(cache.get(document, "k"))
+        # A tail on a payload that names no registered schema is still an unknown schema, which a shard log skips.
+        stranger = msgpack.packb(["tests.stranger.v1", "x"], use_bin_type=True) + b"\xc0"
         with self.assertRaises(KeyError):
-            resolve_document_for_raw_record(documents, "unknown", raw("unknown", head + b"\xc0"))
-        # A registered schema id routes, and the tail is refused by count when the record is applied.
-        cache = CultCache()
-        cache.register_document_type(document)
-        with self.assertRaises(ValueError) as caught:
-            apply_raw_document_record(cache, documents, raw(known, head + b"\xc0"))
-        self.assertIn("1 bytes after its first MessagePack value", str(caught.exception))
-        self.assertIsNone(cache.get(document, "k"))
+            resolve_document_for_raw_record(documents, "unknown", raw("unknown", stranger))
 
     def test_cultnet_replication_helpers_apply_raw_snapshot_and_shard_log(self) -> None:
         document = define_database_entry_type(

tests/GameCult.Networking.Tests/BytesAfterPathTests.cs

@@ -21,7 +21,15 @@ namespace GameCult.Networking.Tests
             [Key(1)] public int N = 3;
         }
 
-        private static readonly CultDocumentRegistry Documents = CultDocumentRegistry.ForTypes(new[] { typeof(TailDoc) });
+        [CultDocument("tests.bytes_after_deep", "tests.bytes_after_deep.v1")]
+        [MessagePackObject]
+        public sealed class DeepDoc
+        {
+            [Key(0)] [CultName] public string Id = "a";
+            [Key(1)] public DeepDoc Inner;
+        }
+
+        private static readonly CultDocumentRegistry Documents = CultDocumentRegistry.ForTypes(new[] { typeof(TailDoc), typeof(DeepDoc) });
 
         private static byte[] WithTail() =>
             CultDocumentMessagePackSerialization.SerializeUntyped(new TailDoc(), typeof(TailDoc), Documents).Concat(new byte[] { 0xc0 }).ToArray();
@@ -58,5 +66,28 @@ namespace GameCult.Networking.Tests
             var registry = new CultNetDocumentRegistry(Documents);
             Refuses(() => registry.DeserializeRawDocument(Raw(registry, WithTail())));
         }
+
+        // The write side of the same rule: a record with no binding is serialized by SerializeUntyped, and a payload nested past
+        // the limit is refused there, before it can reach the wire.
+        [Test]
+        public void ToRawRecordWithNoBindingRefusesAPayloadNestedPastTheLimit()
+        {
+            var registry = new CultNetDocumentRegistry(Documents);
+            var descriptor = Documents.GetRequired(typeof(DeepDoc));
+            DeepDoc Chain(int links)
+            {
+                var head = new DeepDoc();
+                var tip = head;
+                for (var i = 0; i < links; i++) tip = tip.Inner = new DeepDoc();
+                return head;
+            }
+
+            // Each link is one array; the root is depth 1, so 255 links is depth 256 (at the limit) and 256 links is 257.
+            Assert.That(registry.ToRawRecord(descriptor, new CultRecordKey("a"), Chain(255), "1970-01-01T00:00:00.000Z").Payload, Is.Not.Empty);
+            var refused = Assert.Throws<MessagePackSerializationException>(
+                () => registry.ToRawRecord(descriptor, new CultRecordKey("a"), Chain(256), "1970-01-01T00:00:00.000Z"))!;
+            Assert.That(refused.Message, Does.Contain("refused on write"));
+            Assert.That(refused.Message, Does.Contain(typeof(DeepDoc).FullName));
+        }
     }
 }