Rust: a payload is one MessagePack value (bytes-after-rs) (#131)
This commit is contained in:
parent
13ce5ea50a
commit
363a8d6fe5
8 changed files with 346 additions and 30 deletions
|
|
@ -259,7 +259,7 @@ landed in `ead5260c`, Python in `5618d8dd`, Rust in `c028c511`. TypeScript is no
|
|||
| Runtime | Owner of the decision (constant and walker) | Doors routed through it |
|
||||
|---|---|---|
|
||||
| C# | `CultPayloadDepth.Check` (`src/GameCult.Caching.MessagePack/CultPayloadDepth.cs`), limit `CultMessagePackSecurity.MaximumPayloadDepth`. MessagePack's own graph counter is set to the same value as a backstop only. | Write: `CultDocumentMessagePackSerialization.SerializeUntyped`. Read: `DeserializeUntyped`. The cache store (`SerializePayload`/`DeserializePayload`), the directory store, `CultMesh` and single-file documents all call those two; C# has no separate raw-envelope or prepared-batch door. |
|
||||
| Rust | `cultcache_rs::check_payload_depth` and `MAX_PAYLOAD_DEPTH` (`packages/cultcache-rs/src/lib.rs`). | Read: `decode_payload` (get, list, `validate_envelope`/`put_envelope`, `finish_prepared_entry`). Write: `put`/`prepare_entry`/`prepare_entry_named` reach it through `finish_prepared_entry`, which decodes what it just encoded. Raw envelope: `put_raw_envelope`, the door CultMesh network puts use (`cultmesh-rs` `lib.rs`, two call sites). Prepared batch: `put_prepared_batch` via `require_payload_depth`. cultnet-rs: every free-form or skippable wire payload decodes through `cultcache_rs::decode_payload` (message decode in `contracts.rs`, `provider_session`, `media_stream_wire`, `replication`, `service_identity`). |
|
||||
| Rust | `check_record_payload` (private; depth, then no bytes after the first value) over `first_value_end`, and `MAX_PAYLOAD_DEPTH`; `cultcache_rs::check_payload_depth` is the depth-only probe for outside callers (`packages/cultcache-rs/src/lib.rs`). | Read: `decode_payload` (get, list, `validate_envelope`/`put_envelope`, `finish_prepared_entry`). Write: `put`/`prepare_entry`/`prepare_entry_named` reach it through `finish_prepared_entry`, which decodes what it just encoded. Raw envelope: `put_raw_envelope`, the door CultMesh network puts use (`cultmesh-rs` `lib.rs`, two call sites). Prepared batch: `put_prepared_batch` via `require_record_payload`. Load: `pull_all_backing_stores` and `load_soa` run `require_record_payload` on every record, so a store holding a payload past the bound (or with bytes after it) is refused at load, and the cache then refuses every write and removal (`CultCacheLoadRefused`) until a load succeeds. cultnet-rs: every free-form or skippable wire payload decodes through `cultcache_rs::decode_payload` (message decode in `contracts.rs`, `provider_session`, `media_stream_wire`, `replication`, `service_identity`). |
|
||||
| Python | `_refuse_past_depth_limit` and `MAX_PAYLOAD_DEPTH` (`packages/cultcache-py/src/cultcache_py/documents.py`); a msgpack `StackError` on read maps to the same error. | `DocumentDefinition.encode_payload` (write) and `decode_payload` (read). `CultCache.put`, `put_envelope`, `put_envelopes` and load-time hydration go through them; there is no raw-envelope door. cultmesh-py round trips call the same pair. |
|
||||
|
||||
- **Doors deliberately not routed.** cultnet-rs `idunn_lifecycle_brake`, `provider_health_contracts` and
|
||||
|
|
|
|||
|
|
@ -15,6 +15,8 @@ Semver: this is a breaking change in the 0.x line (0.3.x to 0.4.0).
|
|||
- `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 cache whose last `pull_all_backing_stores` was refused (a tail, a payload past the nesting bound, a duplicate unique-index value, a retype conflict, a record from the wrong store) refuses every `put`, `put_envelope`, `put_raw_envelope`, `put_prepared_batch`, `update` and `delete` with `CultCacheLoadRefused` until a pull succeeds. Before, it stayed writable over an empty image, and the next batch write replaced the store with only the new records.
|
||||
- `decode_payload`, `put_raw_envelope` and prepared batches refuse a payload with bytes after its first MessagePack value (`payload has {n} bytes after its first MessagePack value`), by count; the tail is never walked or quoted. They used to read the first value and keep or drop the tail, so a record with a stray tail was served and rewritten without it. A store holding such a record (single-file, redb, `load_soa`) is refused at load, naming the type, the key and the count, and stays unreadable until the record is repaired; the same check now runs on every record at load, as the C# and Python runtimes do at open. The nesting bound is part of that check, so a stored payload nested past it is refused at load too, where it used to load and refuse on read.
|
||||
- A single-file store written as `cultcache.store.v3` is read and written: a store the C# runtime marked v3 (it holds a record with an element id: an owned list element with a `[CultElementId]` member) keeps its header here, and is carried through every rewrite with its ids; this runtime never drops the mark. A runtime from before this change refuses a v3 store by name. Rolling back to an older release after a store has been written as v3 is one-way: there is no carry-back, so restore from a copy taken before upgrading.
|
||||
- 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.
|
||||
|
|
|
|||
|
|
@ -600,14 +600,22 @@ impl std::error::Error for PayloadTooDeep {}
|
|||
|
||||
/// Walks the MessagePack headers of one payload without recursion, counting every array and map and
|
||||
/// refusing the first container past [`MAX_PAYLOAD_DEPTH`]. Scalars are skipped by length and nothing
|
||||
/// is allocated per element. A payload that ends early is the decoder's to refuse.
|
||||
/// is allocated per element. Only the first value is walked; a payload that ends early is the
|
||||
/// decoder's to refuse. A depth-only probe: [`decode_payload`] also refuses bytes after the value.
|
||||
pub fn check_payload_depth(bytes: &[u8]) -> std::result::Result<(), PayloadTooDeep> {
|
||||
first_value_end(bytes).map(|_| ())
|
||||
}
|
||||
|
||||
/// Where the first MessagePack value of `bytes` ends, walking only that value (its tail is never
|
||||
/// 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;
|
||||
let mut position = 0usize;
|
||||
while position < bytes.len() {
|
||||
let Some((advance, children)) = element_header(bytes, position) else {
|
||||
return Ok(());
|
||||
return Ok(None);
|
||||
};
|
||||
position = position.saturating_add(advance);
|
||||
if let Some(children) = children {
|
||||
|
|
@ -629,7 +637,18 @@ pub fn check_payload_depth(bytes: &[u8]) -> std::result::Result<(), PayloadTooDe
|
|||
depth -= 1;
|
||||
}
|
||||
if depth == 0 {
|
||||
return Ok(());
|
||||
return Ok(Some(position));
|
||||
}
|
||||
}
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
/// The one rule for "is this payload one value": the nesting bound, then no bytes after the first
|
||||
/// value. The count is all it reports; the tail is never walked or quoted.
|
||||
fn check_record_payload(bytes: &[u8]) -> Result<()> {
|
||||
if let Some(end) = first_value_end(bytes)? {
|
||||
if end < bytes.len() {
|
||||
bail!("payload has {} bytes after its first MessagePack value", bytes.len() - end);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
|
|
@ -784,7 +803,7 @@ fn decode_grown<T: serde::de::DeserializeOwned>(bytes: &[u8]) -> std::result::Re
|
|||
grown(&mut reader, |grown| T::deserialize(grown))
|
||||
}
|
||||
|
||||
/// The one decode of a document payload: the nesting bound first, then [`decode_grown`], whose
|
||||
/// The one decode of a document payload: the payload check first (nesting bound, one value), then [`decode_grown`], whose
|
||||
/// rmp-serde depth counter (1024) is only a backstop. A failure keeps its kind and never quotes the
|
||||
/// bytes. Readers of stored envelope payloads outside this crate decode through it too.
|
||||
///
|
||||
|
|
@ -796,7 +815,7 @@ fn decode_grown<T: serde::de::DeserializeOwned>(bytes: &[u8]) -> std::result::Re
|
|||
/// verified on x86_64-pc-windows-msvc by hand (`decode_stack`, revision 10a39dac, 13 tests, debug and release); a
|
||||
/// `stacker` bump or a change to the stack-size tests repeats it.
|
||||
pub fn decode_payload<T: serde::de::DeserializeOwned>(payload: &[u8]) -> Result<T> {
|
||||
check_payload_depth(payload)?;
|
||||
check_record_payload(payload)?;
|
||||
decode_grown(payload).map_err(opaque_decode_error)
|
||||
}
|
||||
|
||||
|
|
@ -918,10 +937,11 @@ impl Drop for UnreadScope {
|
|||
}
|
||||
}
|
||||
|
||||
/// [`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.
|
||||
/// [`decode_payload`] with the mark: the same payload check and grown decode, inside a scope the derives' slot
|
||||
/// visitors note into. A payload past the nesting limit or with bytes after its value 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)?;
|
||||
check_record_payload(payload)?;
|
||||
let scope = UnreadScope::open();
|
||||
let value = decode_grown(payload).map_err(opaque_decode_error)?;
|
||||
Ok((value, scope.mark()))
|
||||
|
|
@ -948,10 +968,10 @@ struct Definition {
|
|||
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<()> {
|
||||
check_payload_depth(&entry.payload).with_context(|| {
|
||||
/// The payload check for an envelope admitted without a typed decode, the one [`decode_payload`]
|
||||
/// runs first. Names the entry's type and key, never the payload.
|
||||
fn require_record_payload(entry: &CultCacheEnvelope) -> Result<()> {
|
||||
check_record_payload(&entry.payload).with_context(|| {
|
||||
format!(
|
||||
"refused CultCache envelope {:?} at key {:?}",
|
||||
entry.r#type, entry.key
|
||||
|
|
@ -2917,8 +2937,30 @@ pub struct CultCache {
|
|||
/// Records a store holds that no registered type claims: (store index, the envelope as the store read it). Derived from the last
|
||||
/// pull and replaced with it; written nowhere else. It decides only refusals at its keys.
|
||||
foreign: Vec<(usize, CultCacheEnvelope)>,
|
||||
/// Why the last pull of the backing stores was refused; `None` once a pull succeeds. While set, every write and removal is refused.
|
||||
load_refused: Option<String>,
|
||||
}
|
||||
|
||||
/// A write or removal refused because the cache's last load of its backing stores was refused. The cache holds no
|
||||
/// image of those stores, so any write would persist an image that lacks the records it never read. Repair the store
|
||||
/// named by `reason` (which names a type, a key and a clause, never a payload) and load again.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct CultCacheLoadRefused {
|
||||
pub reason: String,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for CultCacheLoadRefused {
|
||||
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
write!(
|
||||
formatter,
|
||||
"CultCache refuses writes and removals: its load of the backing stores failed ({}); repair the store and load again",
|
||||
self.reason
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for CultCacheLoadRefused {}
|
||||
|
||||
/// A record a store holds that no registered type claims: carried, never served.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct ForeignRecord {
|
||||
|
|
@ -2980,6 +3022,7 @@ impl CultCache {
|
|||
projectors: BTreeMap::new(),
|
||||
stores: Vec::new(),
|
||||
foreign: Vec::new(),
|
||||
load_refused: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -3116,12 +3159,39 @@ impl CultCache {
|
|||
/// Reads every store into a staged image and replaces the cache's view
|
||||
/// only when all of them loaded, so a refused load admits nothing.
|
||||
pub fn pull_all_backing_stores(&mut self) -> Result<()> {
|
||||
match self.stage_pull() {
|
||||
Ok((loaded, foreign, holders)) => {
|
||||
self.entries = loaded;
|
||||
self.foreign = foreign;
|
||||
self.index_holders = holders;
|
||||
self.load_refused = None;
|
||||
Ok(())
|
||||
}
|
||||
Err(error) => {
|
||||
self.load_refused = Some(format!("{error:#}"));
|
||||
Err(error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The one gate of every write and removal: a cache whose load was refused holds no image of its stores.
|
||||
fn require_loaded(&self) -> Result<()> {
|
||||
match &self.load_refused {
|
||||
Some(reason) => Err(CultCacheLoadRefused { reason: reason.clone() }.into()),
|
||||
None => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
/// The image every store holds, judged whole, installed nowhere.
|
||||
#[allow(clippy::type_complexity)]
|
||||
fn stage_pull(&self) -> Result<(BTreeMap<(String, String), CultCacheEnvelope>, Vec<(usize, CultCacheEnvelope)>, IndexHolders)> {
|
||||
let mut loaded = BTreeMap::new();
|
||||
let mut foreign = Vec::new();
|
||||
let known_types: BTreeSet<String> = self.definitions.keys().cloned().collect();
|
||||
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,
|
||||
|
|
@ -3154,10 +3224,7 @@ impl CultCache {
|
|||
}
|
||||
self.refuse_second_globals(&loaded)?;
|
||||
let holders = self.judge_image(&loaded)?;
|
||||
self.entries = loaded;
|
||||
self.foreign = foreign;
|
||||
self.index_holders = holders;
|
||||
Ok(())
|
||||
Ok((loaded, foreign, holders))
|
||||
}
|
||||
|
||||
/// The records the last pull found in a store that no registered type claims, sorted by key (UTF-16 order). The cache does not
|
||||
|
|
@ -3383,6 +3450,7 @@ impl CultCache {
|
|||
}
|
||||
|
||||
pub fn put<T: DatabaseEntry>(&mut self, key: impl Into<String>, value: &T) -> Result<T> {
|
||||
self.require_loaded()?;
|
||||
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)?;
|
||||
|
|
@ -3460,6 +3528,7 @@ impl CultCache {
|
|||
}
|
||||
|
||||
pub fn put_prepared_batch(&mut self, staged: Vec<CultCacheEnvelope>) -> Result<()> {
|
||||
self.require_loaded()?;
|
||||
if staged.is_empty() {
|
||||
return Err(anyhow!("CultCache batch must contain at least one entry"));
|
||||
}
|
||||
|
|
@ -3484,7 +3553,7 @@ impl CultCache {
|
|||
entry.r#type
|
||||
));
|
||||
}
|
||||
require_payload_depth(entry)?;
|
||||
require_record_payload(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 {
|
||||
|
|
@ -3527,6 +3596,7 @@ impl CultCache {
|
|||
}
|
||||
|
||||
pub fn put_envelope<T: DatabaseEntry>(&mut self, entry: CultCacheEnvelope) -> Result<T> {
|
||||
self.require_loaded()?;
|
||||
let parsed = self.validate_envelope::<T>(&entry)?;
|
||||
self.refuse_unread_slots(&entry.r#type, &entry.key)?;
|
||||
self.refuse_claimed_id(&entry)?;
|
||||
|
|
@ -3547,6 +3617,7 @@ impl CultCache {
|
|||
/// admission; CultCache still enforces registered type, identity, timestamp,
|
||||
/// the payload nesting bound ([`MAX_PAYLOAD_DEPTH`]), routing, and persistence.
|
||||
pub fn put_raw_envelope(&mut self, entry: CultCacheEnvelope) -> Result<()> {
|
||||
self.require_loaded()?;
|
||||
if !self.definitions.contains_key(&entry.r#type) {
|
||||
return Err(anyhow!(
|
||||
"No entry type registered for CultCache envelope type {:?}",
|
||||
|
|
@ -3565,7 +3636,7 @@ impl CultCache {
|
|||
entry.r#type
|
||||
));
|
||||
}
|
||||
require_payload_depth(&entry)?;
|
||||
require_record_payload(&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)?;
|
||||
|
|
@ -3621,6 +3692,7 @@ impl CultCache {
|
|||
}
|
||||
|
||||
pub fn delete<T: DatabaseEntry>(&mut self, key: &str) -> Result<bool> {
|
||||
self.require_loaded()?;
|
||||
self.require_entry_type::<T>()?;
|
||||
self.refuse_unread_slots(T::TYPE, key)?;
|
||||
let id = entry_id_parts(T::TYPE, key);
|
||||
|
|
@ -3662,6 +3734,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)?;
|
||||
|
|
@ -6296,6 +6369,34 @@ mod tests {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn put_raw_envelope_refuses_bytes_after_the_value() -> Result<()> {
|
||||
let temp = tempfile::tempdir()?;
|
||||
let store_path = temp.path().join("tail.cc");
|
||||
let mut cache = CultCache::new();
|
||||
cache.register_entry_type::<Settings>()?;
|
||||
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?;
|
||||
let tailed = CultCacheEnvelope {
|
||||
key: "app".to_string(),
|
||||
r#type: Settings::TYPE.to_string(),
|
||||
payload: vec![0x91, 0x01, 0xc0],
|
||||
stored_at: now_utc_second(),
|
||||
schema_id: None,
|
||||
};
|
||||
let error = format!("{:#}", cache.put_raw_envelope(tailed.clone()).unwrap_err());
|
||||
assert!(error.contains("payload has 1 bytes after its first MessagePack value"), "{error}");
|
||||
assert!(error.contains("app") && error.contains(Settings::TYPE), "{error}");
|
||||
assert!(cache.snapshot().is_empty());
|
||||
assert!(!store_path.exists() || SingleFileMessagePackBackingStore::new(&store_path).pull_all()?.is_empty());
|
||||
let error = format!("{:#}", cache.put_prepared_batch(vec![tailed.clone()]).unwrap_err());
|
||||
assert!(error.contains("payload has 1 bytes after its first MessagePack value"), "{error}");
|
||||
assert!(cache.snapshot().is_empty());
|
||||
// The same bytes without the tail are admitted: the refusal is the count, not the shape.
|
||||
cache.put_raw_envelope(CultCacheEnvelope { payload: vec![0x91, 0x01], ..tailed })?;
|
||||
assert_eq!(cache.snapshot().len(), 1);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn soa_snapshot_round_trips_registered_entries() -> Result<()> {
|
||||
let temp = tempfile::tempdir()?;
|
||||
|
|
|
|||
|
|
@ -481,9 +481,11 @@ fn load_soa_refuses_a_same_length_replacement_of_a_marked_record() {
|
|||
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]);
|
||||
// One value (a bin16) of the marked record's length: same length, different bytes. A value followed by padding
|
||||
// would be refused at admission as bytes after the value, before the read-only check is reached.
|
||||
let length = marked.payload.len();
|
||||
let mut same_length = vec![0u8; length];
|
||||
same_length[..3].copy_from_slice(&[0xc5, ((length - 3) >> 8) as u8, (length - 3) as u8]);
|
||||
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;
|
||||
|
|
@ -491,3 +493,47 @@ fn load_soa_refuses_a_same_length_replacement_of_a_marked_record() {
|
|||
assert_refused_with(&error, "marked", &mark("Composition", 3, 2, 1));
|
||||
assert_eq!(cache.snapshot_envelopes(), all);
|
||||
}
|
||||
|
||||
// The payload check runs before the mark scope opens: a tail on a slot-bearing payload is refused by count, never marked.
|
||||
#[test]
|
||||
fn a_tail_on_a_slot_payload_is_refused_before_any_mark() {
|
||||
let mut tailed = vector("extra-slot-record.msgpack");
|
||||
tailed.push(0xc0);
|
||||
let error = decode_payload_marked::<UnionSurface>(&tailed).unwrap_err();
|
||||
assert!(format!("{error:#}").contains("payload has 1 bytes after its first MessagePack value"), "{error:#}");
|
||||
// Nothing leaked: the next marked decode finds exactly its own mark.
|
||||
let found = decode_payload_marked::<UnionSurface>(&vector("extra-slot-record.msgpack")).unwrap().1.unwrap();
|
||||
assert_eq!(found.occurrences, 1);
|
||||
}
|
||||
|
||||
// A store holding marked records and one with a tail is refused at load; the cache then refuses every write with the load
|
||||
// refusal (not a per-key mark conflict), and the store file is untouched.
|
||||
#[test]
|
||||
fn a_store_with_marked_records_and_a_tail_is_refused_and_its_cache_writes_nothing() {
|
||||
use cultcache_rs::CacheBackingStore;
|
||||
use cultcache_rs::CultCacheLoadRefused;
|
||||
let (mut held, _dir, path) = held_marked_and_clean();
|
||||
let mut tailed = vector("union-surface.msgpack");
|
||||
tailed.push(0xc0);
|
||||
SingleFileMessagePackBackingStore::new(&path).push(&envelope("tailed", tailed)).unwrap();
|
||||
let before = std::fs::read(&path).unwrap();
|
||||
// The cache that held the store re-reads it and is refused.
|
||||
assert!(format!("{:#}", held.pull_all_backing_stores().unwrap_err()).contains("bytes after"));
|
||||
let fresh_value = constructed();
|
||||
let mut fresh = registered();
|
||||
fresh.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
|
||||
assert!(fresh.pull_all_backing_stores().is_err());
|
||||
for cache in [&mut held, &mut fresh] {
|
||||
let staged = cache.prepare_entry("new", &fresh_value).unwrap().0;
|
||||
let errors = [
|
||||
cache.put("new", &fresh_value).unwrap_err(),
|
||||
cache.put_prepared_batch(vec![staged.clone()]).unwrap_err(),
|
||||
cache.put_raw_envelope(staged).unwrap_err(),
|
||||
cache.delete::<UnionSurface>("new").unwrap_err(),
|
||||
];
|
||||
for error in errors {
|
||||
assert!(error.downcast_ref::<CultCacheLoadRefused>().is_some(), "{error:#}");
|
||||
}
|
||||
}
|
||||
assert_eq!(std::fs::read(&path).unwrap(), before);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,9 +4,13 @@
|
|||
use cultcache_rs::CultCache;
|
||||
use cultcache_rs::CacheBackingStore;
|
||||
use cultcache_rs::CultCacheEnvelope;
|
||||
use cultcache_rs::CultCacheLoadRefused;
|
||||
use cultcache_rs::CultCacheSoaTable;
|
||||
use cultcache_rs::MAX_PAYLOAD_DEPTH;
|
||||
use cultcache_rs::PayloadTooDeep;
|
||||
use cultcache_rs::check_payload_depth;
|
||||
use cultcache_rs::decode_payload;
|
||||
use cultcache_rs::RedbMessagePackBackingStore;
|
||||
use cultcache_rs::DatabaseEntry;
|
||||
use cultcache_rs::SingleFileMessagePackBackingStore;
|
||||
|
||||
|
|
@ -464,7 +468,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");
|
||||
|
|
@ -472,7 +476,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:#}")),
|
||||
|
|
@ -529,6 +536,157 @@ fn bounds_vectors() {
|
|||
assert_eq!((accepted, refused, read_only), (4, 7, 1));
|
||||
}
|
||||
|
||||
// A payload is one MessagePack value (bytes-after-outcomes.msgpack): every door refuses a tail by
|
||||
// its count, names the clause, never the depth limit (the tail is not walked) and never a value.
|
||||
#[test]
|
||||
fn bytes_after_vectors() {
|
||||
let rows: Vec<serde_json::Value> = rmp_serde::from_slice(&vector("bytes-after-outcomes.msgpack")).unwrap();
|
||||
assert_eq!(rows.len(), 4);
|
||||
let surface = vector("union-surface.msgpack");
|
||||
for row in &rows {
|
||||
let fields = row.as_array().unwrap();
|
||||
let (file, outcome, union, detail) = (fields[0].as_str().unwrap(), fields[1].as_str().unwrap(), fields[2].as_str().unwrap(), fields[4].as_str().unwrap());
|
||||
assert_eq!((outcome, union, detail), ("refused", "UnionSurface", "bytes-after"), "{file}");
|
||||
let bytes = vector(file);
|
||||
assert_eq!(&bytes[..surface.len()], &surface[..], "{file}");
|
||||
let clause = format!("payload has {} bytes after its first MessagePack value", bytes.len() - surface.len());
|
||||
let check = |door: &str, message: &str| {
|
||||
assert!(message.contains(&clause), "{file} {door}: {message}");
|
||||
assert!(!message.contains("the limit is"), "{file} {door} walked the tail: {message}");
|
||||
assert!(!message.contains("CANARY-bytes-after-7d1e"), "{file} {door} echoed the tail: {message}");
|
||||
assert!(!message.contains(CANARY), "{file} {door} echoed a value: {message}");
|
||||
};
|
||||
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 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();
|
||||
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();
|
||||
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();
|
||||
let message = format!("{:#}", raw.put_raw_envelope(envelope("k", &bytes)).unwrap_err());
|
||||
check("put_raw_envelope", &message);
|
||||
assert!(raw.snapshot().is_empty(), "{file}");
|
||||
}
|
||||
}
|
||||
|
||||
// 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:#}");
|
||||
}
|
||||
|
||||
// Depth is refused where the payload is loaded: a store holding one nested past the bound is not read.
|
||||
#[test]
|
||||
fn depth_is_refused_at_load() {
|
||||
let mut deep = vec![0x91u8; MAX_PAYLOAD_DEPTH + 1];
|
||||
deep.push(0xc0);
|
||||
let temp = tempfile::tempdir().unwrap();
|
||||
let path = temp.path().join("deep.cc");
|
||||
SingleFileMessagePackBackingStore::new(&path).push(&envelope("k", &deep)).unwrap();
|
||||
let mut cache = CultCache::new();
|
||||
cache.register_entry_type::<UnionSurface>().unwrap();
|
||||
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
|
||||
let error = cache.pull_all_backing_stores().unwrap_err();
|
||||
assert_eq!(error.downcast_ref::<PayloadTooDeep>(), Some(&TOO_DEEP), "{error:#}");
|
||||
assert!(cache.snapshot().is_empty());
|
||||
}
|
||||
|
||||
// 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]
|
||||
fn a_refused_load_refuses_every_write() {
|
||||
let tail = {
|
||||
let mut bytes = vector("union-surface.msgpack");
|
||||
bytes.push(0xc0);
|
||||
bytes
|
||||
};
|
||||
let good = envelope("good", &vector("union-surface.msgpack"));
|
||||
let bad = envelope("bad", &tail);
|
||||
for redb in [false, true] {
|
||||
let temp = tempfile::tempdir().unwrap();
|
||||
let path = temp.path().join(if redb { "seed.redb" } else { "seed.cc" });
|
||||
let mut cache = CultCache::new();
|
||||
cache.register_entry_type::<UnionSurface>().unwrap();
|
||||
if redb {
|
||||
let mut store = RedbMessagePackBackingStore::new(&path).unwrap();
|
||||
store.push(&good).unwrap();
|
||||
store.push(&bad).unwrap();
|
||||
drop(store);
|
||||
cache.add_generic_backing_store(RedbMessagePackBackingStore::new(&path).unwrap()).unwrap();
|
||||
} else {
|
||||
let mut store = SingleFileMessagePackBackingStore::new(&path);
|
||||
store.push(&good).unwrap();
|
||||
store.push(&bad).unwrap();
|
||||
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
|
||||
}
|
||||
let load = cache.pull_all_backing_stores().unwrap_err();
|
||||
assert!(format!("{load:#}").contains("bytes after"), "{load:#}");
|
||||
let before = std::fs::read(&path).unwrap();
|
||||
let value = constructed();
|
||||
let staged = cache.prepare_entry("new", &value).unwrap().0;
|
||||
let results: Vec<(&str, anyhow::Error)> = vec![
|
||||
("put", cache.put("new", &value).unwrap_err()),
|
||||
("put_prepared_batch", cache.put_prepared_batch(vec![staged.clone()]).unwrap_err()),
|
||||
("put_envelope", cache.put_envelope::<UnionSurface>(staged.clone()).unwrap_err()),
|
||||
("put_raw_envelope", cache.put_raw_envelope(staged.clone()).unwrap_err()),
|
||||
("update", cache.update::<UnionSurface, _>("good", |current| current.unwrap_or_else(constructed)).unwrap_err()),
|
||||
("delete", cache.delete::<UnionSurface>("good").unwrap_err()),
|
||||
("delete a key the image lacks", cache.delete::<UnionSurface>("absent").unwrap_err()),
|
||||
];
|
||||
for (door, error) in results {
|
||||
let refusal = error.downcast_ref::<CultCacheLoadRefused>().unwrap_or_else(|| panic!("{door} redb={redb}: {error:#}"));
|
||||
assert!(refusal.to_string().contains("bytes after"), "{door}: {refusal}");
|
||||
}
|
||||
assert_eq!(std::fs::read(&path).unwrap(), before, "redb={redb}: the store changed");
|
||||
assert!(cache.snapshot().is_empty());
|
||||
// Repairing the store and loading again lifts it.
|
||||
if redb {
|
||||
RedbMessagePackBackingStore::new(&path).unwrap().delete(&bad).unwrap();
|
||||
} else {
|
||||
SingleFileMessagePackBackingStore::new(&path).delete(&bad).unwrap();
|
||||
}
|
||||
cache.pull_all_backing_stores().unwrap();
|
||||
cache.put("new", &value).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
// 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() {
|
||||
let mut truncated = vec![0x91, 0xd9, 0x40];
|
||||
truncated.extend_from_slice(b"short");
|
||||
let message = format!("{:#}", decode_payload::<UnionSurface>(&truncated).unwrap_err());
|
||||
assert!(!message.contains("bytes after"), "{message}");
|
||||
}
|
||||
|
||||
fn wrap_chain(wraps: usize, leaf: Node) -> UnionSurface {
|
||||
let mut node = leaf;
|
||||
for _ in 0..wraps {
|
||||
|
|
|
|||
|
|
@ -15,5 +15,6 @@ only in git history.
|
|||
|
||||
### Breaking
|
||||
|
||||
- A CultNet frame with bytes after its message is refused, by count (through `cultcache_rs::decode_payload`); the decoder used to read the message and drop the tail. No producer emits one. Behaviour change: the next release is a minor bump (0.x breaking).
|
||||
- `CultNetDocumentRegistry::register` returns `Result<&mut Self>` where it returned `&mut Self`. A binding that claims a document type or a schema id another binding already holds is refused with `cultcache_rs::SchemaConflictError` (naming the new binding's schema id and both document types) and nothing is bound; binding a type to the schema id it already holds still replaces that binding. Callers chain `register(...)?`.
|
||||
- `CultMeshCdnChunkRef` gains a public `id` field, the decimal text of `offset` (the reference's element id, slot 4). A manifest written before ids reads `""`; encoding writes the slot.
|
||||
|
|
|
|||
|
|
@ -1810,12 +1810,10 @@ fn a_replicated_payload_nested_past_the_bound_is_refused_typed_on_apply() -> Res
|
|||
};
|
||||
assert_eq!(documents.len(), 1);
|
||||
|
||||
backing.push(&envelope("past", limit + 1))?;
|
||||
let mut cache = CultCache::new();
|
||||
cache.register_entry_type::<GhostlightAgentStateFixture>()?;
|
||||
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store))?;
|
||||
cache.pull_all_backing_stores()?;
|
||||
let past = registry.create_raw_snapshot_response(&cache, "past", None, Some(&["past".to_string()]))?;
|
||||
// The sender cannot load a payload past the bound, so the frame is the at-limit one with its bytes swapped.
|
||||
let mut past = at_limit.clone();
|
||||
let CultNetMessage::SnapshotResponseRaw { documents, .. } = &mut past else { unreachable!() };
|
||||
documents[0].payload = envelope("past", limit + 1).payload;
|
||||
let mut target = CultCache::new();
|
||||
target.register_entry_type::<GhostlightAgentStateFixture>()?;
|
||||
let error = registry
|
||||
|
|
|
|||
|
|
@ -190,6 +190,16 @@ fn message_past_the_bound_is_refused_typed() {
|
|||
});
|
||||
}
|
||||
|
||||
// A frame is one message: cultcache's payload rule refuses bytes after it, by count.
|
||||
#[test]
|
||||
fn frame_with_bytes_after_is_refused() {
|
||||
let mut bytes = hello_with(&nested(0));
|
||||
assert!(decode_message(bytes.clone(), BIG).is_ok());
|
||||
bytes.push(0xc0);
|
||||
let error = format!("{:#}", decode_message(bytes, BIG).unwrap_err());
|
||||
assert!(error.contains("1 bytes after its first MessagePack value"), "{error}");
|
||||
}
|
||||
|
||||
fn raw_v1_bytes(header_unknown: Option<&[u8]>) -> Vec<u8> {
|
||||
let mut header: Vec<(rmpv::Value, rmpv::Value)> = vec![
|
||||
("schemaId".into(), "s".into()),
|
||||
|
|
|
|||
Loading…
Reference in a new issue