GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

cultcache-rs: update tests, content-version vector, docs snippet, README and CHANGELOG

Browse source 
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Libby 2026-11-13 19:56:42 +00:00
commit 2bd4466af4
7 changed files with 816 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

32
docs/start/README.md
View file

@ -18,7 +18,7 @@ this page.
| --- | --- | --- | --- | --- | --- |
| **C#** | [first call, tested](#c-first-call); [update, tested](#c-update-with-retry) | [package](../../packages/cultmath/README.md); snippet not yet | [package](../../src/GameCult.Geometry/README.md); snippet not yet | [first call, tested](#c-cultnet-first-call); [package](../../src/GameCult.Networking) | [quickstart](../../src/GameCult.Mesh/docs/durable-node-quickstart.md); run by `scripts/verify-eve-getting-started.ps1` |
| **TypeScript** | [first call, tested](#typescript-first-call); [update, tested](#typescript-update-with-retry) | not yet | not yet | [first call, tested](#typescript-cultnet-first-call); [package](../../packages/cultnet-ts/README.md) | [package](../../packages/cultmesh-ts/README.md); snippet not yet; [recipe: two-client .io game](io-game.md), tested with two live clients, browser page not yet run |
| **Rust** | [first call, tested](#rust-first-call) | [native core](../../packages/cultmath/native/cultmath-core); snippet not yet | not yet | [first call, tested](#rust-cultnet-first-call); [package](../../packages/cultnet-rs/README.md), [examples](../../packages/cultnet-rs/examples) | [package](../../packages/cultmesh-rs); snippet not yet |
| **Rust** | [first call, tested](#rust-first-call); [update, tested](#rust-update-with-retry) | [native core](../../packages/cultmath/native/cultmath-core); snippet not yet | not yet | [first call, tested](#rust-cultnet-first-call); [package](../../packages/cultnet-rs/README.md), [examples](../../packages/cultnet-rs/examples) | [package](../../packages/cultmesh-rs); snippet not yet |
| **Python** | [first call, tested](#python-first-call) | not yet | not yet | [first call, tested](#python-cultnet-first-call); [package](../../packages/cultnet-py/README.md) | [package](../../packages/cultmesh-py/README.md); snippet not yet |
For the whole
@ -220,6 +220,36 @@ fn first_call() -> anyhow::Result<()> {
}
```
## Rust update with retry
`update` changes a record from what the store holds now and commits only if it is still that record. The source is the
second region of
[`packages/cultcache-rs/tests/first_call.rs`](../../packages/cultcache-rs/tests/first_call.rs), which uses the `Item` of
the first call above; a test in that file fails if the block stops matching the file.
```rust
fn restock(path: &Path) -> anyhow::Result<CultUpdateResult<Item>> {
let mut cache = CultCache::new();
cache.register_entry_type::<Item>()?;
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path))?;
cache.pull_all_backing_stores()?;
cache.update::<Item, _>("item:potion", |item| {
// The store's copy of the record, or nothing if it has none; this may run again on a lost race.
let mut item = item.unwrap_or(Item {
name: "Potion".to_string(),
category: "Consumable".to_string(),
value: 0,
});
item.value += 10;
Some(item)
})
}
```
The write is durable when `update` returns. A lost race re-runs the closure, so keep it free of side effects; when the
attempts run out (five, or the bound `update_with_attempts` takes) the outcome says `LostRace` instead of failing, and
returning `None` writes nothing and says `Declined`. `put` is the blind opt-out: last writer wins.
## Python first call
A document is a definition with an encoder and decoder; `name=` registers the name lookup, `indexes=` unique lookups and

8
packages/cultcache-rs/CHANGELOG.md
View file

@ -6,8 +6,14 @@ only in git history.
## Unreleased
### Added
- `CultCache::update_with_attempts` and the conditional `update`: each attempt reads the record from its home store with its content version in one read, applies a function to a copy and commits only over that version, rerunning on a lost race (default five attempts). The result is `CultUpdateResult<T>`, outcome `Updated`, `Declined` (the function returned `None`) or `LostRace`, with the cache's document. Public types `CultStoredRecord`, `CultCommitIf`, `CultUpdateOutcome`, `CultUpdateResult`.
- `CacheBackingStore::read(key)` and `commit_if(entry, expected)`, provided methods whose defaults refuse with an error naming the store kind; `SingleFileMessagePackBackingStore` implements both. The content version is the lowercase hex SHA-256 of the record as the store file encodes it (key, schema id, `stored_at`, payload), equal to the C# reference's `tests/vectors/content-version`. New dependency `sha2` 0.10 (with `digest`, `block-buffer`, `crypto-common`, `generic-array`, `typenum`, `cpufeatures`).
### Changed
- The single-file store's `push`, `push_all` and `commit_if` stamp a replacing write strictly later than the record it replaces at the same key (a staged stamp stands only when it parses and is strictly later; otherwise the stamp is minted as the replaced stamp plus one microsecond, or now if later), with no identical-content exception, as C# does. The cache holds the stamp the store wrote. The redb stores and the store-level exact-envelope writes (`compare_and_swap_entry` and its family) still write the caller's envelope unchanged.
- The single-file store re-checks for a link once the store lock is held (every write and the try-exchange), so a link put at the path while a writer waited on the lock is refused with `StoreLinkedError` and the lock released, as C# does, instead of being replaced. Pinned by a test that puts the link while a second holder owns the lock.
- `CultCache::is_writable` does not look at store paths: a link at a store path is not predicted, and the write refuses it with `StoreLinkedError`. Pinned by a test through a cache attached to a linked store.
@ -15,6 +21,8 @@ only in git history.
Semver: this is a breaking change in the 0.x line (0.3.x to 0.4.0).
- `CultCache::update` is replaced, not extended: `update::<T, _>(key, apply)` takes `FnMut(Option<T>) -> Option<T>` and returns `Result<CultUpdateResult<T>>`. The blind `FnOnce(Option<T>) -> T` form that returned `T` is gone; use `put` for a blind write. It reads the home store, not the cache's image, so a cache whose store has no `read` and `commit_if` (the redb stores, a custom store) refuses it before `apply` runs.
- `CacheBackingStore::push` returns `Result<String>`, the `stored_at` written, and `push_all` returns `Result<Vec<String>>`, one stamp per entry given in order (a soft write that wrote nothing returns an empty vec). A custom store must return what it wrote; the redb stores return the entries' own stamps. Huginn implements `CacheBackingStore` in four test stores under a pinned git rev: they change at its next bump.
- A write at a store path that is a symbolic link, live or dangling, is refused with the new public `StoreLinkedError` (`error.downcast_ref::<StoreLinkedError>()`) before the lock and the replace, and writes nothing: `push`, `push_all`, `delete`, the exchanges, and a soft `push_all`, which used to skip a link and now refuses. Before, a write at a live link replaced the link with a regular file and a dangling link failed with `io::ErrorKind::NotFound`. The redb stores (`RedbMessagePackBackingStore`, `OwnedRedbMessagePackBackingStore`) are exempt: they are ruled migrated and deleted (`paged-store-redb-migrate-and-delete`) and still open a live link's target. Configure the link's target as the store path. Reads through a live link work.
- A record whose payload carries unread slots is read-only. A payload with a slot past its type's declared slots, at any depth (record, nested record, list element, union arm, unit arm), still decodes, and every write or removal of its key is now refused before any store is touched: `put`, `put_envelope`, `put_raw_envelope`, `put_prepared_batch` (the whole batch) and `delete`, so `update` too. The refusal is `SchemaConflictError` with the new `unread_slots`; its message names the type and both slot counts and never a slot value. Holders of such a record used to overwrite it and drop the extra slots silently. The payload nesting refusal still runs first. A held payload this build cannot read at all (an arm it lacks from a newer writer, or a corrupt record) is read-only too: every write or removal is refused with an error naming the type and key (a newer writer's record is read with that build; a corrupt one is restored from a backup or removed at the store itself, as no cache call removes it), and `is_writable` answers false. `load_soa` is a write of every held key it omits or changes, so it refuses to drop or replace a read-only record. `is_writable` also answers false for a key a store carries as a foreign record and for a type other than the one holding the key, as the writes do. Each write over a held key decodes the held payload once to ask (about 40% of a `put` of a 200,000-element record, measured in release); keys nothing holds pay nothing.
- cultmesh-rs `pull_rudp_catalog_snapshot` applies a snapshot as one cache commit (`put_prepared_batch`), as C# does: a snapshot document for a key the node holds read-only, or any other refusal, applies nothing. It used to apply document by document, so a refusal ended the pull with the earlier documents kept. A snapshot whose types route to different backing stores is now refused whole, as a batch cannot span stores. C# `CultCache.Commit` refuses a cross-store commit the same way (`The batch spans A and B; a commit lands in one home store`), so the two agree; a test pins it.

9
packages/cultcache-rs/README.md
View file

@ -186,7 +186,7 @@ without pretending Rust has runtime reflection hiding under the floorboards.
- `soa::<T>`
- `put::<T>`
- `put_envelope::<T>`
- `update::<T>`
- `update::<T>` and `update_with_attempts::<T>`: the conditional golden-path write (read the stored record, apply, commit only over what was read, rerun on a lost race)
- `delete::<T>`
- `snapshot`
- `snapshot_soa`
@ -304,6 +304,13 @@ The contract, shared with every runtime, is one home store per document type
## Persistence Semantics
`update` is the golden path for changing a record another writer may also change: each attempt reads the record and its content
version from the home store, hands `apply` a copy, and commits only if the store still holds that version. `put` is the blind
write. The single-file store stamps a replacing write (`push`, `push_all` and `update`'s commit) strictly later than the record
it replaces, and the cache holds the stamp the store reports; the redb stores and the exact-envelope family
(`compare_and_swap_entry` and the other store-level conditional writes) write the caller's envelope byte for byte. A store
needs `read` and `commit_if` to serve `update`; the redb stores do not have them.
`put` persists before mutating the in-memory cache. If persistence fails, the
cache does not pretend the write succeeded.

35
packages/cultcache-rs/src/lib.rs
View file

@ -5958,6 +5958,41 @@ mod tests {
Ok(())
}
fn is_minted_text(text: &str) -> bool {
let bytes = text.as_bytes();
bytes.len() == 27
&& bytes.ends_with(b"Z")
&& bytes[19] == b'.'
&& bytes[20..26].iter().all(u8::is_ascii_digit)
&& chrono::DateTime::parse_from_rfc3339(text).is_ok()
}
fn at(text: &str) -> chrono::DateTime<chrono::FixedOffset> {
chrono::DateTime::parse_from_rfc3339(text).unwrap()
}
#[test]
fn stamps_stand_only_when_strictly_later_and_otherwise_mint_over_the_replaced() {
// A strictly later staged stamp stands as written.
assert_eq!(stored_at_over("2099-01-01T00:00:00Z", "2026-01-01T00:00:00Z"), "2099-01-01T00:00:00Z");
// An equal or earlier one is replaced by a minted stamp later than the replaced, over C#'s seven digits too.
for replaced in ["2026-01-01T00:00:00.1234567+00:00", "2026-01-01T00:00:00Z", "2099-01-01T00:00:00Z"] {
for staged in [replaced, "2020-01-01T00:00:00Z"] {
let minted = stored_at_over(staged, replaced);
assert!(is_minted_text(&minted), "{minted}");
assert!(at(&minted) > at(replaced), "{minted} over {replaced}");
}
}
assert_eq!(stored_at_over("2020-01-01T00:00:00Z", "2099-01-01T00:00:00Z"), "2099-01-01T00:00:00.000001Z");
// An unparseable stamp on either side is never kept: the minted stamp is the replaced's successor or now.
let minted = stored_at_over("not a stamp", "2099-01-01T00:00:00Z");
assert_eq!(minted, "2099-01-01T00:00:00.000001Z");
let minted = stored_at_over("2099-01-01T00:00:00Z", "not a stamp");
assert!(is_minted_text(&minted));
assert!((chrono::Utc::now() - at(&minted).with_timezone(&chrono::Utc)).num_seconds().abs() < 60, "{minted}");
assert!(is_minted_text(&mint_stored_at(None, chrono::Utc::now())));
}
#[test]
fn failed_snapshot_publication_preserves_the_committed_destination() -> Result<()> {
let temp = tempfile::tempdir()?;

26
packages/cultcache-rs/tests/content_version_vector.rs Normal file
View file

@ -0,0 +1,26 @@
//! `tests/vectors/content-version`, written only by the C# reference: the content version of a stored record is the lowercase
//! hex SHA-256 of its persisted encoding (`variants:ruling:write-set-content-version-declared-ownership`). The store reports
//! the version of every record of the committed `store.cc` as `versions.msgpack` says.
use std::path::PathBuf;
use cultcache_rs::{CacheBackingStore, SingleFileMessagePackBackingStore};
fn vectors() -> PathBuf {
PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../tests/vectors/content-version")
}
#[test]
fn read_reports_every_version_the_reference_reads() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("store.cc");
std::fs::copy(vectors().join("store.cc"), &path).unwrap();
let rows: Vec<(String, String)> = rmp_serde::from_slice(&std::fs::read(vectors().join("versions.msgpack")).unwrap()).unwrap();
assert!(rows.len() >= 2);
let store = SingleFileMessagePackBackingStore::new(&path);
for (key, version) in rows {
let record = store.read(&key).unwrap().unwrap_or_else(|| panic!("no record {key}"));
assert_eq!(record.version, version, "{key}");
}
assert_eq!(store.read("absent").unwrap(), None);
}

58
packages/cultcache-rs/tests/first_call.rs
View file

@ -1,7 +1,9 @@
//! The Rust CultCache first call from docs/start/README.md. The docs show the region between the
//! `snippet` markers verbatim, and `docs_show_the_tested_snippet` fails when they differ.
//! The Rust CultCache first call and update from docs/start/README.md. The docs show each region between its
//! `snippet` markers verbatim, and `docs_show_the_tested_snippets` fails when they differ.
use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore};
use std::path::Path;
use cultcache_rs::{CultCache, CultUpdateResult, DatabaseEntry, SingleFileMessagePackBackingStore};
// snippet:begin
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
@ -40,17 +42,59 @@ fn first_call() -> anyhow::Result<()> {
}
// snippet:end
// snippet:update:begin
fn restock(path: &Path) -> anyhow::Result<CultUpdateResult<Item>> {
let mut cache = CultCache::new();
cache.register_entry_type::<Item>()?;
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path))?;
cache.pull_all_backing_stores()?;
cache.update::<Item, _>("item:potion", |item| {
// The store's copy of the record, or nothing if it has none; this may run again on a lost race.
let mut item = item.unwrap_or(Item {
name: "Potion".to_string(),
category: "Consumable".to_string(),
value: 0,
});
item.value += 10;
Some(item)
})
}
// snippet:update:end
#[test]
fn first_call_round_trips_through_a_second_cache() {
first_call().unwrap();
}
#[test]
fn docs_show_the_tested_snippet() {
fn restock_twice_reads_twenty_through_a_second_cache() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("restock.cc");
assert_eq!(restock(&path).unwrap().outcome, cultcache_rs::CultUpdateOutcome::Updated);
let second = restock(&path).unwrap();
assert_eq!(second.outcome, cultcache_rs::CultUpdateOutcome::Updated);
assert_eq!(second.document.unwrap().value, 20);
let mut reopened = CultCache::new();
reopened.register_entry_type::<Item>().unwrap();
reopened.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
reopened.pull_all_backing_stores().unwrap();
assert_eq!(reopened.get_required::<Item>("item:potion").unwrap().value, 20);
}
/// The docs show the region between `// snippet:<region>begin` and `end` of this file as one fenced block.
fn docs_show_region(region: &str) {
let source = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/first_call.rs")).unwrap();
let begin = source.find("// snippet:begin\n").unwrap() + "// snippet:begin\n".len();
let end = source.find("// snippet:end\n").unwrap();
let (begin_marker, end_marker) = (format!("// snippet:{region}begin\n"), format!("// snippet:{region}end\n"));
let begin = source.find(&begin_marker).unwrap() + begin_marker.len();
let end = source.find(&end_marker).unwrap();
let docs = std::fs::read_to_string(concat!(env!("CARGO_MANIFEST_DIR"), "/../../docs/start/README.md")).unwrap();
let block = format!("```rust\n{}```", &source[begin..end]);
assert!(docs.contains(&block), "docs/start/README.md no longer shows tests/first_call.rs verbatim");
assert!(docs.contains(&block), "docs/start/README.md no longer shows the {region:?} region of tests/first_call.rs verbatim");
}
#[test]
fn docs_show_the_tested_snippets() {
docs_show_region("");
docs_show_region("update:");
}

657
packages/cultcache-rs/tests/update.rs Normal file
View file

@ -0,0 +1,657 @@
//! `CultCache::update`: read the record as its home store holds it, apply, commit only over what was read, rerun on a lost race
//! (C# `CultCache.Update`, as amended by PR #170). A rival is a second `CultCache` on the same file.
use std::cell::Cell;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Barrier};
use cultcache_rs::{
CacheBackingStore, CultCache, CultCacheEnvelope, CultCommitIf, CultStoredRecord, CultUpdateOutcome, DatabaseEntry,
PushAllOptions, RedbMessagePackBackingStore, SingleFileMessagePackBackingStore,
};
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(type = "counter")]
struct Counter {
#[cultcache(key = 0)]
n: u32,
}
/// The same type with a second slot: a record it writes carries a slot `Counter` does not read.
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(type = "counter")]
struct WideCounter {
#[cultcache(key = 0)]
n: u32,
#[cultcache(key = 1)]
extra: u32,
}
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(type = "note")]
struct Note {
#[cultcache(key = 0)]
text: String,
}
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(type = "unique_sku")]
struct Sku {
#[cultcache(key = 0, index = "Sku")]
sku: String,
}
fn counter(n: u32) -> Counter {
Counter { n }
}
fn open(path: &Path) -> CultCache {
let mut cache = CultCache::new();
cache.register_entry_type::<Counter>().unwrap();
cache.register_entry_type::<Note>().unwrap();
cache.register_entry_type::<Sku>().unwrap();
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(path)).unwrap();
cache.pull_all_backing_stores().unwrap();
cache
}
fn path_in(dir: &tempfile::TempDir) -> PathBuf {
dir.path().join("store.cc")
}
/// What a fresh cache reads at `key`.
fn stored(path: &Path, key: &str) -> Option<u32> {
open(path).get::<Counter>(key).unwrap().map(|counter| counter.n)
}
fn file(path: &Path) -> Option<Vec<u8>> {
std::fs::read(path).ok()
}
#[test]
fn two_writers_updating_lose_no_increment() {
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let calls = Arc::new(AtomicUsize::new(0));
let start = Arc::new(Barrier::new(2));
let threads: Vec<_> = (0..2)
.map(|_| {
let (path, calls, start) = (path.clone(), calls.clone(), start.clone());
std::thread::spawn(move || {
let mut cache = open(&path);
start.wait();
for _ in 0..50 {
let result = cache
.update_with_attempts::<Counter, _>("c", 100, |current| {
calls.fetch_add(1, Ordering::SeqCst);
// The window a rival can land in.
std::thread::sleep(std::time::Duration::from_millis(1));
Some(counter(current.map_or(0, |held| held.n) + 1))
})
.unwrap();
assert_eq!(result.outcome, CultUpdateOutcome::Updated);
}
})
})
.collect();
for thread in threads {
thread.join().unwrap();
}
assert_eq!(stored(&path, "c"), Some(100));
assert!(calls.load(Ordering::SeqCst) > 100, "no apply was rerun, so no race was exercised");
}
#[test]
fn a_lost_attempt_reruns_apply_on_the_fresh_value() {
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut cache = open(&path);
cache.put("c", &counter(1)).unwrap();
let mut rival = open(&path);
let mut seen = Vec::new();
let result = cache
.update::<Counter, _>("c", |current| {
let current = current.unwrap().n;
if seen.is_empty() {
rival.put("c", &counter(100)).unwrap();
}
seen.push(current);
Some(counter(current + 1))
})
.unwrap();
assert_eq!(seen, [1, 100]);
assert_eq!(result.outcome, CultUpdateOutcome::Updated);
assert_eq!(result.document, Some(counter(101)));
assert_eq!(stored(&path, "c"), Some(101));
}
#[test]
fn exhausted_attempts_return_lost_race() {
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut cache = open(&path);
cache.put("c", &counter(1)).unwrap();
let mut rival = open(&path);
let mut calls = 0;
let result = cache
.update_with_attempts::<Counter, _>("c", 3, |current| {
calls += 1;
rival.put("c", &counter(100 * calls)).unwrap();
current
})
.unwrap();
assert_eq!(calls, 3);
assert_eq!(result.outcome, CultUpdateOutcome::LostRace);
// The pull before LostRace shows the value that won.
assert_eq!(result.document, Some(counter(300)));
assert_eq!(stored(&path, "c"), Some(300));
}
#[test]
fn max_attempts_bounds() {
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut cache = open(&path);
cache.put("c", &counter(1)).unwrap();
let mut rival = open(&path);
// Always loses; panics past 10 calls so a weakened bound fails here instead of hanging.
let mut losing = |calls: &Cell<u32>| {
let n = calls.get() + 1;
calls.set(n);
assert!(n <= 10, "apply ran past the bound");
rival.put("c", &counter(1000 + n)).unwrap();
Some(counter(n))
};
let calls = Cell::new(0);
let one = cache.update_with_attempts::<Counter, _>("c", 1, |_| losing(&calls)).unwrap();
assert_eq!((calls.get(), one.outcome), (1, CultUpdateOutcome::LostRace));
let calls = Cell::new(0);
assert!(cache.update_with_attempts::<Counter, _>("c", 0, |_| losing(&calls)).is_err());
assert_eq!(calls.get(), 0);
// The default bound is 5.
let calls = Cell::new(0);
let default = cache.update::<Counter, _>("c", |_| losing(&calls)).unwrap();
assert_eq!((calls.get(), default.outcome), (5, CultUpdateOutcome::LostRace));
}
#[test]
fn none_declines_and_writes_nothing() {
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut cache = open(&path);
let absent = cache.update::<Counter, _>("c", |current| {
assert_eq!(current, None);
None
});
let absent = absent.unwrap();
assert_eq!((absent.outcome, absent.document), (CultUpdateOutcome::Declined, None));
assert_eq!(file(&path), None, "a declined update created the store");
cache.put("c", &counter(1)).unwrap();
let before = file(&path);
let held = cache.update::<Counter, _>("c", |_| None).unwrap();
assert_eq!((held.outcome, held.document), (CultUpdateOutcome::Declined, Some(counter(1))));
assert_eq!(file(&path), before);
}
#[test]
fn a_declined_update_installs_nothing() {
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut cache = open(&path);
cache.put("c", &counter(1)).unwrap();
let mut rival = open(&path);
rival.put("c", &counter(7)).unwrap();
let mut seen = None;
let result = cache
.update::<Counter, _>("c", |current| {
seen = current;
None
})
.unwrap();
assert_eq!(seen, Some(counter(7)), "apply saw the held copy, not the store's");
assert_eq!(result.document, Some(counter(1)));
assert_eq!(cache.get::<Counter>("c").unwrap(), Some(counter(1)));
cache.pull_all_backing_stores().unwrap();
assert_eq!(cache.get::<Counter>("c").unwrap(), Some(counter(7)));
}
#[test]
fn absent_key_rival_removal_and_a_store_holding_nothing() {
// Two writers creating one absent key.
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let start = Arc::new(Barrier::new(2));
let threads: Vec<_> = (0..2)
.map(|_| {
let (path, start) = (path.clone(), start.clone());
std::thread::spawn(move || {
let mut cache = open(&path);
start.wait();
let result = cache
.update_with_attempts::<Counter, _>("new", 20, |current| {
std::thread::sleep(std::time::Duration::from_millis(1));
Some(counter(current.map_or(0, |held| held.n) + 1))
})
.unwrap();
assert_eq!(result.outcome, CultUpdateOutcome::Updated);
})
})
.collect();
for thread in threads {
thread.join().unwrap();
}
assert_eq!(stored(&path, "new"), Some(2));
// A rival deletes the record inside apply: the retry sees nothing.
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut cache = open(&path);
cache.put("c", &counter(1)).unwrap();
let mut rival = open(&path);
let mut seen = Vec::new();
let result = cache
.update::<Counter, _>("c", |current| {
if seen.is_empty() {
assert!(rival.delete::<Counter>("c").unwrap());
}
seen.push(current.clone());
Some(counter(current.map_or(0, |held| held.n) + 1))
})
.unwrap();
assert_eq!(seen, [Some(counter(1)), None]);
assert_eq!(result.outcome, CultUpdateOutcome::Updated);
assert_eq!(stored(&path, "c"), Some(1));
// The cache holds a record the store no longer has: apply is handed None.
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut cache = open(&path);
cache.put("c", &counter(5)).unwrap();
assert!(open(&path).delete::<Counter>("c").unwrap());
let mut seen = Vec::new();
cache
.update::<Counter, _>("c", |current| {
seen.push(current);
Some(counter(1))
})
.unwrap();
assert_eq!(seen, [None]);
assert_eq!(stored(&path, "c"), Some(1));
}
#[test]
fn a_rival_writing_and_restoring_the_value_still_reruns_apply() {
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut cache = open(&path);
cache.put("c", &counter(1)).unwrap();
let mut rival = open(&path);
let mut seen = Vec::new();
cache
.update::<Counter, _>("c", |current| {
let current = current.unwrap().n;
if seen.is_empty() {
rival.put("c", &counter(5)).unwrap();
rival.put("c", &counter(1)).unwrap();
}
seen.push(current);
Some(counter(current + 1))
})
.unwrap();
assert_eq!(seen, [1, 1]);
assert_eq!(stored(&path, "c"), Some(2));
}
#[test]
fn a_stale_caches_first_attempt_reads_the_store() {
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut a = open(&path);
a.put("c", &counter(1)).unwrap();
let mut b = open(&path);
a.put("c", &counter(7)).unwrap();
let mut seen = Vec::new();
b.update::<Counter, _>("c", |current| {
seen.push(current.clone());
current.map(|held| counter(held.n + 1))
})
.unwrap();
assert_eq!(seen, [Some(counter(7))]);
assert_eq!(stored(&path, "c"), Some(8));
}
/// A custom store over the single-file store that has no `read` or `commit_if`.
struct ReadlessStore(SingleFileMessagePackBackingStore);
impl CacheBackingStore for ReadlessStore {
fn pull_all(&self) -> anyhow::Result<Vec<CultCacheEnvelope>> {
self.0.pull_all()
}
fn push(&mut self, entry: &CultCacheEnvelope) -> anyhow::Result<String> {
self.0.push(entry)
}
fn delete(&mut self, entry: &CultCacheEnvelope) -> anyhow::Result<()> {
self.0.delete(entry)
}
fn push_all(&mut self, entries: &[CultCacheEnvelope], options: PushAllOptions) -> anyhow::Result<Vec<String>> {
self.0.push_all(entries, options)
}
}
/// Runs `update` with an apply that counts its calls; the refusal must come before the first.
fn refused(cache: &mut CultCache, path: &Path, key: &str) -> String {
let before = file(path);
let calls = Cell::new(0);
let error = cache
.update::<Counter, _>(key, |current| {
calls.set(calls.get() + 1);
current
})
.expect_err("the update must be refused");
assert_eq!(calls.get(), 0, "apply ran before the refusal: {error:#}");
assert_eq!(file(path), before, "a refused update touched the store");
format!("{error:#}")
}
#[test]
fn refusals_come_before_apply() {
// Another type's record written at the key since the pull: the read resolves it.
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut cache = open(&path);
open(&path).put("k", &Note { text: "n".into() }).unwrap();
let text = refused(&mut cache, &path, "k");
assert!(text.contains("update cannot copy it"), "{text}");
// Another type held at the key.
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut cache = open(&path);
cache.put("k", &Note { text: "n".into() }).unwrap();
refused(&mut cache, &path, "k");
// A cache with no store holds another type at the key.
let mut memory = CultCache::new();
memory.register_entry_type::<Counter>().unwrap();
memory.register_entry_type::<Note>().unwrap();
memory.put("k", &Note { text: "n".into() }).unwrap();
let calls = Cell::new(0);
assert!(memory
.update::<Counter, _>("k", |current| {
calls.set(calls.get() + 1);
current
})
.is_err());
assert_eq!(calls.get(), 0);
// A foreign record (no registered type claims it) held at the key, and one that appears after the pull.
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
open(&path).put("k", &Note { text: "n".into() }).unwrap();
let mut only_counter = CultCache::new();
only_counter.register_entry_type::<Counter>().unwrap();
only_counter.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
only_counter.pull_all_backing_stores().unwrap();
refused(&mut only_counter, &path, "k");
let mut later = CultCache::new();
later.register_entry_type::<Counter>().unwrap();
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
later.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
later.pull_all_backing_stores().unwrap();
open(&path).put("z", &Note { text: "n".into() }).unwrap();
let text = refused(&mut later, &path, "z");
assert!(text.contains("update cannot copy it"), "{text}");
// A redb home and a custom store without read.
let dir = tempfile::tempdir().unwrap();
let redb_path = dir.path().join("store.redb");
let mut redb = CultCache::new();
redb.register_entry_type::<Counter>().unwrap();
redb.add_generic_backing_store(RedbMessagePackBackingStore::new(&redb_path).unwrap()).unwrap();
redb.pull_all_backing_stores().unwrap();
redb.put("c", &counter(1)).unwrap();
let text = refused(&mut redb, &redb_path, "c");
assert!(text.contains("read and commit_if"), "{text}");
let path = path_in(&dir);
let mut custom = CultCache::new();
custom.register_entry_type::<Counter>().unwrap();
custom.add_generic_backing_store(ReadlessStore(SingleFileMessagePackBackingStore::new(&path))).unwrap();
custom.pull_all_backing_stores().unwrap();
custom.put("c", &counter(1)).unwrap();
let text = refused(&mut custom, &path, "c");
assert!(text.contains("read and commit_if"), "{text}");
// A held read-only record, and a store record with slots the cache never loaded.
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut wide = CultCache::new();
wide.register_entry_type::<WideCounter>().unwrap();
wide.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
wide.pull_all_backing_stores().unwrap();
wide.put("held", &WideCounter { n: 1, extra: 2 }).unwrap();
let mut cache = open(&path);
let text = refused(&mut cache, &path, "held");
assert!(text.contains("held"), "{text}");
wide.put("unloaded", &WideCounter { n: 1, extra: 2 }).unwrap();
refused(&mut cache, &path, "unloaded");
// An unpulled cache.
let mut unpulled = CultCache::new();
unpulled.register_entry_type::<Counter>().unwrap();
unpulled.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&path)).unwrap();
refused(&mut unpulled, &path, "held");
}
#[test]
fn a_refusal_after_apply_is_not_retried() {
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut cache = open(&path);
cache.put("a", &Sku { sku: "S".into() }).unwrap();
cache.put("b", &Sku { sku: "T".into() }).unwrap();
let before = file(&path);
let calls = Cell::new(0);
let error = cache
.update::<Sku, _>("b", |_| {
calls.set(calls.get() + 1);
Some(Sku { sku: "S".into() })
})
.expect_err("a unique-index clash");
assert_eq!(calls.get(), 1, "{error:#}");
assert_eq!(file(&path), before);
}
#[test]
fn own_writes_and_the_store_less_cache() {
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let mut cache = open(&path);
cache.put("c", &counter(1)).unwrap();
let calls = Cell::new(0);
for expected in [2, 3] {
let result = cache
.update::<Counter, _>("c", |current| {
calls.set(calls.get() + 1);
current.map(|held| counter(held.n + 1))
})
.unwrap();
assert_eq!(result.outcome, CultUpdateOutcome::Updated);
assert_eq!(result.document, Some(counter(expected)));
assert_eq!(result.document, cache.get::<Counter>("c").unwrap());
}
assert_eq!(calls.get(), 2);
let mut memory = CultCache::new();
memory.register_entry_type::<Counter>().unwrap();
let calls = Cell::new(0);
let result = memory
.update::<Counter, _>("c", |current| {
calls.set(calls.get() + 1);
Some(counter(current.map_or(0, |held| held.n) + 1))
})
.unwrap();
assert_eq!((calls.get(), result.outcome, result.document), (1, CultUpdateOutcome::Updated, Some(counter(1))));
}
#[test]
fn the_ledger_holds_under_contention() {
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let start = Arc::new(Barrier::new(3));
let threads: Vec<_> = (0..3)
.map(|_| {
let (path, start) = (path.clone(), start.clone());
std::thread::spawn(move || {
let mut cache = open(&path);
start.wait();
(0..25)
.filter(|_| {
let result = cache
.update::<Counter, _>("c", |current| Some(counter(current.map_or(0, |held| held.n) + 1)))
.unwrap();
result.outcome == CultUpdateOutcome::Updated
})
.count() as u32
})
})
.collect();
let updated: u32 = threads.into_iter().map(|thread| thread.join().unwrap()).sum();
assert_eq!(stored(&path, "c"), Some(updated));
}
fn envelope_of<T: DatabaseEntry>(cache: &CultCache, key: &str, value: &T, stamp: &str) -> CultCacheEnvelope {
let (mut entry, _) = cache.prepare_entry(key, value).unwrap();
entry.stored_at = stamp.to_string();
entry
}
const FUTURE: &str = "2099-01-01T00:00:00Z";
fn later_than_future(stamp: &str) -> bool {
chrono::DateTime::parse_from_rfc3339(stamp).unwrap() > chrono::DateTime::parse_from_rfc3339(FUTURE).unwrap()
}
#[test]
fn commit_if_writes_only_over_the_version_it_was_given() {
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let cache = open(&path);
let mut store = SingleFileMessagePackBackingStore::new(&path);
assert_eq!(store.read("c").unwrap(), None);
let one = envelope_of(&cache, "c", &counter(1), FUTURE);
// Against an absent key, expecting a version mismatches; expecting nothing commits.
assert_eq!(store.commit_if(&one, Some("0000")).unwrap(), CultCommitIf::Mismatch);
assert_eq!(file(&path), None);
assert_eq!(store.commit_if(&one, None).unwrap(), CultCommitIf::Committed { stored_at: FUTURE.to_string() });
let CultStoredRecord { envelope, version } = store.read("c").unwrap().unwrap();
assert_eq!(envelope.stored_at, FUTURE);
let before = file(&path);
let two = envelope_of(&cache, "c", &counter(2), FUTURE);
assert_eq!(store.commit_if(&two, None).unwrap(), CultCommitIf::Mismatch);
assert_eq!(store.commit_if(&two, Some("not-the-version")).unwrap(), CultCommitIf::Mismatch);
assert_eq!(file(&path), before, "a mismatch wrote");
let CultCommitIf::Committed { stored_at } = store.commit_if(&two, Some(&version)).unwrap() else { panic!("the version matches") };
assert!(later_than_future(&stored_at), "{stored_at}");
assert_eq!(stored(&path, "c"), Some(2));
}
#[test]
fn the_cache_facing_writes_stamp_over_a_replaced_record() {
let seeded = |dir: &tempfile::TempDir| {
let path = path_in(dir);
let seed = open(&path);
let mut store = SingleFileMessagePackBackingStore::new(&path);
store.push(&envelope_of(&seed, "c", &counter(1), FUTURE)).unwrap();
(path, seed)
};
let on_disk = |path: &Path| {
let mut store = SingleFileMessagePackBackingStore::new(path);
let stamp = store.read("c").unwrap().unwrap().envelope.stored_at;
let _ = &mut store;
stamp
};
// The store: push, push_all and commit_if each return the stamp they wrote, strictly later than what they replaced.
let dir = tempfile::tempdir().unwrap();
let (path, seed) = seeded(&dir);
let mut store = SingleFileMessagePackBackingStore::new(&path);
let stamp = store.push(&envelope_of(&seed, "c", &counter(2), FUTURE)).unwrap();
assert!(later_than_future(&stamp) && stamp == on_disk(&path), "push: {stamp}");
let stamps = store.push_all(&[envelope_of(&seed, "c", &counter(3), FUTURE)], PushAllOptions::default()).unwrap();
assert!(later_than_future(&stamps[0]) && stamps[0] == on_disk(&path), "push_all: {stamps:?}");
let version = store.read("c").unwrap().unwrap().version;
let CultCommitIf::Committed { stored_at } = store.commit_if(&envelope_of(&seed, "c", &counter(4), FUTURE), Some(&version)).unwrap()
else {
panic!("matching version")
};
assert!(later_than_future(&stored_at) && stored_at == on_disk(&path), "commit_if: {stored_at}");
// One stamp per input, in order, a duplicate getting the stamp of the entry that landed; a soft write that wrote nothing none.
let stamps = store
.push_all(
&[envelope_of(&seed, "c", &counter(5), FUTURE), envelope_of(&seed, "d", &counter(6), FUTURE), envelope_of(&seed, "c", &counter(7), FUTURE)],
PushAllOptions::default(),
)
.unwrap();
assert_eq!(stamps.len(), 3);
assert_eq!(stamps[0], stamps[2]);
assert_eq!(stamps[1], FUTURE, "a record replacing nothing keeps its stamp");
assert_eq!(stamps[0], on_disk(&path));
assert!(store.push_all(&[envelope_of(&seed, "e", &counter(1), FUTURE)], PushAllOptions { soft: true }).unwrap().is_empty());
// The cache holds the stamp the store wrote after every write entry.
type Write = fn(&mut CultCache) -> ();
let writes: [(&str, Write); 5] = [
("put", |cache| drop(cache.put("c", &counter(9)).unwrap())),
("put_envelope", |cache| {
let entry = envelope_of(cache, "c", &counter(9), FUTURE);
drop(cache.put_envelope::<Counter>(entry).unwrap())
}),
("put_raw_envelope", |cache| {
let entry = envelope_of(cache, "c", &counter(9), FUTURE);
cache.put_raw_envelope(entry).unwrap()
}),
("put_prepared_batch", |cache| {
let entry = envelope_of(cache, "c", &counter(9), FUTURE);
cache.put_prepared_batch(vec![entry]).unwrap()
}),
("update", |cache| drop(cache.update::<Counter, _>("c", |_| Some(counter(9))).unwrap())),
];
for (name, write) in writes {
let dir = tempfile::tempdir().unwrap();
let (path, _) = seeded(&dir);
let mut cache = open(&path);
write(&mut cache);
let held = cache.get_envelope::<Counter>("c").unwrap().unwrap().stored_at;
assert!(later_than_future(&held), "{name}: {held}");
assert_eq!(held, on_disk(&path), "{name}: the cache holds a stamp the store did not write");
}
}
#[test]
fn the_byte_exact_writers_keep_the_callers_stamp() {
let dir = tempfile::tempdir().unwrap();
let path = path_in(&dir);
let cache = open(&path);
let mut store = SingleFileMessagePackBackingStore::new(&path);
let held = envelope_of(&cache, "c", &counter(1), FUTURE);
store.push(&held).unwrap();
let swapped = envelope_of(&cache, "c", &counter(2), "2026-01-01T00:00:00Z");
assert!(store.compare_and_swap_entry(&held, swapped.clone()).unwrap());
assert_eq!(store.read("c").unwrap().unwrap().envelope, swapped, "compare_and_swap_entry restamped");
let redb_path = dir.path().join("store.redb");
let mut redb = RedbMessagePackBackingStore::new(&redb_path).unwrap();
redb.push(&held).unwrap();
let older = envelope_of(&cache, "c", &counter(3), "2026-01-01T00:00:00Z");
assert_eq!(redb.push(&older).unwrap(), older.stored_at);
assert_eq!(redb.pull_all().unwrap(), vec![older.clone()]);
assert_eq!(redb.push_all(std::slice::from_ref(&held), PushAllOptions::default()).unwrap(), vec![held.stored_at.clone()]);
assert_eq!(redb.pull_all().unwrap(), vec![held]);
}