From c37183273699e3f1e15ff7ea4597d0132c3e44e4 Mon Sep 17 00:00:00 2001 From: Libby Date: Thu, 12 Nov 2026 01:05:46 +0000 Subject: [PATCH] Rust cache: one record per key, foreign carry, key-identity vector walked (#123) --- docs/write-ownership-map.md | 26 +- packages/cultcache-rs/CHANGELOG.md | 2 + packages/cultcache-rs/README.md | 12 +- packages/cultcache-rs/src/lib.rs | 497 ++++++++++++++++-- .../cultcache-rs/tests/key_identity_vector.rs | 283 ++++++++++ packages/cultcache-rs/tests/unique_index.rs | 6 +- tests/vectors/key-identity/README.md | 3 + 7 files changed, 776 insertions(+), 53 deletions(-) create mode 100644 packages/cultcache-rs/tests/key_identity_vector.rs diff --git a/docs/write-ownership-map.md b/docs/write-ownership-map.md index 3bce4ad2..44452dc6 100644 --- a/docs/write-ownership-map.md +++ b/docs/write-ownership-map.md @@ -33,8 +33,8 @@ Probes: scratch tests in this worktree, removed. C# filtered runs on Starfire (o | Probe | C# | TypeScript | Rust | Python | |---|---|---|---|---| -| P1 cache {A,B}: B at k, then A at k | **C**: B's record replaced, k now A's id | R `SchemaConflictError` (cultcache-ts: one record per key) | C, both live at k | R `SchemaConflictError`, A's write refused, B kept (was C, both live) | -| P2 cache {A} only, store holds B at k, write A at k | R `CultSchemaConflictException` (B is foreign, carried) | R `SchemaConflictError` at write (foreign loads, carried; cultcache-ts) | R at load, same | R `SchemaConflictError` (B is foreign, carried; the load admits it; was R at load) | +| P1 cache {A,B}: B at k, then A at k | **C**: B's record replaced, k now A's id | R `SchemaConflictError` (cultcache-ts: one record per key) | R `SchemaConflictError` naming k, A not written (leaf cut) | R `SchemaConflictError`, A's write refused, B kept (was C, both live) | +| P2 cache {A} only, store holds B at k, write A at k | R `CultSchemaConflictException` (B is foreign, carried) | R `SchemaConflictError` at write (foreign loads, carried; cultcache-ts) | R `SchemaConflictError` (B loads as foreign, carried; leaf cut) | R `SchemaConflictError` (B is foreign, carried; the load admits it; was R at load) | | P3 cacheless `WriteDocument` over B, registry null / {A} / {A,B} | R / R / R | no cacheless write | no cacheless write | no cacheless write | | P4 cache {A,B}: remove k and upsert A at k in one batch | C (k is A's) | n/a | n/a | n/a | | P5 cache {A,B}: remove, then add in a second commit | C | n/a | n/a | n/a | @@ -55,12 +55,22 @@ What this establishes: - Retype as remove plus add already works in the cache (P4, P5); the cut adds no new primitive. - Directory store: its own judge (`DirectoryMessagePackBackingStore.cs:287-306`) asks `Registry.Owns` for every key, upserts included. It must take the same per-key rule. -- TypeScript, Rust, Python: records are keyed by `(type, key)` (Rust `entries: BTreeMap<(String, String), _>`, - `lib.rs:2659`; TS `#entryId(definition.type, key)`, `cult-cache.ts:323`; Python sorts and stores by `(type, key)`). - A write of A at k never meets B's record: both live at k (P1). None carries a foreign record; a store holding an - unregistered type does not load (P2). So no leaf has a "write meets another type's record" path to fix. Bringing the - ruling to the leaves first needs C#'s key identity (one record per key) and C#'s foreign carry, which are parity changes - of the leaves' store model, not a write-ownership rule. Owned by the new follow-up `variants:follow_up:leaf-key-identity-parity`. +- TypeScript and Python: records are keyed by `(type, key)` (TS `#entryId(definition.type, key)`, `cult-cache.ts:323`; + Python sorts and stores by `(type, key)`). A write of A at k never meets B's record: both live at k (P1). Neither carries + a foreign record; a store holding an unregistered type does not load (P2). So neither has a "write meets another type's + record" path to fix: bringing the ruling to them first needs C#'s key identity (one record per key) and C#'s foreign + carry, parity changes of their store model, not a write-ownership rule (follow-up `variants:follow_up:leaf-key-identity-parity`). +- Rust is that follow-up's first leaf cut (`cut-leaf-key-identity-rs`). The cache keeps its `(type, key)` map; one judge + (`refuse_retype`) refuses another registered type at a key the cache holds, or any write at a key a store carries + foreign, on every write path, pull and SoA load, comparing types, not the ids envelopes claim. At a key the cache does not + hold (a fresh key, or one another writer filled after this cache's pull) the single-file store's commit guard judges by + the stored and claimed schema ids, so an envelope claiming another registered type's id is refused earlier, at every + envelope admission (`refuse_claimed_id`, the pull's own `resolve_registered_type` lookup); an id naming no registered type + passes, trusted as claimed: an envelope under such an id can replace another writer's record stored under it at a key + this cache does not hold, or be served as another type by a reader that registers the type the id names (Soul g8a, g8c, + open: which ids a Rust type owns is not declared to CultCache). Store-level writers called directly (`push`, + `insert_entry_if_absent`, compare-and-swap) trust the claimed id. + The cache carries an unregistered type's records as foreign and walks the vector's 12 rows. - Interop (unproven): a Python store holding A and B at k fails C# decode first on payload encoding (Python's lambda-defined document wrote JSON payloads), so what C# does with two records at one key was not reached. diff --git a/packages/cultcache-rs/CHANGELOG.md b/packages/cultcache-rs/CHANGELOG.md index c32773d9..ed7cd60a 100644 --- a/packages/cultcache-rs/CHANGELOG.md +++ b/packages/cultcache-rs/CHANGELOG.md @@ -17,10 +17,12 @@ only in git history. - 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 of another type name under another id is refused (`push` of the same type name under another id still replaces the record, as before; `compare_and_swap_entry`, and a `push_all` batch holding one type under two ids, refuse it). 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 cache holds one record per key. `put`, `put_envelope`, `put_raw_envelope` and a batch of a second type under a key the cache holds, or a key a store carries foreign, are refused with `SchemaConflictError` even with no store (a memory-only cache used to hold both), whatever schema id the envelope claims: another registered type's record is never owned. `put_envelope`, `put_raw_envelope`, `put_prepared_batch` and `load_soa` also refuse, at any key, an envelope whose schema id names another type registered in this cache (the id a pull would resolve the record by). An id this cache does not register is trusted as the envelope claims it: at a key this cache does not hold, the single-file store decides ownership by that id, so such an envelope can replace another writer's record stored under it, and a reader registering the type the id names serves the record as that type. Store-level writers called directly (`push`, `insert_entry_if_absent`, compare-and-swap) still trust the claimed id. A refused write or pull changes nothing. Two stores holding one key refuse the pull, naming the key and both types, and so does one store holding two records at a key that resolve to one type. A store holding a record no registered type claims now loads (it used to refuse with "No schema is registered for persisted entry type") and the record is carried, through `put_prepared_batch` too, which lays foreign records back byte for byte; `delete` at a foreign key is refused. `load_soa` refuses a second type at a key or a record at a foreign key, and still refuses an unregistered type. The redb stores still key by `(type, key)`, and their own writers (`push`, `insert_entry_if_absent`, compare-and-swap) still put a second type at a held key: a cache now refuses to pull such a redb store with `SchemaConflictError`, where it used to load both records. - A write of the same type under a held key replaces the record by key, and in one `push_all` the last given of several entries under a key and a schema id is the record, whatever their type names. The redb stores are unchanged and still key by type and key. ### Added +- `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::()`. diff --git a/packages/cultcache-rs/README.md b/packages/cultcache-rs/README.md index 45724843..8a8b8f5e 100644 --- a/packages/cultcache-rs/README.md +++ b/packages/cultcache-rs/README.md @@ -21,8 +21,14 @@ CultCache is a domain cache with persistence adapters. - `CultCache` is the query and mutation surface. - Domain structs implement `DatabaseEntry`. -- Entries are stored behind a `type::key` identity, so multiple entry types - can share a logical key without colliding. +- A cache holds one record per key whatever its type. Records are held per + `(type, key)`, but a write of a second type under a key the cache holds, or a + key a store carries foreign, is refused with `SchemaConflictError` naming the + key (no store needed); retyping a key is a delete, then an add. A store + record no registered type claims loads: the cache does not serve it, lists it + in `foreign_records()`, lays it back byte for byte through every write + including `put_prepared_batch`, and refuses a write or delete at its key. + Registering a type that claims its schema id picks it up at the next pull. - Each payload is encoded directly from the known `DatabaseEntry` type into MessagePack bytes. - Backing stores are adapters, not the public data model. @@ -458,7 +464,7 @@ That is the Rust version of the original trick: - C# refuses to store objects outside the `DatabaseEntry` root type. - Runtime-visible `DatabaseEntry` subclasses get dedicated formatters. -- Unknown or unregistered types are rejected. +- A write of an unregistered type is rejected; a stored record of an unregistered type is carried as foreign, not decoded. - MessagePack payload bytes deserialize through known concrete formatters, not an open-world type loader or generic structured-value bridge. diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index f7b8cae0..686547f4 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -2750,6 +2750,18 @@ pub struct CultCache { /// values. Registration data, so the judge decodes only indexed types. projectors: BTreeMap, stores: Vec, + /// 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)>, +} + +/// A record a store holds that no registered type claims: carried, never served. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ForeignRecord { + pub key: String, + pub schema_id: String, + pub schema_name: String, + pub stored_at: String, } type IndexProjector = fn(&[u8]) -> Result>; @@ -2803,6 +2815,7 @@ impl CultCache { index_holders: BTreeMap::new(), projectors: BTreeMap::new(), stores: Vec::new(), + foreign: Vec::new(), } } @@ -2940,6 +2953,7 @@ impl CultCache { /// only when all of them loaded, so a refused load admits nothing. pub fn pull_all_backing_stores(&mut self) -> Result<()> { let mut loaded = BTreeMap::new(); + let mut foreign = Vec::new(); let known_types: BTreeSet = self.definitions.keys().cloned().collect(); let schema_name_definitions = self.schema_name_definitions.clone(); for (index, registration) in self.stores.iter().enumerate() { @@ -2950,10 +2964,10 @@ impl CultCache { entry.schema_id.as_deref(), &entry.r#type, ) else { - return Err(anyhow!( - "No schema is registered for persisted entry type {:?}", - entry.r#type - )); + // No registered type claims the record: the cache carries it, never serves it. + self.refuse_retype(&loaded, &foreign, &entry)?; + foreign.push((index, entry)); + continue; }; let home = self.resolve_route_index(&canonical_type); if home != Some(index) { @@ -2966,16 +2980,39 @@ impl CultCache { )); } entry.r#type = canonical_type; + self.refuse_retype(&loaded, &foreign, &entry)?; + // Two records of a store resolving to one type at one key (one stamped with the other's id): one record per key. + if let Some(held) = loaded.get(&entry_id(&entry)) { + return Err(retype_conflict(held, &entry)); + } loaded.insert(entry_id(&entry), entry); } } self.refuse_second_globals(&loaded)?; let holders = self.judge_image(&loaded)?; self.entries = loaded; + self.foreign = foreign; self.index_holders = holders; Ok(()) } + /// The records the last pull found in a store that no registered type claims, sorted by key (UTF-16 order). The cache does not + /// serve them; it carries them through every write and refuses a write or removal at their keys. + pub fn foreign_records(&self) -> Vec { + let mut records: Vec = self + .foreign + .iter() + .map(|(_, entry)| ForeignRecord { + key: entry.key.clone(), + schema_id: schema_identity(entry).to_string(), + schema_name: entry.r#type.clone(), + stored_at: entry.stored_at.clone(), + }) + .collect(); + records.sort_by(|left, right| utf16_cmp(&left.key, &right.key)); + records + } + /// Returns the exact envelopes in the currently loaded cache image. /// Callers that need optimistic concurrency can validate typed documents /// and capture CAS expectations from one coherent read. @@ -3130,6 +3167,7 @@ impl CultCache { pub fn put(&mut self, key: impl Into, value: &T) -> Result { let (entry, parsed) = self.prepare_entry(key, value)?; 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])?; if let Some(home) = self.home_index(T::TYPE)? { self.stores[home].store.push(&entry)?; @@ -3228,6 +3266,7 @@ impl CultCache { )); } require_payload_depth(entry)?; + self.refuse_claimed_id(entry)?; let Some(home) = self.home_index(&entry.r#type)? else { continue; }; @@ -3244,13 +3283,18 @@ impl CultCache { next.insert(entry_id(entry), entry.clone()); } self.refuse_second_globals(&next)?; + for entry in &staged { + self.refuse_retype(&next, &self.foreign, entry)?; + } let change = self.judge_unique(&staged.iter().collect::>())?; if let Some(store_index) = store_index { - // The home store receives its own records only: those it already - // holds in this cache plus the staged ones. + // The home store receives its own records only: those it already holds in this cache plus the staged ones, and + // the records it carries that no registered type claims: push_all replaces the store's records, so a foreign + // record is laid back as the store read it. let snapshot: Vec<_> = next .values() .filter(|entry| self.resolve_route_index(&entry.r#type) == Some(store_index)) + .chain(self.foreign.iter().filter(|(home, _)| *home == store_index).map(|(_, entry)| entry)) .cloned() .collect(); self.stores[store_index] @@ -3264,7 +3308,9 @@ impl CultCache { pub fn put_envelope(&mut self, entry: CultCacheEnvelope) -> Result { let parsed = self.validate_envelope::(&entry)?; + self.refuse_claimed_id(&entry)?; 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])?; if let Some(home) = self.home_index(T::TYPE)? { self.stores[home].store.push(&entry)?; @@ -3299,7 +3345,9 @@ impl CultCache { )); } require_payload_depth(&entry)?; + self.refuse_claimed_id(&entry)?; 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])?; if let Some(home) = self.home_index(&entry.r#type)? { self.stores[home].store.push(&entry)?; @@ -3354,6 +3402,9 @@ impl CultCache { self.require_entry_type::()?; 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) { + return Err(foreign_conflict(foreign, T::TYPE)); + } return Ok(false); }; let released = self.projected_slots(&entry)?; @@ -3387,6 +3438,8 @@ impl CultCache { )); } self.home_index(&entry.r#type)?; + self.refuse_claimed_id(&entry)?; + self.refuse_retype(&loaded, &self.foreign, &entry)?; loaded.insert(entry_id(&entry), entry); } self.refuse_second_globals(&loaded)?; @@ -3509,6 +3562,46 @@ impl CultCache { } } + /// The one key-identity judge (C# `RefuseUndeclared` and the cache's held-owned check): a cache holds one record per key + /// whatever its type. `written` may not land at a key a store carries foreign, nor at a key another registered type holds + /// in `image`. C# `WrittenOwns` lets a write replace a held record its own type's id owns; a registered type's id here is + /// its entry type and declares no compatible ids, so another type's record is never owned, whatever schema id either + /// envelope claims. Refused with a typed [`SchemaConflictError`] naming the key and both types, never a value; nothing is + /// changed. The single-file store's commit guard judges the same question for the file on disk (another writer's file) + /// and stays. + fn refuse_retype( + &self, + image: &BTreeMap<(String, String), CultCacheEnvelope>, + foreign: &[(usize, CultCacheEnvelope)], + written: &CultCacheEnvelope, + ) -> Result<()> { + if let Some((_, carried)) = foreign.iter().find(|(_, carried)| carried.key == written.key) { + return Err(foreign_conflict(carried, &written.r#type)); + } + for type_id in self.definitions.keys().filter(|type_id| **type_id != written.r#type) { + if let Some(held) = image.get(&(type_id.clone(), written.key.clone())) { + return Err(retype_conflict(held, written)); + } + } + Ok(()) + } + + /// An admitted envelope's schema id may not name another registered type: the pull resolves a record's type from that id + /// (`resolve_registered_type`), so admitting it would store one type that every reload serves as the other, at a key this + /// cache may not hold (another writer's record, or a fresh key). Admission asks the pull's own lookup. An id that names no + /// registered type (a dynamic or versioned id) is the envelope's own. Store-level writers called directly trust the id. + fn refuse_claimed_id(&self, entry: &CultCacheEnvelope) -> Result<()> { + let known_types: BTreeSet = self.definitions.keys().cloned().collect(); + match resolve_registered_type(&known_types, &self.schema_name_definitions, entry.schema_id.as_deref(), &entry.r#type) { + Some(resolved) if resolved != entry.r#type => Err(anyhow::Error::new(SchemaConflictError { + schema_id: schema_identity(entry).to_string(), + schema_names: vec![resolved, entry.r#type.clone()], + record_key: entry.key.clone(), + })), + _ => Ok(()), + } + } + fn refuse_second_globals(&self, entries: &BTreeMap<(String, String), CultCacheEnvelope>) -> Result<()> { for (type_id, key) in entries.keys() { self.refuse_second_global(entries, type_id, key)?; @@ -3742,13 +3835,20 @@ fn require_unique_keys(envelopes: &[CultCacheEnvelope]) -> Result<()> { Ok(()) } -/// 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. +/// What a record's stored schema id is: the id it carries, else its type (a registered type's id is its entry type). fn schema_identity(entry: &CultCacheEnvelope) -> &str { entry.schema_id.as_deref().unwrap_or(&entry.r#type) } +/// The refusal of a write at a key a store carries as a record no registered type claims. +fn foreign_conflict(carried: &CultCacheEnvelope, written_type: &str) -> anyhow::Error { + anyhow::Error::new(SchemaConflictError { + schema_id: schema_identity(carried).to_string(), + schema_names: vec![carried.r#type.clone(), written_type.to_string()], + record_key: carried.key.clone(), + }) +} + /// 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 { @@ -4309,6 +4409,347 @@ mod tests { SingleFileMessagePackBackingStore::new(path).pull_all() } + const SECRET: &str = "payload-text-a-refusal-must-not-echo"; + + fn store_with(path: &Path, records: impl FnOnce(&mut CultCache) -> Result<()>) -> Result<()> { + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path))?; + records(&mut cache) + } + + #[test] + fn retype_memory_cache_refuses_a_second_type_on_every_write_path_and_keeps_the_first() -> Result<()> { + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + cache.put("shared", &settings_one())?; + let before = cache.snapshot(); + let secret = Note { title: SECRET.to_string(), body: SECRET.to_string() }; + let (staged, _) = cache.prepare_entry("shared", &secret)?; + + let mut errors = vec![ + cache.put("shared", &secret).unwrap_err(), + cache.put_envelope::(staged.clone()).unwrap_err(), + cache.put_raw_envelope(staged.clone()).unwrap_err(), + cache.put_prepared_batch(vec![staged.clone()]).unwrap_err(), + ]; + // Two types staged at one key in one batch never land, even at a fresh key. + let (fresh_note, _) = cache.prepare_entry("fresh", &secret)?; + let (fresh_settings, _) = cache.prepare_entry("fresh", &settings_one())?; + errors.push(cache.put_prepared_batch(vec![fresh_note, fresh_settings]).unwrap_err()); + for (error, key) in errors.iter().zip(["shared", "shared", "shared", "shared", "fresh"]) { + assert_eq!(retype_refusal(error).record_key, key); + assert!(!format!("{error:#}").contains(SECRET), "a refusal never echoes a value: {error:#}"); + } + assert_eq!(cache.snapshot(), before); + cache.delete::("shared")?; + cache.put("shared", &secret)?; + Ok(()) + } + + /// Two types whose envelopes carry one schema id (an id naming no registered type, so admission passes it) are still two + /// types: ownership is the written type's own id, never the id an envelope claims (C# `WrittenOwns` reads the written + /// entry's catalog entry). Refused at a key the other type holds on every envelope path, with and without a store, and in + /// an SoA image. + #[test] + fn retype_envelopes_sharing_one_claimed_id_are_still_two_types() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("cache.cc"); + for stored in [false, true] { + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + if stored { + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path))?; + cache.pull_all_backing_stores()?; + } + let (mut settings, _) = cache.prepare_entry("shared", &settings_one())?; + settings.schema_id = Some("dynamic.shared.v1".to_string()); + cache.put_raw_envelope(settings.clone())?; + let before = (cache.snapshot(), if stored { fs::read(&path)? } else { Vec::new() }); + let (mut stamped, _) = cache.prepare_entry("shared", ¬e_one())?; + stamped.schema_id = settings.schema_id.clone(); + for error in [ + cache.put_envelope::(stamped.clone()).unwrap_err(), + cache.put_raw_envelope(stamped.clone()).unwrap_err(), + cache.put_prepared_batch(vec![stamped.clone()]).unwrap_err(), + ] { + assert_eq!(retype_refusal(&error).record_key, "shared", "stored {stored}"); + } + assert_eq!((cache.snapshot(), if stored { fs::read(&path)? } else { Vec::new() }), before, "stored {stored}"); + + let error = cache.load_soa(CultCacheSoaTable::from_envelopes(vec![settings, stamped])).unwrap_err(); + assert_eq!(retype_refusal(&error).record_key, "shared"); + assert_eq!(cache.snapshot(), before.0); + } + Ok(()) + } + + /// An envelope whose schema id names another registered type is refused at admission, whether or not this cache holds the + /// key: a stale cache (another writer put Settings at the key after this pull) and a fresh key both, on every envelope path + /// and in an SoA image, the store unchanged. The pull resolves a record's type from that id, so admitting it would store a + /// Note that every reload serves as Settings. + #[test] + fn retype_an_envelope_claiming_another_registered_types_id_is_refused_at_admission() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("cache.cc"); + let open = || -> Result { + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path))?; + cache.pull_all_backing_stores()?; + Ok(cache) + }; + let mut stale = open()?; + open()?.put("held-elsewhere", &settings_one())?; + let before = fs::read(&path)?; + for key in ["held-elsewhere", "fresh"] { + let (mut stamped, _) = stale.prepare_entry(key, ¬e_one())?; + stamped.schema_id = Some(Settings::TYPE.to_string()); + for error in [ + stale.put_envelope::(stamped.clone()).unwrap_err(), + stale.put_raw_envelope(stamped.clone()).unwrap_err(), + stale.put_prepared_batch(vec![stamped.clone()]).unwrap_err(), + stale.load_soa(CultCacheSoaTable::from_envelopes(vec![stamped.clone()])).unwrap_err(), + ] { + let refusal = retype_refusal(&error); + assert_eq!((refusal.record_key.as_str(), refusal.schema_id.as_str()), (key, Settings::TYPE), "{error:#}"); + } + } + assert!(stale.snapshot().is_empty()); + assert_eq!(fs::read(&path)?, before, "nothing reached the store"); + assert_eq!(open()?.get_required::("held-elsewhere")?, settings_one()); + Ok(()) + } + + /// A store that holds two records at one key, one stamped with the other's schema id, is refused at the pull: both resolve + /// to one type, and the cache would otherwise keep whichever came last. + #[test] + fn retype_a_store_holding_two_records_at_one_key_refuses_the_pull() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("cache.redb"); + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + let (settings, _) = cache.prepare_entry("shared", &settings_one())?; + let (mut stamped, _) = cache.prepare_entry("shared", ¬e_one())?; + stamped.schema_id = Some(Settings::TYPE.to_string()); + { + let mut store = RedbMessagePackBackingStore::new(&path)?; + store.push(&settings)?; + store.push(&stamped)?; + } + cache.add_generic_backing_store(RedbMessagePackBackingStore::new(&path)?)?; + let error = cache.pull_all_backing_stores().unwrap_err(); + assert_eq!(retype_refusal(&error).record_key, "shared"); + assert!(cache.snapshot().is_empty()); + Ok(()) + } + + #[test] + fn retype_two_stores_holding_one_key_refuse_the_pull_and_admit_nothing() -> Result<()> { + let temp = tempfile::tempdir()?; + let (first, second) = (temp.path().join("first.cc"), temp.path().join("second.cc")); + store_with(&first, |cache| cache.put("k", &settings_one()).map(drop))?; + store_with(&second, |cache| cache.put("k", ¬e_one()).map(drop))?; + let (first_bytes, second_bytes) = (fs::read(&first)?, fs::read(&second)?); + + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + cache.add_backing_store(SingleFileMessagePackBackingStore::new(&first), [Settings::TYPE])?; + cache.pull_all_backing_stores()?; + let before = cache.snapshot(); + cache.add_backing_store(SingleFileMessagePackBackingStore::new(&second), [Note::TYPE])?; + let error = cache.pull_all_backing_stores().unwrap_err(); + assert_eq!(retype_refusal(&error).record_key, "k"); + assert_eq!(cache.snapshot(), before, "a refused pull admits nothing"); + assert_eq!((fs::read(&first)?, fs::read(&second)?), (first_bytes, second_bytes)); + Ok(()) + } + + #[test] + fn retype_a_store_with_an_unregistered_type_loads_lists_it_and_keeps_its_bytes() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("cache.cc"); + store_with(&path, |cache| { + cache.put("held", &settings_one())?; + cache.put("carried", ¬e_one()).map(drop) + })?; + let carried = |path: &Path| -> Result { + Ok(reloaded(path)?.into_iter().find(|entry| entry.key == "carried").unwrap()) + }; + let original = carried(&path)?; + + for as_batch in [false, true] { + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path))?; + cache.pull_all_backing_stores()?; + let foreign = cache.foreign_records(); + assert_eq!(foreign.len(), 1); + assert_eq!((foreign[0].key.as_str(), foreign[0].schema_id.as_str(), foreign[0].schema_name.as_str()), ("carried", Note::TYPE, Note::TYPE)); + assert_eq!(cache.snapshot().len(), 1, "a foreign record is listed, not served"); + + if as_batch { + let (staged, _) = cache.prepare_entry("other", &settings_one())?; + cache.put_prepared_batch(vec![staged])?; + } else { + cache.put("other", &settings_one())?; + } + assert!(cache.delete::("other")?); + assert_eq!(carried(&path)?, original, "the foreign record keeps its bytes (batch {as_batch})"); + + let secret = Settings { theme: SECRET.to_string(), retries: 9 }; + let before = fs::read(&path)?; + let mut refusals = vec![cache.put("carried", &secret).unwrap_err(), cache.delete::("carried").unwrap_err()]; + let (staged, _) = cache.prepare_entry("carried", &secret)?; + refusals.push(cache.put_prepared_batch(vec![staged]).unwrap_err()); + for error in &refusals { + assert_eq!(retype_refusal(error).record_key, "carried"); + assert!(!format!("{error:#}").contains(SECRET), "{error:#}"); + } + assert_eq!(fs::read(&path)?, before); + } + Ok(()) + } + + #[test] + fn retype_a_foreign_key_in_another_store_refuses_the_pull_and_every_write_to_it() -> Result<()> { + let temp = tempfile::tempdir()?; + let (routed, generic) = (temp.path().join("routed.cc"), temp.path().join("generic.cc")); + store_with(&routed, |cache| cache.put("k", &settings_one()).map(drop))?; + store_with(&generic, |cache| cache.put("n", ¬e_one()).map(drop))?; + let open = || -> Result { + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.add_backing_store(SingleFileMessagePackBackingStore::new(&routed), [Settings::TYPE])?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&generic))?; + Ok(cache) + }; + + // A registered record in one store and a foreign one at the same key in another: the pull is refused. + store_with(&generic, |cache| cache.put("k", ¬e_one()).map(drop))?; + let mut cache = open()?; + assert_eq!(retype_refusal(&cache.pull_all_backing_stores().unwrap_err()).record_key, "k"); + assert!(cache.snapshot().is_empty() && cache.foreign_records().is_empty()); + + // The foreign key lives in the generic store; a write routed to the other store is refused by the cache alone. + fs::remove_file(&generic)?; + store_with(&generic, |cache| cache.put("n", ¬e_one()).map(drop))?; + let mut cache = open()?; + cache.pull_all_backing_stores()?; + let before = (cache.snapshot(), fs::read(&routed)?, fs::read(&generic)?); + let (staged, _) = cache.prepare_entry("n", &settings_one())?; + for error in [cache.put("n", &settings_one()).unwrap_err(), cache.put_prepared_batch(vec![staged]).unwrap_err()] { + assert_eq!(retype_refusal(&error).record_key, "n"); + } + assert_eq!((cache.snapshot(), fs::read(&routed)?, fs::read(&generic)?), before); + Ok(()) + } + + /// A batch lays back only its home store's foreign records: another store's are never copied into it. + #[test] + fn retype_a_batch_lays_back_only_its_own_stores_foreign_records() -> Result<()> { + let temp = tempfile::tempdir()?; + let (routed, generic) = (temp.path().join("routed.cc"), temp.path().join("generic.cc")); + store_with(&routed, |cache| cache.put("own", &settings_one()).map(drop))?; + store_with(&generic, |cache| cache.put("carried", ¬e_one()).map(drop))?; + let generic_bytes = fs::read(&generic)?; + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.add_backing_store(SingleFileMessagePackBackingStore::new(&routed), [Settings::TYPE])?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&generic))?; + cache.pull_all_backing_stores()?; + let (staged, _) = cache.prepare_entry("other", &settings_one())?; + cache.put_prepared_batch(vec![staged])?; + let keys: Vec = reloaded(&routed)?.into_iter().map(|entry| entry.key).collect(); + assert_eq!(keys, ["other", "own"], "the routed store holds its own records only"); + assert_eq!(fs::read(&generic)?, generic_bytes); + cache.pull_all_backing_stores()?; + Ok(()) + } + + #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] + #[cultcache(type = "only-one", schema = "tests.only_one", global)] + struct OnlyOne { + #[cultcache(key = 0)] + text: String, + } + + /// A pull refused after every record is read (here, a second global) changes nothing, the foreign list included. + #[test] + fn retype_a_pull_refused_after_reading_keeps_the_foreign_list() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("cache.cc"); + store_with(&path, |cache| cache.put("carried", ¬e_one()).map(drop))?; + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.register_entry_type::()?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path))?; + cache.pull_all_backing_stores()?; + let before = (cache.snapshot(), cache.foreign_records()); + + // Behind the cache's back: another foreign record and two records of the global type. + let mut store = SingleFileMessagePackBackingStore::new(&path); + let (mut late, _) = cache.prepare_entry("late", &settings_one())?; + late.r#type = Note::TYPE.to_string(); + late.schema_id = Some(Note::TYPE.to_string()); + store.push(&late)?; + for key in ["g1", "g2"] { + store.push(&cache.prepare_entry(key, &OnlyOne { text: key.to_string() })?.0)?; + } + assert!(cache.pull_all_backing_stores().is_err()); + assert_eq!((cache.snapshot(), cache.foreign_records()), before, "a refused pull changes nothing"); + Ok(()) + } + + /// An SoA image is judged against the foreign keys too: it may not put a record at a key a store carries foreign. + #[test] + fn retype_load_soa_refuses_a_record_at_a_foreign_key() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("cache.cc"); + store_with(&path, |cache| cache.put("carried", ¬e_one()).map(drop))?; + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path))?; + cache.pull_all_backing_stores()?; + let (staged, _) = cache.prepare_entry("carried", &settings_one())?; + let error = cache.load_soa(CultCacheSoaTable::from_envelopes(vec![staged])).unwrap_err(); + assert_eq!(retype_refusal(&error).record_key, "carried"); + assert!(cache.snapshot().is_empty()); + Ok(()) + } + + #[test] + fn retype_load_soa_refuses_a_second_type_at_a_key_and_admits_nothing() -> Result<()> { + let mut cache = CultCache::new(); + cache.register_registry(TestEntries)?; + cache.put("kept", &settings_one())?; + let before = cache.snapshot(); + let (settings, _) = cache.prepare_entry("k", &settings_one())?; + let (note, _) = cache.prepare_entry("k", ¬e_one())?; + let error = cache.load_soa(CultCacheSoaTable::from_envelopes(vec![settings, note])).unwrap_err(); + assert_eq!(retype_refusal(&error).record_key, "k"); + assert_eq!(cache.snapshot(), before); + Ok(()) + } + + #[test] + fn retype_a_later_registration_claims_the_foreign_record_at_the_next_load() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("cache.cc"); + store_with(&path, |cache| cache.put("carried", ¬e_one()).map(drop))?; + + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path))?; + cache.pull_all_backing_stores()?; + assert_eq!(cache.foreign_records().len(), 1); + cache.register_entry_type::()?; + cache.pull_all_backing_stores()?; + assert!(cache.foreign_records().is_empty()); + assert_eq!(cache.get_required::("carried")?, note_one()); + Ok(()) + } + #[test] fn a_second_type_under_a_stored_key_is_refused_and_the_store_and_cache_stay() -> Result<()> { let temp = tempfile::tempdir()?; @@ -5321,24 +5762,26 @@ mod tests { #[test] fn refused_pull_leaves_prior_view_intact() -> Result<()> { let temp = tempfile::tempdir()?; - let store_path = temp.path().join("generic.cc"); + let (settings_path, generic_path) = (temp.path().join("settings.cc"), temp.path().join("generic.cc")); let mut cache = CultCache::new(); cache.register_entry_type::()?; - cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + cache.add_backing_store(SingleFileMessagePackBackingStore::new(&settings_path), [Settings::TYPE])?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&generic_path))?; cache.put("app", &Settings { theme: "before".to_string(), retries: 1 })?; let before = cache.snapshot(); - // A record of a type this cache never registered lands in the store behind its back. - SingleFileMessagePackBackingStore::new(&store_path).push(&CultCacheEnvelope { - key: "stray".into(), + // A record of a type this cache never registered lands, behind its back, in the other store at a held key. + SingleFileMessagePackBackingStore::new(&generic_path).push(&CultCacheEnvelope { + key: "app".into(), r#type: "unregistered".into(), payload: rmp_serde::to_vec(&("x",))?, stored_at: "2026-09-14T00:00:00Z".into(), schema_id: Some("unregistered".into()), })?; let error = cache.pull_all_backing_stores().unwrap_err(); - assert!(error.to_string().contains("\"unregistered\""), "{error}"); + assert_eq!(retype_refusal(&error).record_key, "app"); assert_eq!(cache.snapshot(), before); + assert!(cache.foreign_records().is_empty()); assert_eq!(cache.get_required::("app")?.theme, "before"); Ok(()) } @@ -5466,30 +5909,6 @@ mod tests { Ok(()) } - #[test] - fn pull_rejects_unregistered_persisted_entry_type() -> Result<()> { - let temp = tempfile::tempdir()?; - let store_path = temp.path().join("cache.cc"); - let mut store = SingleFileMessagePackBackingStore::new(&store_path); - store.push(&CultCacheEnvelope { - key: "unknown".to_string(), - r#type: "unregistered".to_string(), - payload: rmp_serde::to_vec(&1_u8)?, - stored_at: now_utc_second(), - schema_id: Some("unregistered".to_string()), - })?; - - let mut cache = CultCache::new(); - cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; - let error = cache.pull_all_backing_stores().unwrap_err(); - assert!( - error - .to_string() - .contains("No schema is registered for persisted entry type") - ); - Ok(()) - } - #[test] fn payload_is_binary_messagepack_not_json_value() -> Result<()> { let temp = tempfile::tempdir()?; diff --git a/packages/cultcache-rs/tests/key_identity_vector.rs b/packages/cultcache-rs/tests/key_identity_vector.rs new file mode 100644 index 00000000..13cc8de5 --- /dev/null +++ b/packages/cultcache-rs/tests/key_identity_vector.rs @@ -0,0 +1,283 @@ +//! `tests/vectors/key-identity`, written only by the C# reference: a cache holds one record per key whatever its type, and a +//! store holding a record no registered type claims loads, lists it as foreign and keeps its bytes +//! (`variants:ruling:cache-refuses-retype`, `write-ownership-by-written-type`, `unheld-removal-refused-at-commit`). +//! Every row is walked against this cache: each put row twice, once as `put` and once as a one-entry `put_prepared_batch`. + +use std::path::{Path, PathBuf}; + +use cultcache_rs::{ + CacheBackingStore, CultCache, DatabaseEntry, SchemaConflictError, SingleFileMessagePackBackingStore, +}; +use serde_json::Value; + +// The ids C# writes in the store catalogs and on every record (the vector's README). +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache(type = "sha256:b20703a98f49c253898404ab50ba2c630720f0b8a5e1a41614d5650be22799fe", schema = "tests.ki_alpha")] +struct Alpha { + #[cultcache(key = 0)] + text: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache(type = "sha256:dee4b727cf2ee04e61ab635ff0c5f3dca7345a411dc5f1018e583ab16009cf94", schema = "tests.ki_beta")] +struct Beta { + #[cultcache(key = 0)] + text: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache(type = "sha256:fbc26e351e00c9e7185e8de171c7486aba707c229c949f0aae6b889288af5f6d", schema = "tests.ki_gamma")] +struct Gamma { + #[cultcache(key = 0)] + text: String, +} + +fn vectors() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../tests/vectors/key-identity") +} + +/// `[refusal key or nil, held [type, key, text], file [key, schema name] or nil, foreign keys, kept foreign keys]`. +type Expect = (Option>, Vec<(String, String, String)>, Option>, Vec, Vec); +type Row = (String, Vec, Vec>, Expect); + +fn text(value: &Value) -> &str { + value.as_str().unwrap() +} + +fn register(cache: &mut CultCache, names: &[String]) { + for name in names { + match name.as_str() { + "alpha" => drop(cache.register_entry_type::().unwrap()), + "beta" => drop(cache.register_entry_type::().unwrap()), + "gamma" => drop(cache.register_entry_type::().unwrap()), + other => panic!("unknown type {other}"), + } + } +} + +fn short(entry_type: &str) -> &'static str { + match entry_type { + t if t == Alpha::TYPE => "alpha", + t if t == Beta::TYPE => "beta", + t if t == Gamma::TYPE => "gamma", + other => panic!("unknown entry type {other}"), + } +} + +fn put(cache: &mut CultCache, as_batch: bool, kind: &str, key: &str, body: &str) -> anyhow::Result<()> { + macro_rules! write { + ($t:ident) => {{ + let value = $t { text: body.to_string() }; + if as_batch { + let (entry, _) = cache.prepare_entry(key, &value)?; + cache.put_prepared_batch(vec![entry]) + } else { + cache.put(key, &value).map(drop) + } + }}; + } + match kind { + "alpha" => write!(Alpha), + "beta" => write!(Beta), + "gamma" => write!(Gamma), + other => panic!("unknown type {other}"), + } +} + +fn delete(cache: &mut CultCache, kind: &str, key: &str) -> anyhow::Result { + match kind { + "alpha" => cache.delete::(key), + "beta" => cache.delete::(key), + "gamma" => cache.delete::(key), + other => panic!("unknown type {other}"), + } +} + +fn open_cache(registered: &[String], path: Option<&Path>) -> anyhow::Result { + let mut cache = CultCache::new(); + register(&mut cache, registered); + if let Some(path) = path { + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path))?; + cache.pull_all_backing_stores()?; + } + Ok(cache) +} + +/// The records a store file holds, read by a fresh store: (key, schema name, schema id, stored-at, payload). +fn file_records(path: &Path) -> Vec<(String, String, Option, String, Vec)> { + let mut records: Vec<_> = SingleFileMessagePackBackingStore::new(path) + .pull_all() + .unwrap() + .into_iter() + .map(|entry| (entry.key, entry.r#type, entry.schema_id, entry.stored_at, entry.payload)) + .collect(); + records.sort_by(|left, right| left.0.encode_utf16().cmp(right.0.encode_utf16())); + records +} + +fn walk(row: &Row, as_batch: bool) -> Result<(), String> { + let (label, registered, ops, expect) = row; + let dir = tempfile::tempdir().unwrap(); + let mut path: Option = None; + let mut before = Vec::new(); + let mut cache: Option = None; + let mut refused: Option = None; + let mut refused_key: Option = None; + let mut shared_key: Option = None; + let mut wrote = false; + for (index, op) in ops.iter().enumerate() { + let last = index == ops.len() - 1; + let result: anyhow::Result<()> = match text(&op[0]) { + "open" => { + drop(cache.take()); + match &op[1] { + Value::String(file) if file == "memory" => open_cache(registered, None).map(|opened| cache = Some(opened)), + file => { + let fresh = dir.path().join("store.cc"); + if let Value::String(file) = file { + std::fs::copy(vectors().join(file), &fresh).unwrap(); + } + before = if fresh.exists() { file_records(&fresh) } else { Vec::new() }; + let opened = open_cache(registered, Some(&fresh)); + path = Some(fresh); + opened.map(|opened| cache = Some(opened)) + } + } + } + "reopen" => { + drop(cache.take()); + open_cache(registered, path.as_deref()).map(|opened| cache = Some(opened)) + } + "attach" => { + let file = text(&op[1]); + let attached = dir.path().join(file); + std::fs::copy(vectors().join(file), &attached).unwrap(); + let held: Vec = cache.as_ref().unwrap().snapshot().into_iter().map(|entry| entry.key).collect(); + let shared: Vec = file_records(&attached).into_iter().map(|record| record.0).filter(|key| held.contains(key)).collect(); + let home: Vec = op[2].as_array().unwrap().iter().map(|name| short_to_type(text(name)).to_string()).collect(); + let cache = cache.as_mut().unwrap(); + let attach = cache + .add_backing_store(SingleFileMessagePackBackingStore::new(&attached), home) + .and_then(|()| cache.pull_all_backing_stores()); + if attach.is_err() && last && shared.len() == 1 { + refused_key = Some(shared[0].clone()); + shared_key = refused_key.clone(); + assert_eq!(std::fs::read(&attached).unwrap(), std::fs::read(vectors().join(file)).unwrap(), "{label}: a refused attach leaves the store as it was"); + } + attach + } + "put" => { + wrote = true; + put(cache.as_mut().unwrap(), as_batch, text(&op[1]), text(&op[2]), text(&op[3])) + } + "delete" => delete(cache.as_mut().unwrap(), text(&op[1]), text(&op[2])).map(drop), + other => panic!("{label}: unknown op {other}"), + }; + match result { + Ok(()) => {} + Err(error) if last => refused = Some(error), + Err(error) => return Err(format!("{label}: op {index} failed: {error:#}")), + } + } + if !wrote && as_batch { + return Ok(()); // not a write row: the batch variant is the same walk + } + let Some(cache) = cache.as_ref() else { + return Err(format!("{label} (batch {as_batch}): the cache did not open: {:#}", refused.unwrap())); + }; + let last = ops.last().unwrap(); + if let Some(error) = &refused { + if let Some(conflict) = error.downcast_ref::() { + refused_key = Some(conflict.record_key.clone()); + let named = if text(&last[0]) == "attach" { shared_key.clone().unwrap() } else { text(&last[2]).to_string() }; + assert_eq!(named, conflict.record_key, "{label}: the refusal names the key written, removed or shared"); + } + let key = refused_key.clone().unwrap_or_default(); + let message = format!("{error:#}"); + let (all, delimited) = occurrences(&message, &key); + assert!(!key.is_empty() && delimited > 0, "{label}: the refusal names the key as a token: {message}"); + if text(&last[0]) == "attach" { + assert_eq!(all, delimited, "{label}: the key occurs in the refusal only as a delimited token: {message}"); + } + } + let mut held: Vec<(String, String, String)> = cache + .snapshot() + .into_iter() + .map(|entry| { + let body = match short(&entry.r#type) { + "alpha" => cache.get::(&entry.key).unwrap().unwrap().text, + "beta" => cache.get::(&entry.key).unwrap().unwrap().text, + _ => cache.get::(&entry.key).unwrap().unwrap().text, + }; + (short(&entry.r#type).to_string(), entry.key, body) + }) + .collect(); + held.sort_by(|a, b| (a.0.as_str(), a.1.encode_utf16().collect::>()).cmp(&(b.0.as_str(), b.1.encode_utf16().collect::>()))); + let foreign: Vec = cache.foreign_records().into_iter().map(|record| record.key).collect(); + let (file, kept) = match &path { + None => (None, Vec::new()), + Some(path) => { + let after = file_records(path); + let kept = foreign + .iter() + .filter(|key| { + let original = before.iter().find(|record| &record.0 == *key); + let now = after.iter().find(|record| &record.0 == *key); + original.is_some() && original.map(|r| (&r.2, &r.3, &r.4)) == now.map(|r| (&r.2, &r.3, &r.4)) + }) + .cloned() + .collect(); + // This writer names a catalog entry by the record's type, which for these types is the id; C# names it by the + // schema name. Which type the record at a key is, is the verdict, so the id reads as its schema name. + let name = |described: String| match described.as_str() { + t if t == Alpha::TYPE => Alpha::SCHEMA_NAME.to_string(), + t if t == Beta::TYPE => Beta::SCHEMA_NAME.to_string(), + t if t == Gamma::TYPE => Gamma::SCHEMA_NAME.to_string(), + _ => described, + }; + (Some(after.into_iter().map(|record| (record.0, name(record.1))).collect::>()), kept) + } + }; + let observed: Expect = (refused_key.map(|key| vec![key]), held, file, foreign, kept); + if &observed == expect { Ok(()) } else { Err(format!("{label} (batch {as_batch}): committed {expect:?} but this cache answers {observed:?}")) } +} + +/// How often `key` occurs in `message`, and how often as a delimited token (not a piece of a longer word, schema name or id). +fn occurrences(message: &str, key: &str) -> (usize, usize) { + let word = |c: Option| c.is_some_and(|c| c.is_ascii_alphanumeric() || c == '_'); + let found: Vec = message.match_indices(key).map(|(at, _)| at).collect(); + let delimited = found + .iter() + .filter(|at| !word(message[..**at].chars().next_back()) && !word(message[**at + key.len()..].chars().next())) + .count(); + (found.len(), delimited) +} + +fn short_to_type(name: &str) -> &'static str { + match name { + "alpha" => Alpha::TYPE, + "beta" => Beta::TYPE, + "gamma" => Gamma::TYPE, + other => panic!("unknown type {other}"), + } +} + +#[test] +fn key_identity_rows_hold_their_verdicts() { + let bytes = std::fs::read(vectors().join("rows.msgpack")).unwrap(); + let rows: Vec = rmp_serde::from_slice(&bytes).unwrap(); + let mut walked = 0; + let mut failed = Vec::new(); + for row in &rows { + let mut ok = true; + for as_batch in [false, true] { + if let Err(message) = walk(row, as_batch) { + failed.push(message); + ok = false; + } + } + walked += usize::from(ok); + } + assert!(failed.is_empty(), "{}", failed.join("\n")); + assert_eq!(walked, 12, "rows walked"); +} diff --git a/packages/cultcache-rs/tests/unique_index.rs b/packages/cultcache-rs/tests/unique_index.rs index 1132b2b5..193ba6fb 100644 --- a/packages/cultcache-rs/tests/unique_index.rs +++ b/packages/cultcache-rs/tests/unique_index.rs @@ -245,9 +245,9 @@ fn delete_releases_a_value() { fn types_sharing_an_alias_and_a_value_do_not_conflict() { let mut cache = memory_cache(); cache.put("a", &unique("S")).unwrap(); - cache.put("a", &Other { sku: "S".into() }).unwrap(); - cache.put("b", &Other { sku: "T".into() }).unwrap(); - assert!(cache.put("b", &Other { sku: "S".into() }).is_err()); + cache.put("oa", &Other { sku: "S".into() }).unwrap(); + cache.put("ob", &Other { sku: "T".into() }).unwrap(); + assert!(cache.put("ob", &Other { sku: "S".into() }).is_err()); assert_eq!(cache.get_by_index::("Sku", "S").unwrap().unwrap(), unique("S")); assert_eq!(cache.get_by_index::("Sku", "S").unwrap().unwrap(), Other { sku: "S".into() }); } diff --git a/tests/vectors/key-identity/README.md b/tests/vectors/key-identity/README.md index f6ad6516..65e137b1 100644 --- a/tests/vectors/key-identity/README.md +++ b/tests/vectors/key-identity/README.md @@ -19,6 +19,9 @@ Registering a type that declares its schema id claims it at the next load. - `tests.ki_alpha`: `sha256:b20703a98f49c253898404ab50ba2c630720f0b8a5e1a41614d5650be22799fe` - `tests.ki_beta`: `sha256:dee4b727cf2ee04e61ab635ff0c5f3dca7345a411dc5f1018e583ab16009cf94` - `tests.ki_gamma`: `sha256:fbc26e351e00c9e7185e8de171c7486aba707c229c949f0aae6b889288af5f6d` + The Rust writer names a catalog entry it writes by the record's type, which for these types is the id, where C# names it by + schema name; the Rust walker (`packages/cultcache-rs/tests/key_identity_vector.rs`) maps each of the three ids, and only + those, back to its schema name before comparing `file`, so the verdict is compared and the spelling divergence is not. - `foreign-store.cc`: a single-file store, alpha `a` at key `j`, gamma `g` at key `k`, stored-at pinned to `2026-01-01T00:00:00Z`. `catalog-store.cc`: alpha `a` at `qx`. `run-store.cc`: beta `b` at `qx` (row 11). The row-11 key is `qx` because it must occur in the refusal's message only as the key, never inside a schema name (`k` is inside `ki_alpha`).