diff --git a/Cargo.lock b/Cargo.lock index b19d785..528ac10 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -391,31 +391,15 @@ version = "0.1.0" dependencies = [ "anyhow", "chrono", - "cultcache-rs-derive 0.1.0", + "cultcache-rs-derive", "fs2", "pretty_assertions", - "rmp-serde", - "serde", - "serde_bytes", - "serde_json", - "tempfile", - "uuid", -] - -[[package]] -name = "cultcache-rs" -version = "0.1.0" -source = "git+https://github.com/GameCult/cultcache-rs.git?rev=0ba87453115e3d44e8fcb7419866cd7dd70fb096#0ba87453115e3d44e8fcb7419866cd7dd70fb096" -dependencies = [ - "anyhow", - "chrono", - "cultcache-rs-derive 0.1.0 (git+https://github.com/GameCult/cultcache-rs.git?rev=0ba87453115e3d44e8fcb7419866cd7dd70fb096)", - "fs2", "redb", "rmp-serde", "serde", "serde_bytes", "serde_json", + "tempfile", "uuid", "windows-sys 0.59.0", ] @@ -429,22 +413,12 @@ dependencies = [ "syn 2.0.119", ] -[[package]] -name = "cultcache-rs-derive" -version = "0.1.0" -source = "git+https://github.com/GameCult/cultcache-rs.git?rev=0ba87453115e3d44e8fcb7419866cd7dd70fb096#0ba87453115e3d44e8fcb7419866cd7dd70fb096" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.119", -] - [[package]] name = "cultmesh-rs" version = "0.1.0" dependencies = [ "anyhow", - "cultcache-rs 0.1.0", + "cultcache-rs", "cultnet-rs", "pretty_assertions", "rmp-serde", @@ -460,7 +434,7 @@ dependencies = [ "anyhow", "base64 0.22.1", "chrono", - "cultcache-rs 0.1.0", + "cultcache-rs", "ed25519-dalek", "hmac", "pretty_assertions", @@ -817,8 +791,7 @@ dependencies = [ "base64 0.22.1", "chrono", "codex-connector", - "cultcache-rs 0.1.0", - "cultcache-rs 0.1.0 (git+https://github.com/GameCult/cultcache-rs.git?rev=0ba87453115e3d44e8fcb7419866cd7dd70fb096)", + "cultcache-rs", "cultmesh-rs", "cultnet-rs", "ghostlight-persona-projection", diff --git a/Cargo.toml b/Cargo.toml index 64d35a5..ce04448 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,7 +23,6 @@ axum = { version = "0.8", features = ["macros"] } chrono = { version = "0.4", features = ["serde"] } base64 = "0.22" codex-connector = { git = "https://github.com/GameCult/CodexConnector.git", rev = "68fe94b9475823de92ed766a97ea962d48723708" } -cultcache-legacy = { package = "cultcache-rs", git = "https://github.com/GameCult/cultcache-rs.git", rev = "0ba87453115e3d44e8fcb7419866cd7dd70fb096" } cultcache-rs = { path = "vendor/cultcache-rs" } cultnet-rs = { path = "vendor/cultnet-rs" } cultmesh-rs = { path = "vendor/cultmesh-rs" } diff --git a/crates/ghostlight-dungeon/Cargo.toml b/crates/ghostlight-dungeon/Cargo.toml index 510c53f..c1204cb 100644 --- a/crates/ghostlight-dungeon/Cargo.toml +++ b/crates/ghostlight-dungeon/Cargo.toml @@ -18,7 +18,6 @@ axum.workspace = true chrono.workspace = true base64.workspace = true codex-connector.workspace = true -cultcache-legacy.workspace = true cultcache-rs.workspace = true cultnet-rs.workspace = true cultmesh-rs.workspace = true diff --git a/crates/ghostlight-dungeon/src/app_session.rs b/crates/ghostlight-dungeon/src/app_session.rs index 769c2a4..7301712 100644 --- a/crates/ghostlight-dungeon/src/app_session.rs +++ b/crates/ghostlight-dungeon/src/app_session.rs @@ -8,7 +8,7 @@ use aes_gcm::{Aes256Gcm, KeyInit, aead::Aead}; use anyhow::{Context, bail}; use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD}; use chrono::{DateTime, Utc}; -use cultcache_legacy::{CacheBackingStore, CultCacheEnvelope, OwnedRedbMessagePackBackingStore}; +use cultcache_rs::{CacheBackingStore, CultCacheEnvelope, OwnedRedbMessagePackBackingStore}; use rand::RngCore; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -92,7 +92,7 @@ impl AppSessionOwner { sessions: BTreeMap::new(), }; let row = envelope(&state)?; - if !store.compare_and_swap_batch(&[], std::slice::from_ref(&row))? { + if !store.compare_and_swap_batch(&[], vec![row.clone()])? { bail!("app-session store changed during initialization"); } (row, state) @@ -341,10 +341,9 @@ impl AppSessionOwner { self.ensure_owned()?; prune_sessions(&mut next, Utc::now()); let next_row = envelope(&next)?; - let swapped = self.store.compare_and_swap_batch( - std::slice::from_ref(&self.row), - std::slice::from_ref(&next_row), - ); + let swapped = self + .store + .compare_and_swap_batch(std::slice::from_ref(&self.row), vec![next_row.clone()]); if !matches!(swapped, Ok(true)) { return self.poison(anyhow::anyhow!( "app-session commit outcome is uncertain; sole ownership was lost" @@ -561,7 +560,7 @@ mod tests { stored_at: Utc::now().to_rfc3339(), schema_id: Some(STORE_SCHEMA.into()), }; - assert!(store.compare_and_swap_batch(&[], &[row]).unwrap()); + assert!(store.compare_and_swap_batch(&[], vec![row]).unwrap()); drop(store); assert!(AppSessionOwner::open(&store_path, &key).is_err()); } @@ -646,7 +645,7 @@ mod tests { stored_at: Utc::now().to_rfc3339(), schema_id: Some("ghostlight.app_session_store.v1".into()), }; - assert!(store.compare_and_swap_batch(&[], &[foreign]).unwrap()); + assert!(store.compare_and_swap_batch(&[], vec![foreign]).unwrap()); assert!(AppSessionOwner::open(&store_path, &key).is_err()); } @@ -724,7 +723,7 @@ mod tests { .unwrap() .encrypted_refresh_claim = "not-an-encrypted-claim".into(); let corrupt = envelope(&state).unwrap(); - assert!(store.compare_and_swap_batch(&rows, &[corrupt]).unwrap()); + assert!(store.compare_and_swap_batch(&rows, vec![corrupt]).unwrap()); drop(store); assert!(AppSessionOwner::open(&path, &key).is_err()); diff --git a/crates/ghostlight-dungeon/src/world/controllers.rs b/crates/ghostlight-dungeon/src/world/controllers.rs index 1f4bcc5..c924c84 100644 --- a/crates/ghostlight-dungeon/src/world/controllers.rs +++ b/crates/ghostlight-dungeon/src/world/controllers.rs @@ -19,7 +19,7 @@ use codex_connector::{ CodexTransportEventPayload, CodexTransportInvocation, CodexTransportOutcome, provider_request_sha256, }; -use cultcache_legacy::{CacheBackingStore, CultCacheEnvelope, OwnedRedbMessagePackBackingStore}; +use cultcache_rs::{CacheBackingStore, CultCacheEnvelope, OwnedRedbMessagePackBackingStore}; use ghostlight_persona_projection::{ CaptureToolFeedback, InterpretationAccumulator, InterpretationFinalization, InterpretationReport, InterpreterPrompt, OperationalAgentPrompt, PersonaPrompt, PersonaTurn, diff --git a/crates/ghostlight-dungeon/src/world/journal.rs b/crates/ghostlight-dungeon/src/world/journal.rs index 0910100..b694310 100644 --- a/crates/ghostlight-dungeon/src/world/journal.rs +++ b/crates/ghostlight-dungeon/src/world/journal.rs @@ -3,7 +3,7 @@ use super::{ WorldEffect, WorldId, WorldState, apply_effect, commit_digest, reduce, state_digest, }; use chrono::Utc; -use cultcache_legacy::{CacheBackingStore, CultCacheEnvelope, OwnedRedbMessagePackBackingStore}; +use cultcache_rs::{CacheBackingStore, CultCacheEnvelope, OwnedRedbMessagePackBackingStore}; use serde::{Serialize, de::DeserializeOwned}; use std::{ cell::Cell, diff --git a/vendor/README.md b/vendor/README.md index ef1968a..5b397de 100644 --- a/vendor/README.md +++ b/vendor/README.md @@ -1,10 +1,11 @@ # Vendored GameCult Rust substrate -`cultcache-rs`, `cultnet-rs`, and `cultmesh-rs` are copied together from Odin -commit `d34cb7e914e017e376d6f5fa98d464659baf1d73`. They are one compatible -protocol generation and must be updated as a set. Ghostlight vendors the set so -an immutable Starfire release can be rebuilt from its own exact Git commit. +`cultcache-rs`, `cultnet-rs`, and `cultmesh-rs` are a locally owned compatible +protocol generation. Ghostlight vendors the set so an immutable release can be +rebuilt from its exact Git commit. -The older CultCache revision named `cultcache-legacy` remains confined to the -campaign persistence adapter until campaign stores are migrated. It does not -own CultMesh documents or publication. +The vendored CultCache crate owns both the snapshot store and the pinned redb +store used by application sessions, the world journal, and controller-work +custody. Its redb table, key, and envelope encoding remain compatible with the +existing Ghostlight stores. There is no second CultCache package or persistence +authority in the workspace. diff --git a/vendor/cultcache-rs/Cargo.toml b/vendor/cultcache-rs/Cargo.toml index 3b4a1d6..ef221d1 100644 --- a/vendor/cultcache-rs/Cargo.toml +++ b/vendor/cultcache-rs/Cargo.toml @@ -13,6 +13,7 @@ anyhow = "1" chrono = { version = "0.4", features = ["serde"] } cultcache-rs-derive = { path = "cultcache-rs-derive" } fs2 = "0.4" +redb = "4.1.0" rmp-serde = "1.3" serde = { version = "1", features = ["derive"] } serde_bytes = "0.11" @@ -22,3 +23,6 @@ uuid = { version = "1", features = ["v4"] } [dev-dependencies] pretty_assertions = "1.4" tempfile = "3" + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.59", features = ["Win32_Foundation", "Win32_Storage_FileSystem"] } diff --git a/vendor/cultcache-rs/src/lib.rs b/vendor/cultcache-rs/src/lib.rs index 0354674..0532ca6 100644 --- a/vendor/cultcache-rs/src/lib.rs +++ b/vendor/cultcache-rs/src/lib.rs @@ -11,9 +11,12 @@ use std::fs::OpenOptions; use std::path::Path; use std::path::PathBuf; +mod owned_redb; + extern crate self as cultcache_rs; pub use cultcache_rs_derive::DatabaseEntry; +pub use owned_redb::OwnedRedbMessagePackBackingStore; pub trait DatabaseEntry: Serialize + DeserializeOwned + Clone + Send + 'static { const TYPE: &'static str; diff --git a/vendor/cultcache-rs/src/owned_redb.rs b/vendor/cultcache-rs/src/owned_redb.rs new file mode 100644 index 0000000..0bca612 --- /dev/null +++ b/vendor/cultcache-rs/src/owned_redb.rs @@ -0,0 +1,564 @@ +use crate::{CacheBackingStore, CultCacheEnvelope, PushAllOptions}; +use anyhow::{Context, Result, anyhow, bail}; +use redb::{Database, ReadableDatabase, ReadableTable, TableDefinition}; +use std::collections::BTreeSet; +use std::fs::{self, File, OpenOptions}; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +const REDB_ENVELOPES: TableDefinition<&[u8], &[u8]> = TableDefinition::new("cultcache_envelopes"); + +struct OwnedRedbInner { + database: Database, + _owner_lock: File, + file_identity: String, +} + +/// A pinned redb store for one long-lived service owner. This handle holds the +/// CultCache external lock, an open file handle, and the redb database for its +/// entire lifetime. Clones share that exact ownership authority. +#[derive(Clone)] +pub struct OwnedRedbMessagePackBackingStore { + path: PathBuf, + inner: Arc, +} + +impl std::fmt::Debug for OwnedRedbMessagePackBackingStore { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("OwnedRedbMessagePackBackingStore") + .field("path", &self.path) + .field("file_identity", &self.inner.file_identity) + .finish_non_exhaustive() + } +} + +impl OwnedRedbMessagePackBackingStore { + pub fn new(path: impl Into) -> Result { + let path = path.into(); + if let Some(parent) = path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + { + fs::create_dir_all(parent) + .with_context(|| format!("failed to create {}", parent.display()))?; + } + let lock_path = redb_lock_path(&path); + let owner_lock = OpenOptions::new() + .create(true) + .read(true) + .truncate(false) + .write(true) + .open(&lock_path) + .with_context(|| format!("failed to open {}", lock_path.display()))?; + fs2::FileExt::try_lock_exclusive(&owner_lock).with_context(|| { + format!( + "redb CultCache {} already has an active owner", + path.display() + ) + })?; + + // Give redb ownership of the already pinned file, structurally closing + // path substitution during database creation. The post-open identity + // comparison separately verifies that the pathname still names it. + let database_file = OpenOptions::new() + .create(true) + .read(true) + .truncate(false) + .write(true) + .open(&path) + .with_context(|| format!("failed to pin redb CultCache {}", path.display()))?; + let held_identity = file_identity_from_file(&database_file)?; + let database = Database::builder() + .create_file(database_file) + .with_context(|| format!("failed to open redb CultCache {}", path.display()))?; + let path_identity_file = File::open(&path)?; + let path_identity = file_identity_from_file(&path_identity_file)?; + if held_identity != path_identity { + bail!( + "redb CultCache path identity changed while opening: held {held_identity}, path {path_identity}" + ); + } + let write = database.begin_write()?; + { + write.open_table(REDB_ENVELOPES)?; + } + write.commit()?; + Ok(Self { + path, + inner: Arc::new(OwnedRedbInner { + database, + _owner_lock: owner_lock, + file_identity: held_identity, + }), + }) + } + + pub fn path(&self) -> &Path { + &self.path + } + + pub fn file_identity(&self) -> &str { + &self.inner.file_identity + } + + pub fn require_file_identity(&self, expected: &str) -> Result<()> { + if expected != self.file_identity() { + bail!( + "redb CultCache file identity mismatch: expected {expected}, owned {}", + self.file_identity() + ); + } + Ok(()) + } + + pub fn validate_path_identity(&self) -> Result<()> { + let current_file = File::open(&self.path) + .with_context(|| format!("owned redb path {} is missing", self.path.display()))?; + let current = file_identity_from_file(¤t_file)?; + self.require_file_identity(¤t).with_context(|| { + format!( + "owned redb path {} no longer names the pinned file", + self.path.display() + ) + }) + } + + pub fn compare_and_swap_batch( + &self, + expected: &[CultCacheEnvelope], + replacements: Vec, + ) -> Result { + if replacements.is_empty() { + return Err(anyhow!( + "conditional batch requires a non-empty replacement set" + )); + } + let expected_ids = unique_batch_ids(expected, "expected")?; + let replacement_ids = unique_batch_ids(&replacements, "replacement")?; + if !expected_ids.is_subset(&replacement_ids) { + return Err(anyhow!( + "conditional batch must replace every expected identity" + )); + } + let write = self.inner.database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let mut valid = true; + for row in expected { + if read_redb_entry(&table, row)? != Some(row.clone()) { + valid = false; + break; + } + } + if valid { + for row in &replacements { + if !expected_ids.contains(&entry_id(row)) + && read_redb_entry(&table, row)?.is_some() + { + valid = false; + break; + } + } + } + if valid { + for row in &replacements { + insert_redb_entry(&mut table, row)?; + } + } + valid + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + } + + pub fn append_if_snapshot_unchanged( + &self, + expected_snapshot: &[CultCacheEnvelope], + additions: Vec, + ) -> Result { + if additions.is_empty() { + return Err(anyhow!("conditional snapshot append requires additions")); + } + unique_batch_ids(expected_snapshot, "expected snapshot")?; + let addition_ids = unique_batch_ids(&additions, "snapshot additions")?; + let write = self.inner.database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let mut current = read_all_redb(&table)?; + let mut expected = expected_snapshot.to_vec(); + current.sort_by_key(entry_id); + expected.sort_by_key(entry_id); + let valid = current == expected + && !current + .iter() + .any(|row| addition_ids.contains(&entry_id(row))); + if valid { + for row in &additions { + insert_redb_entry(&mut table, row)?; + } + } + valid + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + } + + pub fn replace_and_append_if_snapshot_unchanged( + &self, + expected_snapshot: &[CultCacheEnvelope], + replacements: Vec, + ) -> Result { + if replacements.is_empty() { + return Err(anyhow!( + "conditional snapshot replacement requires replacements" + )); + } + unique_batch_ids(expected_snapshot, "expected snapshot")?; + unique_batch_ids(&replacements, "snapshot replacements")?; + let write = self.inner.database.begin_write()?; + let matched = { + let mut table = write.open_table(REDB_ENVELOPES)?; + let mut current = read_all_redb(&table)?; + let mut expected = expected_snapshot.to_vec(); + current.sort_by_key(entry_id); + expected.sort_by_key(entry_id); + let valid = current == expected; + if valid { + for row in &replacements { + insert_redb_entry(&mut table, row)?; + } + } + valid + }; + if matched { + write.commit()?; + } else { + write.abort()?; + } + Ok(matched) + } +} + +impl CacheBackingStore for OwnedRedbMessagePackBackingStore { + fn pull_all(&self) -> Result> { + let read = self.inner.database.begin_read()?; + let table = read.open_table(REDB_ENVELOPES)?; + read_all_redb(&table) + } + + fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + let write = self.inner.database.begin_write()?; + { + let mut table = write.open_table(REDB_ENVELOPES)?; + insert_redb_entry(&mut table, entry)?; + } + write.commit()?; + Ok(()) + } + + fn delete(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + let write = self.inner.database.begin_write()?; + { + let mut table = write.open_table(REDB_ENVELOPES)?; + let key = redb_identity(entry)?; + table.remove(key.as_slice())?; + } + write.commit()?; + Ok(()) + } + + fn push_all(&mut self, entries: &[CultCacheEnvelope], _options: PushAllOptions) -> Result<()> { + unique_batch_ids(entries, "push all")?; + let write = self.inner.database.begin_write()?; + { + let mut table = write.open_table(REDB_ENVELOPES)?; + let keys = table + .iter()? + .map(|row| Ok(row?.0.value().to_vec())) + .collect::>>()?; + for key in keys { + table.remove(key.as_slice())?; + } + for entry in entries { + insert_redb_entry(&mut table, entry)?; + } + } + write.commit()?; + Ok(()) + } +} + +fn unique_batch_ids( + entries: &[CultCacheEnvelope], + label: &str, +) -> Result> { + let ids = entries.iter().map(entry_id).collect::>(); + if ids.len() != entries.len() { + return Err(anyhow!( + "conditional batch {label} set contains duplicate identities" + )); + } + Ok(ids) +} + +fn entry_id(entry: &CultCacheEnvelope) -> (String, String) { + (entry.r#type.clone(), entry.key.clone()) +} + +fn redb_lock_path(path: &Path) -> PathBuf { + let mut lock_name = path + .file_name() + .map(|value| value.to_os_string()) + .unwrap_or_else(|| "cultcache.cc".into()); + lock_name.push(".lock"); + path.with_file_name(lock_name) +} + +#[cfg(unix)] +fn file_identity_from_file(file: &File) -> Result { + use std::os::unix::fs::MetadataExt; + let metadata = file.metadata()?; + Ok(format!( + "unix:{:016x}:{:016x}", + metadata.dev(), + metadata.ino() + )) +} + +#[cfg(windows)] +fn file_identity_from_file(file: &File) -> Result { + use std::mem::zeroed; + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::Storage::FileSystem::{ + BY_HANDLE_FILE_INFORMATION, GetFileInformationByHandle, + }; + let mut information: BY_HANDLE_FILE_INFORMATION = unsafe { zeroed() }; + let success = + unsafe { GetFileInformationByHandle(file.as_raw_handle() as _, &mut information) }; + if success == 0 { + return Err(std::io::Error::last_os_error()) + .context("failed to read Windows file identity"); + } + let index = ((information.nFileIndexHigh as u64) << 32) | information.nFileIndexLow as u64; + Ok(format!( + "windows:{:08x}:{index:016x}", + information.dwVolumeSerialNumber + )) +} + +fn redb_identity(entry: &CultCacheEnvelope) -> Result> { + rmp_serde::to_vec(&entry_id(entry)).context("failed to encode CultCache identity") +} + +fn read_redb_entry( + table: &impl ReadableTable<&'static [u8], &'static [u8]>, + identity: &CultCacheEnvelope, +) -> Result> { + let key = redb_identity(identity)?; + table + .get(key.as_slice())? + .map(|value| { + rmp_serde::from_slice(value.value()).context("failed to decode redb CultCache envelope") + }) + .transpose() +} + +fn insert_redb_entry( + table: &mut redb::Table<&[u8], &[u8]>, + entry: &CultCacheEnvelope, +) -> Result<()> { + let key = redb_identity(entry)?; + let value = rmp_serde::to_vec(entry).context("failed to encode redb CultCache envelope")?; + table.insert(key.as_slice(), value.as_slice())?; + Ok(()) +} + +fn read_all_redb( + table: &impl ReadableTable<&'static [u8], &'static [u8]>, +) -> Result> { + let mut entries = table + .iter()? + .map(|row| { + let (_, value) = row?; + rmp_serde::from_slice(value.value()).context("failed to decode redb CultCache envelope") + }) + .collect::>>()?; + entries.sort_by_key(entry_id); + Ok(entries) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_envelope(r#type: &str, key: &str, payload: &[u8]) -> CultCacheEnvelope { + CultCacheEnvelope { + key: key.to_string(), + r#type: r#type.to_string(), + payload: payload.to_vec(), + stored_at: "2026-07-13T00:00:00Z".to_string(), + schema_id: Some(r#type.to_string()), + } + } + + #[test] + fn legacy_redb_identity_remains_tuple_encoded() -> Result<()> { + let entry = test_envelope("model", "current", b"one"); + assert_eq!( + redb_identity(&entry)?, + vec![ + 0x92, 0xa5, b'm', b'o', b'd', b'e', b'l', 0xa7, b'c', b'u', b'r', b'r', b'e', b'n', + b't', + ] + ); + Ok(()) + } + + #[test] + fn owned_redb_clones_share_authority_and_fresh_owner_is_refused() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("owned.cc"); + let owner = OwnedRedbMessagePackBackingStore::new(&path)?; + let clone = owner.clone(); + assert_eq!(owner.file_identity(), clone.file_identity()); + assert!(OwnedRedbMessagePackBackingStore::new(&path).is_err()); + drop(owner); + assert!(OwnedRedbMessagePackBackingStore::new(&path).is_err()); + drop(clone); + assert!(OwnedRedbMessagePackBackingStore::new(&path).is_ok()); + Ok(()) + } + + #[test] + fn owned_redb_file_identity_is_stable_across_writes_and_reopen() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("identity.cc"); + let identity = { + let mut owner = OwnedRedbMessagePackBackingStore::new(&path)?; + let identity = owner.file_identity().to_string(); + owner.push(&test_envelope("model", "current", b"one"))?; + owner.validate_path_identity()?; + owner.require_file_identity(&identity)?; + identity + }; + let reopened = OwnedRedbMessagePackBackingStore::new(&path)?; + assert_eq!(reopened.file_identity(), identity); + assert_eq!(reopened.pull_all()?.len(), 1); + Ok(()) + } + + #[test] + fn owned_redb_identity_helper_refuses_mismatch() -> Result<()> { + let temp = tempfile::tempdir()?; + let owner = OwnedRedbMessagePackBackingStore::new(temp.path().join("mismatch.cc"))?; + assert!(owner.require_file_identity("not-the-owned-file").is_err()); + owner.require_file_identity(owner.file_identity())?; + Ok(()) + } + + #[test] + fn owned_redb_batch_refuses_stale_member_and_commits_success_atomically() -> Result<()> { + let temp = tempfile::tempdir()?; + let mut owner = OwnedRedbMessagePackBackingStore::new(temp.path().join("batch-owned.cc"))?; + let first = test_envelope("model", "first", b"one"); + let second = test_envelope("model", "second", b"one"); + owner.push(&first)?; + owner.push(&second)?; + let first_two = test_envelope("model", "first", b"two"); + let second_two = test_envelope("model", "second", b"two"); + let companion = test_envelope("receipt", "batch", b"committed"); + let stale_second = test_envelope("model", "second", b"stale"); + assert!(!owner.compare_and_swap_batch( + &[first.clone(), stale_second], + vec![first_two.clone(), second_two.clone(), companion.clone()], + )?); + assert_eq!(owner.pull_all()?, vec![first.clone(), second.clone()]); + assert!(owner.compare_and_swap_batch( + &[first, second], + vec![first_two.clone(), second_two.clone(), companion.clone()], + )?); + assert_eq!(owner.pull_all()?, vec![first_two, second_two, companion]); + Ok(()) + } + + #[test] + fn owned_redb_snapshot_append_refuses_stale_or_colliding_state() -> Result<()> { + let temp = tempfile::tempdir()?; + let mut owner = + OwnedRedbMessagePackBackingStore::new(temp.path().join("snapshot-append.cc"))?; + let model = test_envelope("model", "current", b"revision-1"); + owner.push(&model)?; + let stale = owner.pull_all()?; + owner.push(&test_envelope("event", "concurrent", b"live"))?; + assert!(!owner.append_if_snapshot_unchanged( + &stale, + vec![test_envelope("readiness", "proof", b"ready")], + )?); + let exact = owner.pull_all()?; + assert!(owner.append_if_snapshot_unchanged( + &exact, + vec![test_envelope("readiness", "proof", b"ready")], + )?); + let current = owner.pull_all()?; + assert!(!owner.append_if_snapshot_unchanged( + ¤t, + vec![test_envelope("model", "current", b"collision")], + )?); + Ok(()) + } + + #[test] + fn owned_redb_snapshot_replacement_and_append_refuses_concurrent_rows() -> Result<()> { + let temp = tempfile::tempdir()?; + let mut owner = + OwnedRedbMessagePackBackingStore::new(temp.path().join("snapshot-replace.cc"))?; + let current = test_envelope("session", "current", b"active"); + owner.push(¤t)?; + let stale = owner.pull_all()?; + let concurrent = test_envelope("event", "concurrent", b"live"); + owner.push(&concurrent)?; + let completed = test_envelope("session", "current", b"completed"); + let receipt = test_envelope("receipt", "terminal", b"done"); + assert!(!owner.replace_and_append_if_snapshot_unchanged( + &stale, + vec![completed.clone(), receipt.clone()], + )?); + let exact = owner.pull_all()?; + assert!(owner.replace_and_append_if_snapshot_unchanged( + &exact, + vec![completed.clone(), receipt.clone()], + )?); + assert_eq!(owner.pull_all()?, vec![concurrent, receipt, completed]); + Ok(()) + } + + #[test] + fn owned_redb_path_replacement_cannot_redirect_pinned_authority() -> Result<()> { + let temp = tempfile::tempdir()?; + let path = temp.path().join("pinned.cc"); + let displaced = temp.path().join("displaced.cc"); + let mut owner = OwnedRedbMessagePackBackingStore::new(&path)?; + let identity = owner.file_identity().to_string(); + owner.push(&test_envelope("model", "before", b"one"))?; + fs::rename(&path, &displaced)?; + File::create(&path)?; + assert!(owner.validate_path_identity().is_err()); + assert_eq!(owner.file_identity(), identity); + owner.push(&test_envelope("model", "after", b"two"))?; + assert_eq!(owner.pull_all()?.len(), 2); + assert_ne!( + file_identity_from_file(&File::open(&path)?)?, + owner.file_identity() + ); + Ok(()) + } +}