diff --git a/packages/cultnet-py/src/cultnet_py/transport.py b/packages/cultnet-py/src/cultnet_py/transport.py index 3462b047..0f378613 100644 --- a/packages/cultnet-py/src/cultnet_py/transport.py +++ b/packages/cultnet-py/src/cultnet_py/transport.py @@ -505,6 +505,25 @@ class CultNetRudpSession: self._track_reliable(response, now_ms) return response + def answer_repeated_connect(self, packet: CultNetRudpPacket, now_ms: int = 0) -> CultNetRudpPacket: + """Answers a Connect from a peer this session already accepted. + + The packet's sequence is remembered and nothing is queued, so a Connect + storm cannot grow the reliable queue. The reply is the Accept still awaiting + acknowledgement, or an Ack once it was acknowledged. + """ + self._require_connection(packet) + if packet.packet_type != CultNetRudpPacketType.CONNECT: + raise ValueError(f"Expected RUDP connect packet, got {packet.packet_type.value}") + self._apply_acknowledgements(packet) + self._remember_received(packet.sequence) + self._last_received_at_ms = now_ms + for pending in self._pending_reliable.values(): + if pending.packet.packet_type == CultNetRudpPacketType.ACCEPT: + pending.last_sent_at_ms = now_ms + return pending.packet + return self.create_ack() + def send( self, channel_id: str, @@ -975,6 +994,7 @@ class CultNetRudpSocketTransportConnection: def connect(self, payload: bytes = b"") -> None: if self.mode != CultNetRudpSocketMode.CLIENT: raise ValueError("Only a client RUDP socket transport can initiate connect") + self.disconnect_reason = None self._send_packet(self.session.create_connect(_now_ms(), payload)) def send(self, channel_id: str, payload: bytes) -> None: @@ -1024,12 +1044,18 @@ class CultNetRudpSocketTransportConnection: return None if self.mode == CultNetRudpSocketMode.SERVER and packet.packet_type == CultNetRudpPacketType.CONNECT: + # A Connect from the peer this transport already accepted repeats: + # answer it with the Accept already owed and queue nothing. try: - accept = self.session.accept_connect(packet, _now_ms()) + if self.session.connected: + reply = self.session.answer_repeated_connect(packet, _now_ms()) + else: + reply = self.session.accept_connect(packet, _now_ms()) + self.disconnect_reason = None except ValueError: self._packets_dropped += 1 return None - self._send_packet(accept) + self._send_packet(reply) return None try: @@ -1080,7 +1106,16 @@ class CultNetRudpSocketTransportConnection: self._delivered_frames.clear() self.socket.settimeout(poll_timeout) try: - while self.session.outstanding_reliable_packet_count > 0: + while True: + # An ended session forgot its unacknowledged writes; reporting + # them flushed would be a lie. + if self.disconnect_reason is not None: + raise ConnectionError( + "RUDP session ended before its reliable writes were acknowledged: " + + self.disconnect_reason.decode("utf-8", errors="replace") + ) + if self.session.outstanding_reliable_packet_count == 0: + return if time.monotonic() >= deadline: raise TimeoutError( "RUDP reliable flush timed out with " diff --git a/packages/cultnet-py/tests/test_cultnet.py b/packages/cultnet-py/tests/test_cultnet.py index 5c2680ef..57c24421 100644 --- a/packages/cultnet-py/tests/test_cultnet.py +++ b/packages/cultnet-py/tests/test_cultnet.py @@ -1079,6 +1079,63 @@ class CultNetTests(unittest.TestCase): peer_socket.close() server.close() + def _server_mode_pair(self, connection_id: int): + server_socket = bind_udp_socket() + peer_socket = bind_udp_socket() + server = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-server", + socket=server_socket, + mode=CultNetRudpSocketMode.SERVER, + connection_id=connection_id, + ) + ) + peer = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id)) + + def to_server(packet) -> None: + peer_socket.sendto(encode_rudp_packet(packet), server_socket.getsockname()) + + def drain_peer() -> list: + packets = [] + while True: + try: + wire, _ = peer_socket.recvfrom(65535) + except TimeoutError: + return packets + packets.append(decode_rudp_packet(wire)) + + return server, server_socket, peer_socket, peer, to_server, drain_peer + + def test_cultnet_rudp_flush_fails_for_a_write_the_ended_session_forgot(self) -> None: + server, _, peer_socket, peer, to_server, drain_peer = self._server_mode_pair(0x10203054) + try: + to_server(peer.create_connect(0, b"join")) + server.receive_once() + peer.receive([p for p in drain_peer() if p.packet_type == CultNetRudpPacketType.ACCEPT][0], 0) + server.send("schema", b"never acknowledged") + (frame,) = peer.send_many("schema", b"poison", CultNetRudpSendOptions(reliable=True, ordered=True)) + to_server(replace(frame, fragment_count=2, fragment_id=0)) + self.assertIsNone(server.receive_once()) + with self.assertRaises(ConnectionError): + server.flush_reliable(0.2) + finally: + peer_socket.close() + server.close() + + def test_cultnet_rudp_server_mode_owes_one_reliable_accept_however_many_connects_repeat(self) -> None: + server, _, peer_socket, peer, to_server, drain_peer = self._server_mode_pair(0x10203055) + try: + connect = peer.create_connect(0, b"join") + for _ in range(20): + to_server(connect) + server.receive_once() + accepts = {p.sequence for p in drain_peer() if p.packet_type == CultNetRudpPacketType.ACCEPT} + self.assertEqual(len(accepts), 1) + self.assertLessEqual(server.outstanding_reliable_packet_count, 1) + finally: + peer_socket.close() + server.close() + def test_cultnet_rudp_session_rejects_limits_that_cannot_admit_a_connect(self) -> None: with self.assertRaises(ValueError): CultNetRudpSession(CultNetRudpSessionOptions(connection_id=1, initial_sequence=0xFFFFFFFF))