diff --git a/packages/cultmesh-py/src/cultmesh_py/server.py b/packages/cultmesh-py/src/cultmesh_py/server.py index ff6d998d..41d9d9e9 100644 --- a/packages/cultmesh-py/src/cultmesh_py/server.py +++ b/packages/cultmesh-py/src/cultmesh_py/server.py @@ -20,6 +20,7 @@ from cultnet_py import ( create_tcp_framed_transport_profile, decode_rudp_packet, encode_rudp_packet, + is_permanent_send_error, wire_message_schema_descriptors, ) @@ -79,7 +80,9 @@ class CultMeshLocalServer: 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.""" + peer is affected. A datagram that can never be sent as built + (`is_permanent_send_error`) is not counted here: it ends that peer's + session, and the peer's goodbye names the error.""" return self._rudp_send_failures def __post_init__(self) -> None: @@ -216,7 +219,10 @@ class CultMeshLocalServer: while not self._stop.is_set(): try: wire, remote_addr = rudp_socket.recvfrom(65535) - except TimeoutError: + except (TimeoutError, ConnectionResetError): + # Windows reports an earlier ICMP port-unreachable, from a peer that + # has gone away, as an error on the next receive. It names no datagram + # of this read, so it is idle, not the end of the thread. self._poll_rudp_resends(rudp_socket, peers) continue except OSError: @@ -245,20 +251,23 @@ class CultMeshLocalServer: ) ) peers[remote_addr] = peer - self._send_rudp_packet( + unsendable = self._send_rudp_packet( rudp_socket, remote_addr, peer.session.accept_connect(packet, now_ms, b"cultmesh-local-rudp"), ) + if unsendable is not None: + self._end_unsendable_peer(rudp_socket, peers, remote_addr, unsendable) continue if peer is None: continue result = peer.session.receive(packet, now_ms) - if result.reply is not None: - self._send_rudp_packet(rudp_socket, remote_addr, result.reply) - for ready in result.ready_to_send: - self._send_rudp_packet(rudp_socket, remote_addr, ready) + unsendable: OSError | None = None + for outgoing in (*([result.reply] if result.reply is not None else []), *result.ready_to_send): + unsendable = self._send_rudp_packet(rudp_socket, remote_addr, outgoing) + if unsendable is not None: + break if result.disconnected: peers.pop(remote_addr, None) continue @@ -273,18 +282,23 @@ class CultMeshLocalServer: continue responses = self._handle_connection_message(message, peer.subscriptions) for response in responses: - self._send_rudp_schema_frame( - rudp_socket, - remote_addr, - peer.session, - msgpack.packb(response, use_bin_type=True), - ) - if packet.reliable or packet.packet_type == CultNetRudpPacketType.DATA or result.delivered: - self._send_rudp_packet( + if unsendable is None: + unsendable = self._send_rudp_schema_frame( + rudp_socket, + remote_addr, + peer.session, + msgpack.packb(response, use_bin_type=True), + ) + if unsendable is None and ( + packet.reliable or packet.packet_type == CultNetRudpPacketType.DATA or result.delivered + ): + unsendable = self._send_rudp_packet( rudp_socket, remote_addr, peer.session.create_ack_for_received(packet.sequence), ) + if unsendable is not None: + self._end_unsendable_peer(rudp_socket, peers, remote_addr, unsendable) def _poll_rudp_resends( self, @@ -294,7 +308,10 @@ class CultMeshLocalServer: now_ms = _now_ms() for remote_addr, peer in list(peers.items()): for packet in peer.session.due_resends(now_ms): - self._send_rudp_packet(rudp_socket, remote_addr, packet) + unsendable = self._send_rudp_packet(rudp_socket, remote_addr, packet) + if unsendable is not None: + self._end_unsendable_peer(rudp_socket, peers, remote_addr, unsendable) + break def _send_rudp_schema_frame( self, @@ -302,27 +319,49 @@ class CultMeshLocalServer: remote_addr: tuple[str, int], session: CultNetRudpSession, payload: bytes, - ) -> None: + ) -> OSError | None: for packet in session.send_many( "schema", payload, CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=_now_ms()), ): - self._send_rudp_packet(rudp_socket, remote_addr, packet) + unsendable = self._send_rudp_packet(rudp_socket, remote_addr, packet) + if unsendable is not None: + return unsendable + return None def _send_rudp_packet( self, rudp_socket: socket.socket, remote_addr: tuple[str, int], packet: CultNetRudpPacket, - ) -> None: - # A failed send is that peer's lost datagram: counted, never raised. - # Raising here ended the RUDP thread for every peer. + ) -> OSError | None: + # A transient send failure is that peer's lost datagram: counted, never + # raised (raising here ended the RUDP thread for every peer). A permanent + # one (`is_permanent_send_error`) is returned: the datagram can never be + # sent as built, and the caller ends that peer's session. wire = encode_rudp_packet(packet) try: rudp_socket.sendto(wire, remote_addr) - except OSError: + except OSError as error: + if is_permanent_send_error(error): + return error self._rudp_send_failures += 1 + return None + + def _end_unsendable_peer( + self, + rudp_socket: socket.socket, + peers: dict[tuple[str, int], _RudpPeerConnection], + remote_addr: tuple[str, int], + error: OSError, + ) -> None: + """The session owes its peer a packet that can never be sent as built. It ends, + never the thread, and the peer's goodbye names the error.""" + peer = peers.pop(remote_addr, None) + if peer is None: + return + self._send_rudp_packet(rudp_socket, remote_addr, peer.session.end_unsendable_session(error)) def _handle_connection_message( self, diff --git a/packages/cultmesh-py/tests/test_cultmesh.py b/packages/cultmesh-py/tests/test_cultmesh.py index cccb2ecb..18243923 100644 --- a/packages/cultmesh-py/tests/test_cultmesh.py +++ b/packages/cultmesh-py/tests/test_cultmesh.py @@ -1,5 +1,6 @@ from __future__ import annotations +import errno import json import os import socket @@ -4558,6 +4559,98 @@ class CultMeshRudpSendFailureTests(unittest.TestCase): finally: server.stop() + def _served(self, y: CultNetRudpSocketTransportConnection) -> Any: + import msgpack # type: ignore + + y.send("schema", hello(runtime_id="probe").to_bytes()) + for _ in range(200): + response = y.receive_once() + if response is not None: + return msgpack.unpackb(response.payload, raw=False) + time.sleep(0.005) + return None + + def test_a_vanished_peer_does_not_kill_the_rudp_thread(self) -> None: + # On Windows the closed port of a peer that went away is reported as + # ConnectionResetError (WinError 10054) by the next receive. It can only fail there. + server = CultMesh.serve_node(CultMesh.create_node(runtime_id="mesh-server")) + try: + x, _ = self._client(server, "peer-x") + y, _ = self._client(server, "peer-y") + x.send("schema", hello(runtime_id="probe").to_bytes()) + x.socket.close() + time.sleep(1.0) # the server's reply and resends to X draw ICMP port-unreachable + + self.assertTrue(server._rudp_thread.is_alive()) + served = self._served(y) + self.assertIsNotNone(served, "Y is still served after X vanished") + self.assertEqual(served["runtimeId"], "mesh-server") + finally: + server.stop() + + def test_a_reset_reported_by_a_receive_is_idle_not_the_end_of_the_thread(self) -> None: + server = CultMesh.serve_node(CultMesh.create_node(runtime_id="mesh-server")) + original_recvfrom = socket.socket.recvfrom + resets = [3] + try: + y, _ = self._client(server, "peer-y") + + def recvfrom(sock: socket.socket, *args: Any) -> Any: + if sock is server._rudp_socket and resets[0] > 0: + resets[0] -= 1 + raise ConnectionResetError(errno.ECONNRESET, "injected reset") + return original_recvfrom(sock, *args) + + with patch.object(socket.socket, "recvfrom", recvfrom): + self._wait_for(lambda: resets[0] == 0) + self.assertTrue(server._rudp_thread.is_alive()) + self.assertIsNotNone(self._served(y)) + finally: + server.stop() + + def test_a_permanent_send_failure_ends_only_that_peers_session_and_the_goodbye_names_it(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") + 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): + # The server's response to X can never be sent as built. + x.send("schema", hello(runtime_id="probe").to_bytes()) + for _ in range(200): + x.receive_once() + if x.disconnect_reason is not None: + break + time.sleep(0.005) + self.assertIsNotNone(x.disconnect_reason, "X is told its session ended") + self.assertTrue(x.disconnect_reason.startswith(b"packet could not be sent: ")) + self.assertGreater(len(x.disconnect_reason), len(b"packet could not be sent: ")) + self.assertTrue(server._rudp_thread.is_alive()) + + # Y is unaffected. + self.assertFalse(x.connected) + self.assertIsNotNone(self._served(y)) + self.assertEqual(server.rudp_send_failures, 0) + finally: + server.stop() + + def test_healthy_sends_are_not_failures(self) -> None: + server = CultMesh.serve_node(CultMesh.create_node(runtime_id="mesh-server")) + try: + x, _ = self._client(server, "peer-x") + self.assertIsNotNone(self._served(x)) + self.assertEqual(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: diff --git a/packages/cultnet-py/src/cultnet_py/__init__.py b/packages/cultnet-py/src/cultnet_py/__init__.py index b1734f25..59693f29 100644 --- a/packages/cultnet-py/src/cultnet_py/__init__.py +++ b/packages/cultnet-py/src/cultnet_py/__init__.py @@ -122,6 +122,7 @@ from .transport import ( TcpFramedTransportConnection, compute_reconnect_delay_ms, create_reconnect_policy, + is_permanent_send_error, create_rudp_transport_profile, create_tcp_framed_transport_profile, decode_rudp_packet, @@ -208,6 +209,7 @@ __all__ = [ "CultNetRudpReceiveResult", "CultNetRudpSendOptions", "CultNetRudpSession", + "is_permanent_send_error", "CultNetRudpSessionOptions", "CultNetRudpSocketMode", "CultNetRudpReconnectLoop", diff --git a/packages/cultnet-py/src/cultnet_py/transport.py b/packages/cultnet-py/src/cultnet_py/transport.py index 355faa51..a0b17cd6 100644 --- a/packages/cultnet-py/src/cultnet_py/transport.py +++ b/packages/cultnet-py/src/cultnet_py/transport.py @@ -1,5 +1,6 @@ from __future__ import annotations +import errno import secrets import socket import threading @@ -435,6 +436,32 @@ def create_rudp_transport_profile( } +UNSENDABLE_PACKET_REASON = "packet could not be sent" + +_PERMANENT_SEND_ERRNOS = frozenset( + code + for code in ( + errno.EMSGSIZE, + errno.EINVAL, + errno.EAFNOSUPPORT, + getattr(errno, "WSAEMSGSIZE", None), + getattr(errno, "WSAEINVAL", None), + getattr(errno, "WSAEAFNOSUPPORT", None), + ) + if code is not None +) + + +def is_permanent_send_error(error: OSError) -> bool: + """Whether a failed `sendto` can never succeed for the datagram as built, so resending + it is pointless: it is too large for a datagram (EMSGSIZE), its address is malformed for + the socket (EINVAL), or belongs to another address family (EAFNOSUPPORT), with their + Winsock equivalents. Everything else (no route, full buffers, firewall drops, a refusal + reported for an earlier datagram, an interrupted call) is a property of the path or the + moment and may pass, so it is a lost datagram, not a reason to end a session.""" + return error.errno in _PERMANENT_SEND_ERRNOS + + class CultNetRudpSession: RELIABLE_SEND_WINDOW_PACKETS = 32 RECEIVED_SEQUENCE_WINDOW = 4_096 @@ -833,6 +860,14 @@ class CultNetRudpSession: def create_ping(self, payload: bytes = b"") -> CultNetRudpPacket: return self._create_packet(CultNetRudpPacketType.PING, "control", payload) + def end_unsendable_session(self, error: OSError) -> CultNetRudpPacket: + """Ends a session that owes its peer a packet that can never be sent as built + (see `is_permanent_send_error`). Resending it would fail forever, and dropping it + would leave a reliable sequence the peer waits on for good, so the session cannot + be kept. Returns the goodbye for the peer; its reason names the error.""" + self.reset_peer_state() + return self.create_disconnect(f"{UNSENDABLE_PACKET_REASON}: {error}".encode("utf-8")) + def create_disconnect(self, reason: bytes = b"") -> CultNetRudpPacket: self._end_session() return self._create_packet(CultNetRudpPacketType.DISCONNECT, "control", reason) diff --git a/packages/cultnet-py/tests/test_cultnet.py b/packages/cultnet-py/tests/test_cultnet.py index fb2eff54..41665f46 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/packages/cultnet-py/tests/test_cultnet.py @@ -877,6 +877,28 @@ class CultNetTests(unittest.TestCase): self.assertEqual(old_ack.ack, oldest_sequence) self.assertEqual(old_ack.ack_mask, 0) + def test_cultnet_rudp_only_a_datagram_that_can_never_be_sent_is_a_permanent_failure(self) -> None: + import errno + + from cultnet_py import is_permanent_send_error + + for code in (errno.EMSGSIZE, errno.EINVAL, errno.EAFNOSUPPORT): + self.assertTrue(is_permanent_send_error(OSError(code, "permanent")), errno.errorcode[code]) + for code in ( + errno.EPERM, + errno.EINTR, + errno.EAGAIN, + errno.EACCES, + errno.ENETDOWN, + errno.ENETUNREACH, + errno.EADDRNOTAVAIL, + errno.ENOBUFS, + errno.ECONNREFUSED, + errno.EHOSTUNREACH, + ): + self.assertFalse(is_permanent_send_error(OSError(code, "transient")), errno.errorcode[code]) + self.assertFalse(is_permanent_send_error(OSError("no errno"))) + def test_cultnet_rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames(self) -> None: server_socket = bind_udp_socket() client_socket = bind_udp_socket()