cultnet-py: multi-peer RUDP socket server, with C#-captured transcript parity

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

packages/cultnet-py/CHANGELOG.md

@@ -2,6 +2,12 @@
 
 ## Unreleased (breaking; version 0.x, so the next minor)
 
+- Added `CultNetRudpSocketTransportServer` (`cultnet_py.rudp_server`): one UDP socket serving many RUDP peers, one session per
+  remote endpoint, ported from the C# `CultNetRudpSocketTransportServer` and held to its bytes by
+  `tests/vectors/rudp-multipeer-server/transcript.txt`. The caller drives it (`try_receive_once`, `poll_resends`). Like the C#
+  listener it never reaps idle peers. `CultNetRudpSession` gained `next_sequence`, `withdraw_send` and `end_with`, and
+  `CultNetTransportStats` gained `send_failures`.
+
 - Removed `resolve_document_and_schema_id_for_raw_record` from the package exports. A raw record resolves to one local schema id, the one
   its document registers, so there is no wire id to return. Use `resolve_document_for_raw_record(documents_by_schema_id, schema_id, record)`,
   which returns the document; take the schema id from `document.catalog_entry().schema_id`.

packages/cultnet-py/src/cultnet_py/__init__.py

@@ -103,6 +103,12 @@ from .shard_replication import (
 )
 from .snapshot import CultNetRawDocumentRecord, CultNetRawSnapshotResponse
 from .subscription import CultNetDatabaseChange
+from .rudp_server import (
+    CultNetRudpSocketServerFrame,
+    CultNetRudpSocketServerOptions,
+    CultNetRudpSocketServerPeer,
+    CultNetRudpSocketTransportServer,
+)
 from .transport import (
     CultNetRudpDeliveredFrame,
     CultNetRudpPacket,
@@ -211,6 +217,10 @@ __all__ = [
     "CultNetRudpReceiveResult",
     "CultNetRudpSendOptions",
     "CultNetRudpSession",
+    "CultNetRudpSocketServerFrame",
+    "CultNetRudpSocketServerOptions",
+    "CultNetRudpSocketServerPeer",
+    "CultNetRudpSocketTransportServer",
     "is_permanent_send_error",
     "send_error_code",
     "CultNetRudpSessionOptions",

packages/cultnet-py/src/cultnet_py/rudp_server.py

@@ -0,0 +1,345 @@
+"""Multi-peer RUDP listener: one UDP socket, one reliable session per remote endpoint.
+
+Matches the C# `CultNetRudpSocketTransportServer` (the reference): a Connect opens a peer's
+session, a Connect from the same endpoint that is not a repeat replaces only that peer, and
+a peer that ends, is refused or cannot be answered never disturbs the others. The caller
+drives it: `try_receive_once` consumes one datagram, `poll_resends` sends what is due.
+Everything runs on the caller's thread, so no session needs a lock.
+"""
+from __future__ import annotations
+
+from collections import deque
+from dataclasses import dataclass
+import socket
+
+from .transport import (
+    REFUSED_PACKET_REASON,
+    CultNetRudpPacket,
+    CultNetRudpPacketType,
+    CultNetRudpSession,
+    CultNetRudpSessionOptions,
+    CultNetReconnectPolicy,
+    CultNetTransportFrame,
+    CultNetTransportStats,
+    _channel_send_options,
+    _now_ms,
+    create_rudp_transport_profile,
+    decode_rudp_packet,
+    encode_rudp_packet,
+    is_permanent_send_error,
+)
+
+REPLACED_PEER_REASON = b"replaced by a new Connect generation"
+
+
+@dataclass
+class CultNetRudpSocketServerOptions:
+    runtime_id: str
+    socket: socket.socket
+    connection_id: int
+    initial_sequence: int | None = None
+    resend_delay_ms: int = 250
+    transport_id: str = "rudp"
+    max_payload_bytes: int | None = None
+    max_fragment_bytes: int | None = None
+    max_pending_reliable_packets: int | None = None
+    reconnect_policy: CultNetReconnectPolicy | None = None
+    accept_payload: bytes = b""
+
+
+class CultNetRudpSocketServerPeer:
+    """One remote endpoint's session. A peer that reconnects from the same endpoint is a new object."""
+
+    def __init__(self, remote_addr: tuple, connect_payload: bytes, session: CultNetRudpSession) -> None:
+        self.remote_addr = remote_addr
+        self.connect_payload = connect_payload
+        self.session = session
+        # The last transport-level reason the session ended with, once it has.
+        self.disconnect_reason: bytes | None = None
+
+    @property
+    def connected(self) -> bool:
+        return self.session.connected
+
+    @property
+    def pending_reliable_packet_count(self) -> int:
+        return self.session.outstanding_reliable_packet_count
+
+
+@dataclass(frozen=True)
+class CultNetRudpSocketServerFrame:
+    peer: CultNetRudpSocketServerPeer
+    frame: CultNetTransportFrame
+
+
+def _remote_key(addr: tuple) -> tuple:
+    return tuple(addr[:2])
+
+
+class CultNetRudpSocketTransportServer:
+    def __init__(self, options: CultNetRudpSocketServerOptions) -> None:
+        if not options.runtime_id.strip():
+            raise ValueError("Runtime id is required.")
+        self.socket = options.socket
+        self._options = options
+        host, port = self.socket.getsockname()[:2]
+        self.profile = create_rudp_transport_profile(
+            options.runtime_id,
+            transport_id=options.transport_id,
+            host=host,
+            port=port,
+            max_payload_bytes=options.max_payload_bytes,
+            max_fragment_bytes=options.max_fragment_bytes,
+            max_pending_reliable_packets=options.max_pending_reliable_packets,
+            reconnect_policy=options.reconnect_policy,
+        )
+        self._peers: dict[tuple, CultNetRudpSocketServerPeer] = {}
+        self._delivered: deque[CultNetRudpSocketServerFrame] = deque()
+        self._bytes_received = 0
+        self._bytes_sent = 0
+        self._frames_received = 0
+        self._frames_sent = 0
+        self._packets_dropped = 0
+        self._send_failures = 0
+        self._closed = False
+        self.peer_disconnected: list = []  # callbacks taking the peer
+
+    @property
+    def peers(self) -> tuple[CultNetRudpSocketServerPeer, ...]:
+        return tuple(self._peers.values())
+
+    @property
+    def stats(self) -> CultNetTransportStats:
+        return CultNetTransportStats(
+            bytes_received=self._bytes_received,
+            bytes_sent=self._bytes_sent,
+            frames_received=self._frames_received,
+            frames_sent=self._frames_sent,
+            packets_dropped=self._packets_dropped,
+            send_failures=self._send_failures,
+        )
+
+    def send(self, peer: CultNetRudpSocketServerPeer, channel_id: str, payload: bytes) -> None:
+        """Sends to one peer. Raises when that peer's session has ended or was replaced, or when a
+        datagram can never be sent as built: the first fragment failing queues nothing, a later one
+        ends the session, as the peer has seen part of the frame."""
+        if self._peers.get(_remote_key(peer.remote_addr)) is not peer:
+            raise ConnectionError("RUDP peer is not connected.")
+        first_sequence = peer.session.next_sequence
+        packets = peer.session.send_many(
+            channel_id,
+            payload,
+            _channel_send_options(channel_id, _now_ms()),
+            max_fragment_bytes=self._options.max_fragment_bytes,
+        )
+        for index, packet in enumerate(packets):
+            error = self._send_packet(peer.remote_addr, packet)
+            if error is None:
+                continue
+            if index == 0:
+                peer.session.withdraw_send(first_sequence)
+            else:
+                self._end_unsendable_peer(peer, error)
+            raise error
+        self._frames_sent += 1
+
+    def send_schema(self, peer: CultNetRudpSocketServerPeer, payload: bytes) -> None:
+        self.send(peer, "schema", payload)
+
+    def broadcast(self, channel_id: str, payload: bytes) -> list[CultNetRudpSocketServerPeer]:
+        """Sends to every connected peer and returns the peers that could not be sent to; one
+        peer's failure never stops the others."""
+        failed = []
+        for peer in self.peers:
+            try:
+                self.send(peer, channel_id, payload)
+            except (OSError, ConnectionError, ValueError):
+                failed.append(peer)
+        return failed
+
+    def disconnect(self, peer: CultNetRudpSocketServerPeer, reason: bytes = b"") -> bool:
+        """Ends one peer's session, telling it why. False when the session had already ended."""
+        return self.end_peer_session(peer, reason)
+
+    def end_peer_session(self, peer: CultNetRudpSocketServerPeer, reason: bytes) -> bool:
+        """Ends the session of a peer whose delivered frame the application could not decode or
+        handle, with a goodbye carrying the application's reason. The frame is already
+        acknowledged, so the session cannot be kept; the peer's frames still queued are not
+        delivered. A session that already ended owes no goodbye, since the endpoint may belong
+        to a new session; the call then returns False."""
+        self._packets_dropped += 1
+        if peer.session.ended:
+            return False
+        self._send_packet(peer.remote_addr, peer.session.end_with(reason))
+        self._delivered = deque(item for item in self._delivered if item.peer is not peer)
+        self._retire(peer, reason)
+        return True
+
+    def receive_once(self) -> CultNetRudpSocketServerFrame | None:
+        return self.try_receive_once()[1]
+
+    def try_receive_once(self) -> tuple[bool, CultNetRudpSocketServerFrame | None]:
+        """Consumes one transport work item. Returns whether progress was made, independently of
+        whether it completed an application frame."""
+        if self._delivered:
+            return True, self._delivered.popleft()
+        try:
+            wire, remote = self.socket.recvfrom(65535)
+        except (TimeoutError, BlockingIOError, ConnectionResetError):
+            return False, None
+        self._bytes_received += len(wire)
+
+        try:
+            packet = decode_rudp_packet(wire)
+        except ValueError:
+            self._packets_dropped += 1
+            return True, None
+        if packet.connection_id != self._options.connection_id:
+            self._packets_dropped += 1
+            return True, None
+
+        key = _remote_key(remote)
+        if packet.packet_type == CultNetRudpPacketType.CONNECT:
+            self._receive_connect(packet, remote, key)
+            return True, None
+        peer = self._peers.get(key)
+        if peer is None:
+            self._packets_dropped += 1
+            return True, None
+        self._receive_from_peer(peer, key, packet)
+        return True, (self._delivered.popleft() if self._delivered else None)
+
+    def poll_resends(self) -> None:
+        """Sends the reliable packets whose resend timers are due, for every peer."""
+        for peer in self.peers:
+            if self._peers.get(_remote_key(peer.remote_addr)) is not peer:
+                continue
+            for packet in peer.session.due_resends(_now_ms()):
+                error = self._send_packet(peer.remote_addr, packet)
+                if error is not None:
+                    self._end_unsendable_peer(peer, error)
+                    break
+
+    def close(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        self.socket.close()
+
+    def __enter__(self) -> "CultNetRudpSocketTransportServer":
+        return self
+
+    def __exit__(self, *_: object) -> None:
+        self.close()
+
+    def _receive_connect(self, packet: CultNetRudpPacket, remote: tuple, key: tuple) -> None:
+        admitted = self._peers.get(key)
+        accept_payload = self._options.accept_payload
+        # The admitted session decides whether this Connect repeats the one it accepted: a repeat
+        # gets the Accept already owed and queues nothing. Any other Connect is a new generation
+        # and ends the admitted session here.
+        if admitted is not None:
+            if admitted.session.connect_repeats(packet):
+                try:
+                    reply = admitted.session.accept_connect(packet, _now_ms(), accept_payload)
+                except ValueError:
+                    self._packets_dropped += 1
+                    return
+                error = self._send_packet(admitted.remote_addr, reply)
+                if error is not None:
+                    self._end_unsendable_peer(admitted, error)
+                return
+            admitted.session.reset_peer_state()
+
+        peer = CultNetRudpSocketServerPeer(
+            tuple(remote),
+            packet.payload,
+            CultNetRudpSession(
+                CultNetRudpSessionOptions(
+                    connection_id=self._options.connection_id,
+                    initial_sequence=self._options.initial_sequence,
+                    resend_delay_ms=self._options.resend_delay_ms,
+                    max_pending_reliable_packets=self._options.max_pending_reliable_packets,
+                )
+            ),
+        )
+        try:
+            accept = peer.session.accept_connect(packet, _now_ms(), accept_payload)
+        except ValueError:
+            self._packets_dropped += 1
+            accept = None
+        # A peer that cannot be answered gets no session, and the one it replaced has already ended.
+        if accept is None or self._send_packet(peer.remote_addr, accept) is not None:
+            self._peers.pop(key, None)
+            if accept is not None:
+                self._packets_dropped += 1
+        else:
+            self._peers[key] = peer
+        if admitted is not None:
+            admitted.disconnect_reason = REPLACED_PEER_REASON
+            self._notify_disconnected(admitted)
+
+    def _receive_from_peer(self, peer: CultNetRudpSocketServerPeer, key: tuple, packet: CultNetRudpPacket) -> None:
+        session = peer.session
+        try:
+            result = 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.
+            self._packets_dropped += 1
+            self._send_packet(peer.remote_addr, session.end_with(REFUSED_PACKET_REASON))
+            self._retire(peer, REFUSED_PACKET_REASON)
+            return
+        error = None
+        if result.reply is not None:
+            error = self._send_packet(peer.remote_addr, result.reply)
+        for ready in result.ready_to_send:
+            if error is None:
+                error = self._send_packet(peer.remote_addr, ready)
+        if result.disconnected:
+            self._retire(peer, result.disconnect_reason)
+            return
+        for frame in result.delivered:
+            self._delivered.append(
+                CultNetRudpSocketServerFrame(peer, CultNetTransportFrame(frame.channel_id, frame.payload))
+            )
+            self._frames_received += 1
+        if error is None and packet.packet_type == CultNetRudpPacketType.DATA:
+            error = self._send_packet(peer.remote_addr, session.create_ack_for_received(packet.sequence))
+        if error is not None:
+            self._end_unsendable_peer(peer, error)
+
+    def _end_unsendable_peer(self, peer: CultNetRudpSocketServerPeer, error: OSError) -> None:
+        """An admitted peer owes a packet that can never be sent as built: its session ends, never
+        the poll, the peer is told, and the disconnect reason names the error. A session that has
+        already ended owes no goodbye, since the endpoint may belong to a new session."""
+        if peer.session.ended:
+            return
+        goodbye = peer.session.end_unsendable_session(error)
+        self._send_packet(peer.remote_addr, goodbye)
+        self._retire(peer, goodbye.payload)
+
+    def _retire(self, peer: CultNetRudpSocketServerPeer, reason: bytes | None) -> None:
+        # A peer a new Connect already replaced was reported then; its key now names the successor.
+        key = _remote_key(peer.remote_addr)
+        if self._peers.get(key) is not peer:
+            return
+        del self._peers[key]
+        peer.disconnect_reason = reason
+        self._notify_disconnected(peer)
+
+    def _notify_disconnected(self, peer: CultNetRudpSocketServerPeer) -> None:
+        for callback in list(self.peer_disconnected):
+            callback(peer)
+
+    def _send_packet(self, remote_addr: tuple, packet: CultNetRudpPacket) -> OSError | None:
+        """A transient failure is that peer's lost datagram: counted, never raised. A permanent
+        one (`is_permanent_send_error`) is returned for the caller to refuse or end the session."""
+        wire = encode_rudp_packet(packet)
+        try:
+            self._bytes_sent += self.socket.sendto(wire, remote_addr)
+        except OSError as error:
+            if is_permanent_send_error(error):
+                return error
+            self._send_failures += 1
+        return None

packages/cultnet-py/src/cultnet_py/transport.py

@@ -29,6 +29,9 @@ class CultNetTransportStats:
     # transport: malformed frames, another session's connection id, a sender
     # that is not the peer, or a packet the session refuses.
     packets_dropped: int = 0
+    # Datagrams a multi-peer server could not send to one peer and lost: each is
+    # a transient failure, never a reason to end a session.
+    send_failures: int = 0
 
 
 @dataclass(frozen=True)
@@ -437,6 +440,7 @@ def create_rudp_transport_profile(
 
 
 UNSENDABLE_PACKET_REASON = "packet could not be sent"
+REFUSED_PACKET_REASON = b"session refused a packet"
 
 # The fixed name of each permanent send failure, keyed by errno, Winsock equivalents included.
 _PERMANENT_SEND_ERROR_CODES = {
@@ -600,6 +604,28 @@ class CultNetRudpSession:
         self._fragment_buffers.clear()
         self._fragment_sets_evicted = 0
 
+    @property
+    def next_sequence(self) -> int:
+        return self._next_sequence
+
+    def withdraw_send(self, first_sequence: int) -> None:
+        """Undoes a send that never reached the wire: every packet it created at or after
+        `first_sequence`, pending or queued, leaves no trace and its sequences are reissued,
+        so the peer never sees a gap. Only valid while nothing else has been sent since."""
+        for sequence in [s for s in self._pending_reliable if s >= first_sequence]:
+            del self._pending_reliable[sequence]
+        kept = [p for p in self._queued_reliable if p.sequence < first_sequence]
+        self._queued_reliable.clear()
+        self._queued_reliable.extend(kept)
+        self._next_sequence = first_sequence
+
+    def end_with(self, reason: bytes) -> CultNetRudpPacket:
+        """Ends a session whose peer's frame was refused or failed after the session acknowledged
+        it. The peer's sequences are forgotten before the goodbye is built, or its ack field would
+        acknowledge the very frame that was refused."""
+        self.reset_peer_state()
+        return self.create_disconnect(reason)
+
     def create_connect(self, now_ms: int = 0, payload: bytes = b"") -> CultNetRudpPacket:
         # A session that has had a peer starts a new generation: nothing it
         # learned from that peer describes the one this Connect reaches, and
@@ -1397,11 +1423,8 @@ class CultNetRudpSocketTransportConnection:
         self.socket.close()
 
     def _end_refused_session(self) -> None:
-        reason = b"session refused a packet"
-        # The goodbye is built after the reset, or its ack field would acknowledge
-        # the very frame the session refused.
-        self.session.reset_peer_state()
-        goodbye = self.session.create_disconnect(reason)
+        reason = REFUSED_PACKET_REASON
+        goodbye = self.session.end_with(reason)
         self.disconnect_reason = reason
         try:
             self._send_packet(goodbye)

packages/cultnet-py/tests/test_rudp_multipeer_server.py

@@ -0,0 +1,350 @@
+"""One UDP socket serves many peers, each in its own session, as the C# reference does.
+
+Sockets bind port 0, so no test races another for a port.
+"""
+from __future__ import annotations
+
+import errno
+import socket
+import time
+import unittest
+from pathlib import Path
+
+from cultnet_py.rudp_server import (
+    CultNetRudpSocketServerOptions,
+    CultNetRudpSocketTransportServer,
+)
+from cultnet_py.transport import (
+    CultNetRudpPacketType,
+    CultNetRudpSendOptions,
+    CultNetRudpSession,
+    CultNetRudpSessionOptions,
+    decode_rudp_packet,
+    encode_rudp_packet,
+)
+
+CONNECTION_ID = 0x43554C54
+VECTORS = Path(__file__).resolve().parents[3] / "tests" / "vectors" / "rudp-multipeer-server" / "transcript.txt"
+
+
+def bind() -> socket.socket:
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sock.bind(("127.0.0.1", 0))
+    sock.settimeout(0.03)
+    return sock
+
+
+def make_server(**kwargs) -> CultNetRudpSocketTransportServer:
+    options = dict(runtime_id="py-test", socket=bind(), connection_id=CONNECTION_ID, resend_delay_ms=5)
+    options.update(kwargs)
+    return CultNetRudpSocketTransportServer(CultNetRudpSocketServerOptions(**options))
+
+
+def pump(server: CultNetRudpSocketTransportServer) -> list:
+    """Consumes datagrams until the socket is idle, returning the frames delivered."""
+    frames = []
+    while True:
+        progressed, frame = server.try_receive_once()
+        if frame is not None:
+            frames.append(frame)
+        if not progressed:
+            return frames
+
+
+def drain(sock: socket.socket) -> list[bytes]:
+    wires = []
+    while True:
+        try:
+            wires.append(sock.recvfrom(65535)[0])
+        except TimeoutError:
+            return wires
+
+
+class Client:
+    def __init__(self, server: CultNetRudpSocketTransportServer, initial_sequence: int) -> None:
+        self.server = server
+        self.socket = bind()
+        self.session = CultNetRudpSession(
+            CultNetRudpSessionOptions(connection_id=CONNECTION_ID, initial_sequence=initial_sequence)
+        )
+
+    @property
+    def addr(self) -> tuple:
+        return self.socket.getsockname()
+
+    def transmit(self, packet) -> None:
+        self.socket.sendto(encode_rudp_packet(packet), self.server.socket.getsockname())
+
+    def connect(self, payload: bytes = b"") -> None:
+        self.transmit(self.session.create_connect(0, payload))
+        pump(self.server)
+        self.receive()
+
+    def receive(self) -> list:
+        packets = [decode_rudp_packet(wire) for wire in drain(self.socket)]
+        for packet in packets:
+            if self.session.connected or packet.packet_type == CultNetRudpPacketType.ACCEPT:
+                self.session.receive(packet, 0)
+        return packets
+
+    def write(self, text: str) -> None:
+        self.transmit(
+            self.session.send("schema", text.encode(), CultNetRudpSendOptions(reliable=True, ordered=True))
+        )
+
+    def peer(self):
+        return next(p for p in self.server.peers if p.remote_addr == self.addr)
+
+    def close(self) -> None:
+        self.socket.close()
+
+
+class MultiPeerServerTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.server = make_server(initial_sequence=500, accept_payload=b"acc")
+        self.addCleanup(self.server.close)
+        self.clients: list[Client] = []
+
+    def client(self, initial_sequence: int) -> Client:
+        client = Client(self.server, initial_sequence)
+        self.clients.append(client)
+        self.addCleanup(client.close)
+        return client
+
+    def test_interleaved_peers_each_get_their_own_session_and_frames(self) -> None:
+        x, y = self.client(10), self.client(20)
+        x.connect(b"hx")
+        y.connect(b"hy")
+        self.assertEqual(len(self.server.peers), 2)
+        self.assertEqual(x.peer().connect_payload, b"hx")
+        self.assertEqual(y.peer().connect_payload, b"hy")
+
+        x.write("x1")
+        y.write("y1")
+        x.write("x2")
+        y.write("y2")
+        frames = pump(self.server)
+        by_peer = {}
+        for item in frames:
+            by_peer.setdefault(item.peer.remote_addr, []).append(item.frame.payload)
+        self.assertEqual(by_peer, {x.addr: [b"x1", b"x2"], y.addr: [b"y1", b"y2"]})
+
+        self.server.send(x.peer(), "schema", b"to-x")
+        self.server.send(y.peer(), "schema", b"to-y")
+        for client, expected in ((x, b"to-x"), (y, b"to-y")):
+            data = [p for p in client.receive() if p.packet_type == CultNetRudpPacketType.DATA]
+            self.assertEqual([p.payload for p in data], [expected])
+
+    def test_one_peer_ending_does_not_disturb_the_other(self) -> None:
+        x, y = self.client(10), self.client(20)
+        x.connect()
+        y.connect()
+        y_peer = y.peer()
+        ended = []
+        self.server.peer_disconnected.append(ended.append)
+
+        x.transmit(x.session.create_disconnect(b"bye"))
+        pump(self.server)
+        self.assertEqual(len(self.server.peers), 1)
+        self.assertEqual([p.disconnect_reason for p in ended], [b"bye"])
+        self.assertIs(y.peer(), y_peer)
+
+        # Datagrams from the ended endpoint belong to no session.
+        before = self.server.stats.packets_dropped
+        x_late = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=CONNECTION_ID, initial_sequence=99))
+        x_late.accept_connect(x.session.create_connect(0), 0)
+        x.transmit(x_late.send("schema", b"late", CultNetRudpSendOptions(reliable=True, ordered=True)))
+        pump(self.server)
+        self.assertEqual(self.server.stats.packets_dropped, before + 1)
+
+        y.write("still here")
+        self.assertEqual([f.frame.payload for f in pump(self.server)], [b"still here"])
+        self.server.send(y_peer, "schema", b"reply")
+        self.assertEqual(
+            [p.payload for p in y.receive() if p.packet_type == CultNetRudpPacketType.DATA], [b"reply"]
+        )
+
+    def test_a_new_connect_from_the_same_endpoint_replaces_only_that_peer(self) -> None:
+        x, y = self.client(10), self.client(20)
+        x.connect()
+        y.connect()
+        old_y = y.peer()
+        ended = []
+        self.server.peer_disconnected.append(ended.append)
+
+        y.session = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=CONNECTION_ID, initial_sequence=30))
+        y.connect(b"again")
+        self.assertEqual(ended, [old_y])
+        self.assertEqual(old_y.disconnect_reason, b"replaced by a new Connect generation")
+        self.assertIsNot(y.peer(), old_y)
+        self.assertEqual(y.peer().connect_payload, b"again")
+        with self.assertRaises(ConnectionError):
+            self.server.send(old_y, "schema", b"to the old session")
+        self.assertEqual(len(self.server.peers), 2)
+
+    def test_a_repeated_connect_is_answered_with_the_owed_accept_and_starts_nothing(self) -> None:
+        x = self.client(10)
+        connect = x.session.create_connect(0, b"h")
+        x.transmit(connect)
+        pump(self.server)
+        first = x.peer()
+        x.transmit(connect)
+        pump(self.server)
+        self.assertIs(x.peer(), first)
+        accepts = [p for p in x.receive() if p.packet_type == CultNetRudpPacketType.ACCEPT]
+        self.assertEqual(len(accepts), 2)
+        self.assertEqual(accepts[0].sequence, accepts[1].sequence)
+
+    def test_strays_are_dropped_and_counted(self) -> None:
+        x = self.client(10)
+        x.socket.sendto(b"not rudp", self.server.socket.getsockname())
+        other = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=CONNECTION_ID + 1, initial_sequence=1))
+        x.transmit(other.create_connect(0))
+        x.transmit(
+            CultNetRudpSession(CultNetRudpSessionOptions(connection_id=CONNECTION_ID, initial_sequence=1)).create_ping()
+        )
+        pump(self.server)
+        self.assertEqual(self.server.stats.packets_dropped, 3)
+        self.assertEqual(self.server.peers, ())
+
+    def test_resends_go_only_to_peers_with_unacknowledged_writes(self) -> None:
+        x, y = self.client(10), self.client(20)
+        x.connect()
+        y.connect()
+        self.server.send(x.peer(), "schema", b"owed")
+        x.receive()  # the first transmission is lost to the test: x never acks
+        y.receive()
+        time.sleep(0.02)
+        self.server.poll_resends()
+        x_data = [p for p in x.receive() if p.packet_type == CultNetRudpPacketType.DATA]
+        self.assertEqual([p.payload for p in x_data], [b"owed"])
+        self.assertEqual([p for p in y.receive() if p.packet_type == CultNetRudpPacketType.DATA], [])
+
+    def test_broadcast_reaches_every_peer(self) -> None:
+        x, y = self.client(10), self.client(20)
+        x.connect()
+        y.connect()
+        self.assertEqual(self.server.broadcast("schema", b"all"), [])
+        for client in (x, y):
+            self.assertEqual(
+                [p.payload for p in client.receive() if p.packet_type == CultNetRudpPacketType.DATA], [b"all"]
+            )
+
+    def test_an_application_ending_a_session_tells_the_peer_and_drops_its_queued_frames(self) -> None:
+        x, y = self.client(10), self.client(20)
+        x.connect()
+        y.connect()
+        x.write("x1")
+        y.write("y1")
+        # Both frames are decoded into the server's queue; ending x discards x's.
+        progressed, first = self.server.try_receive_once()
+        self.assertTrue(progressed)
+        pump_rest = pump(self.server)
+        frames = [first] + pump_rest if first else pump_rest
+        peer_x = x.peer()
+        self.assertTrue(self.server.end_peer_session(peer_x, b"server could not handle a message"))
+        self.assertFalse(self.server.end_peer_session(peer_x, b"again"), "an ended session owes no second goodbye")
+        goodbye = [p for p in x.receive() if p.packet_type == CultNetRudpPacketType.DISCONNECT]
+        self.assertEqual([p.payload for p in goodbye], [b"server could not handle a message"])
+        self.assertEqual(len(self.server.peers), 1)
+        self.assertEqual({f.peer.remote_addr for f in frames}, {x.addr, y.addr})
+
+    def test_a_datagram_too_large_to_send_is_that_callers_error_and_queues_nothing(self) -> None:
+        x = self.client(10)
+        x.connect()
+        peer = x.peer()
+        owed = peer.pending_reliable_packet_count  # the Accept, not yet acknowledged
+        with self.assertRaises(OSError) as raised:
+            self.server.send(peer, "schema", bytes(70_000))
+        self.assertEqual(raised.exception.errno, errno.EMSGSIZE)
+        self.assertEqual(peer.pending_reliable_packet_count, owed)
+        self.assertEqual(len(self.server.peers), 1)
+        # The refused write left no gap in the sequence space the peer is waiting on.
+        self.server.send(peer, "schema", b"fine")
+        data = [p for p in x.receive() if p.packet_type == CultNetRudpPacketType.DATA]
+        self.assertEqual([(p.sequence, p.payload) for p in data], [(501, b"fine")])
+
+
+class UnsendableAfterFirstFragmentSocket:
+    """Delegates to a real socket; fails permanently once its budget of sends to one address is spent."""
+
+    def __init__(self, real: socket.socket, addr: tuple, budget: int) -> None:
+        self._real, self._addr, self._budget = real, addr, budget
+
+    def sendto(self, wire: bytes, addr: tuple) -> int:
+        if tuple(addr) == self._addr:
+            if self._budget == 0:
+                raise OSError(errno.EMSGSIZE, "Message too long")
+            self._budget -= 1
+        return self._real.sendto(wire, addr)
+
+    def __getattr__(self, name: str):
+        return getattr(self._real, name)
+
+
+class UnsendableSessionTests(unittest.TestCase):
+    def test_a_later_fragment_that_cannot_be_sent_ends_that_session_only(self) -> None:
+        server = make_server(initial_sequence=500, max_fragment_bytes=8)
+        self.addCleanup(server.close)
+        x, y = Client(server, 10), Client(server, 20)
+        self.addCleanup(x.close)
+        self.addCleanup(y.close)
+        x.connect()
+        y.connect()
+        x_peer = x.peer()
+        server.socket = UnsendableAfterFirstFragmentSocket(server.socket, x.addr, budget=1)
+        ended = []
+        server.peer_disconnected.append(ended.append)
+
+        with self.assertRaises(OSError):
+            server.send(x_peer, "schema", bytes(24))
+        self.assertEqual(ended, [x_peer])
+        self.assertEqual(x_peer.disconnect_reason, b"packet could not be sent: EMSGSIZE")
+        self.assertEqual(len(server.peers), 1)
+        # The goodbye could not be sent either (same address), so the peer is only retired; y is unharmed.
+        server.send(y.peer(), "schema", b"ok")
+        self.assertEqual(
+            [p.payload for p in y.receive() if p.packet_type == CultNetRudpPacketType.DATA], [b"ok"]
+        )
+
+
+class ReferenceTranscriptTests(unittest.TestCase):
+    """Replays the datagrams the C# reference listener was captured answering, byte for byte."""
+
+    def test_the_listener_answers_with_the_bytes_the_reference_does(self) -> None:
+        server = make_server(
+            connection_id=0x43554C54, initial_sequence=500, max_fragment_bytes=8, accept_payload=b"acc", resend_delay_ms=250
+        )
+        self.addCleanup(server.close)
+        sockets = {label: bind() for label in ("X", "Y")}
+        for sock in sockets.values():
+            self.addCleanup(sock.close)
+        target = server.socket.getsockname()
+
+        steps: list[tuple[list[str], list[tuple[str, str]]]] = []
+        for line in VECTORS.read_text().splitlines():
+            if line.startswith("#") or not line.strip():
+                continue
+            fields = line.split(" ")
+            if fields[0] == "out":
+                steps[-1][1].append((fields[1], fields[2]))
+            else:
+                steps.append((fields, []))
+        self.assertGreater(len(steps), 10)
+
+        for action, expected in steps:
+            if action[0] == "in":
+                sockets[action[1]].sendto(bytes.fromhex(action[2]), target)
+                while not server.try_receive_once()[0]:
+                    time.sleep(0.001)
+            else:
+                peer = next(p for p in server.peers if p.remote_addr == sockets[action[1]].getsockname())
+                server.send(peer, action[2], bytes.fromhex(action[3]))
+            actual = [(label, wire.hex()) for label in ("X", "Y") for wire in drain(sockets[label])]
+            self.assertEqual(actual, expected, f"after {' '.join(action[:2])}")
+        self.assertEqual(len(server.peers), 1)
+        self.assertEqual(server.stats.packets_dropped, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/GameCult.Networking.Tests/CultNetRudpMultiPeerVectorTests.cs

@@ -0,0 +1,155 @@
+#nullable enable
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using NUnit.Framework;
+
+namespace GameCult.Networking.Tests
+{
+    /// <summary>
+    /// The multi-peer listener's golden in tests/vectors/rudp-multipeer-server: two peers on one socket, the
+    /// datagrams each sends and the datagrams the C# reference answers with. The C# reference is the only writer
+    /// (CULTNET_WRITE_VECTORS=1); every other run replays the scenario and compares it byte for byte. Other runtimes'
+    /// listeners replay the same transcript and are held to the same answers.
+    /// </summary>
+    public class CultNetRudpMultiPeerVectorTests
+    {
+        private const string WriteVariable = "CULTNET_WRITE_VECTORS";
+        private const uint ConnectionId = 0x43554c54;
+
+        private sealed class Client
+        {
+            public string Label = "";
+            public Socket Socket = null!;
+            public CultNetRudpSession Session = null!;
+        }
+
+        private static string Hex(byte[] bytes) => Convert.ToHexString(bytes).ToLowerInvariant();
+
+        private static Socket Bind()
+        {
+            var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
+            socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
+            return socket;
+        }
+
+        [Test]
+        public void TheReferenceListenerAnswersTwoPeersWithThePinnedBytes()
+        {
+            var transcript = Run();
+            var path = Path.Combine(VectorDirectory(), "transcript.txt");
+            if (Environment.GetEnvironmentVariable(WriteVariable) == "1")
+            {
+                File.WriteAllText(path, transcript);
+                Assert.Pass("wrote " + path);
+            }
+            Assert.That(transcript, Is.EqualTo(File.ReadAllText(path)));
+        }
+
+        private static string VectorDirectory()
+        {
+            var directory = new DirectoryInfo(TestContext.CurrentContext.TestDirectory);
+            while (directory != null && !Directory.Exists(Path.Combine(directory.FullName, "tests", "vectors")))
+                directory = directory.Parent;
+            if (directory == null) throw new InvalidOperationException("tests/vectors not found above the test directory.");
+            return Path.Combine(directory.FullName, "tests", "vectors", "rudp-multipeer-server");
+        }
+
+        private static string Run()
+        {
+            var log = new StringBuilder();
+            log.Append("# config connection_id=0x43554c54 initial_sequence=500 max_fragment_bytes=8 accept_payload=616363\n");
+            log.Append("# in <peer> <datagram> | send <peer> <channel> <payload> | out <peer> <datagram>; peer labels name remote endpoints\n");
+            var serverSocket = Bind();
+            using var server = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions
+            {
+                RuntimeId = "csharp-vector",
+                Socket = serverSocket,
+                ConnectionId = ConnectionId,
+                InitialSequence = 500,
+                MaxFragmentBytes = 8,
+                AcceptPayload = Encoding.ASCII.GetBytes("acc")
+            });
+            var target = serverSocket.LocalEndPoint!;
+            Client Make(string label, uint sequence) => new Client
+            {
+                Label = label,
+                Socket = Bind(),
+                Session = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId, InitialSequence = sequence })
+            };
+            var x = Make("X", 10);
+            var y = Make("Y", 20);
+            var clients = new[] { x, y };
+
+            void Deliver(Client client, CultNetRudpPacket packet)
+            {
+                var wire = CultNetRudpPacketCodec.Encode(packet);
+                log.Append($"in {client.Label} {Hex(wire)}\n");
+                client.Socket.SendTo(wire, target);
+                while (!server.TryReceiveOnce(out _))
+                    System.Threading.Thread.Sleep(1);
+                Flush();
+            }
+
+            void Flush()
+            {
+                foreach (var client in clients)
+                {
+                    var buffer = new byte[65535];
+                    EndPoint from = new IPEndPoint(IPAddress.Any, 0);
+                    while (client.Socket.Poll(30_000, SelectMode.SelectRead))
+                    {
+                        var received = client.Socket.ReceiveFrom(buffer, ref from);
+                        var wire = buffer.AsSpan(0, received).ToArray();
+                        log.Append($"out {client.Label} {Hex(wire)}\n");
+                        // Clients that stay connected take the server's packets so their acks stay true.
+                        var packet = CultNetRudpPacketCodec.Decode(wire);
+                        if (client.Session.Connected || packet.PacketType == CultNetRudpPacketType.Accept)
+                            client.Session.Receive(packet, 0);
+                    }
+                }
+            }
+
+            CultNetRudpSocketServerPeer PeerOf(Client client) =>
+                server.Peers.Single(p => p.RemoteEndPoint.Equals(client.Socket.LocalEndPoint));
+
+            void Send(Client client, string channel, byte[] payload)
+            {
+                log.Append($"send {client.Label} {channel} {Hex(payload)}\n");
+                server.Send(PeerOf(client), channel, payload);
+                Flush();
+            }
+
+            CultNetRudpPacket Data(Client client, string text) =>
+                client.Session.Send("schema", Encoding.ASCII.GetBytes(text), new CultNetRudpSendOptions { Reliable = true, Ordered = true });
+
+            var xConnect = x.Session.CreateConnect(0, Encoding.ASCII.GetBytes("hx"));
+            Deliver(x, xConnect);
+            Deliver(y, y.Session.CreateConnect(0, Encoding.ASCII.GetBytes("hy")));
+            // A repeat of X's accepted Connect is answered with the Accept owed, not a new session.
+            Deliver(x, xConnect);
+            Deliver(x, Data(x, "ping-x"));
+            Deliver(y, Data(y, "ping-y"));
+            Send(x, "schema", Encoding.ASCII.GetBytes("pong"));
+            // Larger than the fragment size: three fragments, one per datagram.
+            Send(y, "schema", Encoding.ASCII.GetBytes("0123456789abcdefghijklmn"));
+            Send(x, "latest", Encoding.ASCII.GetBytes("sample"));
+            // X leaves; Y is undisturbed, and what X sends afterwards belongs to no session.
+            var late = Data(x, "after-bye");
+            Deliver(x, x.Session.CreateDisconnect(Encoding.ASCII.GetBytes("bye")));
+            Deliver(x, late);
+            Send(y, "schema", Encoding.ASCII.GetBytes("still-here"));
+            // Y reconnects from the same endpoint: a new generation replaces only Y.
+            y.Session = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId, InitialSequence = 30 });
+            Deliver(y, y.Session.CreateConnect(0, Encoding.ASCII.GetBytes("hy2")));
+            Send(y, "schema", Encoding.ASCII.GetBytes("new-gen"));
+            log.Append($"# peers {server.Peers.Count} dropped {server.Stats.PacketsDropped}\n");
+            foreach (var client in clients) client.Socket.Dispose();
+            return log.ToString();
+        }
+    }
+}

tests/vectors/rudp-multipeer-server/README.md

@@ -0,0 +1,13 @@
+# rudp-multipeer-server vectors
+
+Written only by the C# reference, `tests/GameCult.Networking.Tests/CultNetRudpMultiPeerVectorTests.cs`, with
+`CULTNET_WRITE_VECTORS=1`. Every other run replays the scenario and compares it byte for byte; a changed vector is a reviewed
+commit. Other runtimes' multi-peer listeners replay the same transcript and are held to the same answers
+(`packages/cultnet-py/tests/test_rudp_multipeer_server.py`).
+
+`transcript.txt` is one listener (`connection_id`, `initial_sequence`, `max_fragment_bytes` and `accept_payload` in its first line)
+serving two loopback peers, X and Y. Each line is an action or what it caused: `in <peer> <hex>` is a datagram the peer sends the
+listener, `send <peer> <channel> <hex>` is a payload the application sends that peer, and the `out <peer> <hex>` lines after it are
+every datagram the listener put on the wire in answer, X's before Y's. Peers are named by label because ports are ephemeral. The
+scenario covers a Connect, a repeat of an accepted Connect, data and acks, a fragmented send, an unreliable `latest` send, one peer's
+Disconnect and the datagram from it that follows, and the other peer's reconnect from the same endpoint.

tests/vectors/rudp-multipeer-server/transcript.txt

@@ -0,0 +1,30 @@
+# config connection_id=0x43554c54 initial_sequence=500 max_fragment_bytes=8 accept_payload=616363
+# in <peer> <datagram> | send <peer> <channel> <payload> | out <peer> <datagram>; peer labels name remote endpoints
+in X 434e52300001032b43554c540000000a0000000000000000000000000000000000020700636f6e74726f6c6878
+out X 434e52300002032b43554c54000001f40000000a00000000000000000000000000030700636f6e74726f6c616363
+in Y 434e52300001032b43554c54000000140000000000000000000000000000000000020700636f6e74726f6c6879
+out Y 434e52300002032b43554c54000001f40000001400000000000000000000000000030700636f6e74726f6c616363
+in X 434e52300001032b43554c540000000a0000000000000000000000000000000000020700636f6e74726f6c6878
+out X 434e52300002032b43554c54000001f40000000a00000000000000000000000000030700636f6e74726f6c616363
+in X 434e52300003032a43554c540000000b000001f400000000000000000000000000060600736368656d6170696e672d78
+out X 434e52300004002b43554c54000000000000000b00000001000000000000000000000700636f6e74726f6c
+in Y 434e52300003032a43554c5400000015000001f400000000000000000000000000060600736368656d6170696e672d79
+out Y 434e52300004002b43554c54000000000000001500000001000000000000000000000700636f6e74726f6c
+send X schema 706f6e67
+out X 434e52300003032a43554c54000001f50000000b00000001000000000000000000040600736368656d61706f6e67
+send Y schema 303132333435363738396162636465666768696a6b6c6d6e
+out Y 434e523000030b2a43554c54000001f50000001500000001000100000003000000080600736368656d613031323334353637
+out Y 434e523000030b2a43554c54000001f60000001500000001000100010003000000080600736368656d613839616263646566
+out Y 434e523000030b2a43554c54000001f70000001500000001000100020003000000080600736368656d616768696a6b6c6d6e
+send X latest 73616d706c65
+out X 434e52300003042a43554c54000000010000000b000000010000000000000000000606006c617465737473616d706c65
+in X 434e52300007002b43554c5400000000000001f500000001000000000000000000030700636f6e74726f6c627965
+in X 434e52300003032a43554c540000000c000001f500000001000000000000000000090600736368656d6161667465722d627965
+send Y schema 7374696c6c2d68657265
+out Y 434e523000030b2a43554c54000001f80000001500000001000200000002000000080600736368656d617374696c6c2d6865
+out Y 434e523000030b2a43554c54000001f90000001500000001000200010002000000020600736368656d617265
+in Y 434e52300001032b43554c540000001e0000000000000000000000000000000000030700636f6e74726f6c687932
+out Y 434e52300002032b43554c54000001f40000001e00000000000000000000000000030700636f6e74726f6c616363
+send Y schema 6e65772d67656e
+out Y 434e52300003032a43554c54000001f50000001e00000000000000000000000000070600736368656d616e65772d67656e
+# peers 1 dropped 1