GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

cultcache-rs: the payload nesting bound at put_raw_envelope and put_prepared_batch

Browse source 
Both raw write doors run the one check_payload_depth that decode_payload runs first, before any
store is touched; decode_payload is public so envelope readers elsewhere decode through the same
helper. Tests: union arms, maps and arrays at 256 and 257 through both doors, a batch that stages a
valid entry beside a deep one persists neither, the read vectors enter by a store file written
behind the cache.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-11 17:52:24 +02:00
commit a5c0f876de
2 changed files with 126 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

18
packages/cultcache-rs/src/lib.rs
View file

@ -597,11 +597,23 @@ fn element_header(bytes: &[u8], position: usize) -> Option<(usize, Option<usize>
/// The one decode of a document payload: the nesting bound first, then the deserializer, whose own
/// depth counter (1024) is only a backstop. A failure keeps its kind and never quotes the bytes.
fn decode_payload<T: serde::de::DeserializeOwned>(payload: &[u8]) -> Result<T> {
/// Readers of stored envelope payloads outside this crate decode through it too.
pub fn decode_payload<T: serde::de::DeserializeOwned>(payload: &[u8]) -> Result<T> {
check_payload_depth(payload)?;
rmp_serde::from_slice(payload).map_err(opaque_decode_error)
}
/// The nesting bound for an envelope admitted without a typed decode, from the same walk that
/// [`decode_payload`] runs first. Names the entry's type and key, never the payload.
fn require_payload_depth(entry: &CultCacheEnvelope) -> Result<()> {
check_payload_depth(&entry.payload).with_context(|| {
format!(
"refused CultCache envelope {:?} at key {:?}",
entry.r#type, entry.key
)
})
}
/// A decoder failure as an error that keeps only its kind. Every decode of stored bytes
/// goes through this or [`describe_undecodable`], so no error text quotes a stored value.
fn opaque_decode_error(error: rmp_serde::decode::Error) -> anyhow::Error {
@ -2717,6 +2729,7 @@ impl CultCache {
entry.r#type
));
}
require_payload_depth(entry)?;
let Some(home) = self.home_index(&entry.r#type)? else {
continue;
};
@ -2761,7 +2774,7 @@ impl CultCache {
/// claiming compile-time knowledge of its payload schema. This is the
/// CultMesh/schema-registry boundary: callers own payload validation before
/// admission; CultCache still enforces registered type, identity, timestamp,
/// routing, and persistence.
/// the payload nesting bound ([`MAX_PAYLOAD_DEPTH`]), routing, and persistence.
pub fn put_raw_envelope(&mut self, entry: CultCacheEnvelope) -> Result<()> {
if !self.definitions.contains_key(&entry.r#type) {
return Err(anyhow!(
@ -2781,6 +2794,7 @@ impl CultCache {
entry.r#type
));
}
require_payload_depth(&entry)?;
if let Some(home) = self.home_index(&entry.r#type)? {
self.stores[home].store.push(&entry)?;
}

119
packages/cultcache-rs/tests/union_parity.rs
View file

@ -4,6 +4,7 @@
use std::path::PathBuf;
use cultcache_rs::CultCache;
use cultcache_rs::CacheBackingStore;
use cultcache_rs::CultCacheEnvelope;
use cultcache_rs::MAX_PAYLOAD_DEPTH;
use cultcache_rs::PayloadTooDeep;
@ -568,18 +569,26 @@ mod generated_names {
// cache because the nesting rule lives in its one decode.
const CANARY: &str = "CANARY-union-golden-4c2a";
fn envelope(key: &str, payload: &[u8]) -> CultCacheEnvelope {
CultCacheEnvelope {
key: key.into(),
r#type: UnionSurface::TYPE.into(),
payload: payload.to_vec(),
stored_at: "2026-01-01T00:00:00Z".into(),
schema_id: None,
}
}
// 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.
fn through_cache(payload: &[u8]) -> Result<UnionSurface, String> {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("foreign.cc");
SingleFileMessagePackBackingStore::new(&path).push(&envelope("k", payload)).unwrap();
let mut cache = CultCache::new();
cache.register_entry_type::<UnionSurface>().unwrap();
cache
.put_raw_envelope(CultCacheEnvelope {
key: "k".into(),
r#type: UnionSurface::TYPE.into(),
payload: payload.to_vec(),
stored_at: "2026-01-01T00:00:00Z".into(),
schema_id: None,
})
.unwrap();
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
cache.pull_all_backing_stores().unwrap();
match cache.get::<UnionSurface>("k") {
Ok(value) => Ok(value.expect("the envelope was stored")),
Err(error) => Err(format!("{error:#}")),
@ -683,6 +692,98 @@ fn write_refused_past_limit() {
assert_eq!(std::fs::read(&path).unwrap(), file_before);
}
// Every door an envelope enters the store by holds the one bound: the typed put above, and the raw
// doors a peer's bytes arrive through. Three nestings: union arms and maps
// written by the production encoder, and the arrays of the committed vector.
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
struct MapNest {
inner: Option<std::boxed::Box<MapNest>>,
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(type = "tests.map_doc", schema = "tests.map_doc.v1")]
struct MapDoc {
#[cultcache(key = 0)]
root: Option<MapNest>,
}
// The document array is depth 1 and each MapNest is one map.
fn map_doc(maps: usize) -> MapDoc {
let mut nest = MapNest { inner: None };
for _ in 1..maps {
nest = MapNest { inner: Some(std::boxed::Box::new(nest)) };
}
MapDoc { root: Some(nest) }
}
#[test]
fn maps_count_one_level_each_on_every_write_door() {
let at_limit = map_doc(MAX_PAYLOAD_DEPTH - 1);
let past_limit = map_doc(MAX_PAYLOAD_DEPTH);
let at_bytes = rmp_serde::to_vec_named(&at_limit).unwrap();
let past_bytes = rmp_serde::to_vec_named(&past_limit).unwrap();
assert_eq!(check_payload_depth(&at_bytes), Ok(()));
assert_eq!(check_payload_depth(&past_bytes), Err(TOO_DEEP));
let mut cache = CultCache::new();
cache.register_entry_type::<MapDoc>().unwrap();
let (prepared, parsed) = cache.prepare_entry_named("at", &at_limit).unwrap();
assert_eq!((prepared.payload.as_slice(), &parsed), (at_bytes.as_slice(), &at_limit));
cache.put_prepared_batch(vec![prepared]).unwrap();
assert_eq!(cache.get::<MapDoc>("at").unwrap(), Some(at_limit));
let error = cache.prepare_entry_named("past", &past_limit).unwrap_err();
assert_eq!(error.downcast_ref::<PayloadTooDeep>(), Some(&TOO_DEEP));
}
fn raw(type_name: &str, key: &str, payload: Vec<u8>) -> CultCacheEnvelope {
CultCacheEnvelope { r#type: type_name.into(), ..envelope(key, &payload) }
}
#[test]
fn the_raw_doors_hold_the_bound_for_arms_maps_and_arrays() {
let arms_at = rmp_serde::to_vec(&wrap_chain(126, boxed("leaf", vec![]))).unwrap();
let arms_past = rmp_serde::to_vec(&wrap_chain(127, text("leaf", CANARY))).unwrap();
let maps_at = rmp_serde::to_vec_named(&map_doc(MAX_PAYLOAD_DEPTH - 1)).unwrap();
let maps_past = rmp_serde::to_vec_named(&map_doc(MAX_PAYLOAD_DEPTH)).unwrap();
let arrays_at = vector("depth-at-limit.msgpack");
let arrays_past = vector("depth-past-limit.msgpack");
let cases = [
(UnionSurface::TYPE, "arms", arms_at, arms_past),
(MapDoc::TYPE, "maps", maps_at, maps_past),
(UnionSurface::TYPE, "arrays", arrays_at, arrays_past),
(UnionSurface::TYPE, "mixed", Vec::new(), mixed_past),
];
for (type_name, name, at, past) in cases {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("doors.cc");
let mut cache = CultCache::new();
cache.register_entry_type::<UnionSurface>().unwrap();
cache.register_entry_type::<MapDoc>().unwrap();
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
if !at.is_empty() {
cache.put_raw_envelope(raw(type_name, "at-raw", at.clone())).unwrap();
cache.put_prepared_batch(vec![raw(type_name, "at-batch", at.clone())]).unwrap();
}
let stored = cache.snapshot_envelopes();
let file_before = std::fs::read(&path).unwrap();
let refusals = [
cache.put_raw_envelope(raw(type_name, "past-raw", past.clone())).unwrap_err(),
// The valid entry staged beside it must not land either.
cache
.put_prepared_batch(vec![raw(type_name, "ok-staged", at.clone()), raw(type_name, "past-batch", past.clone())])
.unwrap_err(),
];
for error in refusals {
assert_eq!(error.downcast_ref::<PayloadTooDeep>(), Some(&TOO_DEEP), "{name}");
let message = format!("{error:#}");
assert!(message.contains(type_name) && !message.contains(CANARY), "{name}: {message}");
}
assert_eq!(cache.snapshot_envelopes(), stored, "{name}");
assert_eq!(std::fs::read(&path).unwrap(), file_before, "{name}");
}
}
// A debug build on a 2 MiB thread carries 405 containers of this document; the bound keeps every
// accepted payload below that, and every refusal is an error, not an abort.
#[test]