From 27cc8156bcfbbe9d581fb8b65725c49ca44ec098 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 4 Sep 2026 14:54:58 +0200 Subject: [PATCH] Consume CultCache and CultNet from CultLib instead of forking them The CultNet submodule and the in-tree CultCache copy are gone. Both now come from CultLib, and vendor/cultmesh-rs is the only fork left -- it too consumes CultLib, so exactly one version of each crate spans the graph. That mattered: mixing sources would have put two cultnet-rs crates in one build, and the types would not have unified. The in-flight CultMesh RUDP document server comes with it because it is a prerequisite, not a coincidence. The previously committed CultMesh calls assume_connected(), a CultNet method Odin's fork carried and CultLib does not, so the old code cannot build against CultLib at all. The newer code no longer calls it. Verified by reverting the file and watching it fail. Three API differences needed adapting: CultLib's transport options carry a reconnect_policy the fork lacked (7 sites), its create_ack builds the packet straight from ack state and so cannot fail, and its transport profile owns each channel's reliable expiry. Everything but odin-daemon builds; odin-daemon is Unix-only and does not compile on this machine either way. muninn and sleipnir pass 162 tests. The one failure, rudp_media_transport_options_follow_low_latency_profile, expects an Unreliable media channel while every CultNet lineage including both revisions Odin had pinned declares it Reliable, so it failed before this change too. The in-flight idunn-daemon work is deliberately left uncommitted; the committed tree was verified to build without it. Also left alone: the untracked odin-daemon crate and deployment recipe, and the root manifest, whose now-dead vendor excludes are inert. That recipe still requires the vendor/cultnet-rs gitlink and will refuse admission until that line is dropped -- the fix is in the working tree, waiting on that directory being tracked. Co-Authored-By: Claude Opus 5 --- .gitmodules | 3 - Cargo.lock | 30 +- crates/idunn-daemon/Cargo.toml | 4 +- crates/muninn-daemon/Cargo.toml | 2 +- crates/muninn-daemon/src/main.rs | 9 +- crates/odin-core/Cargo.toml | 4 +- crates/sleipnir-daemon/Cargo.toml | 4 +- crates/sleipnir-daemon/src/main.rs | 2 + src/Idunn/README.md | 2 +- vendor/cultcache-rs/.gitignore | 2 - vendor/cultcache-rs/Cargo.toml | 30 - vendor/cultcache-rs/README.md | 332 ----- .../cultcache-rs-derive/Cargo.toml | 13 - .../cultcache-rs-derive/src/lib.rs | 278 ---- .../examples/cultcache_interop.rs | 121 -- vendor/cultcache-rs/src/lib.rs | 1241 ----------------- vendor/cultmesh-rs/Cargo.lock | 149 +- vendor/cultmesh-rs/Cargo.toml | 4 +- vendor/cultmesh-rs/src/lib.rs | 572 ++++++-- .../cultmesh-rs/src/rudp_document_server.rs | 647 +++++++++ .../cultmesh-rs/tests/rudp_document_server.rs | 553 ++++++++ vendor/cultnet-rs | 1 - 22 files changed, 1818 insertions(+), 2185 deletions(-) delete mode 100644 .gitmodules delete mode 100644 vendor/cultcache-rs/.gitignore delete mode 100644 vendor/cultcache-rs/Cargo.toml delete mode 100644 vendor/cultcache-rs/README.md delete mode 100644 vendor/cultcache-rs/cultcache-rs-derive/Cargo.toml delete mode 100644 vendor/cultcache-rs/cultcache-rs-derive/src/lib.rs delete mode 100644 vendor/cultcache-rs/examples/cultcache_interop.rs delete mode 100644 vendor/cultcache-rs/src/lib.rs create mode 100644 vendor/cultmesh-rs/src/rudp_document_server.rs create mode 100644 vendor/cultmesh-rs/tests/rudp_document_server.rs delete mode 160000 vendor/cultnet-rs diff --git a/.gitmodules b/.gitmodules deleted file mode 100644 index efa780e..0000000 --- a/.gitmodules +++ /dev/null @@ -1,3 +0,0 @@ -[submodule "vendor/cultnet-rs"] - path = vendor/cultnet-rs - url = https://github.com/GameCult/cultnet-rs.git diff --git a/Cargo.lock b/Cargo.lock index 5606f06..9aa3bfa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -175,22 +175,25 @@ dependencies = [ [[package]] name = "cultcache-rs" version = "0.1.0" +source = "git+https://github.com/GameCult/CultLib.git?rev=cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585#cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585" dependencies = [ "anyhow", "chrono", "cultcache-rs-derive", "fs2", + "redb", "rmp-serde", "serde", "serde_bytes", "serde_json", "uuid", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] name = "cultcache-rs-derive" version = "0.1.0" +source = "git+https://github.com/GameCult/CultLib.git?rev=cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585#cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585" dependencies = [ "proc-macro2", "quote", @@ -210,6 +213,7 @@ dependencies = [ [[package]] name = "cultnet-rs" version = "0.1.0" +source = "git+https://github.com/GameCult/CultLib.git?rev=cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585#cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585" dependencies = [ "aes-gcm", "anyhow", @@ -638,6 +642,21 @@ dependencies = [ "tempfile", ] +[[package]] +name = "odin-daemon" +version = "0.1.0" +dependencies = [ + "anyhow", + "chrono", + "cultcache-rs", + "cultmesh-rs", + "cultnet-rs", + "fs2", + "rmp-serde", + "serde", + "tempfile", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -775,6 +794,15 @@ dependencies = [ "getrandom 0.3.4", ] +[[package]] +name = "redb" +version = "4.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de6c3b63e007e90ce536ec2ae4690826136a20ec8dbbbb400daef1bb999d2e36" +dependencies = [ + "libc", +] + [[package]] name = "rmp" version = "0.8.15" diff --git a/crates/idunn-daemon/Cargo.toml b/crates/idunn-daemon/Cargo.toml index 4288137..e37dca1 100644 --- a/crates/idunn-daemon/Cargo.toml +++ b/crates/idunn-daemon/Cargo.toml @@ -20,8 +20,8 @@ path = "src/lib.rs" [dependencies] anyhow = "1" cultmesh-rs = { path = "../../vendor/cultmesh-rs" } -cultcache-rs = { path = "../../vendor/cultcache-rs" } -cultnet-rs = { path = "../../vendor/cultnet-rs" } +cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585", package = "cultcache-rs" } +cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585", package = "cultnet-rs" } odin-core = { path = "../odin-core" } rmp-serde = "1.3" sha2 = "0.10" diff --git a/crates/muninn-daemon/Cargo.toml b/crates/muninn-daemon/Cargo.toml index 6a88bfa..eca171c 100644 --- a/crates/muninn-daemon/Cargo.toml +++ b/crates/muninn-daemon/Cargo.toml @@ -16,7 +16,7 @@ path = "src/main.rs" [dependencies] anyhow = "1" cultmesh-rs = { path = "../../vendor/cultmesh-rs" } -cultnet-rs = { path = "../../vendor/cultnet-rs" } +cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585", package = "cultnet-rs" } libc = "0.2" muninn-move-tracker = { path = "../muninn-move-tracker" } muninn-psmoveapi-tracker = { path = "../muninn-psmoveapi-tracker", optional = true } diff --git a/crates/muninn-daemon/src/main.rs b/crates/muninn-daemon/src/main.rs index ac29ce7..11e422a 100644 --- a/crates/muninn-daemon/src/main.rs +++ b/crates/muninn-daemon/src/main.rs @@ -7,7 +7,7 @@ use crate::media_packetizer::{ }; use anyhow::{Context, Result, anyhow}; use cultmesh_rs::{ - CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID, CultMesh, CultMeshNodeOptions, + CultMesh, CultMeshNodeOptions, CultMeshRudpDocumentPublishOptions, CultMeshRudpSnapshotOptions, CultMeshSharedMemoryFrameRing, CultMeshStreamBodyTransport, CultMeshStreamCatalog, CultMeshStreamClock, CultMeshStreamDescriptor, CultMeshStreamKind, @@ -688,6 +688,7 @@ fn run_provider_command_ingress( max_fragment_bytes: Some(1200), max_pending_reliable_packets: Some(64), media_reliable_expire_after_ms: None, + reconnect_policy: None, }, )?; loop { @@ -832,7 +833,6 @@ fn pull_odin_obs_catalog_snapshot(node: &mut cultmesh_rs::CultMeshNode, options: connect_timeout: Duration::from_millis(2000), response_timeout: Duration::from_millis(5000), resend_delay_ms: 15, - connection_id: CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID, ..CultMeshRudpSnapshotOptions::default() }) { eprintln!("Muninn could not pull OBS catalog from Odin: {error:#}"); @@ -5475,6 +5475,7 @@ fn run_hid_controller_rudp_ingress( max_fragment_bytes: Some(HID_CONTROLLER_RUDP_MAX_FRAGMENT_BYTES), max_pending_reliable_packets: Some(256), media_reliable_expire_after_ms: Some(25), + reconnect_policy: None, })?; let mut sources = live_move_state_sources(&options) .into_iter() @@ -6019,6 +6020,7 @@ fn create_hid_controller_stream(options: &Options) -> Result Result<()> { "odin-media-receiver-route".to_string(), "muninn.media-rudp".to_string(), ], - connection_id: CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID, ..CultMeshRudpDocumentPublishOptions::default() }, ) @@ -8975,7 +8976,6 @@ fn publish_capture_command_to_odin( "odin-cultmesh-command-route".to_string(), "muninn.capture-stream".to_string(), ], - connection_id: CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID, ..CultMeshRudpDocumentPublishOptions::default() }, ) @@ -12002,6 +12002,7 @@ mod tests { max_fragment_bytes: Some(1200), max_pending_reliable_packets: Some(64), media_reliable_expire_after_ms: None, + reconnect_policy: None, }, ) .unwrap(); diff --git a/crates/odin-core/Cargo.toml b/crates/odin-core/Cargo.toml index 9e8d08c..5752a4d 100644 --- a/crates/odin-core/Cargo.toml +++ b/crates/odin-core/Cargo.toml @@ -11,10 +11,10 @@ path = "src/lib.rs" [dependencies] anyhow = "1" -cultcache-rs = { path = "../../vendor/cultcache-rs" } +cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585", package = "cultcache-rs" } cultmesh-rs = { path = "../../vendor/cultmesh-rs" } serde_json = "1" -cultnet-rs = { path = "../../vendor/cultnet-rs" } +cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585", package = "cultnet-rs" } serde = { version = "1", features = ["derive"] } [dev-dependencies] diff --git a/crates/sleipnir-daemon/Cargo.toml b/crates/sleipnir-daemon/Cargo.toml index 89294ea..f08651a 100644 --- a/crates/sleipnir-daemon/Cargo.toml +++ b/crates/sleipnir-daemon/Cargo.toml @@ -12,8 +12,8 @@ path = "src/main.rs" [dependencies] anyhow = "1" cultmesh-rs = { path = "../../vendor/cultmesh-rs" } -cultnet-rs = { path = "../../vendor/cultnet-rs" } -cultcache-rs = { path = "../../vendor/cultcache-rs" } +cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585", package = "cultnet-rs" } +cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585", package = "cultcache-rs" } odin-core = { path = "../odin-core" } rmp-serde = "1.3" serde_json = "1" diff --git a/crates/sleipnir-daemon/src/main.rs b/crates/sleipnir-daemon/src/main.rs index 1f4189f..29be8d3 100644 --- a/crates/sleipnir-daemon/src/main.rs +++ b/crates/sleipnir-daemon/src/main.rs @@ -2227,6 +2227,7 @@ fn create_rudp_stream( max_fragment_bytes: Some(HID_RUDP_MAX_FRAGMENT_BYTES), max_pending_reliable_packets: Some(256), media_reliable_expire_after_ms: Some(25), + reconnect_policy: None, })?; transport.connect(Vec::new())?; return Ok(Some(ActiveRudpStream { @@ -2267,6 +2268,7 @@ fn create_rudp_stream( max_fragment_bytes: Some(HID_RUDP_MAX_FRAGMENT_BYTES), max_pending_reliable_packets: Some(256), media_reliable_expire_after_ms: Some(25), + reconnect_policy: None, })?, target: None, last_connect_attempt_at: None, diff --git a/src/Idunn/README.md b/src/Idunn/README.md index b404cf8..5f09e8e 100644 --- a/src/Idunn/README.md +++ b/src/Idunn/README.md @@ -299,7 +299,7 @@ From here on out, the work is not "teach another daemon to speak." It is "delete the witness masks and keep ownership where it belongs." Rust no longer gets to claim the transport is imaginary: -`vendor/cultnet-rs` includes the canonical CNR0 `cultnet.transport.rudp.v0` +CultLib's `cultnet-rs` carries the canonical CNR0 `cultnet.transport.rudp.v0` session path for acknowledged CultNet messages over UDP. That is substrate only. Idunn still marks daemon profiles as `migration-required` until each daemon actually publishes health and command diff --git a/vendor/cultcache-rs/.gitignore b/vendor/cultcache-rs/.gitignore deleted file mode 100644 index 96ef6c0..0000000 --- a/vendor/cultcache-rs/.gitignore +++ /dev/null @@ -1,2 +0,0 @@ -/target -Cargo.lock diff --git a/vendor/cultcache-rs/Cargo.toml b/vendor/cultcache-rs/Cargo.toml deleted file mode 100644 index 17ba246..0000000 --- a/vendor/cultcache-rs/Cargo.toml +++ /dev/null @@ -1,30 +0,0 @@ -[package] -name = "cultcache-rs" -version = "0.1.0" -edition = "2024" -publish = false - -[workspace] -members = ["cultcache-rs-derive"] - -[lib] -name = "cultcache_rs" -path = "src/lib.rs" - -[dependencies] -anyhow = "1" -chrono = { version = "0.4", features = ["serde"] } -cultcache-rs-derive = { path = "cultcache-rs-derive" } -fs2 = "0.4" -rmp-serde = "1.3" -serde = { version = "1", features = ["derive"] } -serde_bytes = "0.11" -serde_json = "1" -uuid = { version = "1", features = ["v4"] } - -[target.'cfg(windows)'.dependencies] -windows-sys = { version = "0.61", features = ["Win32_Storage_FileSystem"] } - -[dev-dependencies] -pretty_assertions = "1.4" -tempfile = "3" diff --git a/vendor/cultcache-rs/README.md b/vendor/cultcache-rs/README.md deleted file mode 100644 index b21c324..0000000 --- a/vendor/cultcache-rs/README.md +++ /dev/null @@ -1,332 +0,0 @@ -# CultCache Rust - -`cultcache-rs` is the Rust port of the useful part of GameCult's CultCache: -a polymorphism-aware in-memory cache that remembers through pluggable backing -stores. - -Consumer code should think in domain types: - -```rust -let player = cache.get_required::("player:meta")?; -cache.put("player:meta", &player)?; -``` - -The persistence layer should deal with envelopes, routing, MessagePack payload -bytes, and schema identity. Application code should not paw at loose JSON files -like a sad little bureaucrat with a clipboard. - -## What It Is - -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. -- Each payload is encoded directly from the known `DatabaseEntry` type into - MessagePack bytes. -- Backing stores are adapters, not the public data model. -- Writes persist to the resolved backing store before the in-memory cache is - updated. -- Type-specific backing stores beat generic backing stores. -- `SingleFileMessagePackBackingStore` is the first concrete store and guards - file access with a sidecar lock file. - -This is not an ORM, not a database, and not distributed consensus in a novelty -hat. If multiple processes write the same backing file, use an external lock or -a coordinator. - -## Current API - -```rust -use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore}; - -#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] -#[cultcache(type = "settings")] -struct Settings { - #[cultcache(key = 0)] - theme: String, - - #[cultcache(key = 1, default)] - retries: u32, -} - -let mut cache = CultCache::new(); -cache.register_entry_type::()?; -cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new("cache.cc")); -cache.pull_all_backing_stores()?; - -cache.put("app", &Settings { - theme: "ash".to_string(), - retries: 3, -})?; - -let settings = cache.get_required::("app")?; -# Ok::<(), anyhow::Error>(()) -``` - -For a closed set of entries, generate a registry instead of repeating yourself: - -```rust -use cultcache_rs::cultcache_registry; - -cultcache_registry!(GameCultEntries { - Settings, - PlayerData, -}); - -let mut cache = CultCache::new(); -cache.register_registry(GameCultEntries)?; -``` - -The important part is that callers retrieve domain values directly: - -```rust -let settings: Settings = cache.get_required("app")?; -let all_settings: Vec = cache.get_all()?; -``` - -The cache is the ergonomic surface. The backing store is the memory prosthetic. - -## How This Maps From Original CultCache - -The original C# CultCache relies on runtime reflection: - -- cacheable models inherit from `DatabaseEntry` -- the cache scans child classes at startup -- `Get(id)` returns a `T` -- optional `INamedEntry` enables name lookup -- optional field/property indexes are registered by member name -- backing stores pull/push/delete `DatabaseEntry` values - -Rust does not have C#-style assembly scanning or runtime subclass discovery. -The closest honest Rust translation is: - -- domain structs implement a marker trait, `DatabaseEntry` -- the derive macro generates a positional array formatter from explicit integer - field slots -- the envelope carries the polymorphic type discriminator -- unknown persisted type ids fail closed instead of constructing arbitrary - runtime types -- a generated registry should eventually replace repetitive manual - registration - -Manual `register_entry_type::()` is still accepted, but entry identity no -longer needs a hand-written impl: `#[derive(DatabaseEntry)]` declares it beside -the domain type. `cultcache_registry!` provides the current generated resolver -surface for a closed entry set. - -## Intended Rust Ergonomics - -The target user-facing shape should be: - -```rust -#[derive(Clone, DatabaseEntry)] -#[cultcache(type = "player")] -pub struct PlayerData { - #[cultcache(key = 0)] - pub id: String, - - #[cultcache(key = 1)] - pub name: String, - - #[cultcache(key = 2)] - pub faction: String, -} - -let mut cache = CultCache::builder() - .with_generated_entries(game_entries()) - .with_generic_store(SingleFileMessagePackBackingStore::new("cache.cc")) - .build()?; - -cache.pull_all_backing_stores()?; -let player = cache.get_required::("player:ari")?; -``` - -The current `cultcache_registry!` macro is deliberately simple. The remaining -ergonomic endpoint is automatic inventory or build-script generation so users -do not have to maintain the registry list by hand. - -The registry is the Rust equivalent of C# reflection and MessagePack resolver -setup. It should know: - -- entry type id -- generated MessagePack/serde formatter path -- Rust type name / schema name -- optional store domain route -- optional name key extractor -- optional secondary indexes -- optional global singleton marker - -In other words: code generation should restore the original CultCache feeling -without pretending Rust has runtime reflection hiding under the floorboards. - -## Current Surface - -- `CultCache::new` -- `register_entry_type::` -- `add_generic_backing_store` -- `add_backing_store` -- `pull_all_backing_stores` -- `get::` -- `get_required::` -- `get_envelope::` -- `get_required_envelope::` -- `get_all::` -- `put::` -- `put_envelope::` -- `update::` -- `delete::` -- `snapshot` -- `SingleFileMessagePackBackingStore` - -## Backing Store Routing - -Generic store: - -```rust -cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new("cache.cc")); -``` - -Type-specific store: - -```rust -cache.add_backing_store( - SingleFileMessagePackBackingStore::new("players.cc"), - ["player"], -); -``` - -When writing a `PlayerData`, the cache checks type-specific stores first. If none -match, it writes to the first generic store. Later matching stores are mirrors. - -This mirrors the C# behavior: - -- specific domain stores own their domain -- the first generic store is the primary generic write target -- later generic stores mirror writes -- this is not multi-master - -## Persistence Semantics - -`put` persists before mutating the in-memory cache. If persistence fails, the -cache does not pretend the write succeeded. - -The envelope is MessagePack, and the payload inside each envelope is also raw -MessagePack bytes encoded from the registered concrete `DatabaseEntry` type. -That avoids the old bootstrap path where payloads were normalized through -`serde_json::Value`. - -For bit-compatible neighbors, that also enables a real fast lane: - -- `get_envelope::()` exports the canonical persisted bytes for a typed entry -- `put_envelope::()` ingests the same envelope into another cache instance - without re-encoding the payload first - -It still decodes once for validation and typed reads. We are not pretending one -`Vec` became metaphysically zero-copy because we believed in it harder. But -the stupid decode/re-encode loop is gone. - -The `DatabaseEntry` derive does not trust Rust source field order as the durable -schema. Every persisted member must declare a stable integer slot: - -```rust -#[derive(Clone, DatabaseEntry)] -#[cultcache(type = "player")] -pub struct PlayerData { - #[cultcache(key = 0)] - pub id: String, - - #[cultcache(key = 1)] - pub name: String, - - #[cultcache(key = 2, default)] - pub level: u32, -} -``` - -The derive emits a tuple/array formatter. Gaps serialize as nil. Missing fields -marked `default` use `Default::default()` when older payloads are read. Deleted -field slots should stay reserved until an explicit store migration rewrites the -data. - -The single-file MessagePack store rewrites an atomic snapshot. That is a sane -starting point for small typed state surfaces, settings, Epiphany agent memory, -heartbeat state, and other compact control-plane data. Large corpora should use -a sharded store or a real database instead of asking one file to become a -warehouse and then acting wounded when physics invoices us. - -The single-file store uses a sidecar lock file for shared reads and exclusive -writes. That protects ordinary multi-process access to the same file. It is -still not a multi-master replication protocol, and it is not a substitute for a -coordinator when higher-level write ordering matters. - -## Near-Term Ergonomic Improvements - -1. **Derive macro** - - `#[derive(DatabaseEntry)]` - - `#[cultcache(type = "...")]` - - `#[cultcache(key = N)]` on every persisted field - - tuple/array formatter generation landed - - optional `#[cultcache(name)]`, `#[cultcache(index)]`, and - `#[cultcache(global)]` are still future work - -2. **Generated registry** - - a `CultCacheRegistry` trait - - `cache.register_registry(GameCultEntries)` - - basic macro registry landed - - automatic inventory/build-script generation and store routes are future work - -3. **Name and index lookups** - - `cache.get_by_name::("Potion")` - - `cache.get_by_index::("faction", "Lucent")` - - generated extractors instead of reflection strings - -4. **Schema/version metadata** - - schema version in the envelope or payload metadata - - explicit migration hooks - - refusal on unknown persisted entry types unless a migration/resolver is - installed - -5. **Projection helpers** - - JSON projection for review and git diffs - - vector stripping / large-field elision for agent memory surfaces - -## Why Not Skip Registration Immediately? - -Rust can only deserialize polymorphic data into concrete types if something -maps the persisted `type` discriminator to the Rust type. In C#, reflection and -MessagePack resolvers can discover a closed `DatabaseEntry` inheritance tree at -runtime and route it through known formatters. In Rust, that map has to come from -somewhere: - -- explicit registration -- a generated registry -- a macro inventory crate -- a hand-written resolver - -Explicit registration is the simplest honest implementation. Generated registry -is the ergonomic destination. Hand-written resolver tables are the punishment -we give ourselves if we get lazy. - -## Security Model - -CultCache should never parse arbitrary structured data into arbitrary runtime -types. It only accepts persisted envelopes whose `type` discriminator resolves -to a known `DatabaseEntry` type registered in the cache or generated registry. - -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. -- MessagePack payload bytes deserialize through known concrete formatters, not - an open-world type loader or generic structured-value bridge. - -The derive/registry work should preserve that shape. Its job is to remove -manual registration ceremony, not to loosen the closed-world boundary. - -## License - -Private GameCult infrastructure for now. Public packaging can wait until the -API grows enough teeth to deserve strangers. diff --git a/vendor/cultcache-rs/cultcache-rs-derive/Cargo.toml b/vendor/cultcache-rs/cultcache-rs-derive/Cargo.toml deleted file mode 100644 index 74d52a4..0000000 --- a/vendor/cultcache-rs/cultcache-rs-derive/Cargo.toml +++ /dev/null @@ -1,13 +0,0 @@ -[package] -name = "cultcache-rs-derive" -version = "0.1.0" -edition = "2024" -publish = false - -[lib] -proc-macro = true - -[dependencies] -proc-macro2 = "1" -quote = "1" -syn = { version = "2", features = ["derive", "parsing"] } diff --git a/vendor/cultcache-rs/cultcache-rs-derive/src/lib.rs b/vendor/cultcache-rs/cultcache-rs-derive/src/lib.rs deleted file mode 100644 index 7d61319..0000000 --- a/vendor/cultcache-rs/cultcache-rs-derive/src/lib.rs +++ /dev/null @@ -1,278 +0,0 @@ -use proc_macro::TokenStream; -use quote::format_ident; -use quote::quote; -use syn::Data; -use syn::DeriveInput; -use syn::Fields; -use syn::LitInt; -use syn::LitStr; -use syn::Type; -use syn::parse_macro_input; - -#[proc_macro_derive(DatabaseEntry, attributes(cultcache))] -pub fn derive_database_entry(input: TokenStream) -> TokenStream { - let input = parse_macro_input!(input as DeriveInput); - let ident = input.ident.clone(); - let mut entry_type: Option = None; - let mut schema_name: Option = None; - - for attribute in input - .attrs - .iter() - .filter(|attr| attr.path().is_ident("cultcache")) - { - let parse_result = attribute.parse_nested_meta(|meta| { - if meta.path.is_ident("type") { - entry_type = Some(meta.value()?.parse()?); - Ok(()) - } else if meta.path.is_ident("schema") { - schema_name = Some(meta.value()?.parse()?); - Ok(()) - } else { - Err(meta.error("unsupported cultcache attribute")) - } - }); - if let Err(error) = parse_result { - return error.to_compile_error().into(); - } - } - - let Some(entry_type) = entry_type else { - return syn::Error::new_spanned( - ident, - "DatabaseEntry derive requires #[cultcache(type = \"...\")]", - ) - .to_compile_error() - .into(); - }; - let schema_name = schema_name.unwrap_or_else(|| LitStr::new(&ident.to_string(), ident.span())); - - let fields = match input.data { - Data::Struct(data) => match data.fields { - Fields::Named(fields) => fields.named, - _ => { - return syn::Error::new_spanned( - ident, - "DatabaseEntry derive requires a named struct", - ) - .to_compile_error() - .into(); - } - }, - _ => { - return syn::Error::new_spanned(ident, "DatabaseEntry derive requires a struct") - .to_compile_error() - .into(); - } - }; - - let mut slots: Vec = Vec::new(); - for field in fields { - let Some(field_ident) = field.ident else { - return syn::Error::new_spanned(field.ty, "DatabaseEntry derive requires named fields") - .to_compile_error() - .into(); - }; - let mut key: Option = None; - let mut default = false; - let mut bytes = false; - for attribute in field - .attrs - .iter() - .filter(|attr| attr.path().is_ident("cultcache")) - { - let parse_result = attribute.parse_nested_meta(|meta| { - if meta.path.is_ident("key") { - let value: LitInt = meta.value()?.parse()?; - key = Some(value.base10_parse()?); - Ok(()) - } else if meta.path.is_ident("default") { - default = true; - Ok(()) - } else if meta.path.is_ident("bytes") { - bytes = true; - Ok(()) - } else { - Err(meta.error("unsupported field cultcache attribute")) - } - }); - if let Err(error) = parse_result { - return error.to_compile_error().into(); - } - } - let Some(key) = key else { - return syn::Error::new_spanned( - field_ident, - "DatabaseEntry fields require #[cultcache(key = N)]", - ) - .to_compile_error() - .into(); - }; - if slots.iter().any(|slot| slot.key == key) { - return syn::Error::new_spanned( - field_ident, - format!("duplicate DatabaseEntry slot {key}"), - ) - .to_compile_error() - .into(); - } - slots.push(CultCacheSlot { - key, - ident: field_ident, - ty: field.ty, - default, - bytes, - }); - } - slots.sort_by_key(|slot| slot.key); - let max_slot = slots.iter().map(|slot| slot.key).max().unwrap_or(0); - - let serialize_elements = (0..=max_slot).map(|slot| { - if let Some(field) = slots.iter().find(|field| field.key == slot) { - let field_ident = &field.ident; - if field.bytes { - quote! { - tuple.serialize_element(::serde_bytes::Bytes::new(&self.#field_ident))?; - } - } else { - quote! { - tuple.serialize_element(&self.#field_ident)?; - } - } - } else { - quote! { - tuple.serialize_element(&())?; - } - } - }); - - let option_declarations = slots.iter().map(|field| { - let variable = format_ident!("slot_{}", field.key); - let ty = &field.ty; - if field.bytes { - quote! { - let mut #variable: ::std::option::Option<::serde_bytes::ByteBuf> = ::std::option::Option::None; - } - } else { - quote! { - let mut #variable: ::std::option::Option<#ty> = ::std::option::Option::None; - } - } - }); - - let deserialize_match_arms = (0..=max_slot).map(|slot| { - if let Some(field) = slots.iter().find(|field| field.key == slot) { - let variable = format_ident!("slot_{}", field.key); - let ty = &field.ty; - if field.bytes { - quote! { - #slot => { - #variable = seq.next_element::<::serde_bytes::ByteBuf>()?; - } - } - } else { - quote! { - #slot => { - #variable = seq.next_element::<#ty>()?; - } - } - } - } else { - quote! { - #slot => { - let _ = seq.next_element::<::serde::de::IgnoredAny>()?; - } - } - } - }); - - let build_fields = slots.iter().map(|field| { - let field_ident = &field.ident; - let variable = format_ident!("slot_{}", field.key); - if field.bytes && field.default { - quote! { - #field_ident: #variable.unwrap_or_default().into_vec() - } - } else if field.bytes { - let field_name = field_ident.to_string(); - quote! { - #field_ident: #variable.ok_or_else(|| ::serde::de::Error::missing_field(#field_name))?.into_vec() - } - } else if field.default { - quote! { - #field_ident: #variable.unwrap_or_default() - } - } else { - let field_name = field_ident.to_string(); - quote! { - #field_ident: #variable.ok_or_else(|| ::serde::de::Error::missing_field(#field_name))? - } - } - }); - - quote! { - impl ::cultcache_rs::DatabaseEntry for #ident { - const TYPE: &'static str = #entry_type; - const SCHEMA_NAME: &'static str = #schema_name; - } - - impl ::serde::Serialize for #ident { - fn serialize(&self, serializer: S) -> ::std::result::Result - where - S: ::serde::Serializer, - { - use ::serde::ser::SerializeTuple; - let mut tuple = serializer.serialize_tuple(#max_slot + 1)?; - #(#serialize_elements)* - tuple.end() - } - } - - impl<'de> ::serde::Deserialize<'de> for #ident { - fn deserialize(deserializer: D) -> ::std::result::Result - where - D: ::serde::Deserializer<'de>, - { - struct CultCacheVisitor; - - impl<'de> ::serde::de::Visitor<'de> for CultCacheVisitor { - type Value = #ident; - - fn expecting(&self, formatter: &mut ::std::fmt::Formatter) -> ::std::fmt::Result { - formatter.write_str(concat!("a CultCache array for ", stringify!(#ident))) - } - - fn visit_seq(self, mut seq: A) -> ::std::result::Result - where - A: ::serde::de::SeqAccess<'de>, - { - #(#option_declarations)* - let mut index = 0usize; - while index <= #max_slot { - match index { - #(#deserialize_match_arms)* - _ => unreachable!(), - } - index += 1; - } - while let ::std::option::Option::Some(_) = seq.next_element::<::serde::de::IgnoredAny>()? {} - Ok(#ident { - #(#build_fields),* - }) - } - } - - deserializer.deserialize_tuple(#max_slot + 1, CultCacheVisitor) - } - } - } - .into() -} - -struct CultCacheSlot { - key: usize, - ident: syn::Ident, - ty: Type, - default: bool, - bytes: bool, -} diff --git a/vendor/cultcache-rs/examples/cultcache_interop.rs b/vendor/cultcache-rs/examples/cultcache_interop.rs deleted file mode 100644 index 4049acf..0000000 --- a/vendor/cultcache-rs/examples/cultcache_interop.rs +++ /dev/null @@ -1,121 +0,0 @@ -use anyhow::Result; -use anyhow::anyhow; -use cultcache_rs::CultCache; -use cultcache_rs::DatabaseEntry; -use cultcache_rs::SingleFileMessagePackBackingStore; -use serde_json::json; -use std::collections::BTreeMap; - -const INTEROP_TYPE: &str = "cultcache.interop-note"; -const INTEROP_SCHEMA_VERSION: &str = "cultcache.interop_note.v1"; - -#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] -#[cultcache(type = "cultcache.interop-note", schema = "CultCacheInteropNote")] -struct CultCacheInteropNote { - #[cultcache(key = 0)] - schema_version: String, - #[cultcache(key = 1)] - document_id: String, - #[cultcache(key = 2)] - author_runtime_id: String, - #[cultcache(key = 3)] - title: String, - #[cultcache(key = 4)] - body: String, - #[cultcache(key = 5, default)] - tags: Vec, -} - -fn main() -> Result<()> { - let mut args = std::env::args().skip(1); - let mode = args - .next() - .ok_or_else(|| anyhow!("expected mode: write | read"))?; - let options = parse_args(args.collect()); - let file = require_arg(&options, "file")?; - - match mode.as_str() { - "write" => write_note(file, require_arg(&options, "runtime-id")?)?, - "read" => read_note(file)?, - _ => return Err(anyhow!("unknown mode {mode}")), - } - Ok(()) -} - -fn write_note(file: &str, runtime_id: &str) -> Result<()> { - let mut cache = build_cache(file)?; - cache.pull_all_backing_stores()?; - let note = CultCacheInteropNote { - schema_version: INTEROP_SCHEMA_VERSION.to_string(), - document_id: format!("note:{runtime_id}"), - author_runtime_id: runtime_id.to_string(), - title: format!("{runtime_id} wrote a CultCache note"), - body: "The v1 store format is the contract.".to_string(), - tags: vec![ - runtime_id.to_string(), - "rust".to_string(), - "interop".to_string(), - ], - }; - cache.put(¬e.document_id, ¬e)?; - print_note(¬e) -} - -fn read_note(file: &str) -> Result<()> { - let mut cache = build_cache(file)?; - cache.pull_all_backing_stores()?; - let notes = cache.get_all::()?; - let note = notes - .first() - .ok_or_else(|| anyhow!("no {INTEROP_TYPE} records found"))?; - print_note(note) -} - -fn build_cache(file: &str) -> Result { - let mut cache = CultCache::new(); - cache.register_entry_type::()?; - cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(file)); - Ok(cache) -} - -fn print_note(note: &CultCacheInteropNote) -> Result<()> { - println!( - "{}", - serde_json::to_string(&json!({ - "schemaVersion": note.schema_version, - "documentId": note.document_id, - "authorRuntimeId": note.author_runtime_id, - "title": note.title, - "body": note.body, - "tags": note.tags, - }))? - ); - Ok(()) -} - -fn parse_args(args: Vec) -> BTreeMap { - let mut parsed = BTreeMap::new(); - let mut index = 0; - while index < args.len() { - let token = &args[index]; - if !token.starts_with("--") { - index += 1; - continue; - } - let name = token.trim_start_matches("--").to_string(); - let value = args - .get(index + 1) - .cloned() - .unwrap_or_else(|| panic!("missing value for --{name}")); - parsed.insert(name, value); - index += 2; - } - parsed -} - -fn require_arg<'a>(options: &'a BTreeMap, name: &str) -> Result<&'a str> { - options - .get(name) - .map(String::as_str) - .ok_or_else(|| anyhow!("missing required argument --{name}")) -} diff --git a/vendor/cultcache-rs/src/lib.rs b/vendor/cultcache-rs/src/lib.rs deleted file mode 100644 index 72ebf7e..0000000 --- a/vendor/cultcache-rs/src/lib.rs +++ /dev/null @@ -1,1241 +0,0 @@ -use anyhow::Context; -use anyhow::Result; -use anyhow::anyhow; -use serde::Serialize; -use serde::de::DeserializeOwned; -use std::collections::BTreeMap; -use std::collections::BTreeSet; -use std::fs; -use std::fs::File; -use std::fs::OpenOptions; -use std::io::Write; -use std::path::Path; -use std::path::PathBuf; - -extern crate self as cultcache_rs; - -pub use cultcache_rs_derive::DatabaseEntry; - -pub trait DatabaseEntry: Serialize + DeserializeOwned + Clone + Send + 'static { - const TYPE: &'static str; - const SCHEMA_NAME: &'static str = "DatabaseEntry"; -} - -pub trait CultCacheRegistry { - fn register_entries(&self, cache: &mut CultCache) -> Result<()>; -} - -#[macro_export] -macro_rules! cultcache_registry { - ($name:ident { $($entry:ty),* $(,)? }) => { - #[derive(Clone, Copy, Debug, Default)] - pub struct $name; - - impl $crate::CultCacheRegistry for $name { - fn register_entries(&self, cache: &mut $crate::CultCache) -> ::anyhow::Result<()> { - $( - cache.register_entry_type::<$entry>()?; - )* - Ok(()) - } - } - }; -} - -#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -#[serde(rename_all = "camelCase")] -pub struct CultCacheEnvelope { - pub key: String, - #[serde(rename = "type")] - pub r#type: String, - #[serde(with = "serde_bytes")] - pub payload: Vec, - pub stored_at: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub schema_id: Option, -} - -#[derive(Clone, Debug, PartialEq, Eq)] -pub struct CultCacheExpectedEnvelope { - pub key: String, - pub r#type: String, - pub current: Option, -} - -#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] -struct PersistedStoreSnapshot( - String, - Vec, - Vec, -); - -#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] -struct PersistedSchemaCatalogEntry( - String, - String, - String, - String, - String, - Vec, - Vec, -); - -#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] -struct PersistedSchemaCatalogMember( - u32, - String, - String, - bool, - bool, - Option, - bool, - Option, -); - -#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] -struct PersistedRecord( - String, - String, - String, - #[serde(with = "serde_bytes")] Vec, -); - -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] -pub struct PushAllOptions { - pub soft: bool, -} - -pub trait CacheBackingStore: Send { - fn pull_all(&self) -> Result>; - fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()>; - fn delete(&mut self, entry: &CultCacheEnvelope) -> Result<()>; - - fn push_all(&mut self, entries: &[CultCacheEnvelope], _options: PushAllOptions) -> Result<()> { - let existing = self.pull_all()?; - for entry in existing { - self.delete(&entry)?; - } - for entry in entries { - self.push(entry)?; - } - Ok(()) - } -} - -#[derive(Clone, Debug, PartialEq, Eq)] -pub struct SingleFileMessagePackBackingStore { - path: PathBuf, -} - -impl SingleFileMessagePackBackingStore { - pub fn new(path: impl Into) -> Self { - Self { path: path.into() } - } - - pub fn path(&self) -> &Path { - &self.path - } - - /// Reads one atomically published snapshot without acquiring or creating - /// the sibling writer lock. This is for immutable, externally owned - /// documents such as pinned trust anchors where the reader has no write - /// authority over the containing directory. - pub fn pull_all_read_only_snapshot(&self) -> Result> { - self.read_all_unlocked() - } - - /// Holds the pre-created sibling lock shared while a read-only consumer - /// evaluates and acts on one snapshot. The reader never creates or writes - /// either file, so an authority owner can grant only read access. Writers - /// using `compare_exchange` cannot replace the snapshot until `action` - /// returns. - pub fn with_read_only_shared_snapshot( - &self, - action: impl FnOnce(Vec) -> Result, - ) -> Result { - self.with_read_only_shared_snapshot_and_unlock_diagnostic(action, |_| {}) - } - - pub fn with_read_only_shared_snapshot_and_unlock_diagnostic( - &self, - action: impl FnOnce(Vec) -> Result, - on_unlock_failure: impl FnOnce(anyhow::Error), - ) -> Result { - self.with_read_only_shared_snapshot_using( - action, - |lock| { - fs2::FileExt::unlock(lock) - .with_context(|| format!("failed to unlock {}", self.lock_path().display())) - }, - on_unlock_failure, - ) - } - - fn with_read_only_shared_snapshot_using( - &self, - action: impl FnOnce(Vec) -> Result, - unlock: U, - on_unlock_failure: D, - ) -> Result - where - U: FnOnce(&File) -> Result<()>, - D: FnOnce(anyhow::Error), - { - let lock_path = self.lock_path(); - let lock = OpenOptions::new() - .read(true) - .open(&lock_path) - .with_context(|| format!("failed to open pre-created {}", lock_path.display()))?; - fs2::FileExt::lock_shared(&lock) - .with_context(|| format!("failed to lock {}", lock_path.display()))?; - let guard = ReadOnlySharedLockGuard::new(lock, unlock, on_unlock_failure); - let result = self.read_all_unlocked().and_then(action); - drop(guard); - result - } - - pub fn compare_exchange( - &self, - expected: &[CultCacheExpectedEnvelope], - replacements: &[CultCacheEnvelope], - ) -> Result { - self.with_exclusive_lock(|| { - let mut entries = self.read_all_unlocked()?; - for condition in expected { - let current = entries - .iter() - .find(|entry| entry.r#type == condition.r#type && entry.key == condition.key); - if current != condition.current.as_ref() { - return Ok(false); - } - if let Some(envelope) = condition.current.as_ref() - && (envelope.r#type != condition.r#type || envelope.key != condition.key) - { - return Err(anyhow!( - "CultCache compare-exchange expectation identity differs from its envelope" - )); - } - } - let mut replacement_ids = BTreeSet::new(); - for replacement in replacements { - if replacement.key.trim().is_empty() || replacement.r#type.trim().is_empty() { - return Err(anyhow!( - "CultCache compare-exchange replacement identity is empty" - )); - } - if !replacement_ids.insert(entry_id(replacement)) { - return Err(anyhow!( - "CultCache compare-exchange contains a duplicate replacement" - )); - } - } - entries.retain(|entry| !replacement_ids.contains(&entry_id(entry))); - entries.extend_from_slice(replacements); - entries.sort_by_key(entry_id); - self.write_all_unlocked(&entries)?; - Ok(true) - }) - } - - /// Atomically replaces an entire snapshot only when the persisted entries - /// still exactly match `expected`. This is the bounded migration primitive - /// for removing obsolete identities without racing another writer between - /// a read and a whole-store rewrite. - pub fn compare_exchange_snapshot( - &self, - expected: &[CultCacheEnvelope], - replacements: &[CultCacheEnvelope], - ) -> Result { - self.with_exclusive_lock(|| { - let current = self.read_all_unlocked()?; - if current != expected { - return Ok(false); - } - let mut replacement_ids = BTreeSet::new(); - for replacement in replacements { - if replacement.key.trim().is_empty() || replacement.r#type.trim().is_empty() { - return Err(anyhow!("CultCache snapshot replacement identity is empty")); - } - if !replacement_ids.insert(entry_id(replacement)) { - return Err(anyhow!( - "CultCache snapshot replacement contains a duplicate identity" - )); - } - } - let mut ordered = replacements.to_vec(); - ordered.sort_by_key(entry_id); - self.write_all_unlocked(&ordered)?; - Ok(true) - }) - } - - fn read_all_unlocked(&self) -> Result> { - if !self.path.exists() { - return Ok(Vec::new()); - } - let bytes = fs::read(&self.path) - .with_context(|| format!("failed to read {}", self.path.display()))?; - if bytes.is_empty() { - return Ok(Vec::new()); - } - decode_store_snapshot(&bytes) - .or_else(|_| rmp_serde::from_slice(&bytes)) - .with_context(|| format!("failed to decode MessagePack {}", self.path.display())) - } - - fn write_all_unlocked(&self, entries: &[CultCacheEnvelope]) -> Result<()> { - if let Some(parent) = self.path.parent() { - fs::create_dir_all(parent) - .with_context(|| format!("failed to create {}", parent.display()))?; - } - let bytes = rmp_serde::to_vec(&encode_store_snapshot(entries)) - .context("failed to encode MessagePack")?; - let tmp_path = temporary_path_for(&self.path); - let mut temporary = OpenOptions::new() - .create_new(true) - .write(true) - .open(&tmp_path) - .with_context(|| format!("failed to create {}", tmp_path.display()))?; - temporary - .write_all(&bytes) - .with_context(|| format!("failed to write {}", tmp_path.display()))?; - temporary - .sync_all() - .with_context(|| format!("failed to sync {}", tmp_path.display()))?; - drop(temporary); - replace_file_atomically(&tmp_path, &self.path)?; - #[cfg(unix)] - if let Some(parent) = self.path.parent() { - File::open(parent) - .and_then(|directory| directory.sync_all()) - .with_context(|| format!("failed to sync {}", parent.display()))?; - } - Ok(()) - } - - fn with_shared_lock(&self, action: impl FnOnce() -> Result) -> Result { - let lock = self.open_lock_file()?; - fs2::FileExt::lock_shared(&lock) - .with_context(|| format!("failed to lock {}", self.lock_path().display()))?; - let result = action(); - fs2::FileExt::unlock(&lock) - .with_context(|| format!("failed to unlock {}", self.lock_path().display()))?; - result - } - - fn with_exclusive_lock(&self, action: impl FnOnce() -> Result) -> Result { - let lock = self.open_lock_file()?; - fs2::FileExt::lock_exclusive(&lock) - .with_context(|| format!("failed to lock {}", self.lock_path().display()))?; - let result = action(); - fs2::FileExt::unlock(&lock) - .with_context(|| format!("failed to unlock {}", self.lock_path().display()))?; - result - } - - fn open_lock_file(&self) -> Result { - let lock_path = self.lock_path(); - if let Some(parent) = lock_path.parent() { - fs::create_dir_all(parent) - .with_context(|| format!("failed to create {}", parent.display()))?; - } - OpenOptions::new() - .create(true) - .read(true) - .truncate(false) - .write(true) - .open(&lock_path) - .with_context(|| format!("failed to open {}", lock_path.display())) - } - - fn lock_path(&self) -> PathBuf { - let mut lock_name = self - .path - .file_name() - .map(|value| value.to_os_string()) - .unwrap_or_else(|| "cultcache.cc".into()); - lock_name.push(".lock"); - self.path.with_file_name(lock_name) - } -} - -struct ReadOnlySharedLockGuard -where - U: FnOnce(&File) -> Result<()>, - D: FnOnce(anyhow::Error), -{ - lock: Option, - unlock: Option, - on_unlock_failure: Option, -} - -impl ReadOnlySharedLockGuard -where - U: FnOnce(&File) -> Result<()>, - D: FnOnce(anyhow::Error), -{ - fn new(lock: File, unlock: U, on_unlock_failure: D) -> Self { - Self { lock: Some(lock), unlock: Some(unlock), on_unlock_failure: Some(on_unlock_failure) } - } -} - -impl Drop for ReadOnlySharedLockGuard -where - U: FnOnce(&File) -> Result<()>, - D: FnOnce(anyhow::Error), -{ - fn drop(&mut self) { - let (Some(lock), Some(unlock)) = (self.lock.as_ref(), self.unlock.take()) else { return; }; - if let Err(error) = unlock(lock) - && let Some(on_unlock_failure) = self.on_unlock_failure.take() - { - on_unlock_failure(error); - } - } -} - -#[cfg(unix)] -fn replace_file_atomically(source: &Path, destination: &Path) -> Result<()> { - fs::rename(source, destination).with_context(|| { - format!( - "failed to atomically replace {} with {}", - destination.display(), - source.display() - ) - }) -} - -#[cfg(windows)] -fn replace_file_atomically(source: &Path, destination: &Path) -> Result<()> { - use std::os::windows::ffi::OsStrExt; - use windows_sys::Win32::Storage::FileSystem::{ - MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, MoveFileExW, - }; - let source: Vec = source.as_os_str().encode_wide().chain(Some(0)).collect(); - let destination: Vec = destination - .as_os_str() - .encode_wide() - .chain(Some(0)) - .collect(); - let result = unsafe { - MoveFileExW( - source.as_ptr(), - destination.as_ptr(), - MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH, - ) - }; - if result == 0 { - return Err(std::io::Error::last_os_error()).context("atomic CultCache replacement failed"); - } - Ok(()) -} - -impl CacheBackingStore for SingleFileMessagePackBackingStore { - fn pull_all(&self) -> Result> { - self.with_shared_lock(|| self.read_all_unlocked()) - } - - fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()> { - self.with_exclusive_lock(|| { - let mut entries = self.read_all_unlocked()?; - entries.retain(|candidate| entry_id(candidate) != entry_id(entry)); - entries.push(entry.clone()); - entries.sort_by_key(entry_id); - self.write_all_unlocked(&entries) - }) - } - - fn delete(&mut self, entry: &CultCacheEnvelope) -> Result<()> { - self.with_exclusive_lock(|| { - let mut entries = self.read_all_unlocked()?; - entries.retain(|candidate| entry_id(candidate) != entry_id(entry)); - self.write_all_unlocked(&entries) - }) - } - - fn push_all(&mut self, entries: &[CultCacheEnvelope], _options: PushAllOptions) -> Result<()> { - self.with_exclusive_lock(|| { - let mut entries = entries.to_vec(); - entries.sort_by_key(entry_id); - self.write_all_unlocked(&entries) - }) - } -} - -struct CultCacheStoreRegistration { - store: Box, - types: BTreeSet, -} - -pub struct CultCache { - definitions: BTreeMap, - entries: BTreeMap, - stores: Vec, -} - -impl CultCache { - pub fn new() -> Self { - Self { - definitions: BTreeMap::new(), - entries: BTreeMap::new(), - stores: Vec::new(), - } - } - - pub fn register_entry_type(&mut self) -> Result<()> { - if T::TYPE.trim().is_empty() { - return Err(anyhow!( - "CultCache entry types must declare a non-empty type" - )); - } - if let Some(existing_schema) = self.definitions.get(T::TYPE) - && *existing_schema != T::SCHEMA_NAME - { - return Err(anyhow!( - "CultCache already has a different definition registered for type {:?}", - T::TYPE - )); - } - self.definitions.insert(T::TYPE.to_string(), T::SCHEMA_NAME); - Ok(()) - } - - pub fn register_document_type(&mut self) -> Result<()> { - self.register_entry_type::() - } - - pub fn register_registry(&mut self, registry: R) -> Result<&mut Self> { - registry.register_entries(self)?; - Ok(self) - } - - pub fn add_backing_store( - &mut self, - store: impl CacheBackingStore + 'static, - types: impl IntoIterator>, - ) { - self.stores.push(CultCacheStoreRegistration { - store: Box::new(store), - types: types.into_iter().map(Into::into).collect(), - }); - } - - pub fn add_generic_backing_store(&mut self, store: impl CacheBackingStore + 'static) { - self.add_backing_store(store, Vec::::new()); - } - - pub fn pull_all_backing_stores(&mut self) -> Result<()> { - self.entries.clear(); - let known_types: BTreeSet = self.definitions.keys().cloned().collect(); - for registration in &mut self.stores { - for entry in registration.store.pull_all()? { - if !known_types.contains(&entry.r#type) { - return Err(anyhow!( - "No schema is registered for persisted entry type {:?}", - entry.r#type - )); - } - self.entries.insert(entry_id(&entry), entry); - } - } - Ok(()) - } - - pub fn get(&self, key: &str) -> Result> { - self.require_entry_type::()?; - let Some(entry) = self.entries.get(&entry_id_parts(T::TYPE, key)) else { - return Ok(None); - }; - let payload = rmp_serde::from_slice(&entry.payload).with_context(|| { - format!( - "failed to decode CultCache entry {:?} at key {:?} as {}", - T::TYPE, - key, - T::SCHEMA_NAME - ) - })?; - Ok(Some(payload)) - } - - pub fn get_required(&self, key: &str) -> Result { - self.get::(key)? - .ok_or_else(|| anyhow!("CultCache has no {:?} entry at key {:?}", T::TYPE, key)) - } - - pub fn get_envelope(&self, key: &str) -> Result> { - self.require_entry_type::()?; - Ok(self.entries.get(&entry_id_parts(T::TYPE, key)).cloned()) - } - - pub fn get_required_envelope(&self, key: &str) -> Result { - self.get_envelope::(key)? - .ok_or_else(|| anyhow!("CultCache has no {:?} envelope at key {:?}", T::TYPE, key)) - } - - pub fn get_all(&self) -> Result> { - self.require_entry_type::()?; - let mut values = Vec::new(); - for entry in self.entries.values() { - if entry.r#type != T::TYPE { - continue; - } - values.push(rmp_serde::from_slice(&entry.payload).with_context(|| { - format!( - "failed to decode CultCache entry {:?} at key {:?} as {}", - T::TYPE, - entry.key, - T::SCHEMA_NAME - ) - })?); - } - Ok(values) - } - - pub fn put(&mut self, key: impl Into, value: &T) -> Result { - self.require_entry_type::()?; - let key = key.into(); - let payload = rmp_serde::to_vec(value).with_context(|| { - format!( - "failed to encode CultCache entry {:?} at key {:?} as {}", - T::TYPE, - key, - T::SCHEMA_NAME - ) - })?; - let parsed: T = rmp_serde::from_slice(&payload).with_context(|| { - format!( - "failed to validate CultCache entry {:?} at key {:?} as {}", - T::TYPE, - key, - T::SCHEMA_NAME - ) - })?; - let entry = CultCacheEnvelope { - key: key.clone(), - r#type: T::TYPE.to_string(), - payload, - stored_at: now_utc_second(), - schema_id: Some(T::TYPE.to_string()), - }; - let route = self.resolve_route_indices(T::TYPE); - let Some(primary_index) = route.first().copied() else { - return Err(anyhow!( - "No backing store is registered for entry type {:?}", - T::TYPE - )); - }; - self.stores[primary_index].store.push(&entry)?; - for mirror_index in route.iter().skip(1).copied() { - self.stores[mirror_index].store.push(&entry)?; - } - self.entries.insert(entry_id(&entry), entry); - Ok(parsed) - } - - pub fn put_envelope(&mut self, entry: CultCacheEnvelope) -> Result { - self.require_entry_type::()?; - if entry.r#type != T::TYPE { - return Err(anyhow!( - "CultCache envelope type {:?} does not match registered Rust type {:?}", - entry.r#type, - T::TYPE - )); - } - if entry.key.trim().is_empty() { - return Err(anyhow!( - "CultCache envelope keys for type {:?} must be non-empty", - T::TYPE - )); - } - if entry.stored_at.trim().is_empty() { - return Err(anyhow!( - "CultCache envelope stored_at for type {:?} must be non-empty", - T::TYPE - )); - } - - let parsed: T = rmp_serde::from_slice(&entry.payload).with_context(|| { - format!( - "failed to validate CultCache envelope {:?} at key {:?} as {}", - T::TYPE, - entry.key, - T::SCHEMA_NAME - ) - })?; - let route = self.resolve_route_indices(T::TYPE); - let Some(primary_index) = route.first().copied() else { - return Err(anyhow!( - "No backing store is registered for entry type {:?}", - T::TYPE - )); - }; - self.stores[primary_index].store.push(&entry)?; - for mirror_index in route.iter().skip(1).copied() { - self.stores[mirror_index].store.push(&entry)?; - } - self.entries.insert(entry_id(&entry), entry); - Ok(parsed) - } - - pub fn put_raw_envelope(&mut self, entry: CultCacheEnvelope) -> Result<()> { - if !self.definitions.contains_key(&entry.r#type) { - return Err(anyhow!( - "No entry type registered for CultCache envelope type {:?}", - entry.r#type - )); - } - if entry.key.trim().is_empty() { - return Err(anyhow!( - "CultCache envelope keys for type {:?} must be non-empty", - entry.r#type - )); - } - if entry.stored_at.trim().is_empty() { - return Err(anyhow!( - "CultCache envelope stored_at for type {:?} must be non-empty", - entry.r#type - )); - } - let route = self.resolve_route_indices(&entry.r#type); - let Some(primary_index) = route.first().copied() else { - return Err(anyhow!( - "No backing store is registered for entry type {:?}", - entry.r#type - )); - }; - self.stores[primary_index].store.push(&entry)?; - for mirror_index in route.iter().skip(1).copied() { - self.stores[mirror_index].store.push(&entry)?; - } - self.entries.insert(entry_id(&entry), entry); - Ok(()) - } - - pub fn update(&mut self, key: &str, updater: F) -> Result - where - T: DatabaseEntry, - F: FnOnce(Option) -> T, - { - let current = self.get::(key)?; - self.put::(key.to_string(), &updater(current)) - } - - pub fn delete(&mut self, key: &str) -> Result { - self.require_entry_type::()?; - let id = entry_id_parts(T::TYPE, key); - let Some(entry) = self.entries.get(&id).cloned() else { - return Ok(false); - }; - let route = self.resolve_route_indices(T::TYPE); - let Some(primary_index) = route.first().copied() else { - return Err(anyhow!( - "No backing store is registered for entry type {:?}", - T::TYPE - )); - }; - self.stores[primary_index].store.delete(&entry)?; - for mirror_index in route.iter().skip(1).copied() { - self.stores[mirror_index].store.delete(&entry)?; - } - self.entries.remove(&id); - Ok(true) - } - - pub fn snapshot(&self) -> Vec { - self.entries.values().cloned().collect() - } - - fn require_entry_type(&self) -> Result<()> { - match self.definitions.get(T::TYPE) { - Some(schema_name) if *schema_name == T::SCHEMA_NAME => Ok(()), - _ => Err(anyhow!( - "CultCache entry type {:?} is not registered on this cache instance", - T::TYPE - )), - } - } - - fn resolve_route_indices(&self, type_id: &str) -> Vec { - let type_specific: Vec = self - .stores - .iter() - .enumerate() - .filter_map(|(index, registration)| { - registration.types.contains(type_id).then_some(index) - }) - .collect(); - if !type_specific.is_empty() { - return type_specific; - } - self.stores - .iter() - .enumerate() - .filter_map(|(index, registration)| registration.types.is_empty().then_some(index)) - .collect() - } -} - -impl Default for CultCache { - fn default() -> Self { - Self::new() - } -} - -fn entry_id(entry: &CultCacheEnvelope) -> String { - entry_id_parts(&entry.r#type, &entry.key) -} - -fn entry_id_parts(r#type: &str, key: &str) -> String { - format!("{type}::{key}", type = r#type) -} - -fn now_utc_second() -> String { - chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true) -} - -fn encode_store_snapshot(entries: &[CultCacheEnvelope]) -> PersistedStoreSnapshot { - let mut schema_names = BTreeMap::::new(); - for entry in entries { - schema_names.insert( - entry - .schema_id - .clone() - .unwrap_or_else(|| entry.r#type.clone()), - entry.r#type.clone(), - ); - } - - let catalog = schema_names - .into_iter() - .map(|(schema_id, document_type)| { - PersistedSchemaCatalogEntry( - schema_id.clone(), - document_type, - format!("{schema_id}.v1"), - schema_id.clone(), - format!( - "{{\"schemaName\":\"{}\",\"schemaVersion\":\"{}.v1\",\"members\":[]}}", - escape_json_string(&schema_id), - escape_json_string(&schema_id) - ), - vec![schema_id], - Vec::new(), - ) - }) - .collect(); - let records = entries - .iter() - .map(|entry| { - PersistedRecord( - entry.key.clone(), - entry - .schema_id - .clone() - .unwrap_or_else(|| entry.r#type.clone()), - entry.stored_at.clone(), - entry.payload.clone(), - ) - }) - .collect(); - - PersistedStoreSnapshot("cultcache.store.v1".to_string(), catalog, records) -} - -fn decode_store_snapshot(bytes: &[u8]) -> Result> { - let snapshot: PersistedStoreSnapshot = - rmp_serde::from_slice(bytes).context("failed to decode CultCache v1 snapshot")?; - if snapshot.0 != "cultcache.store.v1" { - return Err(anyhow!("unsupported CultCache snapshot {}", snapshot.0)); - } - - let catalog = snapshot - .1 - .into_iter() - .map(|entry| (entry.0, entry.1)) - .collect::>(); - snapshot - .2 - .into_iter() - .map(|record| { - let r#type = catalog.get(&record.1).cloned().ok_or_else(|| { - anyhow!( - "CultCache record {:?} references missing schema {:?}", - record.0, - record.1 - ) - })?; - Ok(CultCacheEnvelope { - key: record.0, - r#type, - stored_at: record.2, - payload: record.3, - schema_id: Some(record.1), - }) - }) - .collect() -} - -fn escape_json_string(value: &str) -> String { - value.replace('\\', "\\\\").replace('"', "\\\"") -} - -fn temporary_path_for(path: &Path) -> PathBuf { - let mut file_name = path - .file_name() - .map(|value| value.to_os_string()) - .unwrap_or_else(|| "cultcache.cc".into()); - file_name.push(format!(".{}.tmp", uuid::Uuid::new_v4())); - path.with_file_name(file_name) -} - -#[cfg(test)] -mod tests { - use super::*; - use pretty_assertions::assert_eq; - use std::sync::mpsc; - use std::time::Duration; - - #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] - #[cultcache(type = "settings")] - struct Settings { - #[cultcache(key = 0)] - theme: String, - #[cultcache(key = 1, default)] - retries: u32, - } - - #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] - #[cultcache(type = "note")] - struct Note { - #[cultcache(key = 0)] - title: String, - #[cultcache(key = 1)] - body: String, - } - - cultcache_registry!(TestEntries { Settings, Note }); - - #[test] - fn captured_snapshot_cannot_outlive_a_shared_consequence_gate() -> Result<()> { - let temp = tempfile::tempdir()?; - let store_path = temp.path().join("authority.cc"); - let mut store = SingleFileMessagePackBackingStore::new(&store_path); - let released = CultCacheEnvelope { - key: "brake".into(), - r#type: "brake".into(), - payload: b"released".to_vec(), - stored_at: "2026-07-20T00:00:00Z".into(), - schema_id: Some("brake.v0".into()), - }; - store.push(&released)?; - - let writer_path = store_path.clone(); - let (started_tx, started_rx) = mpsc::channel(); - let (finished_tx, finished_rx) = mpsc::channel(); - store.with_read_only_shared_snapshot(|captured| { - assert_eq!(captured, vec![released.clone()]); - std::thread::spawn(move || { - started_tx.send(()).unwrap(); - let engaged = CultCacheEnvelope { - payload: b"engaged".to_vec(), - ..released - }; - let mut writer = SingleFileMessagePackBackingStore::new(writer_path); - writer.push(&engaged).unwrap(); - finished_tx.send(()).unwrap(); - }); - started_rx.recv_timeout(Duration::from_secs(1))?; - assert!(finished_rx.recv_timeout(Duration::from_millis(100)).is_err()); - Ok(()) - })?; - finished_rx.recv_timeout(Duration::from_secs(1))?; - assert_eq!(store.pull_all_read_only_snapshot()?[0].payload, b"engaged"); - Ok(()) - } - - #[test] - fn post_action_unlock_failure_is_diagnostic_not_action_result() -> Result<()> { - let temp = tempfile::tempdir()?; - let store_path = temp.path().join("authority.cc"); - let mut store = SingleFileMessagePackBackingStore::new(&store_path); - store.push(&CultCacheEnvelope { - key: "brake".into(), - r#type: "brake".into(), - payload: b"released".to_vec(), - stored_at: "2026-07-20T00:00:00Z".into(), - schema_id: Some("brake".into()), - })?; - - let mut diagnostic = None; - let result = store.with_read_only_shared_snapshot_using( - |_| Ok(42), - |_| Err(anyhow!("injected post-action unlock failure")), - |error| diagnostic = Some(error.to_string()), - ); - - assert_eq!(result?, 42); - assert_eq!(diagnostic.as_deref(), Some("injected post-action unlock failure")); - Ok(()) - } - - #[test] - fn whole_snapshot_replacement_is_atomic_and_rejects_stale_or_duplicate_inputs() -> Result<()> { - let temp = tempfile::tempdir()?; - let store_path = temp.path().join("compact.cc"); - let mut store = SingleFileMessagePackBackingStore::new(&store_path); - let first = CultCacheEnvelope { - key: "first".into(), - r#type: "note".into(), - payload: b"old".to_vec(), - stored_at: "2026-08-23T00:00:00Z".into(), - schema_id: Some("note.v1".into()), - }; - let obsolete = CultCacheEnvelope { - key: "obsolete".into(), - payload: b"discard".to_vec(), - ..first.clone() - }; - store.push(&first)?; - store.push(&obsolete)?; - let expected = store.pull_all_read_only_snapshot()?; - let replacement = CultCacheEnvelope { - payload: b"current".to_vec(), - ..first.clone() - }; - - assert!(store.compare_exchange_snapshot(&expected, std::slice::from_ref(&replacement))?); - assert_eq!(store.pull_all_read_only_snapshot()?, [replacement.clone()]); - assert!(!store.compare_exchange_snapshot(&expected, &[])?); - assert!( - store - .compare_exchange_snapshot( - std::slice::from_ref(&replacement), - &[replacement.clone(), replacement.clone()] - ) - .is_err() - ); - Ok(()) - } - - #[test] - fn familiar_cultcache_flow_persists_and_reloads_typed_documents() -> Result<()> { - let temp = tempfile::tempdir()?; - let store_path = temp.path().join("cache.cc"); - let settings = Settings { - theme: "ash".to_string(), - retries: 3, - }; - - 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)?; - assert_eq!(cache.get_required::("app")?, settings); - - let mut reloaded = CultCache::new(); - reloaded.register_entry_type::()?; - reloaded.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); - reloaded.pull_all_backing_stores()?; - assert_eq!(reloaded.get_required::("app")?, settings); - Ok(()) - } - - #[test] - fn entry_identity_is_polymorphic_by_type_and_key() -> Result<()> { - let temp = tempfile::tempdir()?; - let store_path = temp.path().join("cache.cc"); - let mut cache = CultCache::new(); - cache.register_registry(TestEntries)?; - cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); - - cache.put( - "shared", - &Settings { - theme: "green".to_string(), - retries: 1, - }, - )?; - cache.put( - "shared", - &Note { - title: "same key".to_string(), - body: "different type".to_string(), - }, - )?; - - assert_eq!(cache.snapshot().len(), 2); - assert_eq!(cache.get_required::("shared")?.title, "same key"); - assert_eq!(cache.get_required::("shared")?.theme, "green"); - Ok(()) - } - - #[test] - fn type_specific_store_routes_before_generic_store() -> Result<()> { - let temp = tempfile::tempdir()?; - let generic_path = temp.path().join("generic.cc"); - let settings_path = temp.path().join("settings.cc"); - let mut cache = CultCache::new(); - cache.register_entry_type::()?; - cache.register_entry_type::()?; - cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&generic_path)); - cache.add_backing_store( - SingleFileMessagePackBackingStore::new(&settings_path), - ["settings"], - ); - - cache.put( - "app", - &Settings { - theme: "ash".to_string(), - retries: 3, - }, - )?; - cache.put( - "memo", - &Note { - title: "hello".to_string(), - body: "world".to_string(), - }, - )?; - - let generic_entries = SingleFileMessagePackBackingStore::new(&generic_path).pull_all()?; - let settings_entries = SingleFileMessagePackBackingStore::new(&settings_path).pull_all()?; - assert_eq!(generic_entries[0].r#type, "note"); - assert_eq!(settings_entries[0].r#type, "settings"); - Ok(()) - } - - #[test] - fn update_and_delete_follow_the_cache_api() -> Result<()> { - let temp = tempfile::tempdir()?; - let store_path = temp.path().join("cache.cc"); - let mut cache = CultCache::new(); - cache.register_entry_type::()?; - cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); - - let updated = cache.update::("app", |current| { - let mut current = current.unwrap_or(Settings { - theme: "ash".to_string(), - retries: 0, - }); - current.retries += 1; - current - })?; - assert_eq!(updated.retries, 1); - assert!(cache.delete::("app")?); - assert!(cache.get::("app")?.is_none()); - 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()?; - let store_path = temp.path().join("cache.cc"); - let mut cache = CultCache::new(); - cache.register_entry_type::()?; - cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); - cache.put( - "app", - &Settings { - theme: "ash".to_string(), - retries: 3, - }, - )?; - - let entry = cache.snapshot().remove(0); - let decoded: Settings = rmp_serde::from_slice(&entry.payload)?; - assert_eq!(decoded.theme, "ash"); - assert!(!entry.payload.is_empty()); - Ok(()) - } - - #[test] - fn corrupted_payload_fails_during_typed_retrieval() -> Result<()> { - let temp = tempfile::tempdir()?; - let store_path = temp.path().join("cache.cc"); - let mut store = SingleFileMessagePackBackingStore::new(&store_path); - store.push(&CultCacheEnvelope { - key: "app".to_string(), - r#type: "settings".to_string(), - payload: vec![0xc1], - stored_at: now_utc_second(), - schema_id: Some("settings".to_string()), - })?; - - let mut cache = CultCache::new(); - cache.register_entry_type::()?; - cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store_path)); - cache.pull_all_backing_stores()?; - let error = cache.get_required::("app").unwrap_err(); - assert!( - error - .to_string() - .contains("failed to decode CultCache entry") - ); - Ok(()) - } - - #[test] - fn put_envelope_reuses_existing_messagepack_payload() -> Result<()> { - let temp = tempfile::tempdir()?; - let origin_store = temp.path().join("origin.cc"); - let target_store = temp.path().join("target.cc"); - - let mut origin = CultCache::new(); - origin.register_entry_type::()?; - origin.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&origin_store)); - origin.put( - "app", - &Settings { - theme: "ash".to_string(), - retries: 3, - }, - )?; - - let envelope = origin.get_required_envelope::("app")?; - - let mut target = CultCache::new(); - target.register_entry_type::()?; - target.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&target_store)); - let applied = target.put_envelope::(envelope.clone())?; - - assert_eq!( - applied, - Settings { - theme: "ash".to_string(), - retries: 3, - } - ); - assert_eq!(target.get_required::("app")?, applied); - assert_eq!( - target.get_required_envelope::("app")?.payload, - envelope.payload - ); - Ok(()) - } -} diff --git a/vendor/cultmesh-rs/Cargo.lock b/vendor/cultmesh-rs/Cargo.lock index 328295c..7f68a9b 100644 --- a/vendor/cultmesh-rs/Cargo.lock +++ b/vendor/cultmesh-rs/Cargo.lock @@ -65,10 +65,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" [[package]] -name = "bitflags" -version = "2.11.1" +name = "base64ct" +version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3" +checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" + +[[package]] +name = "bitflags" +version = "2.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "84d7ced0ae9557296835c32bf1b1e02b44c746701f898460fb000d7eaa84f00a" [[package]] name = "block-buffer" @@ -125,6 +131,12 @@ dependencies = [ "inout", ] +[[package]] +name = "const-oid" +version = "0.9.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" + [[package]] name = "core-foundation-sys" version = "0.8.7" @@ -173,6 +185,7 @@ dependencies = [ "serde_bytes", "serde_json", "uuid", + "windows-sys 0.61.2", ] [[package]] @@ -206,6 +219,7 @@ dependencies = [ "base64", "chrono", "cultcache-rs", + "ed25519-dalek", "hmac", "rand", "rmp-serde", @@ -216,6 +230,45 @@ dependencies = [ "sha2", "socket2", "uuid", + "windows-sys 0.61.2", + "zeroize", +] + +[[package]] +name = "curve25519-dalek" +version = "4.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97fb8b7c4503de7d6ae7b42ab72a5a59857b4c937ec27a3d4539dba95b5ab2be" +dependencies = [ + "cfg-if", + "cpufeatures", + "curve25519-dalek-derive", + "digest", + "fiat-crypto", + "rustc_version", + "subtle", + "zeroize", +] + +[[package]] +name = "curve25519-dalek-derive" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "der" +version = "0.7.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" +dependencies = [ + "const-oid", + "zeroize", ] [[package]] @@ -235,6 +288,30 @@ dependencies = [ "subtle", ] +[[package]] +name = "ed25519" +version = "2.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "115531babc129696a58c64a4fef0a8bf9e9698629fb97e9e40767d235cfbcd53" +dependencies = [ + "pkcs8", + "signature", +] + +[[package]] +name = "ed25519-dalek" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70e796c081cee67dc755e1a36a0a172b897fab85fc3f6bc48307991f64e4eca9" +dependencies = [ + "curve25519-dalek", + "ed25519", + "serde", + "sha2", + "subtle", + "zeroize", +] + [[package]] name = "equivalent" version = "1.0.2" @@ -257,6 +334,12 @@ version = "2.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" +[[package]] +name = "fiat-crypto" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -515,6 +598,16 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "pkcs8" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7" +dependencies = [ + "der", + "spki", +] + [[package]] name = "polyval" version = "0.6.2" @@ -654,6 +747,15 @@ dependencies = [ "serde_bytes", ] +[[package]] +name = "rustc_version" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" +dependencies = [ + "semver", +] + [[package]] name = "rustix" version = "1.1.4" @@ -749,6 +851,15 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +[[package]] +name = "signature" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de" +dependencies = [ + "rand_core 0.6.4", +] + [[package]] name = "slab" version = "0.4.12" @@ -765,6 +876,16 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "spki" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d" +dependencies = [ + "base64ct", + "der", +] + [[package]] name = "subtle" version = "2.6.1" @@ -797,9 +918,9 @@ dependencies = [ [[package]] name = "typenum" -version = "1.20.0" +version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "40ce102ab67701b8526c123c1bab5cbe42d7040ccfd0f64af1a385808d2f43de" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" [[package]] name = "unicode-ident" @@ -825,9 +946,9 @@ dependencies = [ [[package]] name = "uuid" -version = "1.23.1" +version = "1.23.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ddd74a9687298c6858e9b88ec8935ec45d22e8fd5e6394fa1bd4e99a87789c76" +checksum = "d258b83ceec21034727ecee8c382cfa6c3e133699b0742c64571814fb420c9f7" dependencies = [ "getrandom 0.4.2", "js-sys", @@ -1208,24 +1329,30 @@ checksum = "cfe53a6657fd280eaa890a3bc59152892ffa3e30101319d168b781ed6529b049" [[package]] name = "zerocopy" -version = "0.8.48" +version = "0.8.50" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eed437bf9d6692032087e337407a86f04cd8d6a16a37199ed57949d415bd68e9" +checksum = "3b065d4f0e55f82fae73202e189638116a87c55ab6b8e6c2721e13dd9d854ad1" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.48" +version = "0.8.50" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "70e3cd084b1788766f53af483dd21f93881ff30d7320490ec3ef7526d203bad4" +checksum = "0b631b19d36a892ab55420c92dbc83ccd79274f25be714855d3074aa71cab639" dependencies = [ "proc-macro2", "quote", "syn", ] +[[package]] +name = "zeroize" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" + [[package]] name = "zmij" version = "1.0.21" diff --git a/vendor/cultmesh-rs/Cargo.toml b/vendor/cultmesh-rs/Cargo.toml index 9a719d3..441481d 100644 --- a/vendor/cultmesh-rs/Cargo.toml +++ b/vendor/cultmesh-rs/Cargo.toml @@ -10,8 +10,8 @@ path = "src/lib.rs" [dependencies] anyhow = "1" -cultcache-rs = { path = "../cultcache-rs" } -cultnet-rs = { path = "../cultnet-rs" } +cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585", package = "cultcache-rs" } +cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585", package = "cultnet-rs" } serde = { version = "1", features = ["derive"] } [dev-dependencies] diff --git a/vendor/cultmesh-rs/src/lib.rs b/vendor/cultmesh-rs/src/lib.rs index 3217a28..708198f 100644 --- a/vendor/cultmesh-rs/src/lib.rs +++ b/vendor/cultmesh-rs/src/lib.rs @@ -12,7 +12,9 @@ use cultnet_rs::decode_cultnet_message_from_slice; use cultnet_rs::encode_cultnet_message_to_vec; use serde::Serialize; use std::collections::BTreeMap; +use std::collections::hash_map::RandomState; use std::env; +use std::hash::{BuildHasher, Hasher}; use std::net::SocketAddr; use std::net::UdpSocket; use std::path::Path; @@ -20,6 +22,10 @@ use std::thread; use std::time::Duration; use std::time::Instant; +mod rudp_document_server; + +pub use rudp_document_server::*; + pub const CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID: u32 = 0x0d1d_0002; pub trait CultMeshDocumentSet: Clone + Send + Sync + 'static { @@ -72,7 +78,6 @@ pub struct CultMeshNodeOptions { pub struct CultMeshRudpDocumentPublishOptions { pub target: SocketAddr, pub runtime_id: String, - pub connection_id: u32, pub connect_timeout: Duration, pub flush_timeout: Duration, pub poll_interval: Duration, @@ -86,7 +91,10 @@ pub struct CultMeshRudpDocumentPublishOptions { pub struct CultMeshRudpSnapshotOptions { pub target: SocketAddr, pub runtime_id: String, - pub connection_id: u32, + /// Exact correlation id for the SnapshotRequest. Raw callers must supply + /// this; `CultMeshNode::pull_rudp_catalog_snapshot` generates one when it + /// owns the request. + pub message_id: String, pub connect_timeout: Duration, pub response_timeout: Duration, pub poll_interval: Duration, @@ -110,7 +118,7 @@ impl Default for CultMeshRudpSnapshotOptions { Self { target: SocketAddr::from(([0, 0, 0, 0], 0)), runtime_id: "cultmesh-rudp-snapshot-client".to_string(), - connection_id: CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID, + message_id: String::new(), connect_timeout: Duration::from_secs(3), response_timeout: Duration::from_secs(3), poll_interval: Duration::from_millis(10), @@ -136,7 +144,6 @@ impl Default for CultMeshRudpDocumentPublishOptions { Self { target: SocketAddr::from(([0, 0, 0, 0], 0)), runtime_id: "cultmesh-rudp-document-publisher".to_string(), - connection_id: CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID, connect_timeout: Duration::from_secs(3), flush_timeout: Duration::from_millis(300), poll_interval: Duration::from_millis(10), @@ -294,7 +301,10 @@ impl CultMeshStreamDescriptor { } pub fn metadata_schema_id(mut self, metadata_schema_id: impl Into) -> Result { - self.metadata_schema_id = Some(require_non_empty(metadata_schema_id.into(), "metadata_schema_id")?); + self.metadata_schema_id = Some(require_non_empty( + metadata_schema_id.into(), + "metadata_schema_id", + )?); Ok(self) } } @@ -388,7 +398,11 @@ pub struct CultMeshSharedMemoryFrameRing { } impl CultMeshSharedMemoryFrameRing { - pub fn new(stream_id: impl Into, slot_count: usize, slot_byte_length: usize) -> Result { + pub fn new( + stream_id: impl Into, + slot_count: usize, + slot_byte_length: usize, + ) -> Result { let stream_id = require_non_empty(stream_id.into(), "stream_id")?; if slot_count == 0 { anyhow::bail!("slot_count must be greater than zero"); @@ -494,7 +508,11 @@ impl CultMeshStreamCatalog { self.streams.values().collect() } - pub fn find(&self, verse_id: &str, kind: Option) -> Vec<&CultMeshStreamDescriptor> { + pub fn find( + &self, + verse_id: &str, + kind: Option, + ) -> Vec<&CultMeshStreamDescriptor> { self.streams .values() .filter(|stream| stream.verse_id == verse_id && kind.is_none_or(|k| stream.kind == k)) @@ -521,7 +539,9 @@ impl CultMeshStreamCatalog { .iter() .copied() .find(|candidate| consumer.supported_transports.contains(candidate)) - .ok_or_else(|| anyhow::anyhow!("stream and consumer have no compatible body transport"))?; + .ok_or_else(|| { + anyhow::anyhow!("stream and consumer have no compatible body transport") + })?; let producer_max = if stream.max_in_flight_frames == 0 { u32::MAX } else { @@ -671,8 +691,16 @@ impl CultMeshNode { pub fn pull_rudp_catalog_snapshot( &mut self, - options: CultMeshRudpSnapshotOptions, + mut options: CultMeshRudpSnapshotOptions, ) -> Result { + if options.message_id.trim().is_empty() { + options.message_id = format!( + "{}:snapshot:{}:{}", + options.runtime_id, + unix_millis(), + fresh_rudp_connection_epoch() + ); + } let response = request_raw_snapshot_from_rudp_catalog(options)?; let cultnet_rs::CultNetMessage::SnapshotResponseRaw { documents, .. } = response else { anyhow::bail!("expected cultnet.snapshot_response_raw.v0"); @@ -722,58 +750,38 @@ fn strip_schema_version_suffix(value: &str) -> &str { } } -fn request_raw_snapshot_from_rudp_catalog( +/// Request a raw CultNet snapshot without importing it into a local cache. +/// +/// The returned value is the exact validated `SnapshotResponseRaw`, including +/// its correlation `message_id`. Each call binds a fresh source port, chooses a +/// fresh connection epoch, and succeeds only after the reliable request is +/// acknowledged. +pub fn request_raw_snapshot_from_rudp_catalog( options: CultMeshRudpSnapshotOptions, ) -> Result { if options.runtime_id.trim().is_empty() { anyhow::bail!("runtime_id must be non-empty"); } - let bind_addr = if options.target.is_ipv4() { - SocketAddr::from(([0, 0, 0, 0], 0)) - } else { - SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 0], 0)) - }; - let socket = UdpSocket::bind(bind_addr)?; - socket.set_read_timeout(Some(options.poll_interval))?; - let mut client = CultNetRudpSocketTransportConnection::new( - CultNetRudpSocketTransportOptions { - runtime_id: options.runtime_id.clone(), - socket, - mode: cultnet_rs::CultNetRudpSocketMode::Client, - remote_addr: Some(options.target), - connection_id: options.connection_id, - initial_sequence: 1, - resend_delay_ms: options.resend_delay_ms, - transport_id: Some("cultmesh-rudp-snapshot-client".to_string()), - max_payload_bytes: None, - max_fragment_bytes: Some(1200), - max_pending_reliable_packets: None, - media_reliable_expire_after_ms: None, - }, + if options.message_id.trim().is_empty() { + anyhow::bail!("raw CultMesh RUDP snapshot requires a message_id"); + } + let message_id = options.message_id.clone(); + let mut client = connect_rudp_client( + options.target, + &options.runtime_id, + options.connect_timeout, + options.poll_interval, + options.resend_delay_ms, + "cultmesh-rudp-snapshot-client", )?; - client.connect(Vec::new())?; - let connect_deadline = Instant::now() + options.connect_timeout; - while !client.connected() && Instant::now() < connect_deadline { - let _ = client.receive_once()?; - client.poll_resends()?; - thread::sleep(options.poll_interval); - } - if !client.connected() { - anyhow::bail!( - "timed out connecting CultMesh RUDP snapshot client {} to {}", - options.runtime_id, - options.target - ); - } - let message_id = format!("{}:snapshot:{}", options.runtime_id, unix_millis()); let request = cultnet_rs::CultNetMessage::SnapshotRequest { message_id: message_id.clone(), schema_ids: options.schema_ids, record_keys: options.record_keys, }; let payload = encode_cultnet_message_to_vec(&request, CultNetWireContract::CultNetSchemaV0)?; - client.send("schema", payload)?; + let receipt = client.send_reliable("schema", payload)?; let response_deadline = Instant::now() + options.response_timeout; let mut snapshot_response = None; while Instant::now() < response_deadline { @@ -789,7 +797,6 @@ fn request_raw_snapshot_from_rudp_catalog( .. } if received == &message_id => { snapshot_response = Some(message); - break; } cultnet_rs::CultNetMessage::Error { error } => { let _ = client.disconnect(b"snapshot-error".to_vec()); @@ -799,15 +806,32 @@ fn request_raw_snapshot_from_rudp_catalog( } } } + match client.reliable_send_status(&receipt) { + cultnet_rs::CultNetRudpReliableSendStatus::Acknowledged + if snapshot_response.is_some() => + { + let _ = client.disconnect(b"snapshot-complete".to_vec()); + return Ok(snapshot_response.expect("response was checked as present")); + } + cultnet_rs::CultNetRudpReliableSendStatus::Invalidated => { + anyhow::bail!("CultMesh RUDP snapshot request receipt was invalidated") + } + _ => {} + } client.poll_resends()?; thread::sleep(options.poll_interval); } - if let Some(message) = snapshot_response { - let _ = client.disconnect(b"snapshot-complete".to_vec()); - return Ok(message); - } let _ = client.disconnect(b"snapshot-timeout".to_vec()); - anyhow::bail!("timed out waiting for CultMesh RUDP snapshot from {}", options.target) + if snapshot_response.is_some() { + anyhow::bail!( + "timed out waiting for reliable acknowledgement of CultMesh RUDP snapshot request to {}", + options.target + ); + } + anyhow::bail!( + "timed out waiting for CultMesh RUDP snapshot from {}", + options.target + ) } fn unix_millis() -> u128 { @@ -817,56 +841,100 @@ fn unix_millis() -> u128 { .as_millis() } -fn publish_cultnet_message_to_rudp_catalog( +/// Publish one raw CultNet schema message to a RUDP catalog. +/// +/// Each call owns a fresh connection epoch and returns only after the catalog +/// acknowledges reliable application admission. +pub fn publish_cultnet_message_to_rudp_catalog( message: &cultnet_rs::CultNetMessage, options: CultMeshRudpDocumentPublishOptions, ) -> Result<()> { if options.runtime_id.trim().is_empty() { anyhow::bail!("runtime_id must be non-empty"); } - let bind_addr = if options.target.is_ipv4() { + let mut client = connect_rudp_client( + options.target, + &options.runtime_id, + options.connect_timeout, + options.poll_interval, + options.resend_delay_ms, + "cultmesh-rudp-document-publisher", + )?; + + let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; + let receipt = client.send_reliable("schema", payload)?; + let flush_deadline = Instant::now() + options.flush_timeout; + while Instant::now() < flush_deadline { + let _ = client.receive_once()?; + match client.reliable_send_status(&receipt) { + cultnet_rs::CultNetRudpReliableSendStatus::Acknowledged => { + let _ = client.disconnect(b"document-acknowledged".to_vec()); + return Ok(()); + } + cultnet_rs::CultNetRudpReliableSendStatus::Invalidated => { + anyhow::bail!("CultMesh RUDP document receipt was invalidated") + } + cultnet_rs::CultNetRudpReliableSendStatus::Pending => {} + } + client.poll_resends()?; + thread::sleep(options.poll_interval); + } + let _ = client.disconnect(b"document-unacknowledged".to_vec()); + anyhow::bail!( + "timed out waiting for reliable acknowledgement from CultMesh RUDP catalog {}", + options.target + ) +} + +fn connect_rudp_client( + target: SocketAddr, + runtime_id: &str, + connect_timeout: Duration, + poll_interval: Duration, + resend_delay_ms: u64, + transport_id: &str, +) -> Result { + let bind_addr = if target.is_ipv4() { SocketAddr::from(([0, 0, 0, 0], 0)) } else { SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 0], 0)) }; let socket = UdpSocket::bind(bind_addr)?; - socket.set_read_timeout(Some(options.poll_interval))?; - let mut client = CultNetRudpSocketTransportConnection::new( - CultNetRudpSocketTransportOptions { - runtime_id: options.runtime_id.clone(), + socket.set_read_timeout(Some(poll_interval))?; + let mut client = + CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + runtime_id: runtime_id.to_string(), socket, mode: cultnet_rs::CultNetRudpSocketMode::Client, - remote_addr: Some(options.target), - connection_id: options.connection_id, + remote_addr: Some(target), + connection_id: fresh_rudp_connection_epoch(), initial_sequence: 1, - resend_delay_ms: options.resend_delay_ms, - transport_id: Some("cultmesh-rudp-document-publisher".to_string()), + resend_delay_ms, + transport_id: Some(transport_id.to_string()), max_payload_bytes: None, max_fragment_bytes: Some(1200), - max_pending_reliable_packets: None, + max_pending_reliable_packets: Some(1024), media_reliable_expire_after_ms: None, - }, - )?; + reconnect_policy: None, + })?; client.connect(Vec::new())?; - let connect_deadline = Instant::now() + options.connect_timeout; + let connect_deadline = Instant::now() + connect_timeout; while !client.connected() && Instant::now() < connect_deadline { let _ = client.receive_once()?; client.poll_resends()?; - thread::sleep(options.poll_interval); + thread::sleep(poll_interval); } if !client.connected() { - client.assume_connected(); + anyhow::bail!("timed out connecting CultMesh RUDP client {runtime_id} to {target}"); } + Ok(client) +} - let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; - client.send("schema", payload)?; - let flush_deadline = Instant::now() + options.flush_timeout; - while Instant::now() < flush_deadline { - let _ = client.receive_once()?; - client.poll_resends()?; - thread::sleep(options.poll_interval); - } - Ok(()) +fn fresh_rudp_connection_epoch() -> u32 { + let mut hasher = RandomState::new().build_hasher(); + hasher.write_u128(unix_millis()); + let epoch = hasher.finish() as u32; + if epoch == 0 { 1 } else { epoch } } #[derive(Clone, Copy, Debug, Default)] @@ -879,8 +947,9 @@ impl CultMesh { return parse_rudp_socket_addr(&text); } if text.starts_with("cultmesh://") { - let resolved = default_cultmesh_rudp_endpoint_resolver(&text)? - .ok_or_else(|| anyhow::anyhow!("CultMesh URI {text} did not resolve to a RUDP endpoint"))?; + let resolved = default_cultmesh_rudp_endpoint_resolver(&text)?.ok_or_else(|| { + anyhow::anyhow!("CultMesh URI {text} did not resolve to a RUDP endpoint") + })?; return parse_rudp_socket_addr(&resolved); } parse_socket_addr(&text) @@ -959,10 +1028,7 @@ pub fn resolve_rudp_endpoint(endpoint: &str) -> Result { } fn parse_rudp_socket_addr(value: &str) -> Result { - let text = value - .trim() - .strip_prefix("rudp://") - .unwrap_or(value.trim()); + let text = value.trim().strip_prefix("rudp://").unwrap_or(value.trim()); parse_socket_addr(text) } @@ -976,7 +1042,13 @@ fn default_cultmesh_rudp_endpoint_resolver(uri: &str) -> Result> let authority = cultmesh_uri_authority(uri)?; let slug = authority .chars() - .map(|ch| if ch.is_ascii_alphanumeric() { ch.to_ascii_uppercase() } else { '_' }) + .map(|ch| { + if ch.is_ascii_alphanumeric() { + ch.to_ascii_uppercase() + } else { + '_' + } + }) .collect::() .trim_matches('_') .to_string(); @@ -1006,11 +1078,7 @@ fn cultmesh_uri_authority(uri: &str) -> Result { let rest = uri .strip_prefix("cultmesh://") .ok_or_else(|| anyhow::anyhow!("CultMesh URI must start with cultmesh://"))?; - let authority = rest - .split(['/', '?', '#']) - .next() - .unwrap_or("") - .trim(); + let authority = rest.split(['/', '?', '#']).next().unwrap_or("").trim(); if authority.is_empty() { anyhow::bail!("CultMesh URI must include an authority"); } @@ -1084,33 +1152,33 @@ mod tests { socket.set_read_timeout(Some(Duration::from_millis(10)))?; let target = socket.local_addr()?; let (sender, receiver) = std::sync::mpsc::channel(); + let delivered = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let delivered_by_sink = delivered.clone(); let server = thread::spawn(move || -> Result<()> { - let mut server = CultNetRudpSocketTransportConnection::new( - CultNetRudpSocketTransportOptions::server( - "odin-test-catalog", - socket, - CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID, - ), + let mut server = CultMeshRudpDocumentServer::new( + socket, + move |receipt: CultMeshRudpRawDocumentReceipt| { + sender + .send(( + receipt.document.schema_id, + receipt.document.record_key, + receipt.document.source_runtime_id, + )) + .ok(); + delivered_by_sink.store(true, std::sync::atomic::Ordering::SeqCst); + Ok(()) + }, + |_query: &CultMeshRudpSnapshotQuery| Ok(Vec::new()), + CultMeshSystemClock::default(), + CultMeshRudpDocumentServerOptions::default(), )?; let deadline = Instant::now() + Duration::from_secs(2); while Instant::now() < deadline { - if let Some(frame) = server.receive_once()? { - let message = cultnet_rs::decode_cultnet_message_from_slice( - &frame.payload, - CultNetWireContract::CultNetSchemaV0, - )?; - if let cultnet_rs::CultNetMessage::DocumentPutRaw { document, .. } = message { - sender - .send(( - document.schema_id, - document.record_key, - document.source_runtime_id, - )) - .ok(); - return Ok(()); - } + server.poll_once()?; + if delivered.load(std::sync::atomic::Ordering::SeqCst) { + return Ok(()); } - server.poll_resends()?; + thread::sleep(Duration::from_millis(1)); } anyhow::bail!("timed out waiting for RUDP document put") }); @@ -1129,6 +1197,232 @@ mod tests { Ok(()) } + #[test] + fn document_publish_fails_when_catalog_is_unreachable() -> Result<()> { + let temp = tempfile::tempdir()?; + let node = CultMesh::create_node( + temp.path().join("cultmesh.cc"), + TestDocuments, + Default::default(), + )?; + let unused = UdpSocket::bind("127.0.0.1:0")?; + let target = unused.local_addr()?; + drop(unused); + + let error = node + .publish_document_to_rudp_catalog( + "note", + &Note { + body: "no false success".into(), + }, + CultMeshRudpDocumentPublishOptions { + target, + connect_timeout: Duration::from_millis(30), + poll_interval: Duration::from_millis(2), + ..CultMeshRudpDocumentPublishOptions::default() + }, + ) + .expect_err("unreachable catalog must fail closed"); + assert!(error.to_string().contains("timed out connecting")); + Ok(()) + } + + #[test] + fn document_publish_requires_application_acknowledgement() -> Result<()> { + let temp = tempfile::tempdir()?; + let node = CultMesh::create_node( + temp.path().join("cultmesh.cc"), + TestDocuments, + Default::default(), + )?; + let socket = UdpSocket::bind("127.0.0.1:0")?; + let target = socket.local_addr()?; + let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let server_stop = stop.clone(); + let server = thread::spawn(move || -> Result<()> { + let mut server = CultMeshRudpDocumentServer::new( + socket, + |_receipt: CultMeshRudpRawDocumentReceipt| { + anyhow::bail!("injected durable sink failure") + }, + |_query: &CultMeshRudpSnapshotQuery| Ok(Vec::new()), + CultMeshSystemClock::default(), + CultMeshRudpDocumentServerOptions::default(), + )?; + while !server_stop.load(std::sync::atomic::Ordering::SeqCst) { + let _ = server.poll_once(); + thread::sleep(Duration::from_millis(1)); + } + Ok(()) + }); + + let error = node + .publish_document_to_rudp_catalog( + "note", + &Note { + body: "still not accepted".into(), + }, + CultMeshRudpDocumentPublishOptions { + target, + connect_timeout: Duration::from_millis(100), + flush_timeout: Duration::from_millis(50), + poll_interval: Duration::from_millis(2), + resend_delay_ms: 5, + ..CultMeshRudpDocumentPublishOptions::default() + }, + ) + .expect_err("unacknowledged document must fail closed"); + stop.store(true, std::sync::atomic::Ordering::SeqCst); + server.join().expect("server thread should not panic")?; + assert!(error.to_string().contains("reliable acknowledgement")); + Ok(()) + } + + #[test] + fn raw_snapshot_preserves_caller_challenge_and_raw_response() -> Result<()> { + let expected = cultnet_rs::CultNetRawDocumentRecord { + schema_id: "cultmesh.test.raw.v0".into(), + record_key: "route-proof".into(), + stored_at: "2026-09-03T00:00:00Z".into(), + payload_encoding: cultnet_rs::CultNetRawPayloadEncoding::Messagepack, + payload: vec![0xc4, 0x03, 0, 0xff, 1], + source_runtime_id: Some("odin-test".into()), + source_agent_id: None, + source_role: None, + tags: None, + }; + let socket = UdpSocket::bind("127.0.0.1:0")?; + let target = socket.local_addr()?; + let challenge = "9aa63ad8-83b5-4cc4-ae4a-91d361d5f845".to_string(); + let server_document = expected.clone(); + let server_challenge = challenge.clone(); + let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let server_stop = stop.clone(); + let server = thread::spawn(move || -> Result<()> { + let mut server = CultMeshRudpDocumentServer::new( + socket, + |_receipt: CultMeshRudpRawDocumentReceipt| Ok(()), + move |query: &CultMeshRudpSnapshotQuery| { + if query.message_id != server_challenge { + anyhow::bail!("snapshot challenge changed") + } + Ok(vec![server_document.clone()]) + }, + CultMeshSystemClock::default(), + CultMeshRudpDocumentServerOptions::default(), + )?; + while !server_stop.load(std::sync::atomic::Ordering::SeqCst) { + server.poll_once()?; + thread::sleep(Duration::from_millis(1)); + } + Ok(()) + }); + + let response = request_raw_snapshot_from_rudp_catalog(CultMeshRudpSnapshotOptions { + target, + runtime_id: "route-proof-test".into(), + message_id: challenge.clone(), + connect_timeout: Duration::from_millis(100), + response_timeout: Duration::from_millis(500), + poll_interval: Duration::from_millis(2), + resend_delay_ms: 5, + schema_ids: Some(vec!["cultmesh.test.raw.v0".into()]), + record_keys: Some(vec!["route-proof".into()]), + }); + stop.store(true, std::sync::atomic::Ordering::SeqCst); + server.join().expect("server thread should not panic")?; + let response = response?; + assert_eq!( + response, + cultnet_rs::CultNetMessage::SnapshotResponseRaw { + message_id: challenge, + documents: vec![expected], + } + ); + Ok(()) + } + + #[test] + fn raw_snapshot_rejects_response_without_request_acknowledgement() -> Result<()> { + let socket = UdpSocket::bind("127.0.0.1:0")?; + socket.set_read_timeout(Some(Duration::from_millis(200)))?; + let target = socket.local_addr()?; + let server = thread::spawn(move || -> Result<()> { + let mut wire = vec![0_u8; 65_535]; + let (received, peer) = socket.recv_from(&mut wire)?; + let connect = cultnet_rs::decode_rudp_packet(&wire[..received])?; + let mut session = + cultnet_rs::CultNetRudpSession::new(cultnet_rs::CultNetRudpSessionOptions { + connection_id: connect.connection_id, + initial_sequence: 1, + resend_delay_ms: 5, + max_pending_reliable_packets: Some(64), + }); + let accept = session.accept_connect(&connect, 1, Vec::new())?; + socket.send_to(&cultnet_rs::encode_rudp_packet(&accept)?, peer)?; + + loop { + let (received, received_peer) = socket.recv_from(&mut wire)?; + if received_peer != peer { + continue; + } + let packet = cultnet_rs::decode_rudp_packet(&wire[..received])?; + let result = session.receive(&packet, 2)?; + let Some(frame) = result.delivered.into_iter().next() else { + continue; + }; + let request = decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + )?; + let cultnet_rs::CultNetMessage::SnapshotRequest { message_id, .. } = request else { + anyhow::bail!("expected snapshot request") + }; + let payload = encode_cultnet_message_to_vec( + &cultnet_rs::CultNetMessage::SnapshotResponseRaw { + message_id, + documents: Vec::new(), + }, + CultNetWireContract::CultNetSchemaV0, + )?; + let mut packets = session.send_many( + "schema", + payload, + cultnet_rs::CultNetRudpSendOptions { + reliable: true, + ordered: true, + sequenced: false, + now_ms: 2, + reliable_expire_after_ms: None, + }, + Some(1200), + )?; + for packet in &mut packets { + packet.ack = 0; + packet.ack_mask = 0; + socket.send_to(&cultnet_rs::encode_rudp_packet(packet)?, peer)?; + } + return Ok(()); + } + }); + + let error = request_raw_snapshot_from_rudp_catalog(CultMeshRudpSnapshotOptions { + target, + runtime_id: "unacknowledged-snapshot-test".into(), + message_id: "b7987efb-69c5-4c42-9355-b6e09535ad21".into(), + connect_timeout: Duration::from_millis(100), + response_timeout: Duration::from_millis(50), + poll_interval: Duration::from_millis(2), + resend_delay_ms: 5, + schema_ids: None, + record_keys: None, + }) + .expect_err("response without request ACK must fail closed"); + server.join().expect("server thread should not panic")?; + assert!(error.to_string().contains("reliable acknowledgement")); + Ok(()) + } + #[test] fn resolves_rudp_endpoint_from_cultmesh_uri_bootstrap_env() -> Result<()> { let previous = env::var("CULTMESH_URI_ODIN_RUDP").ok(); @@ -1154,8 +1448,9 @@ mod tests { unsafe { env::remove_var("CULTMESH_URI_MISSING_ODIN_RUDP"); } - let error = CultMesh::resolve_rudp_endpoint("cultmesh://missing-odin/rendezvous/provider-catalog") - .expect_err("unresolved CultMesh URI should fail"); + let error = + CultMesh::resolve_rudp_endpoint("cultmesh://missing-odin/rendezvous/provider-catalog") + .expect_err("unresolved CultMesh URI should fail"); unsafe { if let Some(value) = previous { env::set_var("CULTMESH_URI_MISSING_ODIN_RUDP", value); @@ -1180,25 +1475,23 @@ mod tests { body: "catalog truth".to_string(), })?; - node.cache.put_raw_envelope(cultcache_rs::CultCacheEnvelope { - key: "note".to_string(), - r#type: node - .documents - .binding_by_schema_id("cultmesh.test.note.v0") - .expect("test binding") - .document_type - .clone(), - payload, - stored_at: "2026-07-03T00:00:00Z".to_string(), - schema_id: Some("cultmesh.test.note.v0".to_string()), - })?; + node.cache + .put_raw_envelope(cultcache_rs::CultCacheEnvelope { + key: "note".to_string(), + r#type: node + .documents + .binding_by_schema_id("cultmesh.test.note.v0") + .expect("test binding") + .document_type + .clone(), + payload, + stored_at: "2026-07-03T00:00:00Z".to_string(), + schema_id: Some("cultmesh.test.note.v0".to_string()), + })?; node.flush()?; let reloaded = CultMesh::create_node(&store_path, TestDocuments, Default::default())?; - assert_eq!( - reloaded.get_required::("note")?.body, - "catalog truth" - ); + assert_eq!(reloaded.get_required::("note")?.body, "catalog truth"); assert_eq!(reloaded.cache().snapshot()[0].r#type, Note::TYPE); Ok(()) } @@ -1232,15 +1525,18 @@ mod tests { )?; let negotiation = catalog.negotiate("muninn:nightwing:move-evidence", &consumer)?; - assert_eq!(negotiation.transport, CultMeshStreamBodyTransport::SharedMemory); - assert_eq!(negotiation.copy_budget, CultMeshStreamCopyBudget::ZeroCopyTarget); + assert_eq!( + negotiation.transport, + CultMeshStreamBodyTransport::SharedMemory + ); + assert_eq!( + negotiation.copy_budget, + CultMeshStreamCopyBudget::ZeroCopyTarget + ); let handle = { - let ring = catalog.create_shared_memory_ring( - "muninn:nightwing:move-evidence", - 4, - 128, - )?; + let ring = + catalog.create_shared_memory_ring("muninn:nightwing:move-evidence", 4, 128)?; ring.try_publish_copy(&[1, 2, 3], 42, 0)? .expect("frame should publish") }; diff --git a/vendor/cultmesh-rs/src/rudp_document_server.rs b/vendor/cultmesh-rs/src/rudp_document_server.rs new file mode 100644 index 0000000..bebeb6c --- /dev/null +++ b/vendor/cultmesh-rs/src/rudp_document_server.rs @@ -0,0 +1,647 @@ +use anyhow::{Result, anyhow}; +use cultnet_rs::{ + CultNetMessage, CultNetRawDocumentRecord, CultNetRudpPacket, CultNetRudpPacketType, + CultNetRudpSendOptions, CultNetRudpSession, CultNetRudpSessionOptions, CultNetWireContract, + decode_cultnet_message_from_slice, decode_rudp_packet, encode_cultnet_message_to_vec, + encode_rudp_packet, +}; +use std::collections::BTreeMap; +use std::io::ErrorKind; +use std::net::{SocketAddr, UdpSocket}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +const MAX_UDP_DATAGRAM_BYTES: usize = 65_535; + +/// The transport identity of one remote CultNet RUDP session. +/// +/// Connection identifiers are scoped to a remote socket address. Two clients +/// may therefore use the same connection identifier without sharing ordering, +/// acknowledgement, resend, or fragment state. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub struct CultMeshRudpSessionKey { + pub remote_addr: SocketAddr, + pub connection_id: u32, +} + +/// A raw document plus the receipt fact owned by the receiving process. +/// +/// `document.payload` is the exact byte vector decoded from the CultNet raw +/// record. The trusted receipt time lives beside the sender-owned record so the +/// transport never rewrites its metadata or payload. +#[derive(Clone, Debug, PartialEq)] +pub struct CultMeshRudpRawDocumentReceipt { + pub session: CultMeshRudpSessionKey, + pub message_id: String, + pub transport_sequence: u32, + pub received_at_unix_millis: u64, + pub document: CultNetRawDocumentRecord, +} + +/// The caller-visible query for a read-only raw snapshot. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CultMeshRudpSnapshotQuery { + pub session: CultMeshRudpSessionKey, + pub message_id: String, + pub requested_at_unix_millis: u64, + pub schema_ids: Option>, + pub record_keys: Option>, +} + +/// Caller-owned admission/persistence port for received raw documents. +pub trait CultMeshRudpRawDocumentSink { + fn accept_raw_document(&mut self, receipt: CultMeshRudpRawDocumentReceipt) -> Result<()>; +} + +impl CultMeshRudpRawDocumentSink for F +where + F: FnMut(CultMeshRudpRawDocumentReceipt) -> Result<()>, +{ + fn accept_raw_document(&mut self, receipt: CultMeshRudpRawDocumentReceipt) -> Result<()> { + self(receipt) + } +} + +/// Caller-owned catalog port for serving raw snapshot requests. +/// +/// The caller decides which records the requester may see. CultMesh preserves +/// those records as raw CultNet documents and does not interpret their schemas. +pub trait CultMeshRudpSnapshotSource { + fn raw_snapshot( + &mut self, + query: &CultMeshRudpSnapshotQuery, + ) -> Result>; +} + +impl CultMeshRudpSnapshotSource for F +where + F: FnMut(&CultMeshRudpSnapshotQuery) -> Result>, +{ + fn raw_snapshot( + &mut self, + query: &CultMeshRudpSnapshotQuery, + ) -> Result> { + self(query) + } +} + +/// Trusted local clocks for receipt facts and transport maintenance. +/// +/// Receipt time is wall time. Expiry and resend decisions use the monotonic +/// value, so a wall-clock correction cannot resurrect or pin a session. +pub trait CultMeshRudpServerClock { + fn now_unix_millis(&self) -> u64; + fn now_monotonic_millis(&self) -> u64; +} + +#[derive(Clone, Debug)] +pub struct CultMeshSystemClock { + monotonic_origin: Instant, +} + +impl Default for CultMeshSystemClock { + fn default() -> Self { + Self { + monotonic_origin: Instant::now(), + } + } +} + +impl CultMeshRudpServerClock for CultMeshSystemClock { + fn now_unix_millis(&self) -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .try_into() + .unwrap_or(u64::MAX) + } + + fn now_monotonic_millis(&self) -> u64 { + self.monotonic_origin + .elapsed() + .as_millis() + .try_into() + .unwrap_or(u64::MAX) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CultMeshRudpDocumentServerOptions { + pub max_sessions: usize, + pub session_idle_timeout: Duration, + pub session_max_lifetime: Duration, + /// Conservative payload admission budget across all live sessions. + /// + /// Admitted ingress and queued snapshot-response bytes are charged until + /// the session ends even when CultNet releases them sooner. Together with + /// CultNet's per-session receive caps, this makes retained payload memory + /// finite without duplicating its fragment/ordering machinery here. + pub max_admitted_payload_bytes: usize, + pub max_admitted_payload_bytes_per_session: usize, + pub max_snapshot_response_bytes: usize, + pub max_snapshot_documents: usize, + pub resend_delay: Duration, + pub max_pending_reliable_packets_per_session: usize, + pub max_fragment_bytes: usize, +} + +impl Default for CultMeshRudpDocumentServerOptions { + fn default() -> Self { + Self { + max_sessions: 64, + session_idle_timeout: Duration::from_secs(30), + session_max_lifetime: Duration::from_secs(15 * 60), + max_admitted_payload_bytes: 32 * 1024 * 1024, + max_admitted_payload_bytes_per_session: 4 * 1024 * 1024, + max_snapshot_response_bytes: 1024 * 1024, + max_snapshot_documents: 4096, + resend_delay: Duration::from_millis(50), + max_pending_reliable_packets_per_session: 1024, + max_fragment_bytes: 1200, + } + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct CultMeshRudpMaintenance { + pub sessions_expired: usize, + pub packets_resent: usize, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum CultMeshRudpApplicationOperation { + DocumentPutRaw, + SnapshotRequest, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CultMeshRudpApplicationRejection { + pub session: CultMeshRudpSessionKey, + pub operation: CultMeshRudpApplicationOperation, + pub message_id: String, + pub reason: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum CultMeshRudpPollOutcome { + Idle, + Handled, + ApplicationRejected(CultMeshRudpApplicationRejection), +} + +struct SessionEntry { + session: CultNetRudpSession, + created_at_monotonic_millis: u64, + last_activity_monotonic_millis: u64, + admitted_payload_bytes: usize, +} + +/// A synchronous, multi-session CultNet RUDP raw-document server. +/// +/// The server owns exactly one UDP socket. Call `poll_once` from an existing +/// daemon loop; no executor or background thread is created here. +pub struct CultMeshRudpDocumentServer { + socket: UdpSocket, + sessions: BTreeMap, + sink: S, + snapshot_source: Q, + clock: C, + options: CultMeshRudpDocumentServerOptions, +} + +impl CultMeshRudpDocumentServer +where + S: CultMeshRudpRawDocumentSink, + Q: CultMeshRudpSnapshotSource, + C: CultMeshRudpServerClock, +{ + pub fn new( + socket: UdpSocket, + sink: S, + snapshot_source: Q, + clock: C, + options: CultMeshRudpDocumentServerOptions, + ) -> Result { + validate_options(&options)?; + socket.local_addr().map_err(|error| { + anyhow!("CultMesh RUDP server requires a bound UDP socket: {error}") + })?; + socket.set_nonblocking(true)?; + Ok(Self { + socket, + sessions: BTreeMap::new(), + sink, + snapshot_source, + clock, + options, + }) + } + + pub fn local_addr(&self) -> Result { + Ok(self.socket.local_addr()?) + } + + pub fn session_count(&self) -> usize { + self.sessions.len() + } + + /// Poll transport maintenance and receive at most one UDP datagram. + /// + /// Application rejection invalidates only the responsible peer session and + /// is returned as data so a daemon can log it and keep serving. Local + /// socket and server-state failures remain errors. + pub fn poll_once(&mut self) -> Result { + self.maintain()?; + let mut wire = vec![0_u8; MAX_UDP_DATAGRAM_BYTES]; + let (received, remote_addr) = match self.socket.recv_from(&mut wire) { + Ok(value) => value, + Err(error) if error.kind() == ErrorKind::WouldBlock => { + return Ok(CultMeshRudpPollOutcome::Idle); + } + Err(error) => return Err(error.into()), + }; + wire.truncate(received); + + let packet = match decode_rudp_packet(&wire) { + Ok(packet) => packet, + Err(_) => return Ok(CultMeshRudpPollOutcome::Handled), + }; + let now_unix = self.clock.now_unix_millis(); + let now_monotonic = self.clock.now_monotonic_millis(); + let key = CultMeshRudpSessionKey { + remote_addr, + connection_id: packet.connection_id, + }; + + if packet.packet_type == CultNetRudpPacketType::Connect { + self.accept_connection(key, &packet, now_monotonic)?; + return Ok(CultMeshRudpPollOutcome::Handled); + } + + if !self.sessions.contains_key(&key) { + return Ok(CultMeshRudpPollOutcome::Handled); + } + if packet.packet_type == CultNetRudpPacketType::Data + && packet.channel_id == "schema" + && !packet.reliable + { + return Ok(CultMeshRudpPollOutcome::Handled); + } + let payload_charge = if packet.packet_type == CultNetRudpPacketType::Data { + packet.payload.len() + } else { + 0 + }; + if !self.payload_budget_allows(key, payload_charge) { + return Ok(CultMeshRudpPollOutcome::Handled); + } + + let result = { + let entry = self + .sessions + .get_mut(&key) + .expect("checked session must remain present"); + entry.last_activity_monotonic_millis = now_monotonic; + entry.admitted_payload_bytes = + entry.admitted_payload_bytes.saturating_add(payload_charge); + entry.session.receive(&packet, now_monotonic) + }; + let result = match result { + Ok(result) => result, + Err(_) => { + self.sessions.remove(&key); + return Ok(CultMeshRudpPollOutcome::Handled); + } + }; + + if let Some(reply) = result.reply { + self.send_packet(key.remote_addr, &reply)?; + } + + if result.disconnected { + self.sessions.remove(&key); + return Ok(CultMeshRudpPollOutcome::Handled); + } + + for frame in result.delivered { + if frame.channel_id != "schema" { + continue; + } + let message = match decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + ) { + Ok(message) => message, + Err(_) => continue, + }; + if let Some(rejection) = self.deliver_application_message( + key, + frame.sequence, + message, + now_unix, + now_monotonic, + )? { + self.sessions.remove(&key); + return Ok(CultMeshRudpPollOutcome::ApplicationRejected(rejection)); + } + } + + if packet.reliable { + let ack = self + .sessions + .get_mut(&key) + .ok_or_else(|| anyhow!("CultMesh RUDP session disappeared before ACK"))? + .session + .create_ack(); + self.send_packet(key.remote_addr, &ack)?; + } + Ok(CultMeshRudpPollOutcome::Handled) + } + + /// Expire idle sessions and resend reliable packets whose deadline passed. + pub fn maintain(&mut self) -> Result { + let now = self.clock.now_monotonic_millis(); + let idle_timeout_ms = duration_millis(self.options.session_idle_timeout); + let lifetime_ms = duration_millis(self.options.session_max_lifetime); + let before = self.sessions.len(); + self.sessions.retain(|_, entry| { + now.saturating_sub(entry.last_activity_monotonic_millis) <= idle_timeout_ms + && now.saturating_sub(entry.created_at_monotonic_millis) <= lifetime_ms + }); + let expired = before - self.sessions.len(); + + let mut resends = Vec::new(); + for (key, entry) in &mut self.sessions { + for packet in entry.session.due_resends(now) { + resends.push((key.remote_addr, packet)); + } + } + for (remote_addr, packet) in &resends { + self.send_packet(*remote_addr, packet)?; + } + Ok(CultMeshRudpMaintenance { + sessions_expired: expired, + packets_resent: resends.len(), + }) + } + + fn accept_connection( + &mut self, + key: CultMeshRudpSessionKey, + packet: &CultNetRudpPacket, + now: u64, + ) -> Result<()> { + if !self.sessions.contains_key(&key) { + if self.sessions.len() >= self.options.max_sessions { + return Ok(()); + } + self.sessions.insert( + key, + SessionEntry { + session: CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: key.connection_id, + initial_sequence: 1, + resend_delay_ms: duration_millis(self.options.resend_delay), + max_pending_reliable_packets: Some( + self.options.max_pending_reliable_packets_per_session, + ), + }), + created_at_monotonic_millis: now, + last_activity_monotonic_millis: now, + admitted_payload_bytes: 0, + }, + ); + } + + let accept = { + let entry = self + .sessions + .get_mut(&key) + .expect("accepted session must be present"); + // The same key is a retransmitted Connect for the existing epoch. + // A genuinely fresh client incarnation must choose a fresh id. + entry.last_activity_monotonic_millis = now; + entry.session.accept_connect(packet, now, Vec::new())? + }; + self.send_packet(key.remote_addr, &accept) + } + + fn deliver_application_message( + &mut self, + key: CultMeshRudpSessionKey, + transport_sequence: u32, + message: CultNetMessage, + now_unix: u64, + now: u64, + ) -> Result> { + match message { + CultNetMessage::DocumentPutRaw { + message_id, + document, + } => { + let receipt = CultMeshRudpRawDocumentReceipt { + session: key, + message_id: message_id.clone(), + transport_sequence, + received_at_unix_millis: now_unix, + document, + }; + if let Err(error) = self.sink.accept_raw_document(receipt) { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::DocumentPutRaw, + message_id, + reason: format!("{error:#}"), + })); + } + } + CultNetMessage::SnapshotRequest { + message_id, + schema_ids, + record_keys, + } => { + let query = CultMeshRudpSnapshotQuery { + session: key, + message_id: message_id.clone(), + requested_at_unix_millis: now_unix, + schema_ids, + record_keys, + }; + let documents = match self.snapshot_source.raw_snapshot(&query) { + Ok(documents) => documents, + Err(error) => { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::SnapshotRequest, + message_id, + reason: format!("{error:#}"), + })); + } + }; + if documents.len() > self.options.max_snapshot_documents { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::SnapshotRequest, + message_id, + reason: format!( + "snapshot returned {} documents; limit is {}", + documents.len(), + self.options.max_snapshot_documents + ), + })); + } + let response = CultNetMessage::SnapshotResponseRaw { + message_id: message_id.clone(), + documents, + }; + let payload = match encode_cultnet_message_to_vec( + &response, + CultNetWireContract::CultNetSchemaV0, + ) { + Ok(payload) => payload, + Err(error) => { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::SnapshotRequest, + message_id, + reason: format!("snapshot response could not be encoded: {error:#}"), + })); + } + }; + if payload.len() > self.options.max_snapshot_response_bytes { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::SnapshotRequest, + message_id, + reason: format!( + "snapshot response is {} bytes; limit is {}", + payload.len(), + self.options.max_snapshot_response_bytes + ), + })); + } + if !self.payload_budget_allows(key, payload.len()) { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::SnapshotRequest, + message_id, + reason: "retained payload budget is full".into(), + })); + } + let payload_bytes = payload.len(); + let entry = self + .sessions + .get_mut(&key) + .ok_or_else(|| anyhow!("CultMesh RUDP session disappeared before response"))?; + let packets = match entry.session.send_many( + "schema", + payload, + CultNetRudpSendOptions { + reliable: true, + ordered: true, + sequenced: false, + now_ms: now, + reliable_expire_after_ms: None, + }, + Some(self.options.max_fragment_bytes), + ) { + Ok(packets) => packets, + Err(error) => { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::SnapshotRequest, + message_id, + reason: format!("snapshot response could not be queued: {error:#}"), + })); + } + }; + entry.admitted_payload_bytes = + entry.admitted_payload_bytes.saturating_add(payload_bytes); + for packet in &packets { + self.send_packet(key.remote_addr, packet)?; + } + } + _ => {} + } + Ok(None) + } + + fn payload_budget_allows(&self, key: CultMeshRudpSessionKey, bytes: usize) -> bool { + let Some(session_bytes) = self + .sessions + .get(&key) + .map(|entry| entry.admitted_payload_bytes) + else { + return false; + }; + if session_bytes + .checked_add(bytes) + .is_none_or(|total| total > self.options.max_admitted_payload_bytes_per_session) + { + return false; + } + self.sessions + .values() + .try_fold(0_usize, |total, entry| { + total.checked_add(entry.admitted_payload_bytes) + }) + .and_then(|total| total.checked_add(bytes)) + .is_some_and(|total| total <= self.options.max_admitted_payload_bytes) + } + + fn send_packet(&mut self, remote_addr: SocketAddr, packet: &CultNetRudpPacket) -> Result<()> { + let wire = encode_rudp_packet(packet)?; + self.socket.send_to(&wire, remote_addr)?; + Ok(()) + } +} + +fn validate_options(options: &CultMeshRudpDocumentServerOptions) -> Result<()> { + if options.max_sessions == 0 { + return Err(anyhow!("max_sessions must be greater than zero")); + } + if options.session_idle_timeout.is_zero() { + return Err(anyhow!("session_idle_timeout must be greater than zero")); + } + if options.session_max_lifetime.is_zero() { + return Err(anyhow!("session_max_lifetime must be greater than zero")); + } + if options.max_admitted_payload_bytes == 0 { + return Err(anyhow!( + "max_admitted_payload_bytes must be greater than zero" + )); + } + if options.max_admitted_payload_bytes_per_session == 0 + || options.max_admitted_payload_bytes_per_session > options.max_admitted_payload_bytes + { + return Err(anyhow!( + "per-session admitted payload limit must be non-zero and no larger than the server limit" + )); + } + if options.max_snapshot_response_bytes == 0 + || options.max_snapshot_response_bytes > options.max_admitted_payload_bytes_per_session + { + return Err(anyhow!( + "snapshot response limit must be non-zero and fit in one session's payload budget" + )); + } + if options.max_snapshot_documents == 0 { + return Err(anyhow!("max_snapshot_documents must be greater than zero")); + } + if options.resend_delay.is_zero() { + return Err(anyhow!("resend_delay must be greater than zero")); + } + if options.max_pending_reliable_packets_per_session == 0 { + return Err(anyhow!( + "max_pending_reliable_packets_per_session must be greater than zero" + )); + } + if options.max_fragment_bytes == 0 { + return Err(anyhow!("max_fragment_bytes must be greater than zero")); + } + Ok(()) +} + +fn duration_millis(duration: Duration) -> u64 { + duration.as_millis().try_into().unwrap_or(u64::MAX) +} diff --git a/vendor/cultmesh-rs/tests/rudp_document_server.rs b/vendor/cultmesh-rs/tests/rudp_document_server.rs new file mode 100644 index 0000000..1e57aaf --- /dev/null +++ b/vendor/cultmesh-rs/tests/rudp_document_server.rs @@ -0,0 +1,553 @@ +use anyhow::Result; +use cultmesh_rs::{ + CultMeshRudpApplicationOperation, CultMeshRudpDocumentServer, + CultMeshRudpDocumentServerOptions, CultMeshRudpPollOutcome, CultMeshRudpRawDocumentReceipt, + CultMeshRudpRawDocumentSink, CultMeshRudpServerClock, CultMeshRudpSnapshotQuery, + CultMeshRudpSnapshotSource, +}; +use cultnet_rs::{ + CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding, + CultNetRudpReliableSendReceipt, CultNetRudpReliableSendStatus, CultNetRudpSocketMode, + CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions, CultNetWireContract, + decode_cultnet_message_from_slice, encode_cultnet_message_to_vec, +}; +use pretty_assertions::assert_eq; +use std::net::{SocketAddr, UdpSocket}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::Duration; + +#[derive(Default)] +struct SinkState { + receipts: Vec, + failures_remaining: usize, +} + +#[derive(Clone, Default)] +struct Sink(Arc>); + +impl Sink { + fn fail_once() -> Self { + Self(Arc::new(Mutex::new(SinkState { + failures_remaining: 1, + ..Default::default() + }))) + } +} + +impl CultMeshRudpRawDocumentSink for Sink { + fn accept_raw_document(&mut self, receipt: CultMeshRudpRawDocumentReceipt) -> Result<()> { + let mut state = self.0.lock().unwrap(); + if state.failures_remaining > 0 { + state.failures_remaining -= 1; + anyhow::bail!("injected sink failure"); + } + state.receipts.push(receipt); + Ok(()) + } +} + +#[derive(Default)] +struct SourceState { + documents: Vec, + failures_remaining: usize, + calls: usize, +} + +#[derive(Clone, Default)] +struct Source(Arc>); + +impl Source { + fn documents(documents: Vec) -> Self { + Self(Arc::new(Mutex::new(SourceState { + documents, + ..Default::default() + }))) + } + + fn fail_once(documents: Vec) -> Self { + Self(Arc::new(Mutex::new(SourceState { + documents, + failures_remaining: 1, + calls: 0, + }))) + } +} + +impl CultMeshRudpSnapshotSource for Source { + fn raw_snapshot( + &mut self, + _: &CultMeshRudpSnapshotQuery, + ) -> Result> { + let mut state = self.0.lock().unwrap(); + state.calls += 1; + if state.failures_remaining > 0 { + state.failures_remaining -= 1; + anyhow::bail!("injected source failure"); + } + Ok(state.documents.clone()) + } +} + +#[derive(Clone)] +struct Clock { + unix: Arc, + monotonic: Arc, +} + +impl Clock { + fn new(now: u64) -> Self { + Self { + unix: Arc::new(AtomicU64::new(now)), + monotonic: Arc::new(AtomicU64::new(now)), + } + } + + fn set(&self, now: u64) { + self.unix.store(now, Ordering::SeqCst); + self.monotonic.store(now, Ordering::SeqCst); + } + + fn set_unix(&self, now: u64) { + self.unix.store(now, Ordering::SeqCst); + } + + fn set_monotonic(&self, now: u64) { + self.monotonic.store(now, Ordering::SeqCst); + } +} + +impl CultMeshRudpServerClock for Clock { + fn now_unix_millis(&self) -> u64 { + self.unix.load(Ordering::SeqCst) + } + + fn now_monotonic_millis(&self) -> u64 { + self.monotonic.load(Ordering::SeqCst) + } +} + +type Server = CultMeshRudpDocumentServer; + +fn document(key: &str, payload: Vec) -> CultNetRawDocumentRecord { + CultNetRawDocumentRecord { + schema_id: "test.raw.v1".into(), + record_key: key.into(), + stored_at: "2026-09-03T00:00:00Z".into(), + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload, + source_runtime_id: Some(format!("runtime-{key}")), + source_agent_id: None, + source_role: None, + tags: None, + } +} + +fn server( + options: CultMeshRudpDocumentServerOptions, + clock: Clock, + sink: Sink, + source: Source, +) -> Result { + CultMeshRudpDocumentServer::new( + UdpSocket::bind("127.0.0.1:0")?, + sink, + source, + clock, + options, + ) +} + +fn client(target: SocketAddr, id: u32) -> Result { + let socket = UdpSocket::bind("127.0.0.1:0")?; + socket.set_nonblocking(true)?; + CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + runtime_id: "test-client".into(), + socket, + mode: CultNetRudpSocketMode::Client, + remote_addr: Some(target), + connection_id: id, + initial_sequence: 1, + resend_delay_ms: 10, + transport_id: None, + max_payload_bytes: None, + max_fragment_bytes: Some(1200), + max_pending_reliable_packets: Some(64), + media_reliable_expire_after_ms: None, + }) +} + +fn connect( + server: &mut Server, + clients: &mut [&mut CultNetRudpSocketTransportConnection], +) -> Result<()> { + for client in &mut *clients { + client.connect(Vec::new())?; + } + for _ in 0..500 { + server.poll_once()?; + for client in &mut *clients { + client.receive_once()?; + } + if clients.iter().all(|client| client.connected()) { + for _ in 0..clients.len() { + server.poll_once()?; + } + return Ok(()); + } + thread::sleep(Duration::from_millis(1)); + } + anyhow::bail!("client connection timed out") +} + +fn send(client: &mut CultNetRudpSocketTransportConnection, message: &CultNetMessage) -> Result<()> { + client.send( + "schema", + encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?, + ) +} + +fn send_reliable( + client: &mut CultNetRudpSocketTransportConnection, + message: &CultNetMessage, +) -> Result { + client.send_reliable( + "schema", + encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?, + ) +} + +#[test] +fn same_connection_id_is_peer_scoped_and_raw_bytes_are_untouched() -> Result<()> { + let clock = Clock::new(42_000); + let sink = Sink::default(); + let mut server = server(Default::default(), clock, sink.clone(), Source::default())?; + let mut a = client(server.local_addr()?, 77)?; + let mut b = client(server.local_addr()?, 77)?; + connect(&mut server, &mut [&mut a, &mut b])?; + + let a_bytes = vec![0xc4, 0x04, 0x00, 0xff, 0x80, 0x01]; + let b_bytes = vec![0x92, 0x01, 0xc4, 0x02, 0x00, 0xff]; + send( + &mut a, + &CultNetMessage::DocumentPutRaw { + message_id: "a".into(), + document: document("a", a_bytes.clone()), + }, + )?; + send( + &mut b, + &CultNetMessage::DocumentPutRaw { + message_id: "b".into(), + document: document("b", b_bytes.clone()), + }, + )?; + for _ in 0..500 { + server.poll_once()?; + a.receive_once()?; + b.receive_once()?; + if sink.0.lock().unwrap().receipts.len() == 2 { + break; + } + } + + let receipts = sink.0.lock().unwrap(); + assert_eq!(server.session_count(), 2); + assert_eq!(receipts.receipts.len(), 2); + assert_ne!( + receipts.receipts[0].session.remote_addr, + receipts.receipts[1].session.remote_addr + ); + assert_eq!( + receipts + .receipts + .iter() + .find(|r| r.message_id == "a") + .unwrap() + .document + .payload, + a_bytes + ); + assert_eq!( + receipts + .receipts + .iter() + .find(|r| r.message_id == "b") + .unwrap() + .document + .payload, + b_bytes + ); + assert!( + receipts + .receipts + .iter() + .all(|r| r.received_at_unix_millis == 42_000) + ); + Ok(()) +} + +#[test] +fn duplicate_connect_preserves_epoch_and_fresh_epoch_is_separate() -> Result<()> { + let clock = Clock::new(45_000); + let sink = Sink::default(); + let mut server = server(Default::default(), clock, sink.clone(), Source::default())?; + let target = server.local_addr()?; + let mut incumbent = client(target, 91)?; + connect(&mut server, &mut [&mut incumbent])?; + + incumbent.connect(Vec::new())?; + server.poll_once()?; + incumbent.receive_once()?; + assert_eq!(server.session_count(), 1); + + send( + &mut incumbent, + &CultNetMessage::DocumentPutRaw { + message_id: "after-duplicate".into(), + document: document("incumbent", vec![1, 2, 3]), + }, + )?; + for _ in 0..20 { + server.poll_once()?; + if sink.0.lock().unwrap().receipts.len() == 1 { + break; + } + } + assert_eq!(sink.0.lock().unwrap().receipts.len(), 1); + + let mut fresh_epoch = client(target, 92)?; + connect(&mut server, &mut [&mut fresh_epoch])?; + assert_eq!(server.session_count(), 2); + Ok(()) +} + +#[test] +fn application_rejection_is_nonfatal_peer_scoped_and_unacknowledged() -> Result<()> { + let clock = Clock::new(48_000); + let sink = Sink::fail_once(); + let source = Source::fail_once(vec![document("snapshot", vec![4, 5, 6])]); + let mut server = server( + Default::default(), + clock.clone(), + sink.clone(), + source.clone(), + )?; + let target = server.local_addr()?; + + let mut publisher = client(target, 101)?; + let mut snapshot_client = client(target, 102)?; + let mut survivor = client(target, 103)?; + connect( + &mut server, + &mut [&mut publisher, &mut snapshot_client, &mut survivor], + )?; + let publish_receipt = send_reliable( + &mut publisher, + &CultNetMessage::DocumentPutRaw { + message_id: "rejected-put".into(), + document: document("rejected", vec![0, 255, 1]), + }, + )?; + let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? else { + panic!("sink rejection must be returned as a nonfatal poll outcome"); + }; + assert_eq!(rejection.session.connection_id, 101); + assert_eq!( + rejection.operation, + CultMeshRudpApplicationOperation::DocumentPutRaw + ); + assert_eq!(rejection.message_id, "rejected-put"); + assert!(rejection.reason.contains("injected sink failure")); + assert_eq!(server.session_count(), 2); + publisher.receive_once()?; + assert_eq!( + publisher.reliable_send_status(&publish_receipt), + CultNetRudpReliableSendStatus::Pending + ); + + let snapshot_receipt = send_reliable( + &mut snapshot_client, + &CultNetMessage::SnapshotRequest { + message_id: "rejected-snapshot".into(), + schema_ids: None, + record_keys: None, + }, + )?; + let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? else { + panic!("source rejection must be returned as a nonfatal poll outcome"); + }; + assert_eq!(rejection.session.connection_id, 102); + assert_eq!( + rejection.operation, + CultMeshRudpApplicationOperation::SnapshotRequest + ); + assert_eq!(rejection.message_id, "rejected-snapshot"); + assert!(rejection.reason.contains("injected source failure")); + assert_eq!(server.session_count(), 1); + snapshot_client.receive_once()?; + assert_eq!( + snapshot_client.reliable_send_status(&snapshot_receipt), + CultNetRudpReliableSendStatus::Pending + ); + + let survivor_receipt = send_reliable( + &mut survivor, + &CultNetMessage::DocumentPutRaw { + message_id: "accepted-after-rejections".into(), + document: document("survivor", vec![9, 8, 7]), + }, + )?; + assert_eq!(server.poll_once()?, CultMeshRudpPollOutcome::Handled); + for _ in 0..20 { + survivor.receive_once()?; + if survivor.reliable_send_status(&survivor_receipt) + == CultNetRudpReliableSendStatus::Acknowledged + { + break; + } + } + assert_eq!( + survivor.reliable_send_status(&survivor_receipt), + CultNetRudpReliableSendStatus::Acknowledged + ); + assert_eq!(sink.0.lock().unwrap().receipts.len(), 1); + assert_eq!(source.0.lock().unwrap().calls, 1); + Ok(()) +} + +#[test] +fn snapshot_response_resends_until_acknowledged() -> Result<()> { + let clock = Clock::new(51_000); + let expected = document("catalog", vec![0x81, 0xa1, 0x78, 0x2a]); + let source = Source::documents(vec![expected.clone()]); + let options = CultMeshRudpDocumentServerOptions { + resend_delay: Duration::from_millis(10), + ..Default::default() + }; + let mut server = server(options, clock.clone(), Sink::default(), source)?; + let mut client = client(server.local_addr()?, 88)?; + connect(&mut server, &mut [&mut client])?; + send( + &mut client, + &CultNetMessage::SnapshotRequest { + message_id: "snapshot".into(), + schema_ids: Some(vec!["test.raw.v1".into()]), + record_keys: Some(vec!["catalog".into()]), + }, + )?; + server.poll_once()?; + + clock.set(51_011); + assert_eq!(server.maintain()?.packets_resent, 1); + let response = loop { + if let Some(frame) = client.receive_once()? { + break decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + )?; + } + }; + assert_eq!( + response, + CultNetMessage::SnapshotResponseRaw { + message_id: "snapshot".into(), + documents: vec![expected], + } + ); + + for _ in 0..20 { + server.poll_once()?; + } + clock.set(51_022); + assert_eq!(server.maintain()?.packets_resent, 0); + Ok(()) +} + +#[test] +fn payload_and_snapshot_output_budgets_fail_closed() -> Result<()> { + let clock = Clock::new(55_000); + let sink = Sink::default(); + let message = CultNetMessage::DocumentPutRaw { + message_id: "budget".into(), + document: document("budget", vec![7; 64]), + }; + let encoded = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?; + let options = CultMeshRudpDocumentServerOptions { + max_admitted_payload_bytes: encoded.len(), + max_admitted_payload_bytes_per_session: encoded.len(), + max_snapshot_response_bytes: encoded.len(), + ..Default::default() + }; + let mut budget_server = server(options, clock, sink.clone(), Source::default())?; + let target = budget_server.local_addr()?; + let mut first = client(target, 111)?; + let mut second = client(target, 112)?; + connect(&mut budget_server, &mut [&mut first, &mut second])?; + send(&mut first, &message)?; + budget_server.poll_once()?; + send(&mut second, &message)?; + budget_server.poll_once()?; + assert_eq!(sink.0.lock().unwrap().receipts.len(), 1); + + let oversized = Source::documents(vec![document("oversized", vec![8; 512])]); + let options = CultMeshRudpDocumentServerOptions { + max_snapshot_response_bytes: 128, + ..Default::default() + }; + let mut snapshot_server = server(options, Clock::new(56_000), Sink::default(), oversized)?; + let mut snapshot_client = client(snapshot_server.local_addr()?, 113)?; + connect(&mut snapshot_server, &mut [&mut snapshot_client])?; + let receipt = send_reliable( + &mut snapshot_client, + &CultNetMessage::SnapshotRequest { + message_id: "too-large".into(), + schema_ids: None, + record_keys: None, + }, + )?; + let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = snapshot_server.poll_once()? + else { + panic!("oversized snapshot must reject only its session"); + }; + assert_eq!( + rejection.operation, + CultMeshRudpApplicationOperation::SnapshotRequest + ); + assert_eq!(rejection.message_id, "too-large"); + assert!(rejection.reason.contains("snapshot response is")); + assert_eq!(snapshot_server.session_count(), 0); + snapshot_client.receive_once()?; + assert_eq!( + snapshot_client.reliable_send_status(&receipt), + CultNetRudpReliableSendStatus::Pending + ); + Ok(()) +} + +#[test] +fn session_cap_and_expiry_use_monotonic_time() -> Result<()> { + let clock = Clock::new(60_000); + let options = CultMeshRudpDocumentServerOptions { + max_sessions: 1, + session_idle_timeout: Duration::from_secs(1), + session_max_lifetime: Duration::from_millis(100), + ..Default::default() + }; + let mut server = server(options, clock.clone(), Sink::default(), Source::default())?; + let target = server.local_addr()?; + let mut admitted = client(target, 1)?; + connect(&mut server, &mut [&mut admitted])?; + let mut rejected = client(target, 1)?; + rejected.connect(Vec::new())?; + server.poll_once()?; + assert_eq!(server.session_count(), 1); + assert!(!rejected.connected()); + + clock.set_unix(1); + clock.set_monotonic(60_101); + assert_eq!(server.maintain()?.sessions_expired, 1); + assert_eq!(server.session_count(), 0); + Ok(()) +} diff --git a/vendor/cultnet-rs b/vendor/cultnet-rs deleted file mode 160000 index 9dcb88d..0000000 --- a/vendor/cultnet-rs +++ /dev/null @@ -1 +0,0 @@ -Subproject commit 9dcb88df1361b157a48502f43093fe7a25ca22b7