diff --git a/packages/cultmesh-rs/src/lib.rs b/packages/cultmesh-rs/src/lib.rs index e460ac78..7999f18b 100644 --- a/packages/cultmesh-rs/src/lib.rs +++ b/packages/cultmesh-rs/src/lib.rs @@ -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( @@ -706,8 +713,16 @@ impl CultMeshNode { pub fn pull_rudp_catalog_snapshot( &mut self, - options: CultMeshRudpSnapshotOptions, + mut options: CultMeshRudpSnapshotOptions, ) -> Result { + if options.message_id.trim().is_empty() { + options.message_id = format!( + "{}:snapshot:{}:{}", + options.runtime_id, + unix_millis(), + fresh_rudp_connection_epoch() + ); + } let response = request_raw_snapshot_from_rudp_catalog(options)?; let cultnet_rs::CultNetMessage::SnapshotResponseRaw { documents, .. } = response else { anyhow::bail!("expected cultnet.snapshot_response_raw.v0"); @@ -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 { 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 { + 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> { - 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(); diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs new file mode 100644 index 00000000..bebeb6c3 --- /dev/null +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -0,0 +1,647 @@ +use anyhow::{Result, anyhow}; +use cultnet_rs::{ + CultNetMessage, CultNetRawDocumentRecord, CultNetRudpPacket, CultNetRudpPacketType, + CultNetRudpSendOptions, CultNetRudpSession, CultNetRudpSessionOptions, CultNetWireContract, + decode_cultnet_message_from_slice, decode_rudp_packet, encode_cultnet_message_to_vec, + encode_rudp_packet, +}; +use std::collections::BTreeMap; +use std::io::ErrorKind; +use std::net::{SocketAddr, UdpSocket}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +const MAX_UDP_DATAGRAM_BYTES: usize = 65_535; + +/// The transport identity of one remote CultNet RUDP session. +/// +/// Connection identifiers are scoped to a remote socket address. Two clients +/// may therefore use the same connection identifier without sharing ordering, +/// acknowledgement, resend, or fragment state. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub struct CultMeshRudpSessionKey { + pub remote_addr: SocketAddr, + pub connection_id: u32, +} + +/// A raw document plus the receipt fact owned by the receiving process. +/// +/// `document.payload` is the exact byte vector decoded from the CultNet raw +/// record. The trusted receipt time lives beside the sender-owned record so the +/// transport never rewrites its metadata or payload. +#[derive(Clone, Debug, PartialEq)] +pub struct CultMeshRudpRawDocumentReceipt { + pub session: CultMeshRudpSessionKey, + pub message_id: String, + pub transport_sequence: u32, + pub received_at_unix_millis: u64, + pub document: CultNetRawDocumentRecord, +} + +/// The caller-visible query for a read-only raw snapshot. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CultMeshRudpSnapshotQuery { + pub session: CultMeshRudpSessionKey, + pub message_id: String, + pub requested_at_unix_millis: u64, + pub schema_ids: Option>, + pub record_keys: Option>, +} + +/// Caller-owned admission/persistence port for received raw documents. +pub trait CultMeshRudpRawDocumentSink { + fn accept_raw_document(&mut self, receipt: CultMeshRudpRawDocumentReceipt) -> Result<()>; +} + +impl CultMeshRudpRawDocumentSink for F +where + F: FnMut(CultMeshRudpRawDocumentReceipt) -> Result<()>, +{ + fn accept_raw_document(&mut self, receipt: CultMeshRudpRawDocumentReceipt) -> Result<()> { + self(receipt) + } +} + +/// Caller-owned catalog port for serving raw snapshot requests. +/// +/// The caller decides which records the requester may see. CultMesh preserves +/// those records as raw CultNet documents and does not interpret their schemas. +pub trait CultMeshRudpSnapshotSource { + fn raw_snapshot( + &mut self, + query: &CultMeshRudpSnapshotQuery, + ) -> Result>; +} + +impl CultMeshRudpSnapshotSource for F +where + F: FnMut(&CultMeshRudpSnapshotQuery) -> Result>, +{ + fn raw_snapshot( + &mut self, + query: &CultMeshRudpSnapshotQuery, + ) -> Result> { + self(query) + } +} + +/// Trusted local clocks for receipt facts and transport maintenance. +/// +/// Receipt time is wall time. Expiry and resend decisions use the monotonic +/// value, so a wall-clock correction cannot resurrect or pin a session. +pub trait CultMeshRudpServerClock { + fn now_unix_millis(&self) -> u64; + fn now_monotonic_millis(&self) -> u64; +} + +#[derive(Clone, Debug)] +pub struct CultMeshSystemClock { + monotonic_origin: Instant, +} + +impl Default for CultMeshSystemClock { + fn default() -> Self { + Self { + monotonic_origin: Instant::now(), + } + } +} + +impl CultMeshRudpServerClock for CultMeshSystemClock { + fn now_unix_millis(&self) -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .try_into() + .unwrap_or(u64::MAX) + } + + fn now_monotonic_millis(&self) -> u64 { + self.monotonic_origin + .elapsed() + .as_millis() + .try_into() + .unwrap_or(u64::MAX) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CultMeshRudpDocumentServerOptions { + pub max_sessions: usize, + pub session_idle_timeout: Duration, + pub session_max_lifetime: Duration, + /// Conservative payload admission budget across all live sessions. + /// + /// Admitted ingress and queued snapshot-response bytes are charged until + /// the session ends even when CultNet releases them sooner. Together with + /// CultNet's per-session receive caps, this makes retained payload memory + /// finite without duplicating its fragment/ordering machinery here. + pub max_admitted_payload_bytes: usize, + pub max_admitted_payload_bytes_per_session: usize, + pub max_snapshot_response_bytes: usize, + pub max_snapshot_documents: usize, + pub resend_delay: Duration, + pub max_pending_reliable_packets_per_session: usize, + pub max_fragment_bytes: usize, +} + +impl Default for CultMeshRudpDocumentServerOptions { + fn default() -> Self { + Self { + max_sessions: 64, + session_idle_timeout: Duration::from_secs(30), + session_max_lifetime: Duration::from_secs(15 * 60), + max_admitted_payload_bytes: 32 * 1024 * 1024, + max_admitted_payload_bytes_per_session: 4 * 1024 * 1024, + max_snapshot_response_bytes: 1024 * 1024, + max_snapshot_documents: 4096, + resend_delay: Duration::from_millis(50), + max_pending_reliable_packets_per_session: 1024, + max_fragment_bytes: 1200, + } + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct CultMeshRudpMaintenance { + pub sessions_expired: usize, + pub packets_resent: usize, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum CultMeshRudpApplicationOperation { + DocumentPutRaw, + SnapshotRequest, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CultMeshRudpApplicationRejection { + pub session: CultMeshRudpSessionKey, + pub operation: CultMeshRudpApplicationOperation, + pub message_id: String, + pub reason: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum CultMeshRudpPollOutcome { + Idle, + Handled, + ApplicationRejected(CultMeshRudpApplicationRejection), +} + +struct SessionEntry { + session: CultNetRudpSession, + created_at_monotonic_millis: u64, + last_activity_monotonic_millis: u64, + admitted_payload_bytes: usize, +} + +/// A synchronous, multi-session CultNet RUDP raw-document server. +/// +/// The server owns exactly one UDP socket. Call `poll_once` from an existing +/// daemon loop; no executor or background thread is created here. +pub struct CultMeshRudpDocumentServer { + socket: UdpSocket, + sessions: BTreeMap, + sink: S, + snapshot_source: Q, + clock: C, + options: CultMeshRudpDocumentServerOptions, +} + +impl CultMeshRudpDocumentServer +where + S: CultMeshRudpRawDocumentSink, + Q: CultMeshRudpSnapshotSource, + C: CultMeshRudpServerClock, +{ + pub fn new( + socket: UdpSocket, + sink: S, + snapshot_source: Q, + clock: C, + options: CultMeshRudpDocumentServerOptions, + ) -> Result { + validate_options(&options)?; + socket.local_addr().map_err(|error| { + anyhow!("CultMesh RUDP server requires a bound UDP socket: {error}") + })?; + socket.set_nonblocking(true)?; + Ok(Self { + socket, + sessions: BTreeMap::new(), + sink, + snapshot_source, + clock, + options, + }) + } + + pub fn local_addr(&self) -> Result { + Ok(self.socket.local_addr()?) + } + + pub fn session_count(&self) -> usize { + self.sessions.len() + } + + /// Poll transport maintenance and receive at most one UDP datagram. + /// + /// Application rejection invalidates only the responsible peer session and + /// is returned as data so a daemon can log it and keep serving. Local + /// socket and server-state failures remain errors. + pub fn poll_once(&mut self) -> Result { + self.maintain()?; + let mut wire = vec![0_u8; MAX_UDP_DATAGRAM_BYTES]; + let (received, remote_addr) = match self.socket.recv_from(&mut wire) { + Ok(value) => value, + Err(error) if error.kind() == ErrorKind::WouldBlock => { + return Ok(CultMeshRudpPollOutcome::Idle); + } + Err(error) => return Err(error.into()), + }; + wire.truncate(received); + + let packet = match decode_rudp_packet(&wire) { + Ok(packet) => packet, + Err(_) => return Ok(CultMeshRudpPollOutcome::Handled), + }; + let now_unix = self.clock.now_unix_millis(); + let now_monotonic = self.clock.now_monotonic_millis(); + let key = CultMeshRudpSessionKey { + remote_addr, + connection_id: packet.connection_id, + }; + + if packet.packet_type == CultNetRudpPacketType::Connect { + self.accept_connection(key, &packet, now_monotonic)?; + return Ok(CultMeshRudpPollOutcome::Handled); + } + + if !self.sessions.contains_key(&key) { + return Ok(CultMeshRudpPollOutcome::Handled); + } + if packet.packet_type == CultNetRudpPacketType::Data + && packet.channel_id == "schema" + && !packet.reliable + { + return Ok(CultMeshRudpPollOutcome::Handled); + } + let payload_charge = if packet.packet_type == CultNetRudpPacketType::Data { + packet.payload.len() + } else { + 0 + }; + if !self.payload_budget_allows(key, payload_charge) { + return Ok(CultMeshRudpPollOutcome::Handled); + } + + let result = { + let entry = self + .sessions + .get_mut(&key) + .expect("checked session must remain present"); + entry.last_activity_monotonic_millis = now_monotonic; + entry.admitted_payload_bytes = + entry.admitted_payload_bytes.saturating_add(payload_charge); + entry.session.receive(&packet, now_monotonic) + }; + let result = match result { + Ok(result) => result, + Err(_) => { + self.sessions.remove(&key); + return Ok(CultMeshRudpPollOutcome::Handled); + } + }; + + if let Some(reply) = result.reply { + self.send_packet(key.remote_addr, &reply)?; + } + + if result.disconnected { + self.sessions.remove(&key); + return Ok(CultMeshRudpPollOutcome::Handled); + } + + for frame in result.delivered { + if frame.channel_id != "schema" { + continue; + } + let message = match decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + ) { + Ok(message) => message, + Err(_) => continue, + }; + if let Some(rejection) = self.deliver_application_message( + key, + frame.sequence, + message, + now_unix, + now_monotonic, + )? { + self.sessions.remove(&key); + return Ok(CultMeshRudpPollOutcome::ApplicationRejected(rejection)); + } + } + + if packet.reliable { + let ack = self + .sessions + .get_mut(&key) + .ok_or_else(|| anyhow!("CultMesh RUDP session disappeared before ACK"))? + .session + .create_ack(); + self.send_packet(key.remote_addr, &ack)?; + } + Ok(CultMeshRudpPollOutcome::Handled) + } + + /// Expire idle sessions and resend reliable packets whose deadline passed. + pub fn maintain(&mut self) -> Result { + let now = self.clock.now_monotonic_millis(); + let idle_timeout_ms = duration_millis(self.options.session_idle_timeout); + let lifetime_ms = duration_millis(self.options.session_max_lifetime); + let before = self.sessions.len(); + self.sessions.retain(|_, entry| { + now.saturating_sub(entry.last_activity_monotonic_millis) <= idle_timeout_ms + && now.saturating_sub(entry.created_at_monotonic_millis) <= lifetime_ms + }); + let expired = before - self.sessions.len(); + + let mut resends = Vec::new(); + for (key, entry) in &mut self.sessions { + for packet in entry.session.due_resends(now) { + resends.push((key.remote_addr, packet)); + } + } + for (remote_addr, packet) in &resends { + self.send_packet(*remote_addr, packet)?; + } + Ok(CultMeshRudpMaintenance { + sessions_expired: expired, + packets_resent: resends.len(), + }) + } + + fn accept_connection( + &mut self, + key: CultMeshRudpSessionKey, + packet: &CultNetRudpPacket, + now: u64, + ) -> Result<()> { + if !self.sessions.contains_key(&key) { + if self.sessions.len() >= self.options.max_sessions { + return Ok(()); + } + self.sessions.insert( + key, + SessionEntry { + session: CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: key.connection_id, + initial_sequence: 1, + resend_delay_ms: duration_millis(self.options.resend_delay), + max_pending_reliable_packets: Some( + self.options.max_pending_reliable_packets_per_session, + ), + }), + created_at_monotonic_millis: now, + last_activity_monotonic_millis: now, + admitted_payload_bytes: 0, + }, + ); + } + + let accept = { + let entry = self + .sessions + .get_mut(&key) + .expect("accepted session must be present"); + // The same key is a retransmitted Connect for the existing epoch. + // A genuinely fresh client incarnation must choose a fresh id. + entry.last_activity_monotonic_millis = now; + entry.session.accept_connect(packet, now, Vec::new())? + }; + self.send_packet(key.remote_addr, &accept) + } + + fn deliver_application_message( + &mut self, + key: CultMeshRudpSessionKey, + transport_sequence: u32, + message: CultNetMessage, + now_unix: u64, + now: u64, + ) -> Result> { + match message { + CultNetMessage::DocumentPutRaw { + message_id, + document, + } => { + let receipt = CultMeshRudpRawDocumentReceipt { + session: key, + message_id: message_id.clone(), + transport_sequence, + received_at_unix_millis: now_unix, + document, + }; + if let Err(error) = self.sink.accept_raw_document(receipt) { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::DocumentPutRaw, + message_id, + reason: format!("{error:#}"), + })); + } + } + CultNetMessage::SnapshotRequest { + message_id, + schema_ids, + record_keys, + } => { + let query = CultMeshRudpSnapshotQuery { + session: key, + message_id: message_id.clone(), + requested_at_unix_millis: now_unix, + schema_ids, + record_keys, + }; + let documents = match self.snapshot_source.raw_snapshot(&query) { + Ok(documents) => documents, + Err(error) => { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::SnapshotRequest, + message_id, + reason: format!("{error:#}"), + })); + } + }; + if documents.len() > self.options.max_snapshot_documents { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::SnapshotRequest, + message_id, + reason: format!( + "snapshot returned {} documents; limit is {}", + documents.len(), + self.options.max_snapshot_documents + ), + })); + } + let response = CultNetMessage::SnapshotResponseRaw { + message_id: message_id.clone(), + documents, + }; + let payload = match encode_cultnet_message_to_vec( + &response, + CultNetWireContract::CultNetSchemaV0, + ) { + Ok(payload) => payload, + Err(error) => { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::SnapshotRequest, + message_id, + reason: format!("snapshot response could not be encoded: {error:#}"), + })); + } + }; + if payload.len() > self.options.max_snapshot_response_bytes { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::SnapshotRequest, + message_id, + reason: format!( + "snapshot response is {} bytes; limit is {}", + payload.len(), + self.options.max_snapshot_response_bytes + ), + })); + } + if !self.payload_budget_allows(key, payload.len()) { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::SnapshotRequest, + message_id, + reason: "retained payload budget is full".into(), + })); + } + let payload_bytes = payload.len(); + let entry = self + .sessions + .get_mut(&key) + .ok_or_else(|| anyhow!("CultMesh RUDP session disappeared before response"))?; + let packets = match entry.session.send_many( + "schema", + payload, + CultNetRudpSendOptions { + reliable: true, + ordered: true, + sequenced: false, + now_ms: now, + reliable_expire_after_ms: None, + }, + Some(self.options.max_fragment_bytes), + ) { + Ok(packets) => packets, + Err(error) => { + return Ok(Some(CultMeshRudpApplicationRejection { + session: key, + operation: CultMeshRudpApplicationOperation::SnapshotRequest, + message_id, + reason: format!("snapshot response could not be queued: {error:#}"), + })); + } + }; + entry.admitted_payload_bytes = + entry.admitted_payload_bytes.saturating_add(payload_bytes); + for packet in &packets { + self.send_packet(key.remote_addr, packet)?; + } + } + _ => {} + } + Ok(None) + } + + fn payload_budget_allows(&self, key: CultMeshRudpSessionKey, bytes: usize) -> bool { + let Some(session_bytes) = self + .sessions + .get(&key) + .map(|entry| entry.admitted_payload_bytes) + else { + return false; + }; + if session_bytes + .checked_add(bytes) + .is_none_or(|total| total > self.options.max_admitted_payload_bytes_per_session) + { + return false; + } + self.sessions + .values() + .try_fold(0_usize, |total, entry| { + total.checked_add(entry.admitted_payload_bytes) + }) + .and_then(|total| total.checked_add(bytes)) + .is_some_and(|total| total <= self.options.max_admitted_payload_bytes) + } + + fn send_packet(&mut self, remote_addr: SocketAddr, packet: &CultNetRudpPacket) -> Result<()> { + let wire = encode_rudp_packet(packet)?; + self.socket.send_to(&wire, remote_addr)?; + Ok(()) + } +} + +fn validate_options(options: &CultMeshRudpDocumentServerOptions) -> Result<()> { + if options.max_sessions == 0 { + return Err(anyhow!("max_sessions must be greater than zero")); + } + if options.session_idle_timeout.is_zero() { + return Err(anyhow!("session_idle_timeout must be greater than zero")); + } + if options.session_max_lifetime.is_zero() { + return Err(anyhow!("session_max_lifetime must be greater than zero")); + } + if options.max_admitted_payload_bytes == 0 { + return Err(anyhow!( + "max_admitted_payload_bytes must be greater than zero" + )); + } + if options.max_admitted_payload_bytes_per_session == 0 + || options.max_admitted_payload_bytes_per_session > options.max_admitted_payload_bytes + { + return Err(anyhow!( + "per-session admitted payload limit must be non-zero and no larger than the server limit" + )); + } + if options.max_snapshot_response_bytes == 0 + || options.max_snapshot_response_bytes > options.max_admitted_payload_bytes_per_session + { + return Err(anyhow!( + "snapshot response limit must be non-zero and fit in one session's payload budget" + )); + } + if options.max_snapshot_documents == 0 { + return Err(anyhow!("max_snapshot_documents must be greater than zero")); + } + if options.resend_delay.is_zero() { + return Err(anyhow!("resend_delay must be greater than zero")); + } + if options.max_pending_reliable_packets_per_session == 0 { + return Err(anyhow!( + "max_pending_reliable_packets_per_session must be greater than zero" + )); + } + if options.max_fragment_bytes == 0 { + return Err(anyhow!("max_fragment_bytes must be greater than zero")); + } + Ok(()) +} + +fn duration_millis(duration: Duration) -> u64 { + duration.as_millis().try_into().unwrap_or(u64::MAX) +} diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs new file mode 100644 index 00000000..c0f191b4 --- /dev/null +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -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, + failures_remaining: usize, +} + +#[derive(Clone, Default)] +struct Sink(Arc>); + +impl Sink { + fn fail_once() -> Self { + Self(Arc::new(Mutex::new(SinkState { + failures_remaining: 1, + ..Default::default() + }))) + } +} + +impl CultMeshRudpRawDocumentSink for Sink { + fn accept_raw_document(&mut self, receipt: CultMeshRudpRawDocumentReceipt) -> Result<()> { + let mut state = self.0.lock().unwrap(); + if state.failures_remaining > 0 { + state.failures_remaining -= 1; + anyhow::bail!("injected sink failure"); + } + state.receipts.push(receipt); + Ok(()) + } +} + +#[derive(Default)] +struct SourceState { + documents: Vec, + failures_remaining: usize, + calls: usize, +} + +#[derive(Clone, Default)] +struct Source(Arc>); + +impl Source { + fn documents(documents: Vec) -> Self { + Self(Arc::new(Mutex::new(SourceState { + documents, + ..Default::default() + }))) + } + + fn fail_once(documents: Vec) -> Self { + Self(Arc::new(Mutex::new(SourceState { + documents, + failures_remaining: 1, + calls: 0, + }))) + } +} + +impl CultMeshRudpSnapshotSource for Source { + fn raw_snapshot( + &mut self, + _: &CultMeshRudpSnapshotQuery, + ) -> Result> { + let mut state = self.0.lock().unwrap(); + state.calls += 1; + if state.failures_remaining > 0 { + state.failures_remaining -= 1; + anyhow::bail!("injected source failure"); + } + Ok(state.documents.clone()) + } +} + +#[derive(Clone)] +struct Clock { + unix: Arc, + monotonic: Arc, +} + +impl Clock { + fn new(now: u64) -> Self { + Self { + unix: Arc::new(AtomicU64::new(now)), + monotonic: Arc::new(AtomicU64::new(now)), + } + } + + fn set(&self, now: u64) { + self.unix.store(now, Ordering::SeqCst); + self.monotonic.store(now, Ordering::SeqCst); + } + + fn set_unix(&self, now: u64) { + self.unix.store(now, Ordering::SeqCst); + } + + fn set_monotonic(&self, now: u64) { + self.monotonic.store(now, Ordering::SeqCst); + } +} + +impl CultMeshRudpServerClock for Clock { + fn now_unix_millis(&self) -> u64 { + self.unix.load(Ordering::SeqCst) + } + + fn now_monotonic_millis(&self) -> u64 { + self.monotonic.load(Ordering::SeqCst) + } +} + +type Server = CultMeshRudpDocumentServer; + +fn document(key: &str, payload: Vec) -> CultNetRawDocumentRecord { + CultNetRawDocumentRecord { + schema_id: "test.raw.v1".into(), + record_key: key.into(), + stored_at: "2026-09-03T00:00:00Z".into(), + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload, + source_runtime_id: Some(format!("runtime-{key}")), + source_agent_id: None, + source_role: None, + tags: None, + } +} + +fn server( + options: CultMeshRudpDocumentServerOptions, + clock: Clock, + sink: Sink, + source: Source, +) -> Result { + CultMeshRudpDocumentServer::new( + UdpSocket::bind("127.0.0.1:0")?, + sink, + source, + clock, + options, + ) +} + +fn client(target: SocketAddr, id: u32) -> Result { + let socket = UdpSocket::bind("127.0.0.1:0")?; + socket.set_nonblocking(true)?; + CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + runtime_id: "test-client".into(), + socket, + mode: CultNetRudpSocketMode::Client, + remote_addr: Some(target), + connection_id: id, + initial_sequence: 1, + resend_delay_ms: 10, + transport_id: None, + max_payload_bytes: None, + max_fragment_bytes: Some(1200), + max_pending_reliable_packets: Some(64), + media_reliable_expire_after_ms: None, + 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 { + 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(()) +}