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

Reconcile cultmesh-rs: adopt Odin's lineage, keep CultLib's additions

Browse source 
CultLib's packages/cultmesh-rs and Odin's vendor/cultmesh-rs had diverged
independently -- 87 shared symbols, 6 only here, 9 only there. Neither was
simply ahead of the other, so this is a merge rather than a fast-forward.

Odin's copy is the base because it is the deployed code, running in production
via Idunn, and because it carries a deliberate semantic upgrade: connection_id
was replaced by a per-request message_id correlation with validation requiring
raw callers to supply one. That brings in the RUDP document server (647 lines)
and its tests (553), which the original migration into CultLib left behind.

Layered back in from CultLib's lineage: soa(), get_all_with_keys(),
create_rudp_document_message(), and the plural batch publisher. The batch
function was rewritten off the removed connection_id onto the connect helper.

Both publish paths survive deliberately. They are different jobs, not rival
implementations: the singular path polls one receipt for application-level
acknowledgement and can report Invalidated, while the batch path sends and
then flushes at the transport level. That is also why the census framed the
30s-versus-300ms flush_timeout as a value conflict and was wrong -- one field
name, two mechanisms. The distinction is now documented at the call site.

Surviving connection_id references are the RUDP protocol's own connection
epoch and are unrelated to the removed options field.

cargo test: 16 passed, 0 failed -- 10 lib, 6 document-server integration.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-05 17:26:44 +02:00
commit ae0ac0d3c9
3 changed files with 1584 additions and 149 deletions
Download patch file Download diff file Expand all files Collapse all files

568
packages/cultmesh-rs/src/lib.rs
View file

@ -3,8 +3,8 @@ use cultcache_rs::CultCache;
use cultcache_rs::CultCacheEnvelope;
use cultcache_rs::CultSoaTable;
use cultcache_rs::DatabaseEntry;
use cultcache_rs::SoaDocument;
use cultcache_rs::SingleFileMessagePackBackingStore;
use cultcache_rs::SoaDocument;
use cultnet_rs::CultNetDocumentPutOptions;
use cultnet_rs::CultNetDocumentRegistry;
use cultnet_rs::CultNetRudpSocketTransportConnection;
@ -14,7 +14,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;
@ -22,6 +24,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 {
@ -74,7 +80,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,
@ -88,7 +93,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,
@ -112,7 +120,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),
@ -138,9 +146,8 @@ 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_secs(30),
flush_timeout: Duration::from_millis(300),
poll_interval: Duration::from_millis(10),
resend_delay_ms: 50,
source_agent_id: None,
@ -673,7 +680,7 @@ impl CultMeshNode {
T: DatabaseEntry + Serialize,
{
let message = self.create_rudp_document_message(key, value, &options)?;
publish_cultnet_messages_to_rudp_catalog(std::slice::from_ref(&message), options)
publish_cultnet_message_to_rudp_catalog(&message, options)
}
pub fn create_rudp_document_message<T>(
@ -706,8 +713,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");
@ -757,58 +772,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 {
media_reliable_expire_after_ms: None,
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,
reconnect_policy: None,
})?;
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
);
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",
)?;
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 {
@ -824,7 +819,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());
@ -834,14 +828,28 @@ 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());
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
@ -855,6 +863,58 @@ fn unix_millis() -> u128 {
.as_millis()
}
/// 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 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
)
}
/// Publish a batch of raw CultNet schema messages to a RUDP catalog.
///
/// Unlike [`publish_cultnet_message_to_rudp_catalog`], which polls for
/// per-message application-level acknowledgement, this sends every message on
/// one connection and waits only for a transport-level `flush_reliable`. It
/// trades per-receipt confirmation for batch throughput; callers that need a
/// stronger delivery guarantee should use the singular publish path instead.
pub fn publish_cultnet_messages_to_rudp_catalog(
messages: &[cultnet_rs::CultNetMessage],
options: CultMeshRudpDocumentPublishOptions,
@ -862,43 +922,14 @@ pub fn publish_cultnet_messages_to_rudp_catalog(
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 {
media_reliable_expire_after_ms: None,
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-document-publisher".to_string()),
max_payload_bytes: None,
max_fragment_bytes: Some(1200),
max_pending_reliable_packets: None,
reconnect_policy: None,
})?;
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 document publisher {} to {}",
options.runtime_id,
options.target
);
}
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",
)?;
for message in messages {
let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?;
@ -907,6 +938,57 @@ pub fn publish_cultnet_messages_to_rudp_catalog(
client.flush_reliable(options.flush_timeout)
}
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(poll_interval))?;
let mut client =
CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions {
runtime_id: runtime_id.to_string(),
socket,
mode: cultnet_rs::CultNetRudpSocketMode::Client,
remote_addr: Some(target),
connection_id: fresh_rudp_connection_epoch(),
initial_sequence: 1,
resend_delay_ms,
transport_id: Some(transport_id.to_string()),
max_payload_bytes: None,
max_fragment_bytes: Some(1200),
max_pending_reliable_packets: Some(1024),
media_reliable_expire_after_ms: None,
reconnect_policy: None,
})?;
client.connect(Vec::new())?;
let connect_deadline = Instant::now() + connect_timeout;
while !client.connected() && Instant::now() < connect_deadline {
let _ = client.receive_once()?;
client.poll_resends()?;
thread::sleep(poll_interval);
}
if !client.connected() {
anyhow::bail!("timed out connecting CultMesh RUDP client {runtime_id} to {target}");
}
Ok(client)
}
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)]
pub struct CultMesh;
@ -1122,33 +1204,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")
});
@ -1168,79 +1250,231 @@ mod tests {
}
#[test]
fn large_rudp_publication_waits_for_acknowledged_delivery() -> Result<()> {
fn document_publish_fails_when_catalog_is_unreachable() -> Result<()> {
let temp = tempfile::tempdir()?;
let node = CultMesh::create_node(
temp.path().join("cultmesh.cc"),
TestDocuments,
CultMeshNodeOptions {
runtime_id: "large-publisher-test".into(),
..CultMeshNodeOptions::default()
},
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")?;
socket.set_read_timeout(Some(Duration::from_millis(10)))?;
let target = socket.local_addr()?;
let options = CultMeshRudpDocumentPublishOptions::odin(target, "large-publisher-test");
let messages = [
node.create_rudp_document_message(
"large-a",
&Note {
body: "a".repeat(256 * 1024),
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")
},
&options,
)?,
node.create_rudp_document_message(
"large-b",
&Note {
body: "b".repeat(256 * 1024),
},
&options,
)?,
];
let server = thread::spawn(move || -> Result<Vec<String>> {
thread::sleep(Duration::from_millis(500));
let mut server = CultNetRudpSocketTransportConnection::new(
CultNetRudpSocketTransportOptions::server(
"delayed-odin-test",
socket,
CULTMESH_RUDP_DOCUMENT_CATALOG_CONNECTION_ID,
),
|_query: &CultMeshRudpSnapshotQuery| Ok(Vec::new()),
CultMeshSystemClock::default(),
CultMeshRudpDocumentServerOptions::default(),
)?;
let deadline = Instant::now() + Duration::from_secs(5);
let mut record_keys = Vec::new();
while Instant::now() < deadline && record_keys.len() < 2 {
if let Some(frame) = server.receive_once()? {
let message = decode_cultnet_message_from_slice(
&frame.payload,
CultNetWireContract::CultNetSchemaV0,
)?;
if let cultnet_rs::CultNetMessage::DocumentPutRaw { document, .. } = message {
record_keys.push(document.record_key);
}
}
server.poll_resends()?;
while !server_stop.load(std::sync::atomic::Ordering::SeqCst) {
let _ = server.poll_once();
thread::sleep(Duration::from_millis(1));
}
let linger_deadline = Instant::now() + Duration::from_millis(250);
while Instant::now() < linger_deadline {
let _ = server.receive_once()?;
server.poll_resends()?;
}
if record_keys.len() != 2 {
anyhow::bail!("timed out waiting for both large RUDP documents");
}
Ok(record_keys)
Ok(())
});
publish_cultnet_messages_to_rudp_catalog(&messages, options)?;
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!(
server.join().expect("server thread should not panic")?,
vec!["large-a", "large-b"]
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();

647
packages/cultmesh-rs/src/rudp_document_server.rs Normal file
View 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)
}

554
packages/cultmesh-rs/tests/rudp_document_server.rs Normal file
View file

@ -0,0 +1,554 @@
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,
reconnect_policy: 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(())
}