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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
515a864ef0
commit
27cc8156bc
22 changed files with 1817 additions and 2184 deletions
3
.gitmodules
vendored
3
.gitmodules
vendored
|
|
@ -1,3 +0,0 @@
|
|||
[submodule "vendor/cultnet-rs"]
|
||||
path = vendor/cultnet-rs
|
||||
url = https://github.com/GameCult/cultnet-rs.git
|
||||
30
Cargo.lock
generated
30
Cargo.lock
generated
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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 }
|
||||
|
|
|
|||
|
|
@ -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<Option<ActiveHidCon
|
|||
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,
|
||||
})?;
|
||||
Ok(Some(ActiveHidControllerStream {
|
||||
target,
|
||||
|
|
@ -8948,7 +8950,6 @@ fn publish_obs_media_receiver_to_odin(options: &Options) -> 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();
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
2
vendor/cultcache-rs/.gitignore
vendored
2
vendor/cultcache-rs/.gitignore
vendored
|
|
@ -1,2 +0,0 @@
|
|||
/target
|
||||
Cargo.lock
|
||||
30
vendor/cultcache-rs/Cargo.toml
vendored
30
vendor/cultcache-rs/Cargo.toml
vendored
|
|
@ -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"
|
||||
332
vendor/cultcache-rs/README.md
vendored
332
vendor/cultcache-rs/README.md
vendored
|
|
@ -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::<PlayerData>("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::<Settings>()?;
|
||||
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::<Settings>("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<Settings> = 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<T>(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::<T>()` 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::<PlayerData>("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::<T>`
|
||||
- `add_generic_backing_store`
|
||||
- `add_backing_store`
|
||||
- `pull_all_backing_stores`
|
||||
- `get::<T>`
|
||||
- `get_required::<T>`
|
||||
- `get_envelope::<T>`
|
||||
- `get_required_envelope::<T>`
|
||||
- `get_all::<T>`
|
||||
- `put::<T>`
|
||||
- `put_envelope::<T>`
|
||||
- `update::<T>`
|
||||
- `delete::<T>`
|
||||
- `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::<T>()` exports the canonical persisted bytes for a typed entry
|
||||
- `put_envelope::<T>()` 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<u8>` 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::<T>("Potion")`
|
||||
- `cache.get_by_index::<T>("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.
|
||||
|
|
@ -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"] }
|
||||
278
vendor/cultcache-rs/cultcache-rs-derive/src/lib.rs
vendored
278
vendor/cultcache-rs/cultcache-rs-derive/src/lib.rs
vendored
|
|
@ -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<LitStr> = None;
|
||||
let mut schema_name: Option<LitStr> = 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<CultCacheSlot> = 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<usize> = 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<S>(&self, serializer: S) -> ::std::result::Result<S::Ok, S::Error>
|
||||
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<D>(deserializer: D) -> ::std::result::Result<Self, D::Error>
|
||||
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<A>(self, mut seq: A) -> ::std::result::Result<Self::Value, A::Error>
|
||||
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,
|
||||
}
|
||||
121
vendor/cultcache-rs/examples/cultcache_interop.rs
vendored
121
vendor/cultcache-rs/examples/cultcache_interop.rs
vendored
|
|
@ -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<String>,
|
||||
}
|
||||
|
||||
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::<CultCacheInteropNote>()?;
|
||||
let note = notes
|
||||
.first()
|
||||
.ok_or_else(|| anyhow!("no {INTEROP_TYPE} records found"))?;
|
||||
print_note(note)
|
||||
}
|
||||
|
||||
fn build_cache(file: &str) -> Result<CultCache> {
|
||||
let mut cache = CultCache::new();
|
||||
cache.register_entry_type::<CultCacheInteropNote>()?;
|
||||
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<String>) -> BTreeMap<String, String> {
|
||||
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<String, String>, name: &str) -> Result<&'a str> {
|
||||
options
|
||||
.get(name)
|
||||
.map(String::as_str)
|
||||
.ok_or_else(|| anyhow!("missing required argument --{name}"))
|
||||
}
|
||||
1241
vendor/cultcache-rs/src/lib.rs
vendored
1241
vendor/cultcache-rs/src/lib.rs
vendored
File diff suppressed because it is too large
Load diff
149
vendor/cultmesh-rs/Cargo.lock
generated
vendored
149
vendor/cultmesh-rs/Cargo.lock
generated
vendored
|
|
@ -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"
|
||||
|
|
|
|||
4
vendor/cultmesh-rs/Cargo.toml
vendored
4
vendor/cultmesh-rs/Cargo.toml
vendored
|
|
@ -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]
|
||||
|
|
|
|||
572
vendor/cultmesh-rs/src/lib.rs
vendored
572
vendor/cultmesh-rs/src/lib.rs
vendored
|
|
@ -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<String>) -> Result<Self> {
|
||||
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<String>, slot_count: usize, slot_byte_length: usize) -> Result<Self> {
|
||||
pub fn new(
|
||||
stream_id: impl Into<String>,
|
||||
slot_count: usize,
|
||||
slot_byte_length: usize,
|
||||
) -> Result<Self> {
|
||||
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<CultMeshStreamKind>) -> Vec<&CultMeshStreamDescriptor> {
|
||||
pub fn find(
|
||||
&self,
|
||||
verse_id: &str,
|
||||
kind: Option<CultMeshStreamKind>,
|
||||
) -> 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<usize> {
|
||||
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<cultnet_rs::CultNetMessage> {
|
||||
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<CultNetRudpSocketTransportConnection> {
|
||||
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<SocketAddr> {
|
|||
}
|
||||
|
||||
fn parse_rudp_socket_addr(value: &str) -> Result<SocketAddr> {
|
||||
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<Option<String>>
|
|||
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::<String>()
|
||||
.trim_matches('_')
|
||||
.to_string();
|
||||
|
|
@ -1006,11 +1078,7 @@ fn cultmesh_uri_authority(uri: &str) -> Result<String> {
|
|||
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>("note")?.body,
|
||||
"catalog truth"
|
||||
);
|
||||
assert_eq!(reloaded.get_required::<Note>("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")
|
||||
};
|
||||
|
|
|
|||
647
vendor/cultmesh-rs/src/rudp_document_server.rs
vendored
Normal file
647
vendor/cultmesh-rs/src/rudp_document_server.rs
vendored
Normal file
|
|
@ -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<Vec<String>>,
|
||||
pub record_keys: Option<Vec<String>>,
|
||||
}
|
||||
|
||||
/// Caller-owned admission/persistence port for received raw documents.
|
||||
pub trait CultMeshRudpRawDocumentSink {
|
||||
fn accept_raw_document(&mut self, receipt: CultMeshRudpRawDocumentReceipt) -> Result<()>;
|
||||
}
|
||||
|
||||
impl<F> 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<Vec<CultNetRawDocumentRecord>>;
|
||||
}
|
||||
|
||||
impl<F> CultMeshRudpSnapshotSource for F
|
||||
where
|
||||
F: FnMut(&CultMeshRudpSnapshotQuery) -> Result<Vec<CultNetRawDocumentRecord>>,
|
||||
{
|
||||
fn raw_snapshot(
|
||||
&mut self,
|
||||
query: &CultMeshRudpSnapshotQuery,
|
||||
) -> Result<Vec<CultNetRawDocumentRecord>> {
|
||||
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<S, Q, C> {
|
||||
socket: UdpSocket,
|
||||
sessions: BTreeMap<CultMeshRudpSessionKey, SessionEntry>,
|
||||
sink: S,
|
||||
snapshot_source: Q,
|
||||
clock: C,
|
||||
options: CultMeshRudpDocumentServerOptions,
|
||||
}
|
||||
|
||||
impl<S, Q, C> CultMeshRudpDocumentServer<S, Q, C>
|
||||
where
|
||||
S: CultMeshRudpRawDocumentSink,
|
||||
Q: CultMeshRudpSnapshotSource,
|
||||
C: CultMeshRudpServerClock,
|
||||
{
|
||||
pub fn new(
|
||||
socket: UdpSocket,
|
||||
sink: S,
|
||||
snapshot_source: Q,
|
||||
clock: C,
|
||||
options: CultMeshRudpDocumentServerOptions,
|
||||
) -> Result<Self> {
|
||||
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<SocketAddr> {
|
||||
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<CultMeshRudpPollOutcome> {
|
||||
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<CultMeshRudpMaintenance> {
|
||||
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<Option<CultMeshRudpApplicationRejection>> {
|
||||
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)
|
||||
}
|
||||
553
vendor/cultmesh-rs/tests/rudp_document_server.rs
vendored
Normal file
553
vendor/cultmesh-rs/tests/rudp_document_server.rs
vendored
Normal file
|
|
@ -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<CultMeshRudpRawDocumentReceipt>,
|
||||
failures_remaining: usize,
|
||||
}
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
struct Sink(Arc<Mutex<SinkState>>);
|
||||
|
||||
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<CultNetRawDocumentRecord>,
|
||||
failures_remaining: usize,
|
||||
calls: usize,
|
||||
}
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
struct Source(Arc<Mutex<SourceState>>);
|
||||
|
||||
impl Source {
|
||||
fn documents(documents: Vec<CultNetRawDocumentRecord>) -> Self {
|
||||
Self(Arc::new(Mutex::new(SourceState {
|
||||
documents,
|
||||
..Default::default()
|
||||
})))
|
||||
}
|
||||
|
||||
fn fail_once(documents: Vec<CultNetRawDocumentRecord>) -> Self {
|
||||
Self(Arc::new(Mutex::new(SourceState {
|
||||
documents,
|
||||
failures_remaining: 1,
|
||||
calls: 0,
|
||||
})))
|
||||
}
|
||||
}
|
||||
|
||||
impl CultMeshRudpSnapshotSource for Source {
|
||||
fn raw_snapshot(
|
||||
&mut self,
|
||||
_: &CultMeshRudpSnapshotQuery,
|
||||
) -> Result<Vec<CultNetRawDocumentRecord>> {
|
||||
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<AtomicU64>,
|
||||
monotonic: Arc<AtomicU64>,
|
||||
}
|
||||
|
||||
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<Sink, Source, Clock>;
|
||||
|
||||
fn document(key: &str, payload: Vec<u8>) -> 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<Server> {
|
||||
CultMeshRudpDocumentServer::new(
|
||||
UdpSocket::bind("127.0.0.1:0")?,
|
||||
sink,
|
||||
source,
|
||||
clock,
|
||||
options,
|
||||
)
|
||||
}
|
||||
|
||||
fn client(target: SocketAddr, id: u32) -> Result<CultNetRudpSocketTransportConnection> {
|
||||
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<CultNetRudpReliableSendReceipt> {
|
||||
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(())
|
||||
}
|
||||
1
vendor/cultnet-rs
vendored
1
vendor/cultnet-rs
vendored
|
|
@ -1 +0,0 @@
|
|||
Subproject commit 9dcb88df1361b157a48502f43093fe7a25ca22b7
|
||||
Loading…
Reference in a new issue