diff --git a/packages/cultnet-py/CHANGELOG.md b/packages/cultnet-py/CHANGELOG.md
index cca85a79..04df5ddb 100644
--- a/packages/cultnet-py/CHANGELOG.md
+++ b/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`.
diff --git a/packages/cultnet-py/src/cultnet_py/__init__.py b/packages/cultnet-py/src/cultnet_py/__init__.py
index 3fc36683..d590759a 100644
--- a/packages/cultnet-py/src/cultnet_py/__init__.py
+++ b/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",
diff --git a/packages/cultnet-py/src/cultnet_py/rudp_server.py b/packages/cultnet-py/src/cultnet_py/rudp_server.py
new file mode 100644
index 00000000..2b162043
--- /dev/null
+++ b/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
diff --git a/packages/cultnet-py/src/cultnet_py/transport.py b/packages/cultnet-py/src/cultnet_py/transport.py
index 76b38bf5..3ecfa31b 100644
--- a/packages/cultnet-py/src/cultnet_py/transport.py
+++ b/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)
diff --git a/packages/cultnet-py/tests/test_rudp_multipeer_server.py b/packages/cultnet-py/tests/test_rudp_multipeer_server.py
new file mode 100644
index 00000000..3bb14640
--- /dev/null
+++ b/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()
diff --git a/tests/GameCult.Networking.Tests/CultNetRudpMultiPeerVectorTests.cs b/tests/GameCult.Networking.Tests/CultNetRudpMultiPeerVectorTests.cs
new file mode 100644
index 00000000..0985d516
--- /dev/null
+++ b/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
+{
+ ///
+ /// 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.
+ ///
+ 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 | send | out ; 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();
+ }
+ }
+}
diff --git a/tests/vectors/rudp-multipeer-server/README.md b/tests/vectors/rudp-multipeer-server/README.md
new file mode 100644
index 00000000..21231461
--- /dev/null
+++ b/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 ` is a datagram the peer sends the
+listener, `send ` is a payload the application sends that peer, and the `out ` 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.
diff --git a/tests/vectors/rudp-multipeer-server/transcript.txt b/tests/vectors/rudp-multipeer-server/transcript.txt
new file mode 100644
index 00000000..50f2f64f
--- /dev/null
+++ b/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 | send | out ; 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