diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index 5b4eec96..597198cd 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -930,7 +930,8 @@ fn store_header(bytes: &[u8]) -> StoreHeader { } /// A write the catalog cannot describe: records of different types under one schema id. This writer derives one entry per carried -/// schema id from its records, so that is its only conflict. Nothing is written. It names the id, the two names and a record key. +/// schema id from its records, so that is its only conflict besides a single-file write that would put a second type under a +/// key the store already holds (the key is the record's identity there). Nothing is written. It names the id, the two names and a record key. #[derive(Debug)] pub struct SchemaConflictError { pub schema_id: String, @@ -1746,7 +1747,10 @@ impl CacheBackingStore for SingleFileMessagePackBackingStore { fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()> { self.with_exclusive_lock(|| { let mut entries = self.read_all_unlocked()?; - // A store holds one record per key whatever its type, so a write replaces the record under its key. + // A store holds one record per key: a write replaces the record of its own type under its key and refuses to retype it. + if let Some(held) = entries.iter().find(|candidate| candidate.key == entry.key) { + refuse_retype(held, entry)?; + } entries.retain(|candidate| candidate.key != entry.key); entries.push(entry.clone()); entries.sort_by_key(entry_id); @@ -1770,10 +1774,13 @@ impl CacheBackingStore for SingleFileMessagePackBackingStore { return Ok(()); } - // One record per key: of entries sharing a key the last one given is the record. + // One record per key: of same-type entries sharing a key the last one given is the record; a second type under a + // key in one batch is refused, as a push of it is. let mut by_key = BTreeMap::<&str, &CultCacheEnvelope>::new(); for entry in entries { - by_key.insert(entry.key.as_str(), entry); + if let Some(held) = by_key.insert(entry.key.as_str(), entry) { + refuse_retype(held, entry)?; + } } let mut entries: Vec = by_key.into_values().cloned().collect(); entries.sort_by_key(entry_id); @@ -3716,7 +3723,35 @@ impl DiskStore { /// The legacy envelope array: an array of envelope maps. Its acceptance is the one verdict on an array whose first slot /// is not a store header, for reading it and for replacing it. fn decode_legacy_envelopes(bytes: &[u8]) -> Result> { - decode_grown(bytes).context("failed to decode legacy CultCache envelope array") + let envelopes: Vec = + decode_grown(bytes).context("failed to decode legacy CultCache envelope array")?; + require_unique_keys(&envelopes)?; + Ok(envelopes) +} + +/// A record's identity in a store is its key string alone, compared ordinal: a store holding a key twice is unreadable, +/// whatever the format of the file that holds it. +fn require_unique_keys(envelopes: &[CultCacheEnvelope]) -> Result<()> { + let mut keys = HashSet::<&str>::new(); + for envelope in envelopes { + if !keys.insert(envelope.key.as_str()) { + bail!("CultCache record {:?} appears more than once", envelope.key); + } + } + Ok(()) +} + +/// A store holds one record per key, so a write may replace a record of its own type and may not retype it: retyping is +/// remove-then-add. The refusal names the key and both types; nothing is written. +fn refuse_retype(held: &CultCacheEnvelope, incoming: &CultCacheEnvelope) -> Result<()> { + if held.r#type == incoming.r#type { + return Ok(()); + } + Err(anyhow::Error::new(SchemaConflictError { + schema_id: incoming.schema_id.clone().unwrap_or_else(|| incoming.r#type.clone()), + schema_names: vec![held.r#type.clone(), incoming.r#type.clone()], + record_key: incoming.key.clone(), + })) } /// Decodes a store whose header the caller has already read as a readable one: its envelopes and its catalog. @@ -3755,13 +3790,7 @@ fn decode_store_snapshot(bytes: &[u8]) -> Result<(Vec, Vec>>()?; - // A record's identity is its key string alone, compared ordinal: a store holding a key twice is unreadable. - let mut keys = HashSet::<&str>::new(); - for envelope in &envelopes { - if !keys.insert(envelope.key.as_str()) { - bail!("CultCache record {:?} appears more than once", envelope.key); - } - } + require_unique_keys(&envelopes)?; Ok((envelopes, snapshot.1)) } @@ -4233,32 +4262,114 @@ mod tests { Ok(()) } + fn settings_one() -> Settings { + Settings { theme: "green".to_string(), retries: 1 } + } + + fn note_one() -> Note { + Note { title: "same key".to_string(), body: "different type".to_string() } + } + + fn retype_refusal(error: &anyhow::Error) -> &SchemaConflictError { + error.downcast_ref::().unwrap_or_else(|| panic!("{error:#}")) + } + + fn reloaded(path: &Path) -> Result> { + SingleFileMessagePackBackingStore::new(path).pull_all() + } + #[test] - fn entry_identity_is_polymorphic_by_type_and_key() -> Result<()> { + fn a_second_type_under_a_stored_key_is_refused_and_the_store_and_cache_stay() -> Result<()> { let temp = tempfile::tempdir()?; let store_path = temp.path().join("cache.cc"); let mut cache = CultCache::new(); cache.register_registry(TestEntries)?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; - cache.put( - "shared", - &Settings { - theme: "green".to_string(), - retries: 1, - }, - )?; - cache.put( - "shared", - &Note { - title: "same key".to_string(), - body: "different type".to_string(), - }, - )?; - - assert_eq!(cache.snapshot().len(), 2); - assert_eq!(cache.get_required::("shared")?.title, "same key"); + cache.put("shared", &settings_one())?; + let before = std::fs::read(&store_path)?; + let error = cache.put("shared", ¬e_one()).unwrap_err(); + assert_eq!(retype_refusal(&error).record_key, "shared"); + assert_eq!(std::fs::read(&store_path)?, before, "the refused write changed the file"); + assert_eq!(cache.snapshot().len(), 1); assert_eq!(cache.get_required::("shared")?.theme, "green"); + assert!(cache.get::("shared")?.is_none()); + + // Retyping is remove-then-add, and what was written reads back. + assert!(cache.delete::("shared")?); + cache.put("shared", ¬e_one())?; + let held = reloaded(&store_path)?; + assert_eq!(held.len(), 1); + assert_eq!(held[0].r#type, Note::TYPE); + + // The same type under the key replaces it. + cache.put("shared", &Note { title: "again".to_string(), body: String::new() })?; + let held = reloaded(&store_path)?; + assert_eq!(held.len(), 1); + assert_eq!(cache.get_required::("shared")?.title, "again"); + Ok(()) + } + + #[test] + fn a_batch_holding_a_second_type_under_a_key_is_refused_whole() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("batch.cc"); + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + cache.put("kept", &settings_one())?; + let before = std::fs::read(&store_path)?; + + let (settings, _) = cache.prepare_entry("shared", &settings_one())?; + let (note, _) = cache.prepare_entry("shared", ¬e_one())?; + let error = cache.put_prepared_batch(vec![settings.clone(), note.clone()]).unwrap_err(); + assert_eq!(retype_refusal(&error).record_key, "shared"); + assert_eq!(std::fs::read(&store_path)?, before); + assert_eq!(cache.snapshot().len(), 1); + + // The store's own batch write refuses it too, in either order. + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + for batch in [[settings.clone(), note.clone()], [note, settings]] { + let error = store.push_all(&batch, PushAllOptions::default()).unwrap_err(); + assert_eq!(retype_refusal(&error).record_key, "shared"); + assert_eq!(std::fs::read(&store_path)?, before); + } + Ok(()) + } + + #[test] + fn the_stores_own_push_refuses_a_second_type_under_a_key() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("push.cc"); + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + let (settings, _) = cache.prepare_entry("shared", &settings_one())?; + let (note, _) = cache.prepare_entry("shared", ¬e_one())?; + let mut store = SingleFileMessagePackBackingStore::new(&store_path); + store.push(&settings)?; + let before = std::fs::read(&store_path)?; + let error = store.push(¬e).unwrap_err(); + assert_eq!(retype_refusal(&error).schema_names, vec![Settings::TYPE.to_string(), Note::TYPE.to_string()]); + assert_eq!(std::fs::read(&store_path)?, before); + Ok(()) + } + + #[test] + fn a_legacy_array_with_a_key_under_two_types_is_not_decodable() -> Result<()> { + let temp = tempfile::tempdir()?; + let store_path = temp.path().join("legacy.cc"); + let record = |kind: &str| { + CultCacheEnvelope { + key: "a".to_string(), + r#type: kind.to_string(), + payload: rmp_serde::to_vec(&Vec::::new()).unwrap(), + stored_at: "2026-01-01T00:00:00Z".to_string(), + schema_id: None, + } + }; + std::fs::write(&store_path, rmp_serde::to_vec_named(&vec![record("old.one"), record("old.two")])?)?; + let error = SingleFileMessagePackBackingStore::new(&store_path).pull_all().unwrap_err(); + assert_eq!(unreadable_kind(&error), Some(CultCacheStoreUnreadableKind::Undecodable), "{error:#}"); Ok(()) } @@ -5922,9 +6033,9 @@ mod tests { } - // The writer keeps one record per key: a write of another type to a key replaces the record, so the file it leaves reads. + // The writer keeps one record per key: another type under a held key is refused and the file stays; the same type replaces. #[test] - fn a_write_of_another_type_under_an_existing_key_replaces_the_record() -> Result<()> { + fn a_write_of_another_type_under_an_existing_key_is_refused_and_the_same_type_replaces() -> Result<()> { let temp = tempfile::tempdir()?; let path = temp.path().join("store.msgpack"); let mut store = SingleFileMessagePackBackingStore::new(&path); @@ -5932,13 +6043,13 @@ mod tests { other.r#type = "snapshot.other".to_string(); other.schema_id = Some("snapshot.other".to_string()); store.push(&snapshot_envelope("k", b"one"))?; - store.push(&other)?; + let before = std::fs::read(&path)?; + assert!(store.push(&other).unwrap_err().downcast_ref::().is_some()); + assert!(store.push_all(&[snapshot_envelope("k", b"one"), other.clone()], PushAllOptions::default()).is_err()); + assert_eq!(std::fs::read(&path)?, before); + store.push(&snapshot_envelope("k", b"three"))?; let read = store.pull_all()?; - assert_eq!(read.len(), 1); - assert_eq!(read[0].r#type, "snapshot.other"); - store.push_all(&[snapshot_envelope("k", b"one"), other.clone()], PushAllOptions::default())?; - let read = store.pull_all()?; - assert_eq!((read.len(), read[0].payload.as_slice()), (1, b"two".as_slice())); + assert_eq!((read.len(), read[0].payload.as_slice()), (1, b"three".as_slice())); Ok(()) }