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

Rust: a record with unread slots is held read-only (slot-readonly-rs) #127

Merged
libby merged 6 commits from hands/slot-readonly-rs into main 2026-11-12 04:44:04 +00:00
Conversation 0 Commits 6 Files changed 9 +1063 -242
9 changed files with 1063 additions and 242 deletions
Download patch file Download diff file Expand all files Collapse all files

8
packages/cultcache-rs/CHANGELOG.md
View file

@ -8,6 +8,13 @@ only in git history.
### Breaking
Semver: this is a breaking change in the 0.x line (0.3.x to 0.4.0).
- A record whose payload carries unread slots is read-only. A payload with a slot past its type's declared slots, at any depth (record, nested record, list element, union arm, unit arm), still decodes, and every write or removal of its key is now refused before any store is touched: `put`, `put_envelope`, `put_raw_envelope`, `put_prepared_batch` (the whole batch) and `delete`, so `update` too. The refusal is `SchemaConflictError` with the new `unread_slots`; its message names the type and both slot counts and never a slot value. Holders of such a record used to overwrite it and drop the extra slots silently. The payload nesting refusal still runs first. A held payload this build cannot read at all (an arm it lacks from a newer writer, or a corrupt record) is read-only too: every write or removal is refused with an error naming the type and key (a newer writer's record is read with that build; a corrupt one is restored from a backup or removed at the store itself, as no cache call removes it), and `is_writable` answers false. `load_soa` is a write of every held key it omits or changes, so it refuses to drop or replace a read-only record. `is_writable` also answers false for a key a store carries as a foreign record and for a type other than the one holding the key, as the writes do. Each write over a held key decodes the held payload once to ask (about 40% of a `put` of a 200,000-element record, measured in release); keys nothing holds pay nothing.
- cultmesh-rs `pull_rudp_catalog_snapshot` applies documents one by one with `put_raw_envelope`, so a snapshot document for a key the node holds read-only ends the pull at that document and the documents before it stay applied. C# applies a raw snapshot as one cache commit, so there a refusal applies nothing: this is a divergence, owed by follow-up `mesh-snapshot-atomic` (stage the pull through `put_prepared_batch`).
- `SchemaConflictError` gains the public field `unread_slots: Option<UnreadSlots>`; code that builds it with a struct literal must set it (`None`).
- 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 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.
@ -22,6 +29,7 @@ only in git history.
### Added
- `UnreadSlots { type_name, slot_count, declared_count, occurrences }`, `decode_payload_marked` (the decode that returns the mark) and `CultCache::is_writable(type, key)`. The derives' slot visitors note into a thread-local scope the decode entry opens (`__enter_slots`, `__note_unread_slots`, `__UnitArmSlots`, doc-hidden); a decode outside that entry notes nothing.
- `CultCache::foreign_records()` and `ForeignRecord { key, schema_id, schema_name, stored_at }`: the records the last pull found that no registered type claims, sorted by key.
- Lookups, matching the C# `GetByName`, `GetByIndex` and `GetGlobal`. `#[cultcache(name)]` and `#[cultcache(index)]` / `#[cultcache(index = "alias")]` on `DatabaseEntry` fields and `#[cultcache(global)]` on the struct give `CultCache::get_by_name`, `get_by_index` and `get_global`. An empty name and a blank index value are not found; records of one type sharing a name find the one with the smallest key (an index value is unique, below). A second record of a global type under another key is refused on every write path and on load with `CultCacheSecondGlobal`. Two `name` members, or two members with one index alias, are refused at compile time. A member's text comes from `CultLookupValue` (`String`, `Option` of one, integers, `bool`, `char`).
- `SchemaConflictError`, the typed refusal of a write the catalog cannot describe: records of different types under one schema id, or an arrived schema and a registered one that share an id and disagree on the schema name. A whole-store write (`push_all`) refuses with it, names the schema id, the names involved and a record key, writes nothing and leaves the file as it is. A caller tells it apart with `error.downcast_ref::<SchemaConflictError>()`.

11
packages/cultcache-rs/README.md
View file

@ -236,6 +236,17 @@ assert_eq!(rmp_serde::from_slice::<Node>(&bytes)?, tree);
# Ok::<(), Box<dyn std::error::Error>>(())
```
## Records written by a newer build
A payload that carries a slot past the slots its type declares, in a record, a nested record, a list element or a union
arm (a unit arm declares none), still decodes: the extra slots are skipped. The cache then holds that record read-only,
because writing it would drop what the newer writer stored. `put`, `put_envelope`, `put_raw_envelope`,
`put_prepared_batch` and `delete` of its key are refused with `SchemaConflictError`, whose `unread_slots` names the type
and the slot counts and never a value; `CultCache::is_writable(type, key)` answers the same check. An untouched
read-only record keeps its bytes through every other write. A short payload (missing trailing slots) is not marked, and
a payload nested past `MAX_PAYLOAD_DEPTH` is refused before anything is read. `decode_payload_marked` is the decode that
returns the `UnreadSlots` mark. This runtime has no compatible schema ids, so no mark is consented.
## Structure Of Arrays
The C# CultCache hot path exposes CPU-local SoA scans as:

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

@ -381,6 +381,7 @@ fn slot_codec(ident: &syn::Ident, slots: &[CultCacheSlot]) -> proc_macro2::Token
where
__A: ::serde::de::SeqAccess<'de>,
{
let __ticket = ::cultcache_rs::__enter_slots();
#(#option_declarations)*
let mut index = 0usize;
while index <= #max_slot {
@ -390,7 +391,15 @@ fn slot_codec(ident: &syn::Ident, slots: &[CultCacheSlot]) -> proc_macro2::Token
}
index += 1;
}
while let ::std::option::Option::Some(_) = seq.next_element::<::serde::de::IgnoredAny>()? {}
// The slots past the declared ones are skipped and counted: the decode entry marks the record
// that carried them read-only.
let mut __extra = 0usize;
while seq.next_element::<::serde::de::IgnoredAny>()?.is_some() {
__extra += 1;
}
if __extra > 0 {
::cultcache_rs::__note_unread_slots(__ticket, stringify!(#ident), #max_slot + 1 + __extra, #max_slot + 1);
}
Ok(#ident {
#(#build_fields),*
})
@ -567,10 +576,13 @@ fn union_codec(input: DeriveInput) -> syn::Result<proc_macro2::TokenStream> {
quote! {
#key => {
match seq
.next_element::<::std::option::Option<[(); 0]>>()
.next_element::<::std::option::Option<::cultcache_rs::__UnitArmSlots>>()
.map_err(|_| ::serde::de::Error::custom(#arity_message))?
{
::std::option::Option::Some(::std::option::Option::Some(_)) => #ident::#variant,
::std::option::Option::Some(::std::option::Option::Some(slots)) => {
slots.note(stringify!(#variant));
#ident::#variant
}
::std::option::Option::Some(::std::option::Option::None) => {
return ::std::result::Result::Err(::serde::de::Error::custom(
format!(#nil_slots_message, #key),

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

@ -800,6 +800,154 @@ pub fn decode_payload<T: serde::de::DeserializeOwned>(payload: &[u8]) -> Result<
decode_grown(payload).map_err(opaque_decode_error)
}
/// What a payload decode found past the slots this build declares, at any depth: the first type read with extra slots,
/// how many slots it carried against how many it declares, and how many objects in the payload did. It holds no slot
/// value, so a refusal can name it. A record whose payload carries it is read-only in the cache
/// ([`CultCache::is_writable`]); this runtime has no compatible schema ids, so every mark is unconsented.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct UnreadSlots {
/// The first type read with extra slots (a unit arm is named by its variant).
pub type_name: String,
/// The slots that object carried.
pub slot_count: usize,
/// The slots this build declares for it.
pub declared_count: usize,
/// The objects in the payload that carried extra slots.
pub occurrences: usize,
}
/// The open mark scope of a decode: how many slot objects it has entered, and the mark of the first entered one that
/// carried extra slots (the reference names the first type read, which is the outermost: it is entered before its members).
#[derive(Default)]
struct MarkScope {
entered: usize,
mark: Option<(usize, UnreadSlots)>,
}
thread_local! {
/// The open mark scope of a decode on this thread: `None` when no decode entry has one open, so a decode through
/// serde alone notes nothing. The slot visitors the derives write note into it.
static UNREAD_SLOTS: std::cell::RefCell<Option<MarkScope>> = const { std::cell::RefCell::new(None) };
}
/// The derives' sensor, first half: a slot object is entered. Its ticket orders it among the objects of the decode.
#[doc(hidden)]
pub fn __enter_slots() -> usize {
UNREAD_SLOTS.with(|scope| {
scope.borrow_mut().as_mut().map_or(0, |open| {
open.entered += 1;
open.entered
})
})
}
/// The derives' sensor, second half: the object entered as `ticket`, of `type_name`, carried `carried` slots where this
/// build declares `declared`.
#[doc(hidden)]
pub fn __note_unread_slots(ticket: usize, type_name: &str, carried: usize, declared: usize) {
UNREAD_SLOTS.with(|scope| {
let mut scope = scope.borrow_mut();
let Some(open) = scope.as_mut() else {
return;
};
let occurrences = open.mark.as_ref().map_or(0, |(_, found)| found.occurrences) + 1;
if open.mark.as_ref().is_none_or(|(first, _)| ticket < *first) {
let named = UnreadSlots { type_name: type_name.to_string(), slot_count: carried, declared_count: declared, occurrences };
open.mark = Some((ticket, named));
} else if let Some((_, found)) = open.mark.as_mut() {
found.occurrences = occurrences;
}
});
}
/// A unit arm is a zero-slot record: its slots array is read and counted, not refused.
#[doc(hidden)]
pub struct __UnitArmSlots {
ticket: usize,
extra: usize,
}
impl __UnitArmSlots {
/// Notes the arm if its array held any element.
pub fn note(self, variant: &str) {
if self.extra > 0 {
__note_unread_slots(self.ticket, variant, self.extra, 0);
}
}
}
impl<'de> serde::Deserialize<'de> for __UnitArmSlots {
fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
struct Count;
impl<'de> serde::de::Visitor<'de> for Count {
type Value = __UnitArmSlots;
fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
formatter.write_str("a slots array")
}
fn visit_seq<A: serde::de::SeqAccess<'de>>(self, mut seq: A) -> std::result::Result<Self::Value, A::Error> {
let ticket = __enter_slots();
let mut extra = 0usize;
while seq.next_element::<serde::de::IgnoredAny>()?.is_some() {
extra += 1;
}
Ok(__UnitArmSlots { ticket, extra })
}
}
deserializer.deserialize_seq(Count)
}
}
/// The scope one marked decode opens on this thread; closed when it ends, whatever the decode did. A marked decode is
/// never re-entered (only the cache's own checks open one), so scopes do not nest.
struct UnreadScope;
impl UnreadScope {
fn open() -> Self {
UNREAD_SLOTS.with(|scope| *scope.borrow_mut() = Some(MarkScope::default()));
Self
}
fn mark(&self) -> Option<UnreadSlots> {
UNREAD_SLOTS.with(|scope| scope.borrow_mut().as_mut().and_then(|open| open.mark.take()).map(|(_, mark)| mark))
}
}
impl Drop for UnreadScope {
fn drop(&mut self) {
UNREAD_SLOTS.with(|scope| *scope.borrow_mut() = None);
}
}
/// [`decode_payload`] with the mark: the same depth check and grown decode, inside a scope the derives' slot
/// visitors note into. A payload past the nesting limit is refused before the decode, never marked.
pub fn decode_payload_marked<T: serde::de::DeserializeOwned>(payload: &[u8]) -> Result<(T, Option<UnreadSlots>)> {
check_payload_depth(payload)?;
let scope = UnreadScope::open();
let value = decode_grown(payload).map_err(opaque_decode_error)?;
Ok((value, scope.mark()))
}
/// The registered check of a held payload: monomorphized by [`CultCache::register_entry_type`] so the untyped write
/// entries can decode a payload they cannot name.
/// A payload the type cannot decode at all (an arm this build lacks, a corrupt record) is an error here, and the held
/// record is read-only: the cache cannot tell what a writer of another build stored in it.
fn unread_slots_of<T: serde::de::DeserializeOwned>(payload: &[u8]) -> Result<Option<UnreadSlots>> {
decode_payload_marked::<T>(payload).map(|(_, mark)| mark)
}
/// What the cache finds in a held payload when asked to write or remove its key.
enum HeldFinding {
Free,
Marked(UnreadSlots),
Unreadable(anyhow::Error),
}
/// What a registered type is to the cache: its schema name and how to read the slots a held payload carries.
struct Definition {
schema_name: &'static str,
unread_slots: fn(&[u8]) -> Result<Option<UnreadSlots>>,
}
/// 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<()> {
@ -938,10 +1086,26 @@ pub struct SchemaConflictError {
pub schema_id: String,
pub schema_names: Vec<String>,
pub record_key: String,
/// Set when the refusal is a write or removal of a record held read-only for the slots its payload carries past
/// this build's; `None` for every other refusal.
pub unread_slots: Option<UnreadSlots>,
}
impl std::fmt::Display for SchemaConflictError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
if let Some(unread) = &self.unread_slots {
return write!(
formatter,
"{} record {} is read-only: {} was read with {} slots where this build declares {} ({} object(s) in the record). \
Writing it would drop data a newer writer stored; read it with types that declare those slots, or write it from that build.",
self.schema_names.first().map_or("", String::as_str),
self.record_key,
unread.type_name,
unread.slot_count,
unread.declared_count,
unread.occurrences
);
}
write!(
formatter,
"schema id {:?} cannot describe record {:?}: {:?}",
@ -2738,7 +2902,7 @@ struct CultCacheStoreRegistration {
}
pub struct CultCache {
definitions: BTreeMap<String, &'static str>,
definitions: BTreeMap<String, Definition>,
/// The registered types declared `#[cultcache(global)]`.
globals: BTreeSet<String>,
schema_name_definitions: BTreeMap<String, String>,
@ -2825,8 +2989,8 @@ impl CultCache {
"CultCache entry types must declare a non-empty type"
));
}
if let Some(existing_schema) = self.definitions.get(T::TYPE)
&& *existing_schema != T::SCHEMA_NAME
if let Some(existing) = self.definitions.get(T::TYPE)
&& existing.schema_name != T::SCHEMA_NAME
{
return Err(anyhow!(
"CultCache already has a different definition registered for type {:?}",
@ -2875,7 +3039,7 @@ impl CultCache {
}
}
}
self.definitions.insert(T::TYPE.to_string(), T::SCHEMA_NAME);
self.definitions.insert(T::TYPE.to_string(), Definition { schema_name: T::SCHEMA_NAME, unread_slots: unread_slots_of::<T> });
self.schema_name_definitions
.insert(T::SCHEMA_NAME.to_string(), T::TYPE.to_string());
Ok(())
@ -3036,6 +3200,60 @@ impl CultCache {
Ok(Some(payload))
}
/// Whether a write or removal of `(type_id, key)` would be admitted: false when the held payload carries slots past
/// its registered type's declared slots, at any depth, or cannot be read by this build at all (the record is
/// read-only), when a store carries a foreign record at the key, or when another registered type holds it. True for a
/// key nothing holds. An unregistered type is an error. Advisory: the write entries ask the same checks themselves.
pub fn is_writable(&self, type_id: &str, key: &str) -> Result<bool> {
if !matches!(self.held_finding(type_id, key)?, HeldFinding::Free) {
return Ok(false);
}
let probe = CultCacheEnvelope {
key: key.to_string(),
r#type: type_id.to_string(),
payload: Vec::new(),
stored_at: String::new(),
schema_id: None,
};
Ok(self.home_index(type_id).is_ok() && self.refuse_retype(&self.entries, &self.foreign, &probe).is_ok())
}
/// The one decode behind every write entry and [`CultCache::is_writable`]: the held envelope at `(type_id, key)`
/// read under the registered type's mark.
fn held_finding(&self, type_id: &str, key: &str) -> Result<HeldFinding> {
let Some(definition) = self.definitions.get(type_id) else {
bail!("No entry type registered for CultCache envelope type {type_id:?}");
};
let Some(held) = self.entries.get(&entry_id_parts(type_id, key)) else {
return Ok(HeldFinding::Free);
};
Ok(match (definition.unread_slots)(&held.payload) {
Ok(None) => HeldFinding::Free,
Ok(Some(unread)) => HeldFinding::Marked(unread),
Err(error) => HeldFinding::Unreadable(error),
})
}
/// Refuses a write or removal of a record held read-only, before any store is touched: for the slots its payload
/// carries past this build's, or because this build cannot read the payload at all.
fn refuse_unread_slots(&self, type_id: &str, key: &str) -> Result<()> {
match self.held_finding(type_id, key)? {
HeldFinding::Free => Ok(()),
HeldFinding::Unreadable(error) => Err(error.context(format!(
"CultCache entry {type_id:?} at key {key:?} is held read-only: this build cannot read its payload. A newer writer's record is read with that build; a corrupt one is restored from a backup or removed at the store itself, not through the cache"
))),
HeldFinding::Marked(unread) => {
let held = &self.entries[&entry_id_parts(type_id, key)];
Err(anyhow::Error::new(SchemaConflictError {
schema_id: schema_identity(held).to_string(),
schema_names: vec![self.definitions[type_id].schema_name.to_string()],
record_key: key.to_string(),
unread_slots: Some(unread),
}))
}
}
}
pub fn get_required<T: DatabaseEntry>(&self, key: &str) -> Result<T> {
self.get::<T>(key)?
.ok_or_else(|| anyhow!("CultCache has no {:?} entry at key {:?}", T::TYPE, key))
@ -3166,6 +3384,7 @@ impl CultCache {
pub fn put<T: DatabaseEntry>(&mut self, key: impl Into<String>, value: &T) -> Result<T> {
let (entry, parsed) = self.prepare_entry(key, value)?;
self.refuse_unread_slots(&entry.r#type, &entry.key)?;
self.refuse_second_global(&self.entries, &entry.r#type, &entry.key)?;
self.refuse_retype(&self.entries, &self.foreign, &entry)?;
let change = self.judge_unique(&[&entry])?;
@ -3266,6 +3485,7 @@ impl CultCache {
));
}
require_payload_depth(entry)?;
self.refuse_unread_slots(&entry.r#type, &entry.key)?;
self.refuse_claimed_id(entry)?;
let Some(home) = self.home_index(&entry.r#type)? else {
continue;
@ -3308,6 +3528,7 @@ impl CultCache {
pub fn put_envelope<T: DatabaseEntry>(&mut self, entry: CultCacheEnvelope) -> Result<T> {
let parsed = self.validate_envelope::<T>(&entry)?;
self.refuse_unread_slots(&entry.r#type, &entry.key)?;
self.refuse_claimed_id(&entry)?;
self.refuse_second_global(&self.entries, &entry.r#type, &entry.key)?;
self.refuse_retype(&self.entries, &self.foreign, &entry)?;
@ -3345,6 +3566,7 @@ impl CultCache {
));
}
require_payload_depth(&entry)?;
self.refuse_unread_slots(&entry.r#type, &entry.key)?;
self.refuse_claimed_id(&entry)?;
self.refuse_second_global(&self.entries, &entry.r#type, &entry.key)?;
self.refuse_retype(&self.entries, &self.foreign, &entry)?;
@ -3400,6 +3622,7 @@ impl CultCache {
pub fn delete<T: DatabaseEntry>(&mut self, key: &str) -> Result<bool> {
self.require_entry_type::<T>()?;
self.refuse_unread_slots(T::TYPE, key)?;
let id = entry_id_parts(T::TYPE, key);
let Some(entry) = self.entries.get(&id).cloned() else {
if let Some((_, foreign)) = self.foreign.iter().find(|(_, foreign)| foreign.key == key) {
@ -3427,7 +3650,9 @@ impl CultCache {
/// Replaces the cache's view with the table's records. Every record is
/// checked (registered type, and a home store once any store is attached)
/// into a local image first, so a refused load admits nothing and leaves
/// the prior view intact. Loading writes no store.
/// the prior view intact. Loading writes no store, but a held record that is read-only (unread slots, or a payload
/// this build cannot read) may not be omitted or replaced by the table: the next batch would lay the image over the
/// store.
pub fn load_soa(&mut self, table: CultCacheSoaTable) -> Result<()> {
let mut loaded = BTreeMap::new();
for entry in table.into_envelopes()? {
@ -3442,6 +3667,12 @@ impl CultCache {
self.refuse_retype(&loaded, &self.foreign, &entry)?;
loaded.insert(entry_id(&entry), entry);
}
// A load is a write of every held key it omits or changes: the one check refuses those held read-only.
for ((type_id, key), held) in &self.entries {
if loaded.get(&(type_id.clone(), key.clone())).is_none_or(|incoming| incoming.payload != held.payload) {
self.refuse_unread_slots(type_id, key)?;
}
}
self.refuse_second_globals(&loaded)?;
let holders = self.judge_image(&loaded)?;
self.entries = loaded;
@ -3518,7 +3749,7 @@ impl CultCache {
}
for (slot, keys) in &ordered {
if let [first, second, ..] = keys.as_slice() {
let schema = self.definitions.get(&slot.0).copied().unwrap_or(&slot.0);
let schema = self.definitions.get(&slot.0).map_or(slot.0.as_str(), |definition| definition.schema_name);
let violation = [slot.1.as_str(), first.as_str(), second.as_str(), schema];
if least.is_none_or(|best| violation_cmp(&violation, &best).is_lt()) {
least = Some(violation);
@ -3553,7 +3784,7 @@ impl CultCache {
.find(|held_key| held_key.as_str() != key);
match held_other {
Some(existing_key) => Err(CultCacheSecondGlobal {
schema_name: self.definitions.get(type_id).copied().unwrap_or(type_id).to_string(),
schema_name: self.definitions.get(type_id).map_or(type_id, |definition| definition.schema_name).to_string(),
existing_key: existing_key.clone(),
key: key.to_string(),
}
@ -3597,6 +3828,7 @@ impl CultCache {
schema_id: schema_identity(entry).to_string(),
schema_names: vec![resolved, entry.r#type.clone()],
record_key: entry.key.clone(),
unread_slots: None,
})),
_ => Ok(()),
}
@ -3611,7 +3843,7 @@ impl CultCache {
fn require_entry_type<T: DatabaseEntry>(&self) -> Result<()> {
match self.definitions.get(T::TYPE) {
Some(schema_name) if *schema_name == T::SCHEMA_NAME => Ok(()),
Some(definition) if definition.schema_name == T::SCHEMA_NAME => Ok(()),
_ => Err(anyhow!(
"CultCache entry type {:?} is not registered on this cache instance",
T::TYPE
@ -3693,6 +3925,7 @@ fn encode_store_snapshot(entries: &[CultCacheEnvelope], disk: &DiskStore) -> Res
schema_id,
schema_names: vec![existing_type, entry.r#type.clone()],
record_key: entry.key.clone(),
unread_slots: None,
}));
}
}
@ -3710,6 +3943,7 @@ fn encode_store_snapshot(entries: &[CultCacheEnvelope], disk: &DiskStore) -> Res
schema_id,
schema_names: vec![arrived_type, document_type],
record_key: key,
unread_slots: None,
}));
}
}
@ -3846,6 +4080,7 @@ fn foreign_conflict(carried: &CultCacheEnvelope, written_type: &str) -> anyhow::
schema_id: schema_identity(carried).to_string(),
schema_names: vec![carried.r#type.clone(), written_type.to_string()],
record_key: carried.key.clone(),
unread_slots: None,
})
}
@ -3856,6 +4091,7 @@ fn retype_conflict(held: &CultCacheEnvelope, written: &CultCacheEnvelope) -> any
schema_id: schema_identity(written).to_string(),
schema_names: vec![held.r#type.clone(), written.r#type.clone()],
record_key: written.key.clone(),
unread_slots: None,
})
}

235
packages/cultcache-rs/tests/common/mod.rs Normal file
View file

@ -0,0 +1,235 @@
//! The union-parity vectors' types and constructed value, shared by the test files that walk them.
#![allow(dead_code)]
use std::path::PathBuf;
use cultcache_rs::CultSlots;
use cultcache_rs::CultUnion;
use cultcache_rs::DatabaseEntry;
#[derive(Clone, Debug, PartialEq, CultUnion)]
pub enum Node {
#[cultcache(key = 0)]
Box(BoxArm),
#[cultcache(key = 1)]
Text(TextArm),
#[cultcache(key = 2)]
Use(UseArm),
#[cultcache(key = 5)]
Image(ImageArm),
#[cultcache(key = 6)]
Wrap(WrapArm),
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
pub struct WrapArm {
#[cultcache(key = 0)]
pub inner: Option<std::boxed::Box<Node>>,
#[cultcache(key = 1)]
pub id: String,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
pub struct BoxArm {
#[cultcache(key = 0)]
pub id: String,
#[cultcache(key = 1)]
pub children: Vec<Node>,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
pub struct TextArm {
#[cultcache(key = 0)]
pub id: String,
#[cultcache(key = 1)]
pub content: String,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
pub struct UseArm {
#[cultcache(key = 0)]
pub id: String,
#[cultcache(key = 1)]
pub composition: Composition,
#[cultcache(key = 2)]
pub slots: Vec<Option<Node>>,
}
// Slot 2 is retired: a gap written nil.
#[derive(Clone, Debug, PartialEq, CultSlots)]
pub struct ImageArm {
#[cultcache(key = 0)]
pub id: String,
#[cultcache(key = 1)]
pub uri: String,
#[cultcache(key = 3)]
pub alt: String,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
pub struct Composition {
#[cultcache(key = 0)]
pub name: String,
#[cultcache(key = 1)]
pub body: Option<std::boxed::Box<Node>>,
}
#[derive(Clone, Debug, PartialEq, CultUnion)]
pub enum Outcome {
#[cultcache(key = 0)]
Pending(PendingArm),
#[cultcache(key = 1)]
Completed(CompletedArm),
#[cultcache(key = 2)]
Denied(IdArm),
#[cultcache(key = -2147483648)]
Floor(IdArm),
#[cultcache(key = 2147483647)]
Ceiling(IdArm),
}
// The codec's unit arm (no members, written [key, []]), kept apart from the golden: a document list's arms carry ids.
#[derive(Clone, Debug, PartialEq, CultUnion)]
pub enum UnitOutcome {
#[cultcache(key = 2)]
Denied,
#[cultcache(key = -2147483648)]
Floor,
#[cultcache(key = 2147483647)]
Ceiling,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
pub struct PendingArm {
#[cultcache(key = 0)]
pub message: String,
#[cultcache(key = 1)]
pub id: String,
}
// An arm with no members of its own: just its element id.
#[derive(Clone, Debug, PartialEq, CultSlots)]
pub struct IdArm {
#[cultcache(key = 0)]
pub id: String,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
pub struct CompletedArm {
#[cultcache(key = 0)]
pub source_version: i32,
#[cultcache(key = 1)]
pub id: String,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
pub struct NamedArm {
#[cultcache(key = 0)]
pub name: String,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
pub struct UsesArm {
#[cultcache(key = 0)]
pub name: String,
#[cultcache(key = 1)]
pub uses: i32,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
pub struct CountArm {
#[cultcache(key = 0)]
pub name: String,
#[cultcache(key = 1)]
pub count: i32,
}
#[derive(Clone, Debug, PartialEq, CultUnion)]
pub enum Item {
#[cultcache(key = 0)]
Gem(NamedArm),
#[cultcache(key = 1)]
Tool(UsesArm),
#[cultcache(key = 2)]
Part(CountArm),
}
// The same arms as Item under its own keys: key 2 is unknown to it.
#[derive(Clone, Debug, PartialEq, CultUnion)]
pub enum CraftedItem {
#[cultcache(key = 0)]
Tool(UsesArm),
#[cultcache(key = 1)]
Part(CountArm),
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(type = "tests.union_surface", schema = "tests.union_surface.v1")]
pub struct UnionSurface {
#[cultcache(key = 0)]
pub id: String,
#[cultcache(key = 1)]
pub version: i32,
#[cultcache(key = 2)]
pub root: Option<Node>,
#[cultcache(key = 3)]
pub outcomes: Vec<Outcome>,
#[cultcache(key = 4)]
pub part: Option<Item>,
#[cultcache(key = 5)]
pub crafted: Option<CraftedItem>,
#[cultcache(key = 6)]
pub missing: Option<Node>,
}
pub fn text(id: &str, content: &str) -> Node {
Node::Text(TextArm { id: id.into(), content: content.into() })
}
pub fn image(id: &str, uri: &str, alt: &str) -> Node {
Node::Image(ImageArm { id: id.into(), uri: uri.into(), alt: alt.into() })
}
pub fn boxed(id: &str, children: Vec<Node>) -> Node {
Node::Box(BoxArm { id: id.into(), children })
}
pub fn constructed() -> UnionSurface {
let inner = boxed("0000000000d0", vec![text("0000000000d1", "x"), image("0000000000d2", "uri://a", "alt")]);
UnionSurface {
id: "surface".into(),
version: 3,
root: Some(boxed(
"0000000000c0",
vec![
text("0000000000c1", "hello"),
Node::Use(UseArm {
id: "0000000000c2".into(),
composition: Composition { name: "comp".into(), body: Some(std::boxed::Box::new(inner)) },
slots: vec![Some(text("0000000000e1", "slot")), None, Some(boxed("0000000000e2", vec![]))],
}),
image("0000000000c3", "uri://b", ""),
],
)),
outcomes: vec![
Outcome::Pending(PendingArm { message: "wait".into(), id: "0000000000a1".into() }),
Outcome::Completed(CompletedArm { source_version: 7, id: "0000000000a2".into() }),
Outcome::Denied(id_arm("0000000000a3")),
],
part: Some(Item::Tool(UsesArm { name: "hammer".into(), uses: 12 })),
crafted: Some(CraftedItem::Part(CountArm { name: "gear".into(), count: 4 })),
missing: None,
}
}
pub fn id_arm(id: &str) -> IdArm {
IdArm { id: id.into() }
}
pub fn vector(name: &str) -> Vec<u8> {
let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../../tests/vectors/union-parity")
.join(name);
std::fs::read(&path).unwrap_or_else(|error| panic!("{}: {error}", path.display()))
}

493
packages/cultcache-rs/tests/slot_readonly.rs Normal file
View file

@ -0,0 +1,493 @@
//! A record whose payload carries unread slots at any depth is read-only: the C# reference's union-parity vectors
//! (README "Extra trailing slot" and "Slot stores") walked through the cache. This runtime has no compatible schema
//! ids, so every mark is unconsented and no `explained` outcome can arise.
mod common;
use std::path::Path;
use common::*;
use cultcache_rs::CultCache;
use cultcache_rs::CultCacheEnvelope;
use cultcache_rs::DatabaseEntry;
use cultcache_rs::SchemaConflictError;
use cultcache_rs::SingleFileMessagePackBackingStore;
use cultcache_rs::UnreadSlots;
use cultcache_rs::decode_payload_marked;
/// A row of `slot-outcomes.msgpack`: store, key, writable, explained, type, slots, declared, occurrences, flattenRefusedBy.
type SlotRow = (String, String, bool, Option<bool>, Option<String>, Option<usize>, Option<usize>, Option<usize>, Option<String>);
const SLOT_VALUE: &str = "extra-slot";
/// `tests.slot_unit`, a document `[Name, Outcome: UnitOutcome]`.
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(type = "tests.slot_unit", schema = "tests.slot_unit.v1")]
struct SlotUnit {
#[cultcache(key = 0)]
name: String,
#[cultcache(key = 1)]
outcome: UnitOutcome,
}
/// Two declared slots, the second defaulted: a payload of one slot is short, not marked.
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(type = "tests.shortened", schema = "tests.shortened.v1")]
struct Shortened {
#[cultcache(key = 0)]
id: String,
#[cultcache(key = 1, default)]
note: String,
}
fn registered() -> CultCache {
let mut cache = CultCache::new();
cache.register_entry_type::<UnionSurface>().unwrap();
cache.register_entry_type::<SlotUnit>().unwrap();
cache
}
/// Any MessagePack value a store holds, enough to drop one record from a store and write the rest back.
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(untagged)]
enum Packed {
Nil(()),
Bool(bool),
Int(i64),
Text(String),
Bin(#[serde(with = "serde_bytes")] Vec<u8>),
Array(Vec<Packed>),
}
/// The committed store without its `marked-variant` record. A variant is a five-slot record, which this runtime's
/// store reader refuses whole (`the_variant_row_...`), so the other rows are walked from a copy without it.
fn without_variant_row(store: &[u8]) -> Vec<u8> {
let Packed::Array(mut parts) = rmp_serde::from_slice(store).unwrap() else { panic!("a store is an array") };
let Packed::Array(records) = parts.last_mut().unwrap() else { panic!("a store ends in its records") };
let before = records.len();
records.retain(|record| !matches!(record, Packed::Array(fields) if matches!(fields.first(), Some(Packed::Text(key)) if key == "marked-variant")));
assert_eq!(records.len(), before - 1);
rmp_serde::to_vec(&parts).unwrap()
}
/// A copy of a committed store, opened with both types registered; the vector itself is never written. The slot store
/// is read without its variant row.
fn open_copy(file: &str, dir: &Path) -> (CultCache, std::path::PathBuf) {
let path = dir.join(file);
let bytes = if file == "extra-slot-store.cc" { without_variant_row(&vector(file)) } else { vector(file) };
std::fs::write(&path, bytes).unwrap();
let mut cache = registered();
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
cache.pull_all_backing_stores().unwrap();
(cache, path)
}
fn envelope(key: &str, payload: Vec<u8>) -> CultCacheEnvelope {
CultCacheEnvelope {
key: key.into(),
r#type: UnionSurface::TYPE.into(),
payload,
stored_at: "2026-01-01T00:00:00Z".into(),
schema_id: None,
}
}
fn conflict(error: &anyhow::Error) -> &SchemaConflictError {
error.downcast_ref::<SchemaConflictError>().unwrap_or_else(|| panic!("not a schema conflict: {error:#}"))
}
fn mark(type_name: &str, slot_count: usize, declared_count: usize, occurrences: usize) -> UnreadSlots {
UnreadSlots { type_name: type_name.into(), slot_count, declared_count, occurrences }
}
/// The marks the README table pins, by `detail`. The reference names an arm by its class; here an arm is the
/// `...Arm` struct the union wraps.
fn expected_mark(detail: &str) -> UnreadSlots {
match detail {
"record" | "depth" => mark("UnionSurface", 8, 7, 1),
"nested-record" => mark("Composition", 3, 2, 1),
"arm" => mark("BoxArm", 3, 2, 1),
"list-element" => mark("TextArm", 3, 2, 1),
other => panic!("unknown detail {other}"),
}
}
fn assert_refused_with(error: &anyhow::Error, key: &str, expected: &UnreadSlots) {
let refusal = conflict(error);
assert_eq!(refusal.unread_slots.as_ref(), Some(expected));
assert_eq!(refusal.record_key, key);
assert_eq!(refusal.schema_names.len(), 1);
let message = error.to_string();
assert!(message.contains(&expected.type_name), "{message}");
assert!(message.contains(&expected.slot_count.to_string()) && message.contains(&expected.declared_count.to_string()), "{message}");
assert!(!message.contains(SLOT_VALUE), "a refusal echoed the slot value: {message}");
}
#[test]
fn every_read_only_vector_decodes_marked_and_is_not_writable() {
let rows: Vec<(String, String, String, Option<i64>, String)> = rmp_serde::from_slice(&vector("outcomes.msgpack")).unwrap();
assert_eq!(rows.len(), 9);
let bounds: Vec<serde_json::Value> = rmp_serde::from_slice(&vector("bounds-outcomes.msgpack")).unwrap();
let mut cases: Vec<(String, String)> = rows
.into_iter()
.filter(|(_, outcome, ..)| outcome == "read-only")
.map(|(file, _, _, _, detail)| (file, detail))
.collect();
for entry in &bounds[1..] {
let fields: Vec<&str> = entry.as_array().unwrap().iter().filter_map(|field| field.as_str()).collect();
if fields[1] == "read-only" {
cases.push((fields[0].to_string(), fields[3].to_string()));
}
}
assert_eq!(cases.len(), 5);
for (file, detail) in cases {
let payload = vector(&file);
let (value, found) = decode_payload_marked::<UnionSurface>(&payload).unwrap_or_else(|error| panic!("{file}: {error:#}"));
assert_eq!(value, constructed(), "{file}");
assert_eq!(found, Some(expected_mark(&detail)), "{file}");
let mut cache = registered();
let dir = tempfile::tempdir().unwrap();
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(dir.path().join("s.cc"))).unwrap();
cache.put_raw_envelope(envelope("k", payload.clone())).unwrap();
assert_eq!(cache.get::<UnionSurface>("k").unwrap(), Some(constructed()), "{file}");
assert!(!cache.is_writable(UnionSurface::TYPE, "k").unwrap(), "{file}");
assert!(cache.is_writable(UnionSurface::TYPE, "unheld").unwrap(), "{file}");
let error = cache.put("k", &constructed()).unwrap_err();
assert_refused_with(&error, "k", &expected_mark(&detail));
}
}
#[test]
fn clean_and_short_payloads_are_writable_and_unmarked() {
let (value, found) = decode_payload_marked::<UnionSurface>(&vector("union-surface.msgpack")).unwrap();
assert_eq!((value, found), (constructed(), None));
// A payload short of its declared slots keeps its defaults and does not mark.
let (value, found) = decode_payload_marked::<Shortened>(&rmp_serde::to_vec(&("only",)).unwrap()).unwrap();
assert_eq!((value.id.as_str(), value.note.as_str(), found), ("only", "", None));
}
#[test]
fn a_payload_past_the_nesting_limit_is_refused_never_marked() {
for file in ["depth-skipped-past-limit.msgpack", "depth-past-limit.msgpack"] {
let error = decode_payload_marked::<UnionSurface>(&vector(file)).unwrap_err();
assert!(error.to_string().contains("the limit is 256"), "{file}: {error:#}");
}
}
#[test]
fn a_plain_serde_decode_marks_nothing_and_does_not_leak_a_scope() {
// No scope is open: the visitors note into nothing.
let value: UnionSurface = rmp_serde::from_slice(&vector("extra-slot-record.msgpack")).unwrap();
assert_eq!(value, constructed());
// A marked decode followed by a clean one: the first mark does not survive into the second.
assert!(decode_payload_marked::<UnionSurface>(&vector("extra-slot-record.msgpack")).unwrap().1.is_some());
assert_eq!(decode_payload_marked::<UnionSurface>(&vector("union-surface.msgpack")).unwrap().1, None);
}
#[test]
fn the_slot_store_rows_walk_through_the_cache() {
let rows: Vec<SlotRow> =
rmp_serde::from_slice(&vector("slot-outcomes.msgpack")).unwrap();
assert_eq!(rows.len(), 8);
let dir = tempfile::tempdir().unwrap();
let mut walked = 0;
for (store, key, writable, _explained, type_name, slots, declared, occurrences, flatten_refused_by) in rows {
// The variant row is a record this runtime cannot hold: see the variant test. The consent store's `writable` and
// `explained` answer the compatible-id declaration, which this runtime has none of; its marks are walked below.
if flatten_refused_by.is_some() || store == "extra-slot-consent.cc" {
continue;
}
let case = dir.path().join(&key);
std::fs::create_dir_all(&case).unwrap();
let (mut cache, _) = open_copy(&store, &case);
let type_id = if key == "unit-arm" { SlotUnit::TYPE } else { UnionSurface::TYPE };
assert_eq!(cache.is_writable(type_id, &key).unwrap(), writable, "{store} {key}");
let removal = if key == "unit-arm" { cache.delete::<SlotUnit>(&key) } else { cache.delete::<UnionSurface>(&key) };
match writable {
true => assert!(removal.unwrap()),
false => {
let found = conflict(&removal.unwrap_err()).unread_slots.clone().unwrap();
// The reference names the unit arm's class, Denied2; here the variant.
let expected_type = match type_name.as_deref().unwrap() {
"Denied2" => "Denied",
other => other,
};
assert_eq!((found.type_name.as_str(), Some(found.slot_count), Some(found.declared_count), Some(found.occurrences)), (expected_type, slots, declared, occurrences));
assert!(cache.snapshot_envelopes().iter().any(|held| held.key == key), "a refused removal removed {key}");
}
}
walked += 1;
}
assert_eq!(walked, 3);
}
#[test]
fn the_consent_store_marks_are_unconsented_here() {
let rows: Vec<SlotRow> =
rmp_serde::from_slice(&vector("slot-outcomes.msgpack")).unwrap();
let dir = tempfile::tempdir().unwrap();
let mut found = 0;
for (store, key, _writable, _explained, _type_name, slots, declared, occurrences, _) in rows {
if store != "extra-slot-consent.cc" {
continue;
}
let (cache, _) = open_copy(&store, dir.path());
let held = cache.snapshot_envelopes().into_iter().find(|held| held.key == key).unwrap();
let (_, mark) = decode_payload_marked::<UnionSurface>(&held.payload).unwrap();
match slots {
// The reference's `consented` row is writable there (a listed compatible id covers it); here every mark refuses.
None => assert_eq!(mark, None, "{key}"),
Some(slots) => {
let mark = mark.unwrap_or_else(|| panic!("{key} carried no mark"));
// The first type read is the document, whose reference name is SurfaceDeclaring.
assert_eq!(mark.type_name, "UnionSurface", "{key}");
assert_eq!((Some(mark.slot_count), Some(mark.declared_count), Some(mark.occurrences)), (Some(slots), declared, occurrences), "{key}");
assert!(!cache.is_writable(UnionSurface::TYPE, &key).unwrap(), "{key}");
}
}
found += 1;
}
assert_eq!(found, 4);
}
#[test]
fn every_write_entry_refuses_a_marked_key_before_any_store_is_touched() {
let dir = tempfile::tempdir().unwrap();
let (mut cache, path) = open_copy("extra-slot-store.cc", dir.path());
let expected = mark("Composition", 3, 2, 1);
let before = std::fs::read(&path).unwrap();
let held = cache.snapshot_envelopes().into_iter().find(|held| held.key == "marked").unwrap();
let clean = cache.snapshot_envelopes().into_iter().find(|held| held.key == "clean").unwrap();
assert!(!cache.is_writable(UnionSurface::TYPE, "marked").unwrap());
assert!(cache.is_writable(UnionSurface::TYPE, "clean").unwrap());
let replacing = envelope("marked", rmp_serde::to_vec(&constructed()).unwrap());
assert_refused_with(&cache.put("marked", &constructed()).unwrap_err(), "marked", &expected);
assert_refused_with(&cache.put_envelope::<UnionSurface>(replacing.clone()).unwrap_err(), "marked", &expected);
assert_refused_with(&cache.put_raw_envelope(replacing.clone()).unwrap_err(), "marked", &expected);
assert_refused_with(&cache.put_prepared_batch(vec![replacing]).unwrap_err(), "marked", &expected);
assert_refused_with(&cache.delete::<UnionSurface>("marked").unwrap_err(), "marked", &expected);
assert_refused_with(
&cache.update::<UnionSurface, _>("marked", |current| current.unwrap()).unwrap_err(),
"marked",
&expected,
);
// A batch with a marked key refuses whole: its clean member is not written either.
let mut rewritten = clean.clone();
rewritten.payload = rmp_serde::to_vec(&UnionSurface { version: 9, ..constructed() }).unwrap();
let batch = vec![rewritten, envelope("marked", rmp_serde::to_vec(&constructed()).unwrap())];
assert_refused_with(&cache.put_prepared_batch(batch).unwrap_err(), "marked", &expected);
assert_eq!(std::fs::read(&path).unwrap(), before, "a refused write touched the store");
assert_eq!(cache.snapshot_envelopes().into_iter().find(|entry| entry.key == "marked").unwrap(), held);
assert_eq!(cache.snapshot_envelopes().into_iter().find(|entry| entry.key == "clean").unwrap(), clean);
// A write to another key succeeds, and the marked record's bytes, schema id and stored-at are carried unchanged.
cache.put("clean", &UnionSurface { version: 9, ..constructed() }).unwrap();
cache.put("other", &constructed()).unwrap();
assert!(cache.delete::<UnionSurface>("other").unwrap());
let mut reopened = registered();
reopened.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
reopened.pull_all_backing_stores().unwrap();
let carried = reopened.snapshot_envelopes().into_iter().find(|entry| entry.key == "marked").unwrap();
assert_eq!(carried, held);
assert_eq!(carried.payload, vector("extra-slot-nested-record.msgpack"));
assert_eq!(reopened.get::<UnionSurface>("clean").unwrap().unwrap().version, 9);
}
#[test]
fn a_unit_arm_with_slots_decodes_as_the_unit_arm_and_marks() {
let (document, found) = decode_payload_marked::<SlotUnit>(&unit_arm_payload()).unwrap();
assert_eq!(document, SlotUnit { name: "unit-arm".into(), outcome: UnitOutcome::Denied });
assert_eq!(found, Some(mark("Denied", 1, 0, 1)));
let dir = tempfile::tempdir().unwrap();
let (mut cache, _) = open_copy("extra-slot-store.cc", dir.path());
assert_eq!(cache.get::<SlotUnit>("unit-arm").unwrap(), Some(document.clone()));
assert!(!cache.is_writable(SlotUnit::TYPE, "unit-arm").unwrap());
let error = cache.put("unit-arm", &document).unwrap_err();
assert_refused_with(&error, "unit-arm", &mark("Denied", 1, 0, 1));
assert_refused_with(&cache.delete::<SlotUnit>("unit-arm").unwrap_err(), "unit-arm", &mark("Denied", 1, 0, 1));
// Several elements are still one marked object, and a clean unit arm is unmarked.
let (_, found) = decode_payload_marked::<UnitOutcome>(&rmp_serde::to_vec(&(2u64, ("a", "b", "c"))).unwrap()).unwrap();
assert_eq!(found, Some(mark("Denied", 3, 0, 1)));
let (_, found) = decode_payload_marked::<UnitOutcome>(&vector("unit-arm-denied.msgpack")).unwrap();
assert_eq!(found, None);
}
fn unit_arm_payload() -> Vec<u8> {
let dir = tempfile::tempdir().unwrap();
let (cache, _) = open_copy("extra-slot-store.cc", dir.path());
cache.snapshot_envelopes().into_iter().find(|held| held.key == "unit-arm").unwrap().payload
}
#[test]
fn the_variant_row_of_the_slot_store_is_refused_whole_by_the_store_reader() {
// `marked-variant` is a five-slot record (a variant): this runtime holds no variants, and its v1 snapshot reader
// refuses such a record, so the store as written cannot load here and the other rows are walked without it.
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("extra-slot-store.cc");
std::fs::write(&path, vector("extra-slot-store.cc")).unwrap();
let mut cache = registered();
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
let error = cache.pull_all_backing_stores().unwrap_err();
assert!(format!("{error:#}").contains("is not decodable"), "{error:#}");
}
fn slot_unit_envelope(key: &str, payload: Vec<u8>) -> CultCacheEnvelope {
CultCacheEnvelope { r#type: SlotUnit::TYPE.into(), ..envelope(key, payload) }
}
#[test]
fn a_held_payload_the_type_cannot_read_is_read_only_and_refused_by_name() {
let mut cache = registered();
// A newer writer's record: an arm this build lacks and an extra root slot. get cannot read it, and it is the data
// read-only exists to protect.
let newer = rmp_serde::to_vec(&("CANARY-name", (99u64, ("x",)), "CANARY-extra")).unwrap();
cache.put_raw_envelope(slot_unit_envelope("n", newer)).unwrap();
assert!(cache.get::<SlotUnit>("n").is_err());
assert!(!cache.is_writable(SlotUnit::TYPE, "n").unwrap());
let replacement = SlotUnit { name: "n".into(), outcome: UnitOutcome::Denied };
let held = cache.snapshot_envelopes();
for error in [
cache.put("n", &replacement).unwrap_err(),
cache.put_raw_envelope(slot_unit_envelope("n", rmp_serde::to_vec(&replacement).unwrap())).unwrap_err(),
cache.delete::<SlotUnit>("n").unwrap_err(),
] {
let message = format!("{error:#}");
assert!(message.contains("\"n\"") && message.contains("tests.slot_unit") && message.contains("read-only"), "{message}");
assert!(!message.contains("CANARY"), "{message}");
}
assert_eq!(cache.snapshot_envelopes(), held);
}
#[test]
fn a_corrupt_held_payload_is_refused_naming_the_key_never_its_bytes() {
let mut cache = registered();
cache.put_raw_envelope(envelope("broken", rmp_serde::to_vec(&("CANARY-id", "CANARY-version")).unwrap())).unwrap();
assert!(!cache.is_writable(UnionSurface::TYPE, "broken").unwrap());
let error = cache.delete::<UnionSurface>("broken").unwrap_err();
let message = format!("{error:#}");
assert!(message.contains("\"broken\"") && message.contains("backup") && message.contains("store"), "{message}");
assert!(!message.contains("build that wrote it"), "{message}");
assert!(!message.contains("CANARY"), "{message}");
assert!(cache.is_writable("tests.unregistered", "broken").is_err());
}
fn held_marked_and_clean() -> (CultCache, tempfile::TempDir, std::path::PathBuf) {
let dir = tempfile::tempdir().unwrap();
let (cache, path) = open_copy("extra-slot-store.cc", dir.path());
(cache, dir, path)
}
#[test]
fn load_soa_does_not_replace_or_drop_a_marked_record() {
use cultcache_rs::CultCacheSoaTable;
let (mut cache, _dir, path) = held_marked_and_clean();
let all = cache.snapshot_envelopes();
let held = all.iter().find(|held| held.key == "marked").unwrap().clone();
let without: Vec<_> = all.iter().filter(|held| held.key != "marked").cloned().collect();
let mut replaced = all.clone();
replaced.iter_mut().find(|held| held.key == "marked").unwrap().payload = rmp_serde::to_vec(&constructed()).unwrap();
for table in [without, replaced] {
let error = cache.load_soa(CultCacheSoaTable::from_envelopes(table)).unwrap_err();
assert_refused_with(&error, "marked", &mark("Composition", 3, 2, 1));
assert_eq!(cache.snapshot_envelopes(), all, "a refused load changed the image");
}
// The same record carried unchanged, with another key added, loads.
let mut kept = all.clone();
kept.push(envelope("added", rmp_serde::to_vec(&constructed()).unwrap()));
cache.load_soa(CultCacheSoaTable::from_envelopes(kept)).unwrap();
assert_eq!(cache.snapshot_envelopes().iter().find(|held| held.key == "marked"), Some(&held));
// A clean record may be dropped by a load.
let no_clean: Vec<_> = cache.snapshot_envelopes().into_iter().filter(|held| held.key != "clean").collect();
cache.load_soa(CultCacheSoaTable::from_envelopes(no_clean)).unwrap();
// The marked record is still on disk after the next batch.
let rewrite = envelope("added", rmp_serde::to_vec(&UnionSurface { version: 5, ..constructed() }).unwrap());
cache.put_prepared_batch(vec![rewrite]).unwrap();
let mut reopened = registered();
reopened.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
reopened.pull_all_backing_stores().unwrap();
assert_eq!(reopened.snapshot_envelopes().iter().find(|held| held.key == "marked"), Some(&held));
}
#[test]
fn is_writable_agrees_with_the_writes_for_foreign_keys_and_other_types() {
let dir = tempfile::tempdir().unwrap();
// Only UnionSurface registered: the store's unit-arm record is foreign here.
let path = dir.path().join("extra-slot-store.cc");
std::fs::write(&path, without_variant_row(&vector("extra-slot-store.cc"))).unwrap();
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();
assert!(cache.put("unit-arm", &constructed()).is_err());
assert!(!cache.is_writable(UnionSurface::TYPE, "unit-arm").unwrap());
assert!(cache.is_writable(UnionSurface::TYPE, "unheld").unwrap());
// Another registered type at a key a record of this one holds.
let (mut cache, _dir, _) = held_marked_and_clean();
let other = SlotUnit { name: "clean".into(), outcome: UnitOutcome::Denied };
assert!(cache.put("clean", &other).is_err());
assert!(!cache.is_writable(SlotUnit::TYPE, "clean").unwrap());
assert!(!cache.is_writable(SlotUnit::TYPE, "marked").unwrap());
assert!(cache.is_writable(UnionSurface::TYPE, "clean").unwrap());
}
#[test]
fn a_marked_key_refuses_a_batch_wherever_it_sits() {
let (mut cache, _dir, path) = held_marked_and_clean();
let before = std::fs::read(&path).unwrap();
let marked = envelope("marked", rmp_serde::to_vec(&constructed()).unwrap());
let clean = |key: &str| envelope(key, rmp_serde::to_vec(&constructed()).unwrap());
for batch in [
vec![marked.clone(), clean("a"), clean("b")],
vec![clean("a"), marked.clone(), clean("b")],
vec![clean("a"), clean("b"), marked.clone()],
] {
let error = cache.put_prepared_batch(batch).unwrap_err();
assert_refused_with(&error, "marked", &mark("Composition", 3, 2, 1));
assert_eq!(std::fs::read(&path).unwrap(), before);
assert!(cache.snapshot_envelopes().iter().all(|held| held.key != "a" && held.key != "b"));
}
}
#[test]
fn a_marked_decode_leaves_no_scope_open() {
assert_eq!(cultcache_rs::__enter_slots(), 0);
decode_payload_marked::<UnionSurface>(&vector("extra-slot-record.msgpack")).unwrap();
assert_eq!(cultcache_rs::__enter_slots(), 0, "a marked decode left its scope open on this thread");
let _ = decode_payload_marked::<UnionSurface>(&vector("depth-skipped-past-limit.msgpack"));
assert_eq!(cultcache_rs::__enter_slots(), 0);
}
#[test]
fn is_writable_is_false_for_a_type_with_no_home_store() {
let dir = tempfile::tempdir().unwrap();
let mut cache = registered();
// One store, routed to SlotUnit only: UnionSurface has no home.
cache.add_backing_store(SingleFileMessagePackBackingStore::new(dir.path().join("s.cc")), [SlotUnit::TYPE]).unwrap();
assert!(cache.put("k", &constructed()).is_err());
assert!(!cache.is_writable(UnionSurface::TYPE, "k").unwrap());
assert!(cache.is_writable(SlotUnit::TYPE, "k").unwrap());
}
#[test]
fn load_soa_refuses_a_same_length_replacement_of_a_marked_record() {
use cultcache_rs::CultCacheSoaTable;
let (mut cache, _dir, _) = held_marked_and_clean();
let all = cache.snapshot_envelopes();
let marked = all.iter().find(|held| held.key == "marked").unwrap();
// The clean record's bytes padded to the marked record's length: same length, different bytes.
let mut same_length = vec![0xc0; marked.payload.len()];
same_length[..2].copy_from_slice(&[0x91, 0xa1]);
assert_eq!(same_length.len(), marked.payload.len());
let mut replaced = all.clone();
replaced.iter_mut().find(|held| held.key == "marked").unwrap().payload = same_length;
let error = cache.load_soa(CultCacheSoaTable::from_envelopes(replaced)).unwrap_err();
assert_refused_with(&error, "marked", &mark("Composition", 3, 2, 1));
assert_eq!(cache.snapshot_envelopes(), all);
}

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

@ -1,244 +1,17 @@
//! The Rust runtime held to tests/vectors/union-parity, written by the C# reference.
//! Bytes and outcomes are read from the committed files; only the constructed value is spelled here.
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;
use cultcache_rs::check_payload_depth;
use cultcache_rs::CultSlots;
use cultcache_rs::CultUnion;
use cultcache_rs::DatabaseEntry;
use cultcache_rs::SingleFileMessagePackBackingStore;
#[derive(Clone, Debug, PartialEq, CultUnion)]
enum Node {
#[cultcache(key = 0)]
Box(BoxArm),
#[cultcache(key = 1)]
Text(TextArm),
#[cultcache(key = 2)]
Use(UseArm),
#[cultcache(key = 5)]
Image(ImageArm),
#[cultcache(key = 6)]
Wrap(WrapArm),
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
struct WrapArm {
#[cultcache(key = 0)]
inner: Option<std::boxed::Box<Node>>,
#[cultcache(key = 1)]
id: String,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
struct BoxArm {
#[cultcache(key = 0)]
id: String,
#[cultcache(key = 1)]
children: Vec<Node>,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
struct TextArm {
#[cultcache(key = 0)]
id: String,
#[cultcache(key = 1)]
content: String,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
struct UseArm {
#[cultcache(key = 0)]
id: String,
#[cultcache(key = 1)]
composition: Composition,
#[cultcache(key = 2)]
slots: Vec<Option<Node>>,
}
// Slot 2 is retired: a gap written nil.
#[derive(Clone, Debug, PartialEq, CultSlots)]
struct ImageArm {
#[cultcache(key = 0)]
id: String,
#[cultcache(key = 1)]
uri: String,
#[cultcache(key = 3)]
alt: String,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
struct Composition {
#[cultcache(key = 0)]
name: String,
#[cultcache(key = 1)]
body: Option<std::boxed::Box<Node>>,
}
#[derive(Clone, Debug, PartialEq, CultUnion)]
enum Outcome {
#[cultcache(key = 0)]
Pending(PendingArm),
#[cultcache(key = 1)]
Completed(CompletedArm),
#[cultcache(key = 2)]
Denied(IdArm),
#[cultcache(key = -2147483648)]
Floor(IdArm),
#[cultcache(key = 2147483647)]
Ceiling(IdArm),
}
// The codec's unit arm (no members, written [key, []]), kept apart from the golden: a document list's arms carry ids.
#[derive(Clone, Debug, PartialEq, CultUnion)]
enum UnitOutcome {
#[cultcache(key = 2)]
Denied,
#[cultcache(key = -2147483648)]
Floor,
#[cultcache(key = 2147483647)]
Ceiling,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
struct PendingArm {
#[cultcache(key = 0)]
message: String,
#[cultcache(key = 1)]
id: String,
}
// An arm with no members of its own: just its element id.
#[derive(Clone, Debug, PartialEq, CultSlots)]
struct IdArm {
#[cultcache(key = 0)]
id: String,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
struct CompletedArm {
#[cultcache(key = 0)]
source_version: i32,
#[cultcache(key = 1)]
id: String,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
struct NamedArm {
#[cultcache(key = 0)]
name: String,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
struct UsesArm {
#[cultcache(key = 0)]
name: String,
#[cultcache(key = 1)]
uses: i32,
}
#[derive(Clone, Debug, PartialEq, CultSlots)]
struct CountArm {
#[cultcache(key = 0)]
name: String,
#[cultcache(key = 1)]
count: i32,
}
#[derive(Clone, Debug, PartialEq, CultUnion)]
enum Item {
#[cultcache(key = 0)]
Gem(NamedArm),
#[cultcache(key = 1)]
Tool(UsesArm),
#[cultcache(key = 2)]
Part(CountArm),
}
// The same arms as Item under its own keys: key 2 is unknown to it.
#[derive(Clone, Debug, PartialEq, CultUnion)]
enum CraftedItem {
#[cultcache(key = 0)]
Tool(UsesArm),
#[cultcache(key = 1)]
Part(CountArm),
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(type = "tests.union_surface", schema = "tests.union_surface.v1")]
struct UnionSurface {
#[cultcache(key = 0)]
id: String,
#[cultcache(key = 1)]
version: i32,
#[cultcache(key = 2)]
root: Option<Node>,
#[cultcache(key = 3)]
outcomes: Vec<Outcome>,
#[cultcache(key = 4)]
part: Option<Item>,
#[cultcache(key = 5)]
crafted: Option<CraftedItem>,
#[cultcache(key = 6)]
missing: Option<Node>,
}
fn text(id: &str, content: &str) -> Node {
Node::Text(TextArm { id: id.into(), content: content.into() })
}
fn image(id: &str, uri: &str, alt: &str) -> Node {
Node::Image(ImageArm { id: id.into(), uri: uri.into(), alt: alt.into() })
}
fn boxed(id: &str, children: Vec<Node>) -> Node {
Node::Box(BoxArm { id: id.into(), children })
}
fn constructed() -> UnionSurface {
let inner = boxed("0000000000d0", vec![text("0000000000d1", "x"), image("0000000000d2", "uri://a", "alt")]);
UnionSurface {
id: "surface".into(),
version: 3,
root: Some(boxed(
"0000000000c0",
vec![
text("0000000000c1", "hello"),
Node::Use(UseArm {
id: "0000000000c2".into(),
composition: Composition { name: "comp".into(), body: Some(std::boxed::Box::new(inner)) },
slots: vec![Some(text("0000000000e1", "slot")), None, Some(boxed("0000000000e2", vec![]))],
}),
image("0000000000c3", "uri://b", ""),
],
)),
outcomes: vec![
Outcome::Pending(PendingArm { message: "wait".into(), id: "0000000000a1".into() }),
Outcome::Completed(CompletedArm { source_version: 7, id: "0000000000a2".into() }),
Outcome::Denied(id_arm("0000000000a3")),
],
part: Some(Item::Tool(UsesArm { name: "hammer".into(), uses: 12 })),
crafted: Some(CraftedItem::Part(CountArm { name: "gear".into(), count: 4 })),
missing: None,
}
}
fn id_arm(id: &str) -> IdArm {
IdArm { id: id.into() }
}
fn vector(name: &str) -> Vec<u8> {
let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.join("../../tests/vectors/union-parity")
.join(name);
std::fs::read(&path).unwrap_or_else(|error| panic!("{}: {error}", path.display()))
}
mod common;
use common::*;
fn decode(bytes: &[u8]) -> Result<UnionSurface, String> {
rmp_serde::from_slice(bytes).map_err(|error| error.to_string())
@ -325,11 +98,10 @@ fn an_unknown_arm_lists_the_declared_keys_and_never_echoes_the_payload() {
#[test]
fn a_union_that_is_not_a_key_and_slots_pair_is_refused_naming_the_enum() {
let cases: [(&str, Vec<u8>); 4] = [
let cases: [(&str, Vec<u8>); 3] = [
("an empty array", vec![0x90]),
("a trailing element", rmp_serde::to_vec(&(1u64, (7u8,), 0u8)).unwrap()),
("a string key", rmp_serde::to_vec(&("CANARY-key", (7u8,))).unwrap()),
("a unit arm with slots", rmp_serde::to_vec(&(2u64, ("CANARY-slot",))).unwrap()),
];
for (label, bytes) in cases {
let message = rmp_serde::from_slice::<UnitOutcome>(&bytes).unwrap_err().to_string();

53
packages/cultmesh-rs/src/lib.rs
View file

@ -732,6 +732,18 @@ impl CultMeshNode {
let cultnet_rs::CultNetMessage::SnapshotResponseRaw { documents, .. } = response else {
anyhow::bail!("expected cultnet.snapshot_response_raw.v0");
};
self.apply_raw_snapshot_documents(documents)
}
/// Applies each snapshot document of a registered schema with `put_raw_envelope`. The first refusal ends the pull
/// and the documents before it stay applied; a key this node holds read-only (unread slots from a newer peer, or a
/// payload this build cannot read) is refused like any other. C# applies a raw snapshot as one cache commit, so a
/// refusal there applies nothing: this is a divergence, owed by follow-up mesh-snapshot-atomic (stage the pull through
/// `put_prepared_batch`).
fn apply_raw_snapshot_documents(
&mut self,
documents: Vec<cultnet_rs::CultNetRawDocumentRecord>,
) -> Result<usize> {
let mut applied = 0usize;
for document in documents {
let Some(document_type) =
@ -1199,6 +1211,47 @@ mod tests {
Note => "cultmesh.test.note.v0",
});
fn raw_note(key: &str, payload: Vec<u8>) -> cultnet_rs::CultNetRawDocumentRecord {
cultnet_rs::CultNetRawDocumentRecord {
schema_id: "cultmesh.test.note.v0".to_string(),
record_key: key.to_string(),
stored_at: "2026-01-01T00:00:00Z".to_string(),
payload_encoding: cultnet_rs::CultNetRawPayloadEncoding::Messagepack,
payload,
source_runtime_id: None,
source_agent_id: None,
source_role: None,
tags: None,
}
}
#[test]
fn a_snapshot_document_for_a_read_only_key_ends_the_pull_unlike_csharp_which_applies_nothing() -> Result<()> {
let temp = tempfile::tempdir()?;
let mut node = CultMesh::create_node(temp.path().join("m.cc"), TestDocuments, Default::default())?;
// A newer peer's note: one slot past the declared one.
let newer = rmp_serde::to_vec(&("newer body", "extra slot"))?;
let plain = |body: &str| rmp_serde::to_vec(&(body,));
assert_eq!(node.apply_raw_snapshot_documents(vec![raw_note("held", newer)])?, 1);
assert_eq!(node.get_required::<Note>("held")?.body, "newer body");
let error = node
.apply_raw_snapshot_documents(vec![
raw_note("before", plain("b")?),
raw_note("held", plain("overwrite")?),
raw_note("after", plain("a")?),
])
.unwrap_err();
let conflict = error.downcast_ref::<cultcache_rs::SchemaConflictError>().expect("typed refusal");
assert_eq!(conflict.record_key, "held");
assert!(conflict.unread_slots.is_some());
assert!(!error.to_string().contains("extra slot"));
assert_eq!(node.get_required::<Note>("before")?.body, "b");
assert_eq!(node.get_required::<Note>("held")?.body, "newer body");
assert!(node.get::<Note>("after")?.is_none());
Ok(())
}
#[test]
fn node_round_trips_registered_documents() -> Result<()> {
let temp = tempfile::tempdir()?;

1
packages/cultnet-rs/src/replication.rs
View file

@ -146,6 +146,7 @@ impl CultNetDocumentRegistry {
schema_id: binding.schema_id.clone(),
schema_names: vec![holder.document_type.clone(), binding.document_type.clone()],
record_key: String::new(),
unread_slots: None,
}));
}
self.bindings_by_schema_id