From 4f34c5b9e141f51998c657c3eeb949488753df27 Mon Sep 17 00:00:00 2001 From: Libby Date: Wed, 11 Nov 2026 15:14:44 +0000 Subject: [PATCH] cultcache-rs: retype refusal asks schema-id ownership; compare_and_swap_entry refuses in-place retype; push_all last-wins pinned; changelog for #110 The commit guard becomes settle_one_record_per_key: a written entry that owns the held record's schema id replaces it, another id is a SchemaConflictError. The unreachable same-type branch is gone. Co-Authored-By: Claude Sonnet 5.5 --- packages/cultcache-rs/CHANGELOG.md | 4 ++ packages/cultcache-rs/src/lib.rs | 102 ++++++++++++++++++++++++----- 2 files changed, 90 insertions(+), 16 deletions(-) diff --git a/packages/cultcache-rs/CHANGELOG.md b/packages/cultcache-rs/CHANGELOG.md index 15011491..86fe3eda 100644 --- a/packages/cultcache-rs/CHANGELOG.md +++ b/packages/cultcache-rs/CHANGELOG.md @@ -14,6 +14,10 @@ only in git history. - `bool` index and lookup text renders `True` / `False`, as C# does, where it rendered `true` / `false`. - `DatabaseEntry::UNIQUE_INDEXES` replaces `cult_indexes` as the one index declaration and changes type (its members are the unique aliases); a hand implementation overriding `cult_indexes` no longer compiles. - `register_entry_type` over a type that already holds records re-judges them under the new declaration (unique indexes and the global rule) and refuses, changing nothing, when they violate it. A declaration with an index also refuses when a held record cannot be decoded as the new type; a declaration without one does not read the records, so it does not. +- A single-file store holds one record per key. The v1 and legacy readers refuse a store with two records under one key (`CultCacheStoreUnreadable`, the record `Undecodable`): the store does not open, and a write over it refuses and leaves the file. A consumer that holds such a file must repair it with the runtime that wrote it before opening it here. +- Every single-file writer (`push`, `push_all`, `insert_entry_if_absent`, the compare-exchange and snapshot-append family, and the cache's puts and batches) refuses to put a second type under a key the store holds, with `SchemaConflictError` naming the key and both types, writing nothing. Retyping a key is a delete, then an add (one batch may hold both). A write owns the held record when its schema id equals the id the held record is stored under, as C# `WrittenOwns` asks; a type renamed under a stable id therefore replaces its own record, and a write under another id is refused. Registered types here declare no compatible ids, so equality is the whole test. Writers used to leave two records under the key. +- `compare_and_swap_entry` over a single file refuses an in-place retype (expected `A/k`, replacement `B/k` under another schema id) with the same `SchemaConflictError`, as `compare_exchange` does, where it used to retype the record and return `true`. A lost race still returns `false` first. +- A write of the same type under a held key replaces the record by key, and in one `push_all` the last of several same-type entries under a key is the record. The redb stores are unchanged and still key by type and key. ### Added diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index 8984cd29..440af704 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -1187,6 +1187,10 @@ impl SingleFileMessagePackBackingStore { if entries[index] != *expected { return Ok(false); } + // An in-place retype is a retype: the replacement must own the schema id the held record is stored under. + if schema_identity(&entries[index]) != schema_identity(&replacement) { + return Err(retype_conflict(&entries[index], &replacement)); + } entries[index] = replacement; entries.sort_by_key(entry_id); self.write_all_unlocked(&entries)?; @@ -1479,7 +1483,6 @@ impl SingleFileMessagePackBackingStore { } fn write_all_unlocked(&self, entries: &[CultCacheEnvelope]) -> Result<()> { - require_one_record_per_key(entries)?; let failed = |kind| CultCacheStoreWriteFailed { path: self.path.clone(), kind, @@ -1495,7 +1498,8 @@ impl SingleFileMessagePackBackingStore { Err(error) if io_not_found(&error) => DiskStore::absent(), other => other?, }; - let bytes = encode_store_snapshot(entries, &disk) + let entries = settle_one_record_per_key(entries, &disk)?; + let bytes = encode_store_snapshot(&entries, &disk) .and_then(|snapshot| rmp_serde::to_vec(&snapshot).context("failed to encode MessagePack")) .map_err(|error| error.context(failed(CultCacheStoreWriteFailedKind::Rejected)))?; self.replace_unlocked(&bytes) @@ -3738,23 +3742,47 @@ fn require_unique_keys(envelopes: &[CultCacheEnvelope]) -> Result<()> { Ok(()) } -/// The commit guard every single-file write passes: a store holds one record per key, so a snapshot about to be written -/// may not hold a key twice. Two types under a key are a retype, refused with the key and both types (retyping is -/// remove-then-add); the same type twice is a malformed snapshot. Nothing is written. -fn require_one_record_per_key(entries: &[CultCacheEnvelope]) -> Result<()> { - let mut held = HashMap::<&str, &CultCacheEnvelope>::new(); +/// What a record's stored schema id is: the id it carries, else its type (a registered type's id is its entry type). This +/// runtime's registered types declare no compatible ids, so a written record owns a held one exactly when the ids are equal: +/// the question C# `WrittenOwns` asks of the stored id, not a comparison of type names. +fn schema_identity(entry: &CultCacheEnvelope) -> &str { + entry.schema_id.as_deref().unwrap_or(&entry.r#type) +} + +/// The refusal of a write that would retype a key: the written record does not own the schema id the held one is stored under. +/// Retyping is remove-then-add (variants:ruling:cache-refuses-retype). +fn retype_conflict(held: &CultCacheEnvelope, written: &CultCacheEnvelope) -> anyhow::Error { + anyhow::Error::new(SchemaConflictError { + schema_id: schema_identity(written).to_string(), + schema_names: vec![held.r#type.clone(), written.r#type.clone()], + record_key: written.key.clone(), + }) +} + +/// The commit guard every single-file write passes: a store holds one record per key. Two entries under a key whose schema +/// ids differ are a retype and are refused with the key and both types, nothing written. Two under one schema id are one +/// record: a written entry replaces the held one it owns (a type renamed under a stable id), the held one being what the file +/// already carries. +fn settle_one_record_per_key(entries: &[CultCacheEnvelope], disk: &DiskStore) -> Result> { + let arrived: BTreeMap<(String, String), &CultCacheEnvelope> = disk.envelopes.iter().map(|entry| (entry_id(entry), entry)).collect(); + let held = |entry: &CultCacheEnvelope| arrived.get(&entry_id(entry)) == Some(&entry); + let mut settled = Vec::<&CultCacheEnvelope>::new(); + let mut at_key = HashMap::<&str, usize>::new(); for entry in entries { - let Some(first) = held.insert(entry.key.as_str(), entry) else { continue }; - if first.r#type == entry.r#type { - bail!("CultCache snapshot holds record {:?} of type {:?} more than once", entry.key, entry.r#type); + match at_key.get(entry.key.as_str()) { + None => { + at_key.insert(entry.key.as_str(), settled.len()); + settled.push(entry); + } + Some(&slot) if schema_identity(settled[slot]) != schema_identity(entry) => return Err(retype_conflict(settled[slot], entry)), + Some(&slot) => { + if !held(entry) || held(settled[slot]) { + settled[slot] = entry; + } + } } - return Err(anyhow::Error::new(SchemaConflictError { - schema_id: entry.schema_id.clone().unwrap_or_else(|| entry.r#type.clone()), - schema_names: vec![first.r#type.clone(), entry.r#type.clone()], - record_key: entry.key.clone(), - })); } - Ok(()) + Ok(settled.into_iter().cloned().collect()) } /// Decodes a store whose header the caller has already read as a readable one: its envelopes and its catalog. @@ -4372,6 +4400,7 @@ mod tests { ("push", |s, e, _| s.push(&e[2]).map(|()| true)), ("push_all", |s, e, _| s.push_all(&[e[0].clone(), e[2].clone()], PushAllOptions::default()).map(|()| true)), ("insert_entry_if_absent", |s, e, _| s.insert_entry_if_absent(e[2].clone())), + ("compare_and_swap_entry", |s, e, _| s.compare_and_swap_entry(&e[0], e[2].clone())), ("compare_exchange", |s, e, x| s.compare_exchange(std::slice::from_ref(x), &[e[2].clone()])), ("compare_and_swap_batch", |s, e, _| s.compare_and_swap_batch(&[], vec![e[2].clone()])), ("append_if_snapshot_unchanged", |s, e, _| s.append_if_snapshot_unchanged(&[e[0].clone(), e[1].clone()], vec![e[2].clone()])), @@ -6624,6 +6653,47 @@ mod tests { Ok(()) } + // The retype refusal asks what C# WrittenOwns asks: does the written record own the schema id the held one is stored under. + // A type renamed under a stable id owns it, so its write replaces the record; another id does not, and is refused. + #[test] + fn a_write_owning_the_stored_schema_id_replaces_the_record_under_a_new_name() -> Result<()> { + let envelope = |r#type: &str, schema_id: &str, payload: &[u8]| CultCacheEnvelope { + key: "a".into(), + r#type: r#type.into(), + payload: payload.to_vec(), + stored_at: "2026-09-30T00:00:00Z".into(), + schema_id: Some(schema_id.into()), + }; + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.msgpack"); + store_file(&path, &[("t.s", "t.old")], &[("a", "t.s", b"one".to_vec())])?; + let before = std::fs::read(&path)?; + let error = SingleFileMessagePackBackingStore::new(&path).push(&envelope("t.other", "t.other", b"x")).unwrap_err(); + assert_eq!(retype_refusal(&error).record_key, "a"); + assert_eq!(std::fs::read(&path)?, before); + + SingleFileMessagePackBackingStore::new(&path).push(&envelope("t.s", "t.s", b"two"))?; + let held = SingleFileMessagePackBackingStore::new(&path).pull_all()?; + assert_eq!(held.len(), 1); + assert_eq!((held[0].r#type.as_str(), held[0].payload.as_slice()), ("t.s", b"two".as_slice())); + Ok(()) + } + + // Of same-type entries sharing a key in one push_all the last one given is the record. + #[test] + fn push_all_keeps_the_last_of_two_same_type_entries_under_a_key() -> Result<()> { + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + let (first, _) = cache.prepare_entry("k", &settings_one())?; + let (last, _) = cache.prepare_entry("k", &Settings { theme: "red".to_string(), retries: 2 })?; + let temp = tempfile::tempdir()?; + let path = temp.path().join("store.cc"); + let mut store = SingleFileMessagePackBackingStore::new(&path); + store.push_all(&[first, last.clone()], PushAllOptions::default())?; + assert_eq!(store.pull_all()?.into_iter().map(|e| e.payload).collect::>(), vec![last.payload]); + Ok(()) + } + #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] #[cultcache(type = "tests.stable", schema = "tests.old_name")] struct StableId {