Ack Cut D (Python): a failed send to one peer no longer kills the RUDP thread

CultMeshLocalServer._send_rudp_packet let an OSError from sendto escape
_rudp_loop, which ended the RUDP thread for every peer. It now counts the
failure (rudp_send_failures) and returns. The test patches socket.sendto to
fail for one peer and serves another.

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

packages/cultmesh-py/src/cultmesh_py/server.py

@@ -73,6 +73,14 @@ class CultMeshLocalServer:
     _stop: threading.Event = field(default_factory=threading.Event, init=False, repr=False)
     _thread: threading.Thread | None = field(default=None, init=False, repr=False)
     _rudp_thread: threading.Thread | None = field(default=None, init=False, repr=False)
+    _rudp_send_failures: int = field(default=0, init=False, repr=False)
+
+    @property
+    def rudp_send_failures(self) -> int:
+        """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 no other
+        peer is affected."""
+        return self._rudp_send_failures
 
     def __post_init__(self) -> None:
         if self.max_snapshot_documents is not None and self.max_snapshot_documents < 0:
@@ -298,13 +306,19 @@ class CultMeshLocalServer:
         ):
             self._send_rudp_packet(rudp_socket, remote_addr, packet)
 
-    @staticmethod
     def _send_rudp_packet(
+        self,
         rudp_socket: socket.socket,
         remote_addr: tuple[str, int],
         packet: CultNetRudpPacket,
     ) -> None:
-        rudp_socket.sendto(encode_rudp_packet(packet), remote_addr)
+        # A failed send is that peer's lost datagram: counted, never raised.
+        # Raising here ended the RUDP thread for every peer.
+        wire = encode_rudp_packet(packet)
+        try:
+            rudp_socket.sendto(wire, remote_addr)
+        except OSError:
+            self._rudp_send_failures += 1
 
     def _handle_connection_message(
         self,

packages/cultmesh-py/tests/test_cultmesh.py

@@ -4498,5 +4498,74 @@ class CultMeshTests(unittest.TestCase):
         self.assertEqual(streams.latest_frame("mimir:kiyo-pro").sequence, 42)
 
 
+class CultMeshRudpSendFailureTests(unittest.TestCase):
+    """One peer whose datagrams cannot be sent must not stop the RUDP thread
+    serving the others."""
+
+    def _client(self, server: Any, name: str) -> tuple[CultNetRudpSocketTransportConnection, tuple[str, int]]:
+        sock = bind_udp_socket()
+        client = CultNetRudpSocketTransportConnection(
+            CultNetRudpSocketTransportOptions(
+                runtime_id=name,
+                socket=sock,
+                mode=CultNetRudpSocketMode.CLIENT,
+                connection_id=server.rudp_connection_id,
+                remote_addr=("127.0.0.1", server.port),
+                resend_delay_ms=25,
+            )
+        )
+        client.connect(b"join")
+        for _ in range(200):
+            client.receive_once()
+            if client.connected:
+                break
+        self.assertTrue(client.connected)
+        return client, sock.getsockname()[:2]
+
+    def test_a_failed_send_to_one_peer_does_not_kill_the_rudp_thread(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")
+            y, _ = self._client(server, "peer-y")
+
+            def sendto(sock: socket.socket, data: bytes, *args: Any) -> int:
+                if args and tuple(args[-1]) == tuple(x_addr):
+                    raise OSError("injected send failure")
+                return original_sendto(sock, data, *args)
+
+            request = msgpack.packb(hello(runtime_id="probe"), use_bin_type=True)
+            with patch.object(socket.socket, "sendto", sendto):
+                # The server's reply and ack to X both fail to send.
+                x.send("schema", request)
+                self._wait_for(lambda: server.rudp_send_failures > 0)
+                self.assertTrue(server._rudp_thread.is_alive())
+
+                y.send("schema", request)
+                response = None
+                for _ in range(200):
+                    response = y.receive_once()
+                    if response is not None:
+                        break
+                    time.sleep(0.005)
+
+            self.assertIsNotNone(response, "Y is still served after X's send failed")
+            self.assertEqual(msgpack.unpackb(response.payload, raw=False)["runtimeId"], "mesh-server")
+            self.assertTrue(server._rudp_thread.is_alive())
+            self.assertGreater(server.rudp_send_failures, 0)
+        finally:
+            server.stop()
+
+    def _wait_for(self, predicate: Callable[[], bool]) -> None:
+        deadline = time.monotonic() + 2.0
+        while time.monotonic() < deadline:
+            if predicate():
+                return
+            time.sleep(0.005)
+        self.fail("condition not reached")
+
+
 if __name__ == "__main__":
     unittest.main()