GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Ack Cut D D1 and D2 (Python): a vanished peer is idle, and a send that can never succeed ends only that peer's session

Browse source 
D1: on Windows the closed port of a vanished peer is reported by the next
recvfrom as ConnectionResetError (WinError 10054), and the RUDP thread returned
on it. It is now idle, as in the Rust document server, the Rust hub and the C#
listener. Both tests fail on Windows without the fix and pass with it.

D2: is_permanent_send_error (EMSGSIZE, EINVAL, EAFNOSUPPORT and their Winsock
equivalents) separates a datagram that can never be sent as built from a lost
one. The server ends that peer's session with a goodbye whose reason names the
error and never the thread; transient failures stay counted losses.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 12:19:47 +02:00
commit 1c71d98c71
5 changed files with 213 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

83
packages/cultmesh-py/src/cultmesh_py/server.py
View file

@ -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,

93
packages/cultmesh-py/tests/test_cultmesh.py
View file

@ -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:

2
packages/cultnet-py/src/cultnet_py/__init__.py
View file

@ -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",

35
packages/cultnet-py/src/cultnet_py/transport.py
View file

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

22
packages/cultnet-py/tests/test_cultnet.py
View file

@ -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()