From def93c5d894f20779635636e74f1b4bb730ee819 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 00:47:53 +0200 Subject: [PATCH] Ack Cut D (Rust): a failed send to one peer is that peer's lost datagram The cultmesh-rs document server and the cultnet-rs hub returned the first failed send_to from maintain, poll_once, poll_resends and receive_event_once, so one peer whose datagrams cannot be sent failed every poll for every peer. send_packet now counts a send failure (send_failures() on the document server, CultNetTransportStats.send_failures on the hub) and returns; only an encode failure stays an error. The document server treats recv_from ConnectionReset as Idle, as the socket transport does. Tests inject the failure at the one send function through a cfg(test) set of failing peers and prove the other peer is still served. Co-Authored-By: Claude Sonnet 5.5 --- .../cultmesh-rs/src/rudp_document_server.rs | 43 +++- .../send_failure_tests.rs | 211 ++++++++++++++++++ packages/cultnet-rs/src/rudp.rs | 27 ++- .../cultnet-rs/src/rudp/send_failure_tests.rs | 124 ++++++++++ packages/cultnet-rs/src/transport.rs | 5 + 5 files changed, 406 insertions(+), 4 deletions(-) create mode 100644 packages/cultmesh-rs/src/rudp_document_server/send_failure_tests.rs create mode 100644 packages/cultnet-rs/src/rudp/send_failure_tests.rs diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index b99c7145..9ce04b58 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -6,6 +6,8 @@ use cultnet_rs::{ encode_rudp_packet, }; use std::collections::BTreeMap; +#[cfg(test)] +use std::collections::BTreeSet; use std::io::ErrorKind; use std::net::{SocketAddr, UdpSocket}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -208,6 +210,11 @@ pub struct CultMeshRudpDocumentServer { clock: C, options: CultMeshRudpDocumentServerOptions, packets_dropped: u64, + send_failures: u64, + /// Peers whose every datagram fails to send, standing in for an unroutable + /// or full path that a loopback peer cannot be made to have. + #[cfg(test)] + failing_peers: BTreeSet, } impl CultMeshRudpDocumentServer @@ -236,6 +243,9 @@ where clock, options, packets_dropped: 0, + send_failures: 0, + #[cfg(test)] + failing_peers: BTreeSet::new(), }) } @@ -247,6 +257,15 @@ where self.packets_dropped } + /// Datagrams that could not be sent to a peer. Each is that peer's lost + /// datagram: a reliable packet stays pending and is resent, and the peer's + /// session ends by idle timeout, lifetime, refusal or Disconnect, never by + /// the failure. Never an error: one unreachable peer must not stop the + /// poll that serves the others. + pub fn send_failures(&self) -> u64 { + self.send_failures + } + pub fn local_addr(&self) -> Result { Ok(self.socket.local_addr()?) } @@ -265,7 +284,14 @@ where 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 => { + // Windows reports an earlier ICMP port-unreachable on the next + // receive; it names no datagram of this poll. + Err(error) + if matches!( + error.kind(), + ErrorKind::WouldBlock | ErrorKind::ConnectionReset + ) => + { return Ok(CultMeshRudpPollOutcome::Idle); } Err(error) => return Err(error.into()), @@ -627,11 +653,24 @@ where fn send_packet(&mut self, remote_addr: SocketAddr, packet: &CultNetRudpPacket) -> Result<()> { let wire = encode_rudp_packet(packet)?; - self.socket.send_to(&wire, remote_addr)?; + #[cfg(test)] + let sent = if self.failing_peers.contains(&remote_addr) { + Err(std::io::Error::other("injected send failure")) + } else { + self.socket.send_to(&wire, remote_addr) + }; + #[cfg(not(test))] + let sent = self.socket.send_to(&wire, remote_addr); + if sent.is_err() { + self.send_failures += 1; + } Ok(()) } } +#[cfg(test)] +mod send_failure_tests; + fn validate_options(options: &CultMeshRudpDocumentServerOptions) -> Result<()> { if options.max_sessions == 0 { return Err(anyhow!("max_sessions must be greater than zero")); diff --git a/packages/cultmesh-rs/src/rudp_document_server/send_failure_tests.rs b/packages/cultmesh-rs/src/rudp_document_server/send_failure_tests.rs new file mode 100644 index 00000000..cbcf588b --- /dev/null +++ b/packages/cultmesh-rs/src/rudp_document_server/send_failure_tests.rs @@ -0,0 +1,211 @@ +//! One peer whose datagrams cannot be sent must not stop the server serving +//! the others. The failure is injected in `send_packet` through +//! `failing_peers`; a loopback peer cannot be made unroutable on demand. + +use super::*; +use cultnet_rs::{ + CultNetRawPayloadEncoding, CultNetRudpSocketMode, CultNetRudpSocketTransportConnection, + CultNetRudpSocketTransportOptions, +}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; + +#[derive(Clone)] +struct Clock(Arc); + +impl CultMeshRudpServerClock for Clock { + fn now_unix_millis(&self) -> u64 { + self.0.load(Ordering::SeqCst) + } + fn now_monotonic_millis(&self) -> u64 { + self.0.load(Ordering::SeqCst) + } +} + +type Received = Arc>>; +type Server = CultMeshRudpDocumentServer< + Box Result<()>>, + Box Result>>, + Clock, +>; +type Client = CultNetRudpSocketTransportConnection; + +fn document(key: &str) -> 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: vec![0x81, 0xa1, 0x78, 0x2a], + source_runtime_id: None, + source_agent_id: None, + source_role: None, + tags: None, + } +} + +fn client(target: SocketAddr, id: u32) -> (Client, SocketAddr) { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket.set_nonblocking(true).unwrap(); + let addr = socket.local_addr().unwrap(); + let client = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, + 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, + }) + .unwrap(); + (client, addr) +} + +fn send(client: &mut Client, message: &CultNetMessage) { + client + .send( + "schema", + encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0).unwrap(), + ) + .unwrap(); +} + +/// A server with a 1 s idle timeout and two connected peers, the first +/// returned being the one whose address sorts first. +fn two_peers() -> ( + Server, + Clock, + Received, + (Client, SocketAddr), + (Client, SocketAddr), +) { + let clock = Clock(Arc::new(AtomicU64::new(1_000))); + let received: Received = Arc::default(); + let sink_received = received.clone(); + let sink: Box Result<()>> = + Box::new(move |receipt| { + sink_received.lock().unwrap().push(receipt.message_id); + Ok(()) + }); + let source: Box< + dyn FnMut(&CultMeshRudpSnapshotQuery) -> Result>, + > = Box::new(|_| Ok(vec![document("catalog")])); + let mut server = CultMeshRudpDocumentServer::new( + UdpSocket::bind("127.0.0.1:0").unwrap(), + sink, + source, + clock.clone(), + CultMeshRudpDocumentServerOptions { + session_idle_timeout: Duration::from_secs(1), + resend_delay: Duration::from_millis(10), + ..Default::default() + }, + ) + .unwrap(); + let target = server.local_addr().unwrap(); + let (mut a, a_addr) = client(target, 7); + let (mut b, b_addr) = client(target, 7); + a.connect(Vec::new()).unwrap(); + b.connect(Vec::new()).unwrap(); + for _ in 0..500 { + server.poll_once().unwrap(); + a.receive_once().unwrap(); + b.receive_once().unwrap(); + if a.connected() && b.connected() { + break; + } + thread::sleep(Duration::from_millis(1)); + } + assert!(a.connected() && b.connected()); + for _ in 0..4 { + server.poll_once().unwrap(); + } + assert_eq!(server.session_count(), 2); + if a_addr < b_addr { + (server, clock, received, (a, a_addr), (b, b_addr)) + } else { + (server, clock, received, (b, b_addr), (a, a_addr)) + } +} + +#[test] +fn one_peer_whose_sends_fail_does_not_stop_the_server_serving_the_others() { + let (mut server, clock, received, (mut x, x_addr), (mut y, _)) = two_peers(); + server.failing_peers.insert(x_addr); + + // X's reply, ack and snapshot response all fail to send. X is visited first. + send( + &mut x, + &CultNetMessage::SnapshotRequest { + message_id: "x-snapshot".into(), + schema_ids: None, + record_keys: None, + }, + ); + send( + &mut y, + &CultNetMessage::DocumentPutRaw { + message_id: "y-put".into(), + document: document("y"), + }, + ); + send( + &mut y, + &CultNetMessage::SnapshotRequest { + message_id: "y-snapshot".into(), + schema_ids: None, + record_keys: None, + }, + ); + let mut y_response = None; + for step in 0..200 { + // The clock moves past X's resend delay so `maintain` has X's resend due. + clock.0.store(1_000 + step * 15, Ordering::SeqCst); + // Never `Err`: X's failed sends are its own lost datagrams. + server.poll_once().unwrap(); + if let Some(frame) = y.receive_once().unwrap() { + y_response = Some(frame); + break; + } + thread::sleep(Duration::from_millis(1)); + } + let frame = y_response.expect("Y receives its snapshot response"); + let response = + decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0) + .unwrap(); + assert!(matches!( + response, + CultNetMessage::SnapshotResponseRaw { ref message_id, .. } if message_id == "y-snapshot" + )); + assert!(server.send_failures() > 0); + assert_eq!(received.lock().unwrap().as_slice(), ["y-put"]); + assert!(server.maintain().is_ok()); + + // X's session ends by the idle rule that already ends sessions, not by the + // failure: Y stays active, X does not. + clock.0.store(1_900, Ordering::SeqCst); + send( + &mut y, + &CultNetMessage::DocumentPutRaw { + message_id: "y-put-2".into(), + document: document("y2"), + }, + ); + for _ in 0..20 { + server.poll_once().unwrap(); + thread::sleep(Duration::from_millis(1)); + } + assert_eq!(server.session_count(), 2, "the failure ended no session"); + clock.0.store(2_500, Ordering::SeqCst); + let maintenance = server.maintain().unwrap(); + assert_eq!(maintenance.sessions_expired, 1); + assert_eq!(server.session_count(), 1); +} diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 3b348fe3..1d73216d 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -27,6 +27,9 @@ use crate::create_reconnect_policy; use crate::decode_cultnet_message_from_slice; use crate::encode_cultnet_message_to_vec; +#[cfg(test)] +mod send_failure_tests; + const RUDP_MAGIC: [u8; 4] = [0x43, 0x4e, 0x52, 0x30]; const RUDP_VERSION: u8 = 0; const RUDP_FIXED_HEADER_BYTES: usize = 36; @@ -1440,6 +1443,10 @@ pub struct CultNetRudpServerHub { pending_events: VecDeque, pub profile: CultNetTransportProfile, stats: CultNetTransportStats, + /// Peers whose every datagram fails to send, standing in for an unroutable + /// or full path that a loopback peer cannot be made to have. + #[cfg(test)] + failing_peers: BTreeSet, } impl CultNetRudpServerHub { @@ -1488,6 +1495,8 @@ impl CultNetRudpServerHub { pending_events: VecDeque::new(), profile, stats: CultNetTransportStats::default(), + #[cfg(test)] + failing_peers: BTreeSet::new(), }) } @@ -1796,12 +1805,26 @@ impl CultNetRudpServerHub { timed_out.into_iter().map(|(_, context)| context).collect() } + /// A failed send is that peer's lost datagram: counted, never an error. A + /// reliable packet stays pending and is resent; an unreliable one was + /// allowed to be lost. The peer's session ends by the rules that already + /// end sessions. Only an encode failure, the hub's own bug, is an error. fn send_packet(&mut self, remote_addr: SocketAddr, packet: &CultNetRudpPacket) -> Result<()> { let wire = encode_rudp_packet(packet)?; - let sent = self.socket.send_to(&wire, remote_addr)?; - self.stats.bytes_sent += sent as u64; + match self.send_datagram(&wire, remote_addr) { + Ok(sent) => self.stats.bytes_sent += sent as u64, + Err(_) => self.stats.send_failures += 1, + } Ok(()) } + + fn send_datagram(&self, wire: &[u8], remote_addr: SocketAddr) -> std::io::Result { + #[cfg(test)] + if self.failing_peers.contains(&remote_addr) { + return Err(std::io::Error::other("injected send failure")); + } + self.socket.send_to(wire, remote_addr) + } } impl CultNetRudpSocketTransportConnection { diff --git a/packages/cultnet-rs/src/rudp/send_failure_tests.rs b/packages/cultnet-rs/src/rudp/send_failure_tests.rs new file mode 100644 index 00000000..1a9b66a7 --- /dev/null +++ b/packages/cultnet-rs/src/rudp/send_failure_tests.rs @@ -0,0 +1,124 @@ +//! One peer whose datagrams cannot be sent must not stop the hub serving the +//! others. The failure is injected at `send_datagram` through `failing_peers`. + +use super::*; +use std::time::Duration; + +const CONNECTION_ID: u32 = 0x4355_4c54; + +fn socket() -> UdpSocket { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_millis(20))) + .unwrap(); + socket +} + +fn client(remote: SocketAddr) -> CultNetRudpSocketTransportConnection { + let mut options = + CultNetRudpSocketTransportOptions::client("peer", socket(), remote, CONNECTION_ID); + options.resend_delay_ms = 10; + CultNetRudpSocketTransportConnection::new(options).unwrap() +} + +fn connect( + hub: &mut CultNetRudpServerHub, + client: &mut CultNetRudpSocketTransportConnection, +) -> CultNetRudpServerSessionContext { + client.connect(Vec::new()).unwrap(); + let mut session = None; + for _ in 0..20 { + if let Some(CultNetRudpServerEvent::Connected { session: found }) = + hub.receive_event_once().unwrap() + { + session = Some(found); + break; + } + } + client.receive_once().unwrap(); + assert!(client.connected()); + session.expect("hub saw the Connect") +} + +/// Two connected peers, ordered so the first is the one the hub visits first +/// (the peer map is ordered by address). +fn two_peers() -> ( + CultNetRudpServerHub, + ( + CultNetRudpSocketTransportConnection, + CultNetRudpServerSessionContext, + ), + ( + CultNetRudpSocketTransportConnection, + CultNetRudpServerSessionContext, + ), +) { + let server_socket = socket(); + let server_addr = server_socket.local_addr().unwrap(); + let mut options = CultNetRudpServerHubOptions::new("hub", server_socket, CONNECTION_ID); + options.resend_delay_ms = 10; + let mut hub = CultNetRudpServerHub::new(options).unwrap(); + let mut a = client(server_addr); + let mut b = client(server_addr); + let a_session = connect(&mut hub, &mut a); + let b_session = connect(&mut hub, &mut b); + if a_session.remote_addr < b_session.remote_addr { + (hub, (a, a_session), (b, b_session)) + } else { + (hub, (b, b_session), (a, a_session)) + } +} + +#[test] +fn a_failed_resend_to_one_peer_does_not_stop_the_others_resends() { + let (mut hub, (_x, x_session), (mut y, y_session)) = two_peers(); + // Both peers' first datagram is lost, so both have a reliable resend due. + hub.failing_peers.insert(x_session.remote_addr); + hub.failing_peers.insert(y_session.remote_addr); + hub.send(&x_session, "schema", b"for-x".to_vec()).unwrap(); + hub.send(&y_session, "schema", b"for-y".to_vec()).unwrap(); + assert_eq!(hub.stats().send_failures, 2); + + // Y's path recovers; X's stays dead. X is visited first. + hub.failing_peers.remove(&y_session.remote_addr); + std::thread::sleep(Duration::from_millis(40)); + hub.poll_resends().unwrap(); + + let mut delivered = None; + for _ in 0..20 { + if let Some(frame) = y.receive_once().unwrap() { + delivered = Some(frame); + break; + } + } + assert_eq!(delivered.expect("Y receives its resend").payload, b"for-y"); + assert!(hub.stats().send_failures >= 3); + assert_eq!(hub.sessions().len(), 2, "a send failure ends no session"); +} + +#[test] +fn a_failed_reply_to_one_peer_does_not_stop_the_hub_serving_the_others() { + let (mut hub, (mut x, x_session), (mut y, _y_session)) = two_peers(); + hub.failing_peers.insert(x_session.remote_addr); + + x.send("schema", b"from-x".to_vec()).unwrap(); + y.send("schema", b"from-y".to_vec()).unwrap(); + + let mut frames = Vec::new(); + for _ in 0..20 { + if let Some(CultNetRudpServerEvent::Frame { frame, .. }) = hub.receive_event_once().unwrap() + { + frames.push(frame.payload); + } + if frames.len() == 2 { + break; + } + } + frames.sort(); + assert_eq!(frames, vec![b"from-x".to_vec(), b"from-y".to_vec()]); + assert!( + hub.stats().send_failures >= 1, + "X's ack was counted as lost" + ); + assert_eq!(hub.sessions().len(), 2); +} diff --git a/packages/cultnet-rs/src/transport.rs b/packages/cultnet-rs/src/transport.rs index 71e714c4..f2c196d4 100644 --- a/packages/cultnet-rs/src/transport.rs +++ b/packages/cultnet-rs/src/transport.rs @@ -31,6 +31,11 @@ pub struct CultNetTransportStats { /// unadmitted sender, or a packet the session refused. A steady rate means /// a flow is being moved or scanned, not that this transport is failing. pub packets_dropped: u64, + /// Datagrams a multi-peer server could not send to one peer. Each is a lost + /// datagram attributed to that peer: reliable packets stay pending and are + /// resent, and no other peer's poll is affected. Zero for single-peer + /// transports, whose callers get the send error itself. + pub send_failures: u64, } #[derive(Clone, Debug, PartialEq, Eq)]