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 <noreply@anthropic.com>

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<S, Q, C> {
     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<SocketAddr>,
 }
 
 impl<S, Q, C> CultMeshRudpDocumentServer<S, Q, C>
@@ -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<SocketAddr> {
         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"));

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<AtomicU64>);
+
+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<Mutex<Vec<String>>>;
+type Server = CultMeshRudpDocumentServer<
+    Box<dyn FnMut(CultMeshRudpRawDocumentReceipt) -> Result<()>>,
+    Box<dyn FnMut(&CultMeshRudpSnapshotQuery) -> Result<Vec<CultNetRawDocumentRecord>>>,
+    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<dyn FnMut(CultMeshRudpRawDocumentReceipt) -> Result<()>> =
+        Box::new(move |receipt| {
+            sink_received.lock().unwrap().push(receipt.message_id);
+            Ok(())
+        });
+    let source: Box<
+        dyn FnMut(&CultMeshRudpSnapshotQuery) -> Result<Vec<CultNetRawDocumentRecord>>,
+    > = 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);
+}

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<CultNetRudpServerEvent>,
     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<SocketAddr>,
 }
 
 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<usize> {
+        #[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 {

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);
+}

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)]