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 (Python): a packet one peer's session refuses ends that session, never the RUDP thread

Browse source 
session.receive raises ValueError on bad fragment metadata, and nothing
in _rudp_loop caught it, so one admitted peer could end the thread for
every peer. The server now ends that peer's session with "session refused
a packet", as the socket transport and the Rust servers do.

Tests pin that, and two D2 rules that were unpinned: a resend loop sends
nothing more to a peer it just ended (P1), and responses after one that
can never be sent are not sent (P3).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 15:03:11 +02:00
commit 06f8a31cb5
2 changed files with 146 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -262,7 +262,14 @@ class CultMeshLocalServer:
if peer is None:
continue
result = peer.session.receive(packet, now_ms)
try:
result = peer.session.receive(packet, now_ms)
except ValueError:
# receive() has already recorded the packet's reliable sequence, so the
# session cannot be kept: a retransmit would be acknowledged and the
# frame silently lost. It ends, never the thread, and the peer is told.
self._end_refused_peer(rudp_socket, peers, remote_addr)
continue
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)
@ -349,6 +356,19 @@ class CultMeshLocalServer:
self._rudp_send_failures += 1
return None
def _end_refused_peer(
self,
rudp_socket: socket.socket,
peers: dict[tuple[str, int], _RudpPeerConnection],
remote_addr: tuple[str, int],
) -> None:
peer = peers.pop(remote_addr)
# The goodbye is built after the reset, or its ack field would acknowledge the
# very frame the session refused.
peer.session.reset_peer_state()
goodbye = peer.session.create_disconnect(b"session refused a packet")
self._send_rudp_packet(rudp_socket, remote_addr, goodbye)
def _end_unsendable_peer(
self,
rudp_socket: socket.socket,

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

@ -4759,6 +4759,131 @@ class CultMeshRudpSendFailureTests(unittest.TestCase):
finally:
server.stop()
def test_a_packet_the_session_refuses_ends_only_that_peers_session(self) -> None:
from cultnet_py import CultNetRudpPacket, encode_rudp_packet
server = CultMesh.serve_node(CultMesh.create_node(runtime_id="mesh-server"))
try:
x, _ = self._client(server, "peer-x")
y, _ = self._client(server, "peer-y")
# A fragment whose metadata the session refuses: fragment id 0 names no set.
refused = CultNetRudpPacket(
packet_type=CultNetRudpPacketType.DATA,
connection_id=server.rudp_connection_id,
sequence=0,
ack=0,
ack_mask=0,
channel_id="schema",
reliable=False,
fragment_id=0,
fragment_index=0,
fragment_count=2,
payload=b"x",
)
x.socket.sendto(encode_rudp_packet(refused), ("127.0.0.1", server.port))
for _ in range(200):
x.receive_once()
if x.disconnect_reason is not None:
break
time.sleep(0.005)
self.assertEqual(x.disconnect_reason, b"session refused a packet")
self.assertTrue(server._rudp_thread.is_alive())
self.assertIsNotNone(self._served(y), "Y is still served")
finally:
server.stop()
def _raw_peer(self, server: Any) -> tuple[socket.socket, Any]:
"""A peer driven packet by packet, to deliver frames in an order a client would not."""
from cultnet_py import CultNetRudpSession, CultNetRudpSessionOptions, encode_rudp_packet
sock = bind_udp_socket()
sock.settimeout(0.05)
session = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=server.rudp_connection_id))
sock.sendto(encode_rudp_packet(session.create_connect(0, b"join")), ("127.0.0.1", server.port))
wire, _ = sock.recvfrom(65535)
accept = decode_rudp_packet(wire)
self.assertEqual(accept.packet_type, CultNetRudpPacketType.ACCEPT)
session.receive(accept, 0)
return sock, session
def test_responses_after_one_that_can_never_be_sent_are_not_sent(self) -> None:
from cultnet_py import CultNetRudpSendOptions, encode_rudp_packet
server = CultMesh.serve_node(CultMesh.create_node(runtime_id="mesh-server"))
original_sendto = socket.socket.sendto
try:
sock, session = self._raw_peer(server)
addr = sock.getsockname()[:2]
request = hello(runtime_id="probe").to_bytes()
options = CultNetRudpSendOptions(reliable=True, ordered=True)
first = session.send("schema", request, options)
second = session.send("schema", request, options)
attempts: list[CultNetRudpPacketType] = []
failures = [1]
def sendto(sock_: socket.socket, data: bytes, *args: Any) -> int:
if args and tuple(args[-1]) == tuple(addr):
packet_type = decode_rudp_packet(data).packet_type
attempts.append(packet_type)
if packet_type == CultNetRudpPacketType.DATA 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 second frame waits for the first, so one receive delivers both, and
# the first of the two responses can never be sent as built.
sock.sendto(encode_rudp_packet(second), ("127.0.0.1", server.port))
time.sleep(0.1)
attempts.clear()
sock.sendto(encode_rudp_packet(first), ("127.0.0.1", server.port))
self._wait_for(lambda: CultNetRudpPacketType.DISCONNECT in attempts)
time.sleep(0.1)
# The unsendable response, then only the goodbye: no second response, no ack.
self.assertEqual(attempts, [CultNetRudpPacketType.DATA, CultNetRudpPacketType.DISCONNECT])
self.assertTrue(server._rudp_thread.is_alive())
finally:
server.stop()
def test_a_peer_ended_in_the_resend_loop_is_sent_nothing_more_from_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")
mode = ["transient"]
attempts: list[CultNetRudpPacketType] = []
def sendto(sock: socket.socket, data: bytes, *args: Any) -> int:
if args and tuple(args[-1]) == tuple(x_addr):
packet_type = decode_rudp_packet(data).packet_type
if mode[0] != "transient":
attempts.append(packet_type)
# Only responses fail; X's requests are acknowledged, so X goes quiet.
if packet_type == CultNetRudpPacketType.DATA and mode[0] == "transient":
raise OSError(errno.ENOBUFS, "injected transient failure")
if packet_type == CultNetRudpPacketType.DATA 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.send("schema", hello(runtime_id="one").to_bytes())
x.send("schema", hello(runtime_id="two").to_bytes())
for _ in range(40):
x.receive_once()
time.sleep(0.005)
self._wait_for(lambda: server.rudp_send_failures >= 2)
# Both responses are owed; the next resend of the first can never be sent.
mode[0] = "permanent"
self._wait_for(lambda: CultNetRudpPacketType.DISCONNECT in attempts)
time.sleep(0.2)
self.assertEqual(attempts, [CultNetRudpPacketType.DATA, CultNetRudpPacketType.DISCONNECT])
self.assertTrue(server._rudp_thread.is_alive())
finally:
server.stop()
def _wait_for(self, predicate: Callable[[], bool]) -> None:
deadline = time.monotonic() + 2.0
while time.monotonic() < deadline: