diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index 5a782221..648bd1d0 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -929,16 +929,23 @@ pub fn decode_payload_marked(payload: &[u8]) -> /// 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 carries no finding: replacing or removing it stays allowed, as it was before -/// the mark, so a corrupt record can still be repaired. -fn unread_slots_of(payload: &[u8]) -> Option { - decode_payload_marked::(payload).ok().and_then(|(_, mark)| mark) +/// 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(payload: &[u8]) -> Result> { + decode_payload_marked::(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]) -> Option, + unread_slots: fn(&[u8]) -> Result>, } /// The nesting bound for an envelope admitted without a typed decode, from the same walk that @@ -3193,38 +3200,58 @@ impl CultCache { Ok(Some(payload)) } - /// Whether a write or removal of `(type_id, key)` would be admitted by the unread-slots rule: false when the held - /// payload carries slots past its registered type's declared slots, at any depth, which makes the record read-only. - /// True for a key the cache does not hold, and for a held payload the registered type cannot decode at all (it - /// carries no finding). An unregistered type is an error. Advisory: the write entries ask the same check themselves. + /// 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 { - Ok(self.unread_slots_held(type_id, key)?.is_none()) + 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.refuse_retype(&self.entries, &self.foreign, &probe).is_ok()) } - /// The one check behind every write entry and [`CultCache::is_writable`]: the held envelope at `(type_id, key)` - /// decoded under the registered type's mark. - fn unread_slots_held(&self, type_id: &str, key: &str) -> Result> { + /// 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 { 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(None); + return Ok(HeldFinding::Free); }; - Ok((definition.unread_slots)(&held.payload)) + 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 for its unread slots, before any store is touched. + /// 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<()> { - let Some(unread) = self.unread_slots_held(type_id, key)? else { - return Ok(()); - }; - 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), - })) + 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; repair it with the build that wrote it" + ))), + 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(&self, key: &str) -> Result { @@ -3623,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()? { @@ -3638,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; diff --git a/packages/cultcache-rs/tests/slot_readonly.rs b/packages/cultcache-rs/tests/slot_readonly.rs index 6bf34375..cc2e1847 100644 --- a/packages/cultcache-rs/tests/slot_readonly.rs +++ b/packages/cultcache-rs/tests/slot_readonly.rs @@ -337,14 +337,128 @@ fn the_variant_row_of_the_slot_store_is_refused_whole_by_the_store_reader() { assert!(format!("{error:#}").contains("is not decodable"), "{error:#}"); } -#[test] -fn a_held_payload_the_type_cannot_decode_carries_no_finding_and_stays_repairable() { - let mut cache = registered(); - // A text where the document's version belongs: not a record this type reads, marked or otherwise. - let broken = rmp_serde::to_vec(&("surface", "not-a-version")).unwrap(); - cache.put_raw_envelope(envelope("k", broken)).unwrap(); - assert!(cache.is_writable(UnionSurface::TYPE, "k").unwrap()); - cache.put("k", &constructed()).unwrap(); - assert!(cache.delete::("k").unwrap()); - assert!(cache.is_writable("tests.unregistered", "k").is_err()); +fn slot_unit_envelope(key: &str, payload: Vec) -> 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(&("newer", (99u64, ("x",)), "extra")).unwrap(); + cache.put_raw_envelope(slot_unit_envelope("n", newer)).unwrap(); + assert!(cache.get::("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::("n").unwrap_err(), + ] { + let message = format!("{error:#}"); + assert!(message.contains("\"n\"") && message.contains("tests.slot_unit") && message.contains("read-only"), "{message}"); + assert!(!message.contains("newer") && !message.contains("extra"), "{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::("broken").unwrap_err(); + let message = format!("{error:#}"); + assert!(message.contains("\"broken\"") && message.contains("repair"), "{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::().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::(&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::(&vector("depth-skipped-past-limit.msgpack")); + assert_eq!(cultcache_rs::__enter_slots(), 0); }