diff --git a/packages/cultcache-rs/CHANGELOG.md b/packages/cultcache-rs/CHANGELOG.md index 4e97588c..c6d72492 100644 --- a/packages/cultcache-rs/CHANGELOG.md +++ b/packages/cultcache-rs/CHANGELOG.md @@ -16,6 +16,7 @@ Semver: this is a breaking change in the 0.x line (0.3.x to 0.4.0). - 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. +- A cache with a backing store also refuses every write and removal, with `CultCacheLoadRefused`, until it has pulled: a cache that never pulled, or that gained a store since its last pull, holds no image of its stores, and a batch from it replaced a populated store with only the new records. `is_writable` answers false meanwhile. A cache with no store needs no pull. Callers that attach a store and write (cultmesh nodes created with `pull_on_start: false`) must call `pull_all_backing_stores` first. - `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). diff --git a/packages/cultcache-rs/src/lib.rs b/packages/cultcache-rs/src/lib.rs index 522e226e..47a3dd18 100644 --- a/packages/cultcache-rs/src/lib.rs +++ b/packages/cultcache-rs/src/lib.rs @@ -2937,13 +2937,24 @@ 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, + /// Whether the cache holds an image of its stores. While it is not `Pulled`, every write and removal is refused. + load: LoadState, } -/// 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. +/// What the cache knows of its stores. `Unpulled` is the state of a cache that has never pulled, or that gained a store +/// since: its image lacks whatever the stores hold, and a batch would replace them with it. +#[derive(Clone, Debug, PartialEq, Eq)] +enum LoadState { + Unpulled, + Pulled, + /// Why the last pull was refused. + Refused(String), +} + +/// A write or removal refused because the cache holds no image of its backing stores: its last load was refused, or it has +/// not pulled since its stores were attached. Any write would persist an image that lacks the records it never read. +/// Pull; for a refused load, first repair the store named by `reason` (which names a type, a key and a clause, never a +/// payload). #[derive(Clone, Debug, PartialEq, Eq)] pub struct CultCacheLoadRefused { pub reason: String, @@ -2953,7 +2964,7 @@ 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", + "CultCache refuses writes and removals: it holds no image of its backing stores ({}); repair the store if it was refused and pull again", self.reason ) } @@ -3022,7 +3033,7 @@ impl CultCache { projectors: BTreeMap::new(), stores: Vec::new(), foreign: Vec::new(), - load_refused: None, + load: LoadState::Unpulled, } } @@ -3128,6 +3139,10 @@ impl CultCache { store: Box::new(store), types, }); + let loaded_before = self.load.clone(); + if self.load == LoadState::Pulled { + self.load = LoadState::Unpulled; + } // A record's home cannot change after it is admitted. let candidate = self.stores.len() - 1; let held: BTreeSet = self.entries.keys().map(|(type_id, _)| type_id.clone()).collect(); @@ -3146,6 +3161,7 @@ impl CultCache { self.describe_store(after) ); self.stores.pop(); + self.load = loaded_before; return Err(error); } } @@ -3164,21 +3180,27 @@ impl CultCache { self.entries = loaded; self.foreign = foreign; self.index_holders = holders; - self.load_refused = None; + self.load = LoadState::Pulled; Ok(()) } Err(error) => { - self.load_refused = Some(format!("{error:#}")); + self.load = LoadState::Refused(format!("{error:#}")); Err(error) } } } /// The one gate of every write and removal: a cache whose load was refused holds no image of its stores. + /// A cache without stores has nothing to lose and needs no pull. fn require_loaded(&self) -> Result<()> { - match &self.load_refused { - Some(reason) => Err(CultCacheLoadRefused { reason: reason.clone() }.into()), - None => Ok(()), + match &self.load { + _ if self.stores.is_empty() => Ok(()), + LoadState::Pulled => Ok(()), + LoadState::Refused(reason) => Err(CultCacheLoadRefused { reason: reason.clone() }.into()), + LoadState::Unpulled => Err(CultCacheLoadRefused { + reason: "pull_all_backing_stores has not run since the stores were attached".to_string(), + } + .into()), } } @@ -3269,9 +3291,12 @@ impl CultCache { /// 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 + /// read-only), when the cache holds no image of its stores (see [`CultCacheLoadRefused`]), 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 { + if self.require_loaded().is_err() { + return Ok(false); + } if !matches!(self.held_finding(type_id, key)?, HeldFinding::Free) { return Ok(false); } @@ -4724,6 +4749,7 @@ mod tests { let mut cache = CultCache::new(); cache.register_registry(TestEntries)?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path))?; + cache.pull_all_backing_stores()?; records(&mut cache) } @@ -4965,6 +4991,7 @@ mod tests { let mut cache = CultCache::new(); cache.register_entry_type::()?; cache.add_backing_store(SingleFileMessagePackBackingStore::new(&routed), [Settings::TYPE])?; + cache.pull_all_backing_stores()?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&generic))?; cache.pull_all_backing_stores()?; let (staged, _) = cache.prepare_entry("other", &settings_one())?; @@ -5066,6 +5093,7 @@ mod tests { let mut cache = CultCache::new(); cache.register_registry(TestEntries)?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + cache.pull_all_backing_stores()?; cache.put("shared", &settings_one())?; let before = std::fs::read(&store_path)?; @@ -5098,6 +5126,7 @@ mod tests { let mut cache = CultCache::new(); cache.register_registry(TestEntries)?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + cache.pull_all_backing_stores()?; cache.put("kept", &settings_one())?; let before = std::fs::read(&store_path)?; @@ -5206,6 +5235,7 @@ mod tests { let mut cache = CultCache::new(); cache.register_registry(TestEntries)?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + cache.pull_all_backing_stores()?; let (settings, _) = cache.prepare_entry( "app", @@ -5912,6 +5942,7 @@ mod tests { inner: SingleFileMessagePackBackingStore::new(&store_path), refuse_batch: refusal.clone(), })?; + cache.pull_all_backing_stores()?; cache.put( "app", &Settings { @@ -5952,6 +5983,7 @@ mod tests { ["settings"], )?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&generic_path))?; + cache.pull_all_backing_stores()?; cache.put( "app", @@ -5980,6 +6012,7 @@ mod tests { let temp = tempfile::tempdir()?; let mut cache = CultCache::new(); cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(temp.path().join("a.cc")))?; + cache.pull_all_backing_stores()?; let error = cache .add_generic_backing_store(SingleFileMessagePackBackingStore::new(temp.path().join("b.cc"))) .unwrap_err(); @@ -5993,6 +6026,7 @@ mod tests { let temp = tempfile::tempdir()?; let mut cache = CultCache::new(); cache.add_backing_store(SingleFileMessagePackBackingStore::new(temp.path().join("a.cc")), ["settings"])?; + cache.pull_all_backing_stores()?; let error = cache .add_backing_store( SingleFileMessagePackBackingStore::new(temp.path().join("b.cc")), @@ -6002,6 +6036,7 @@ mod tests { assert!(error.to_string().contains("already routed"), "{error}"); // A refused registration claims nothing, so "note" is still free. cache.add_backing_store(SingleFileMessagePackBackingStore::new(temp.path().join("c.cc")), ["note"])?; + cache.pull_all_backing_stores()?; assert_eq!(cache.stores.len(), 2); Ok(()) } @@ -6014,6 +6049,7 @@ mod tests { let mut cache = CultCache::new(); cache.register_entry_type::()?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&generic_path))?; + cache.pull_all_backing_stores()?; cache.put("app", &Settings { theme: "v1-generic".to_string(), retries: 1 })?; let error = cache @@ -6043,15 +6079,17 @@ mod tests { let mut writer = CultCache::new(); writer.register_entry_type::()?; writer.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&generic_path))?; + writer.pull_all_backing_stores()?; writer.put("app", &Settings { theme: "stray".to_string(), retries: 0 })?; let generic_bytes = fs::read(&generic_path)?; let mut cache = CultCache::new(); cache.register_entry_type::()?; cache.add_backing_store(SingleFileMessagePackBackingStore::new(&settings_path), ["settings"])?; - cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&generic_path))?; + cache.pull_all_backing_stores()?; // The cache already holds a record, in its home store, before the refused load. cache.put("held", &Settings { theme: "home".to_string(), retries: 0 })?; + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&generic_path))?; let before = cache.snapshot(); let settings_bytes = fs::read(&settings_path)?; let error = cache.pull_all_backing_stores().unwrap_err(); @@ -6075,7 +6113,9 @@ mod tests { let mut cache = CultCache::new(); cache.register_entry_type::()?; cache.add_backing_store(SingleFileMessagePackBackingStore::new(&settings_path), [Settings::TYPE])?; + cache.pull_all_backing_stores()?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&generic_path))?; + cache.pull_all_backing_stores()?; cache.put("app", &Settings { theme: "before".to_string(), retries: 1 })?; let before = cache.snapshot(); @@ -6117,6 +6157,7 @@ mod tests { let mut homeless = CultCache::new(); homeless.register_registry(TestEntries)?; homeless.add_backing_store(SingleFileMessagePackBackingStore::new(temp.path().join("settings.cc")), ["settings"])?; + homeless.pull_all_backing_stores()?; homeless.put("held", &Settings { theme: "held".to_string(), retries: 0 })?; let before = homeless.snapshot(); let error = homeless.load_soa(table).unwrap_err(); @@ -6133,7 +6174,9 @@ mod tests { let mut cache = CultCache::new(); cache.register_registry(TestEntries)?; cache.add_backing_store(SingleFileMessagePackBackingStore::new(&settings_path), ["settings"])?; + cache.pull_all_backing_stores()?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&generic_path))?; + cache.pull_all_backing_stores()?; cache.put("memo", &Note { title: "generic".to_string(), body: "home".to_string() })?; cache.put("old", &Settings { theme: "kept".to_string(), retries: 0 })?; @@ -6164,6 +6207,7 @@ mod tests { let mut cache = CultCache::new(); cache.register_registry(TestEntries)?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + cache.pull_all_backing_stores()?; cache.put("app", &Settings { theme: "before".to_string(), retries: 1 })?; let before = cache.snapshot(); let store_bytes = fs::read(&store_path)?; @@ -6203,6 +6247,7 @@ mod tests { let mut cache = CultCache::new(); cache.register_entry_type::()?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + cache.pull_all_backing_stores()?; let updated = cache.update::("app", |current| { let mut current = current.unwrap_or(Settings { @@ -6225,6 +6270,7 @@ mod tests { let mut cache = CultCache::new(); cache.register_entry_type::()?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + cache.pull_all_backing_stores()?; cache.put( "app", &Settings { @@ -6288,6 +6334,7 @@ mod tests { let mut origin = CultCache::new(); origin.register_entry_type::()?; origin.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&origin_store))?; + origin.pull_all_backing_stores()?; origin.put( "app", &Settings { @@ -6301,6 +6348,7 @@ mod tests { let mut target = CultCache::new(); target.register_entry_type::()?; target.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&target_store))?; + target.pull_all_backing_stores()?; let applied = target.put_envelope::(envelope.clone())?; assert_eq!( @@ -6334,6 +6382,7 @@ mod tests { cache.register_entry_type::()?; cache.register_entry_type::()?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + cache.pull_all_backing_stores()?; cache.put_raw_envelope(envelope.clone())?; assert_eq!(cache.snapshot(), vec![envelope.clone()]); @@ -6376,6 +6425,7 @@ mod tests { let mut cache = CultCache::new(); cache.register_entry_type::()?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + cache.pull_all_backing_stores()?; let tailed = CultCacheEnvelope { key: "app".to_string(), r#type: Settings::TYPE.to_string(), @@ -6404,6 +6454,7 @@ mod tests { let mut cache = CultCache::new(); cache.register_registry(TestEntries)?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + cache.pull_all_backing_stores()?; cache.put( "app", &Settings { @@ -6441,6 +6492,7 @@ mod tests { let mut cache = CultCache::new(); cache.register_registry(TestEntries)?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + cache.pull_all_backing_stores()?; cache.put( "settings:a", &Settings { @@ -6523,6 +6575,7 @@ mod tests { let mut cache = CultCache::new(); cache.register_entry_type::()?; cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path))?; + cache.pull_all_backing_stores()?; cache.put( "primary", &Settings { @@ -8051,4 +8104,91 @@ mod tests { assert_eq!(fs::read(&path)?, before); Ok(()) } -} \ No newline at end of file + + /// A store already holding `Settings:"held"`, then a fresh cache attached to it that has not pulled. + fn unpulled_cache_over_populated_stores() -> Result<(tempfile::TempDir, Vec, Vec)> { + let temp = tempfile::tempdir()?; + let single = temp.path().join("single.cc"); + let redb = temp.path().join("redb.db"); + let held = Settings { theme: "held".to_string(), retries: 1 }; + let mut seeder = CultCache::new(); + seeder.register_registry(TestEntries)?; + seeder.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&single))?; + seeder.pull_all_backing_stores()?; + seeder.put("held", &held)?; + let mut seeder = CultCache::new(); + seeder.register_registry(TestEntries)?; + seeder.add_generic_backing_store(RedbMessagePackBackingStore::new(&redb)?)?; + seeder.pull_all_backing_stores()?; + seeder.put("held", &held)?; + + let mut over_single = CultCache::new(); + over_single.register_registry(TestEntries)?; + over_single.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&single))?; + let mut over_redb = CultCache::new(); + over_redb.register_registry(TestEntries)?; + over_redb.add_generic_backing_store(RedbMessagePackBackingStore::new(&redb)?)?; + Ok((temp, vec![over_single, over_redb], vec![single, redb])) + } + + #[test] + fn a_cache_that_never_pulled_refuses_every_write_and_removal() -> Result<()> { + let (_temp, caches, _paths) = unpulled_cache_over_populated_stores()?; + for mut cache in caches { + let fresh = Note { title: "t".to_string(), body: "b".to_string() }; + let (staged, _) = cache.prepare_entry("fresh", &fresh)?; + for error in [ + cache.put_prepared_batch(vec![staged.clone()]).unwrap_err(), + cache.put("fresh", &fresh).unwrap_err(), + cache.put_raw_envelope(staged.clone()).unwrap_err(), + cache.delete::("held").unwrap_err(), + ] { + assert!(error.downcast_ref::().is_some(), "{error:#}"); + } + assert!(cache.snapshot().is_empty()); + } + Ok(()) + } + + #[test] + fn a_batch_from_a_cache_that_never_pulled_cannot_wipe_the_store() -> Result<()> { + let (_temp, caches, paths) = unpulled_cache_over_populated_stores()?; + for (mut cache, path) in caches.into_iter().zip(paths) { + let (staged, _) = cache.prepare_entry("fresh", &Note { title: "t".to_string(), body: "b".to_string() })?; + let _ = cache.put_prepared_batch(vec![staged]); + let mut reader = CultCache::new(); + reader.register_registry(TestEntries)?; + if path.extension().is_some_and(|extension| extension == "db") { + reader.add_generic_backing_store(RedbMessagePackBackingStore::new(&path)?)?; + reader.pull_all_backing_stores()?; + } else { + reader.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path))?; + reader.pull_all_backing_stores()?; + } + reader.pull_all_backing_stores()?; + assert_eq!(reader.get_required::("held")?.theme, "held", "{path:?}"); + } + Ok(()) + } + + #[test] + fn pulling_lifts_the_gate_and_a_store_attached_later_closes_it_again() -> Result<()> { + let (temp, mut caches, _paths) = unpulled_cache_over_populated_stores()?; + let mut cache = caches.remove(0); + cache.pull_all_backing_stores()?; + cache.put("fresh", &Note { title: "t".to_string(), body: "b".to_string() })?; + assert_eq!(cache.get_required::("held")?.theme, "held"); + + let mut routed = CultCache::new(); + routed.register_registry(TestEntries)?; + let settings = Settings { theme: "t".to_string(), retries: 0 }; + routed.add_backing_store(SingleFileMessagePackBackingStore::new(temp.path().join("routed.cc")), [Settings::TYPE])?; + routed.pull_all_backing_stores()?; + routed.put("fresh", &settings)?; + routed.add_generic_backing_store(SingleFileMessagePackBackingStore::new(temp.path().join("other.cc")))?; + assert!(routed.put("again", &settings).unwrap_err().downcast_ref::().is_some()); + routed.pull_all_backing_stores()?; + routed.put("again", &settings)?; + Ok(()) + } +} diff --git a/packages/cultcache-rs/tests/slot_readonly.rs b/packages/cultcache-rs/tests/slot_readonly.rs index 0031e372..cea0853a 100644 --- a/packages/cultcache-rs/tests/slot_readonly.rs +++ b/packages/cultcache-rs/tests/slot_readonly.rs @@ -149,6 +149,7 @@ fn every_read_only_vector_decodes_marked_and_is_not_writable() { let mut cache = registered(); let dir = tempfile::tempdir().unwrap(); cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(dir.path().join("s.cc"))).unwrap(); + cache.pull_all_backing_stores().unwrap(); cache.put_raw_envelope(envelope("k", payload.clone())).unwrap(); assert_eq!(cache.get::("k").unwrap(), Some(constructed()), "{file}"); assert!(!cache.is_writable(UnionSurface::TYPE, "k").unwrap(), "{file}"); @@ -470,6 +471,7 @@ fn is_writable_is_false_for_a_type_with_no_home_store() { let mut cache = registered(); // One store, routed to SlotUnit only: UnionSurface has no home. cache.add_backing_store(SingleFileMessagePackBackingStore::new(dir.path().join("s.cc")), [SlotUnit::TYPE]).unwrap(); + cache.pull_all_backing_stores().unwrap(); assert!(cache.put("k", &constructed()).is_err()); assert!(!cache.is_writable(UnionSurface::TYPE, "k").unwrap()); assert!(cache.is_writable(SlotUnit::TYPE, "k").unwrap()); diff --git a/packages/cultcache-rs/tests/union_parity.rs b/packages/cultcache-rs/tests/union_parity.rs index 67f2b389..bc3aba6d 100644 --- a/packages/cultcache-rs/tests/union_parity.rs +++ b/packages/cultcache-rs/tests/union_parity.rs @@ -718,6 +718,7 @@ fn write_refused_past_limit() { 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(); let at_limit = wrap_chain(126, boxed("0000000000f0", vec![])); assert_eq!(cache.put("at", &at_limit).unwrap(), at_limit); @@ -801,6 +802,7 @@ fn the_raw_doors_hold_the_bound_for_arms_maps_and_arrays() { cache.register_entry_type::().unwrap(); cache.register_entry_type::().unwrap(); cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap(); + cache.pull_all_backing_stores().unwrap(); cache.put_raw_envelope(raw(type_name, "at-raw", at.clone())).unwrap(); cache.put_prepared_batch(vec![raw(type_name, "at-batch", at.clone())]).unwrap(); let stored = cache.snapshot_envelopes();