Ack Cut D D2: pin the disconnect, the resend loop, the Python resend, ack and accept paths

The injected permanent failure now takes precedence over the injected transient
one, so a test can end a session on a peer that is also losing datagrams and
count exactly which datagrams were attempted afterwards. New tests: a hub
disconnect that can never be sent returns the error; a peer or session ended
in the resend loop is sent nothing more (Rust hub, document server, C#); the
goodbye reaches the peer; the Python server ends the session on a permanent
failure in a resend, an ack and an accept. The oversized-send test compares the
outstanding count with the one the hub's own unacknowledged Accept already
holds.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

packages/cultmesh-py/tests/test_cultmesh.py

@@ -4651,6 +4651,106 @@ class CultMeshRudpSendFailureTests(unittest.TestCase):
         finally:
             server.stop()
 
+    def test_a_permanent_failure_in_a_resend_ends_only_that_peers_session(self) -> None:
+        server = CultMesh.serve_node(CultMesh.create_node(runtime_id="mesh-server"))
+        original_sendto = socket.socket.sendto
+        try:
+            x, x_addr = self._client(server, "peer-x")
+            y, _ = self._client(server, "peer-y")
+            mode = ["transient"]
+
+            def sendto(sock: socket.socket, data: bytes, *args: Any) -> int:
+                if args and tuple(args[-1]) == tuple(x_addr) and mode[0] == "transient":
+                    raise OSError(errno.ENOBUFS, "injected transient failure")
+                if args and tuple(args[-1]) == tuple(x_addr) and mode[0] == "permanent":
+                    mode[0] = "healthy"
+                    raise OSError(errno.EMSGSIZE, "Message too long")
+                return original_sendto(sock, data, *args)
+
+            with patch.object(socket.socket, "sendto", sendto):
+                # X's response is lost, so a resend is pending; then it can never be sent as built.
+                x.send("schema", hello(runtime_id="probe").to_bytes())
+                self._wait_for(lambda: server.rudp_send_failures > 0)
+                self.assertFalse(x.disconnect_reason, "a transient failure ends no session")
+                mode[0] = "permanent"
+                for _ in range(200):
+                    x.receive_once()
+                    if x.disconnect_reason is not None:
+                        break
+                    time.sleep(0.005)
+                self.assertIsNotNone(x.disconnect_reason)
+                self.assertTrue(x.disconnect_reason.startswith(b"packet could not be sent: "))
+                self.assertTrue(server._rudp_thread.is_alive())
+                self.assertIsNotNone(self._served(y))
+        finally:
+            server.stop()
+
+    def test_a_permanent_failure_acknowledging_a_frame_ends_the_session(self) -> None:
+        import msgpack  # type: ignore
+
+        server = CultMesh.serve_node(CultMesh.create_node(runtime_id="mesh-server"))
+        original_sendto = socket.socket.sendto
+        try:
+            x, x_addr = self._client(server, "peer-x")
+            failures = [1]
+
+            def sendto(sock: socket.socket, data: bytes, *args: Any) -> int:
+                if args and tuple(args[-1]) == tuple(x_addr) and failures[0] > 0:
+                    failures[0] -= 1
+                    raise OSError(errno.EMSGSIZE, "Message too long")
+                return original_sendto(sock, data, *args)
+
+            with patch.object(socket.socket, "sendto", sendto):
+                # Not a message the server answers, so its ack is the only datagram owed.
+                x.send("schema", msgpack.packb(1))
+                for _ in range(200):
+                    x.receive_once()
+                    if x.disconnect_reason is not None:
+                        break
+                    time.sleep(0.005)
+                self.assertIsNotNone(x.disconnect_reason)
+                self.assertTrue(x.disconnect_reason.startswith(b"packet could not be sent: "))
+                self.assertTrue(server._rudp_thread.is_alive())
+        finally:
+            server.stop()
+
+    def test_a_connect_whose_accept_can_never_be_sent_starts_no_session(self) -> None:
+        server = CultMesh.serve_node(CultMesh.create_node(runtime_id="mesh-server"))
+        original_sendto = socket.socket.sendto
+        try:
+            y, _ = self._client(server, "peer-y")
+            z_socket = bind_udp_socket()
+            z_addr = z_socket.getsockname()[:2]
+            failures = [1]
+
+            def sendto(sock: socket.socket, data: bytes, *args: Any) -> int:
+                if args and tuple(args[-1]) == tuple(z_addr) and failures[0] > 0:
+                    failures[0] -= 1
+                    raise OSError(errno.EMSGSIZE, "Message too long")
+                return original_sendto(sock, data, *args)
+
+            with patch.object(socket.socket, "sendto", sendto):
+                z = CultNetRudpSocketTransportConnection(
+                    CultNetRudpSocketTransportOptions(
+                        runtime_id="peer-z",
+                        socket=z_socket,
+                        mode=CultNetRudpSocketMode.CLIENT,
+                        connection_id=server.rudp_connection_id,
+                        remote_addr=("127.0.0.1", server.port),
+                        resend_delay_ms=25,
+                    )
+                )
+                z.connect(b"join")
+                # A session that started would resend its Accept, and the resend would get through.
+                time.sleep(0.5)
+                z_socket.settimeout(0.05)
+                with self.assertRaises(TimeoutError):
+                    z_socket.recvfrom(65535)
+                self.assertTrue(server._rudp_thread.is_alive())
+                self.assertIsNotNone(self._served(y))
+        finally:
+            server.stop()
+
     def _wait_for(self, predicate: Callable[[], bool]) -> None:
         deadline = time.monotonic() + 2.0
         while time.monotonic() < deadline:

packages/cultmesh-rs/src/rudp_document_server.rs

@@ -724,9 +724,6 @@ where
     fn send_datagram(&mut 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"));
-            }
             match self.unsendable_after.get(&remote_addr).copied() {
                 Some(0) => {
                     self.unsendable_after.remove(&remote_addr);
@@ -737,6 +734,9 @@ where
                 }
                 None => {}
             }
+            if self.failing_peers.contains(&remote_addr) {
+                return Err(std::io::Error::other("injected send failure"));
+            }
         }
         self.socket.send_to(wire, remote_addr)
     }

packages/cultmesh-rs/src/rudp_document_server/send_failure_tests.rs

@@ -344,3 +344,25 @@ fn healthy_sends_are_not_failures() {
     assert_eq!(server.send_failures(), 0);
     assert_eq!(server.session_count(), 2);
 }
+
+#[test]
+fn a_session_ended_in_the_resend_loop_is_sent_nothing_more_from_it() {
+    let (mut server, clock, _received, (mut x, x_addr), (_y, _)) = two_peers();
+    server.failing_peers.insert(x_addr);
+    send(&mut x, &snapshot("one"));
+    send(&mut x, &snapshot("two"));
+    for _ in 0..10 {
+        server.poll_once().unwrap();
+    }
+    let before = server.send_failures();
+
+    server.unsendable_after.insert(x_addr, 0);
+    clock.0.store(1_200, Ordering::SeqCst);
+    let maintenance = server.maintain().unwrap();
+
+    // Both responses are due. The first can never be sent and ends the session;
+    // the goodbye is the only other datagram attempted at X.
+    assert_eq!(maintenance.packets_resent, 2);
+    assert_eq!(server.send_failures() - before, 1);
+    assert_eq!(server.session_count(), 1);
+}

packages/cultnet-rs/src/rudp.rs

@@ -2032,9 +2032,6 @@ impl CultNetRudpServerHub {
     fn send_datagram(&mut 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"));
-            }
             match self.unsendable_after.get(&remote_addr).copied() {
                 Some(0) => {
                     self.unsendable_after.remove(&remote_addr);
@@ -2045,6 +2042,9 @@ impl CultNetRudpServerHub {
                 }
                 None => {}
             }
+            if self.failing_peers.contains(&remote_addr) {
+                return Err(std::io::Error::other("injected send failure"));
+            }
         }
         self.socket.send_to(wire, remote_addr)
     }

packages/cultnet-rs/src/rudp/send_failure_tests.rs

@@ -162,6 +162,10 @@ fn one_peer(
 #[test]
 fn an_oversized_send_returns_the_error_and_queues_nothing() {
     let (mut hub, mut peer, session) = one_peer(None);
+    // The hub's own Accept is still awaiting its ack; the refused send adds nothing to it.
+    let outstanding = hub.peers[&session.remote_addr]
+        .session
+        .outstanding_reliable_packet_count();
 
     let error = hub
         .send(&session, "schema", vec![7; 70_000])
@@ -176,7 +180,7 @@ fn an_oversized_send_returns_the_error_and_queues_nothing() {
     let remote = session.remote_addr;
     assert_eq!(
         hub.peers[&remote].session.outstanding_reliable_packet_count(),
-        0,
+        outstanding,
         "nothing is pending or queued to resend"
     );
 
@@ -213,7 +217,7 @@ fn a_send_that_fails_permanently_after_a_fragment_left_ends_the_session() {
 
 #[test]
 fn a_permanent_failure_in_resends_ends_only_that_peers_session() {
-    let (mut hub, (_x, x_session), (mut y, y_session)) = two_peers();
+    let (mut hub, (mut 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);
@@ -240,6 +244,16 @@ fn a_permanent_failure_in_resends_ends_only_that_peers_session() {
         }
     }
     assert_eq!(delivered.expect("Y receives its resend").payload, b"for-y");
+    // The goodbye reaches X and names the error.
+    for _ in 0..20 {
+        x.receive_once().unwrap();
+        if x.disconnect_reason().is_some() {
+            break;
+        }
+    }
+    assert!(is_unsendable_reason(
+        x.disconnect_reason().expect("X is told its session ended")
+    ));
 }
 
 #[test]
@@ -369,3 +383,34 @@ fn only_a_datagram_that_can_never_be_sent_is_a_permanent_failure() {
         }
     }
 }
+
+#[test]
+fn a_permanent_failure_sending_a_disconnect_is_returned_to_the_caller() {
+    let (mut hub, _peer, session) = one_peer(None);
+    hub.unsendable_after.insert(session.remote_addr, 0);
+
+    let error = hub
+        .disconnect(&session, b"bye".to_vec())
+        .expect_err("the goodbye can never be sent");
+    assert!(is_permanent_send_error(
+        error.downcast_ref::<std::io::Error>().unwrap()
+    ));
+}
+
+#[test]
+fn a_peer_ended_in_the_resend_loop_is_sent_nothing_more_from_it() {
+    let (mut hub, (_x, x_session), (_y, _y_session)) = two_peers();
+    hub.failing_peers.insert(x_session.remote_addr);
+    hub.send(&x_session, "schema", b"one".to_vec()).unwrap();
+    hub.send(&x_session, "schema", b"two".to_vec()).unwrap();
+    assert_eq!(hub.stats().send_failures, 2);
+
+    hub.unsendable_after.insert(x_session.remote_addr, 0);
+    std::thread::sleep(Duration::from_millis(40));
+    hub.poll_resends().unwrap();
+
+    // Both resends are due. The first can never be sent and ends the session;
+    // the goodbye is the only other datagram attempted at X.
+    assert_eq!(hub.stats().send_failures, 3);
+    assert_eq!(hub.sessions().len(), 1);
+}

src/GameCult.Networking/CultNetTransport.cs

@@ -3116,8 +3116,6 @@ namespace GameCult.Networking
             var wire = CultNetRudpPacketCodec.Encode(packet);
             try
             {
-                if (FailingSendPeers.Contains(remoteEndPoint))
-                    throw new SocketException((int)SocketError.HostUnreachable);
                 if (UnsendableAfter.TryGetValue(remoteEndPoint, out var remaining))
                 {
                     if (remaining > 0)
@@ -3130,6 +3128,8 @@ namespace GameCult.Networking
                         throw new SocketException((int)SocketError.MessageSize);
                     }
                 }
+                if (FailingSendPeers.Contains(remoteEndPoint))
+                    throw new SocketException((int)SocketError.HostUnreachable);
                 var sent = _socket.SendTo(wire, remoteEndPoint);
                 _stats.BytesSent += sent;
             }

tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs

@@ -293,6 +293,28 @@ namespace GameCult.Networking.Tests
             Assert.That(server.Stats.PacketsDropped, Is.EqualTo(1));
         }
 
+        [Test]
+        public void APeerEndedInTheResendLoopIsSentNothingMoreFromIt()
+        {
+            var (server, _, x, _) = TwoPeers();
+            using (server)
+            {
+                server.FailingSendPeers.Add(x.EndPoint);
+                server.SendSchema(x.Server, "one");
+                server.SendSchema(x.Server, "two");
+                Assert.That(server.Stats.SendFailures, Is.EqualTo(2));
+
+                server.UnsendableAfter[x.EndPoint] = 0;
+                Thread.Sleep(30);
+                server.PollResends();
+
+                // Both resends are due. The first can never be sent and ends the session; the
+                // goodbye is the only other datagram attempted at X.
+                Assert.That(server.Stats.SendFailures, Is.EqualTo(3));
+                Assert.That(server.Peers, Has.Count.EqualTo(1));
+            }
+        }
+
         [Test]
         public void AHealthySendCountsItsBytesAndNoFailure()
         {