From 343802dacf099bb3cac770e3a15bb1ea22b260df Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 14 Jun 2026 22:53:00 +0100 Subject: [PATCH] Port RUDP reliability core to Python --- docs/cultnet-transport-parity.md | 20 +- .../cultcache-py/src/cultnet_py/__init__.py | 10 + .../cultcache-py/src/cultnet_py/transport.py | 260 +++++++++++++++++- packages/cultcache-py/tests/test_cultcache.py | 120 +++++++- 4 files changed, 398 insertions(+), 12 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 069eb0de..37ea4a03 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -26,9 +26,9 @@ CultNet owns cross-runtime transport semantics: | --- | --- | --- | --- | | C# `GameCult.Networking` | LiteNetLib UDP `NetManager` / `NetPeer`; sends legacy union messages and schema-v0 messages with `DeliveryMethod.ReliableOrdered` | LiteNetLib connection requests and app-level peer/catalog surfaces | C# owns the production-shaped realtime semantics | | C# interop peer | TCP stream with 4-byte length-prefixed MessagePack frames | UDP multicast probe/announce | Test harness only | -| TypeScript `cultnet-ts` | `CultNetPeer` over any Node `Duplex`; interop uses TCP | UDP multicast probe/announce in the interop peer | Byte-stream abstraction; no UDP reliability owner | -| Rust `cultnet-rs` | Interop example uses TCP length-prefixed MessagePack frames | UDP multicast probe/announce | Example harness; no UDP reliability owner | -| Python `cultcache-py` | TCP sockets with 4-byte length-prefixed MessagePack frames for local CultMesh/CultNet server and client | Endpoint lists and CultMesh peer/Verse catalogs | Local framed service body; no UDP reliability owner | +| TypeScript `cultnet-ts` | `CultNetPeer` over any Node `Duplex`; interop uses TCP | UDP multicast probe/announce in the interop peer | Byte-stream abstraction plus socket-free RUDP reliability owner; no UDP socket binding yet | +| Rust `cultnet-rs` | Interop example uses TCP length-prefixed MessagePack frames | UDP multicast probe/announce | Example harness plus socket-free RUDP reliability owner; no UDP socket binding yet | +| Python `cultcache-py` | TCP sockets with 4-byte length-prefixed MessagePack frames for local CultMesh/CultNet server and client | Endpoint lists and CultMesh peer/Verse catalogs | Local framed service body plus socket-free RUDP reliability owner; no UDP socket binding yet | | Kotlin `cultmesh-kotlin` | Minimal WebSocket-like lane over TCP socket | None found in the live package | Thin client surface | The live split is therefore: payload language is converging, transport language @@ -160,15 +160,15 @@ Current progress: the RUDP runtime yet; it is the binary packet language the runtimes must converge on before resend loops, windows, and timeout behavior are allowed to claim parity. -- TypeScript, C#, and Rust now share the first deterministic RUDP reliability state - machine: connect/accept handshake, packet-level ack/ack-mask accounting, - reliable resend scheduling, duplicate suppression, and reliable ordered - channel delivery. It is in-memory and socket-free so the behavior can be - ported before any runtime binds it to UDP I/O. +- TypeScript, C#, Rust, and Python now share the first deterministic RUDP + reliability state machine: connect/accept handshake, packet-level + ack/ack-mask accounting, reliable resend scheduling, duplicate suppression, + and reliable ordered channel delivery. It is in-memory and socket-free so the + behavior can be tested before any runtime binds it to UDP I/O. - The remaining parity work is to add the equivalent port to Kotlin, deepen Python's server-side use of its port, and move each runtime's existing - TCP/LiteNetLib/WebSocket bodies behind those ports, port the RUDP reliability - state machine into Python, then bind it to UDP sockets. + TCP/LiteNetLib/WebSocket bodies behind those ports, then bind the shared RUDP + reliability state machine to UDP sockets. ## RUDP Packet Contract V0 diff --git a/packages/cultcache-py/src/cultnet_py/__init__.py b/packages/cultcache-py/src/cultnet_py/__init__.py index b2b14fb8..e96f50da 100644 --- a/packages/cultcache-py/src/cultnet_py/__init__.py +++ b/packages/cultcache-py/src/cultnet_py/__init__.py @@ -66,8 +66,13 @@ from .shard_replication import ( from .snapshot import CultNetRawDocumentRecord, CultNetRawSnapshotResponse from .subscription import CultNetDatabaseChange from .transport import ( + CultNetRudpDeliveredFrame, CultNetRudpPacket, CultNetRudpPacketType, + CultNetRudpReceiveResult, + CultNetRudpSendOptions, + CultNetRudpSession, + CultNetRudpSessionOptions, CultNetTransportFrame, CultNetTransportStats, TcpFramedTransportConnection, @@ -123,8 +128,13 @@ __all__ = [ "CultNetSimulationConsensusOptions", "CultNetSimulationObservation", "CultNetSimulationObservationHub", + "CultNetRudpDeliveredFrame", "CultNetRudpPacket", "CultNetRudpPacketType", + "CultNetRudpReceiveResult", + "CultNetRudpSendOptions", + "CultNetRudpSession", + "CultNetRudpSessionOptions", "CultNetTransportFrame", "CultNetTransportStats", "TcpFramedTransportConnection", diff --git a/packages/cultcache-py/src/cultnet_py/transport.py b/packages/cultcache-py/src/cultnet_py/transport.py index ccd92f2b..2c77176b 100644 --- a/packages/cultcache-py/src/cultnet_py/transport.py +++ b/packages/cultcache-py/src/cultnet_py/transport.py @@ -1,6 +1,6 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field from enum import Enum from typing import Any, BinaryIO @@ -90,6 +90,40 @@ class CultNetRudpPacket: payload: bytes = b"" +@dataclass(frozen=True) +class CultNetRudpDeliveredFrame: + channel_id: str + payload: bytes + sequence: int + + +@dataclass(frozen=True) +class CultNetRudpReceiveResult: + delivered: tuple[CultNetRudpDeliveredFrame, ...] = field(default_factory=tuple) + reply: CultNetRudpPacket | None = None + + +@dataclass(frozen=True) +class CultNetRudpSessionOptions: + connection_id: int + initial_sequence: int = 1 + resend_delay_ms: int = 250 + + +@dataclass(frozen=True) +class CultNetRudpSendOptions: + reliable: bool = False + ordered: bool = False + sequenced: bool = False + now_ms: int = 0 + + +@dataclass +class _PendingReliablePacket: + packet: CultNetRudpPacket + last_sent_at_ms: int + + _RUDP_PACKET_TYPE_TO_CODE = { CultNetRudpPacketType.CONNECT: 1, CultNetRudpPacketType.ACCEPT: 2, @@ -145,6 +179,230 @@ def create_rudp_transport_profile( } +class CultNetRudpSession: + def __init__(self, options: CultNetRudpSessionOptions) -> None: + self.connection_id = _uint32(options.connection_id, "connection_id") + self.resend_delay_ms = options.resend_delay_ms + self._next_sequence = _uint32(options.initial_sequence, "initial_sequence") + self._connected = False + self._highest_received_sequence: int | None = None + self._received_sequences: set[int] = set() + self._pending_reliable: dict[int, _PendingReliablePacket] = {} + self._ordered_next_sequence_by_channel: dict[str, int] = {} + self._ordered_buffers: dict[str, dict[int, CultNetRudpDeliveredFrame]] = {} + + @property + def connected(self) -> bool: + return self._connected + + @property + def pending_reliable_sequences(self) -> tuple[int, ...]: + return tuple(sorted(self._pending_reliable)) + + def create_connect(self, now_ms: int = 0, payload: bytes = b"") -> CultNetRudpPacket: + packet = self._create_packet( + CultNetRudpPacketType.CONNECT, + "control", + payload, + reliable=True, + ordered=True, + ) + self._track_reliable(packet, now_ms) + return packet + + def accept_connect( + self, + packet: CultNetRudpPacket, + now_ms: int = 0, + payload: bytes = b"", + ) -> CultNetRudpPacket: + self._require_connection(packet) + if packet.packet_type != CultNetRudpPacketType.CONNECT: + raise ValueError(f"Expected RUDP connect packet, got {packet.packet_type.value}") + + self._remember_received(packet.sequence) + self._connected = True + response = self._create_packet( + CultNetRudpPacketType.ACCEPT, + "control", + payload, + reliable=True, + ordered=True, + ) + self._track_reliable(response, now_ms) + return response + + def send( + self, + channel_id: str, + payload: bytes, + options: CultNetRudpSendOptions | None = None, + ) -> CultNetRudpPacket: + if not self._connected: + raise ValueError("Cannot send RUDP data before the session is connected") + + options = options or CultNetRudpSendOptions() + packet = self._create_packet( + CultNetRudpPacketType.DATA, + channel_id, + payload, + reliable=options.reliable, + ordered=options.ordered, + sequenced=options.sequenced, + ) + if packet.reliable: + self._track_reliable(packet, options.now_ms) + return packet + + def receive(self, packet: CultNetRudpPacket, now_ms: int = 0) -> CultNetRudpReceiveResult: + del now_ms + self._require_connection(packet) + self._apply_acknowledgements(packet) + + if packet.packet_type == CultNetRudpPacketType.ACCEPT: + self._remember_received(packet.sequence) + self._connected = True + return CultNetRudpReceiveResult() + + if packet.packet_type == CultNetRudpPacketType.PING: + self._remember_received(packet.sequence) + return CultNetRudpReceiveResult( + reply=self._create_packet(CultNetRudpPacketType.PONG, "control", packet.payload) + ) + + if packet.packet_type in (CultNetRudpPacketType.ACK, CultNetRudpPacketType.PONG): + self._remember_received(packet.sequence) + return CultNetRudpReceiveResult() + + if packet.packet_type != CultNetRudpPacketType.DATA: + return CultNetRudpReceiveResult() + + duplicate = packet.sequence in self._received_sequences + self._remember_received(packet.sequence) + if duplicate: + return CultNetRudpReceiveResult() + + frame = CultNetRudpDeliveredFrame( + channel_id=packet.channel_id, + payload=bytes(packet.payload), + sequence=packet.sequence, + ) + if not packet.ordered: + return CultNetRudpReceiveResult(delivered=(frame,)) + return CultNetRudpReceiveResult(delivered=tuple(self._deliver_ordered(frame))) + + def create_ack(self) -> CultNetRudpPacket: + return self._create_packet(CultNetRudpPacketType.ACK, "control", b"") + + def due_resends(self, now_ms: int) -> tuple[CultNetRudpPacket, ...]: + due: list[CultNetRudpPacket] = [] + for pending in self._pending_reliable.values(): + if now_ms - pending.last_sent_at_ms >= self.resend_delay_ms: + pending.last_sent_at_ms = now_ms + due.append(pending.packet) + return tuple(sorted(due, key=lambda packet: packet.sequence)) + + def _create_packet( + self, + packet_type: CultNetRudpPacketType, + channel_id: str, + payload: bytes, + *, + reliable: bool = False, + ordered: bool = False, + sequenced: bool = False, + ) -> CultNetRudpPacket: + sequence = self._next_sequence + self._next_sequence = _uint32(self._next_sequence + 1, "sequence") + ack, ack_mask = self._ack_state() + return CultNetRudpPacket( + packet_type=packet_type, + connection_id=self.connection_id, + sequence=sequence, + ack=ack, + ack_mask=ack_mask, + channel_id=channel_id, + reliable=reliable, + ordered=ordered, + sequenced=sequenced, + payload=bytes(payload), + ) + + def _track_reliable(self, packet: CultNetRudpPacket, now_ms: int) -> None: + self._pending_reliable[packet.sequence] = _PendingReliablePacket( + packet=CultNetRudpPacket( + packet_type=packet.packet_type, + connection_id=packet.connection_id, + sequence=packet.sequence, + ack=packet.ack, + ack_mask=packet.ack_mask, + channel_id=packet.channel_id, + reliable=packet.reliable, + ordered=packet.ordered, + sequenced=packet.sequenced, + fragment_id=packet.fragment_id, + fragment_index=packet.fragment_index, + fragment_count=packet.fragment_count, + payload=bytes(packet.payload), + ), + last_sent_at_ms=now_ms, + ) + + def _apply_acknowledgements(self, packet: CultNetRudpPacket) -> None: + self._pending_reliable.pop(packet.ack, None) + for bit in range(32): + if packet.ack_mask & (1 << bit): + self._pending_reliable.pop(packet.ack - bit - 1, None) + + def _remember_received(self, sequence: int) -> None: + self._received_sequences.add(sequence) + if self._highest_received_sequence is None or sequence > self._highest_received_sequence: + self._highest_received_sequence = sequence + + def _ack_state(self) -> tuple[int, int]: + ack = self._highest_received_sequence or 0 + ack_mask = 0 + for bit in range(32): + if ack > bit and (ack - bit - 1) in self._received_sequences: + ack_mask |= 1 << bit + return ack, ack_mask + + def _deliver_ordered(self, frame: CultNetRudpDeliveredFrame) -> list[CultNetRudpDeliveredFrame]: + next_sequence = self._ordered_next_sequence_by_channel.get(frame.channel_id) + if next_sequence is None: + self._ordered_next_sequence_by_channel[frame.channel_id] = frame.sequence + 1 + return [frame, *self._drain_ordered(frame.channel_id)] + if frame.sequence < next_sequence: + return [] + if frame.sequence > next_sequence: + self._ordered_buffers.setdefault(frame.channel_id, {})[frame.sequence] = frame + return [] + + self._ordered_next_sequence_by_channel[frame.channel_id] = next_sequence + 1 + return [frame, *self._drain_ordered(frame.channel_id)] + + def _drain_ordered(self, channel_id: str) -> list[CultNetRudpDeliveredFrame]: + delivered: list[CultNetRudpDeliveredFrame] = [] + buffer = self._ordered_buffers.get(channel_id) + if buffer is None: + return delivered + + while True: + next_sequence = self._ordered_next_sequence_by_channel[channel_id] + frame = buffer.pop(next_sequence, None) + if frame is None: + break + delivered.append(frame) + self._ordered_next_sequence_by_channel[channel_id] = next_sequence + 1 + return delivered + + def _require_connection(self, packet: CultNetRudpPacket) -> None: + if packet.connection_id != self.connection_id: + raise ValueError( + f"RUDP packet belongs to connection {packet.connection_id}, expected {self.connection_id}" + ) + + def encode_rudp_packet(packet: CultNetRudpPacket) -> bytes: channel_id = packet.channel_id.encode("utf-8") if len(channel_id) > MAX_RUDP_CHANNEL_ID_BYTES: diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 3fd06289..c13660dc 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -69,6 +69,9 @@ from cultnet_py import ( CultNetSimulationObservationHub, CultNetRudpPacket, CultNetRudpPacketType, + CultNetRudpSendOptions, + CultNetRudpSession, + CultNetRudpSessionOptions, TcpFramedTransportConnection, CultNetWitnessArtifactBundle, apply_raw_document_record, @@ -505,7 +508,8 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(validated.expires_at_utc, expires) self.assertEqual(validated.session_version, 7) - tampered = token[:-1] + ("A" if token[-1] != "A" else "B") + tampered_payload = CultNetSecret.to_base64url(CultNetSecret.from_base64url(payload_b64) + b"tamper") + tampered = f"{tampered_payload}.{signature_b64}" self.assertIsNone(CultNetSecret.try_validate_session_token(tampered, security)) self.assertIsNone(CultNetSecret.try_validate_session_token(token, security, now_utc=expires)) @@ -646,6 +650,120 @@ class CultCacheTests(unittest.TestCase): ], ) + def test_cultnet_rudp_session_handshake_acks_reliable_connect_and_accept_packets(self) -> None: + client = CultNetRudpSession( + CultNetRudpSessionOptions(connection_id=0x0A0B0C0D, initial_sequence=1, resend_delay_ms=50) + ) + server = CultNetRudpSession( + CultNetRudpSessionOptions(connection_id=0x0A0B0C0D, initial_sequence=100, resend_delay_ms=50) + ) + + connect = client.create_connect(0, b"join") + self.assertEqual(connect.packet_type, CultNetRudpPacketType.CONNECT) + self.assertEqual(connect.sequence, 1) + self.assertEqual(client.pending_reliable_sequences, (1,)) + + accept = server.accept_connect(connect, 10, b"ok") + self.assertEqual(accept.packet_type, CultNetRudpPacketType.ACCEPT) + self.assertEqual(accept.ack, 1) + self.assertTrue(server.connected) + self.assertEqual(server.pending_reliable_sequences, (100,)) + + client.receive(accept, 20) + self.assertTrue(client.connected) + self.assertEqual(client.pending_reliable_sequences, ()) + + ack = client.create_ack() + self.assertEqual(ack.ack, 100) + server.receive(ack, 30) + self.assertEqual(server.pending_reliable_sequences, ()) + + def test_cultnet_rudp_session_computes_ack_masks_and_clears_pending_reliable_packets(self) -> None: + sender = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=7, initial_sequence=10, resend_delay_ms=100)) + receiver = CultNetRudpSession( + CultNetRudpSessionOptions(connection_id=7, initial_sequence=200, resend_delay_ms=100) + ) + sender.receive( + CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 7, 1, 0, 0, "control") + ) + receiver.receive( + CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 7, 2, 0, 0, "control") + ) + + first = sender.send( + "schema", + b"first", + CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=0), + ) + second = sender.send( + "schema", + b"second", + CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=0), + ) + third = sender.send( + "schema", + b"third", + CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=0), + ) + self.assertEqual(sender.pending_reliable_sequences, (10, 11, 12)) + + receiver.receive(first) + receiver.receive(third) + ack_with_gap = receiver.create_ack() + self.assertEqual(ack_with_gap.ack, 12) + self.assertEqual(ack_with_gap.ack_mask, 0b10 | (1 << 9)) + sender.receive(ack_with_gap) + self.assertEqual(sender.pending_reliable_sequences, (11,)) + + receiver.receive(second) + full_ack = receiver.create_ack() + self.assertEqual(full_ack.ack, 12) + self.assertEqual(full_ack.ack_mask, 0b11 | (1 << 9)) + sender.receive(full_ack) + self.assertEqual(sender.pending_reliable_sequences, ()) + + def test_cultnet_rudp_session_schedules_reliable_resends_until_acked(self) -> None: + session = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=99, initial_sequence=1, resend_delay_ms=100)) + session.receive( + CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 99, 50, 0, 0, "control") + ) + sent = session.send( + "schema", + b"payload", + CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=10), + ) + + self.assertEqual(session.due_resends(90), ()) + self.assertEqual(tuple(packet.sequence for packet in session.due_resends(110)), (sent.sequence,)) + self.assertEqual(session.due_resends(150), ()) + + session.receive( + CultNetRudpPacket(CultNetRudpPacketType.ACK, 99, 51, sent.sequence, 0, "control") + ) + self.assertEqual(session.due_resends(250), ()) + + def test_cultnet_rudp_session_suppresses_duplicates_and_delivers_reliable_ordered_payloads(self) -> None: + sender = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=123, initial_sequence=1)) + receiver = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=123, initial_sequence=100)) + sender.receive( + CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 123, 90, 0, 0, "control") + ) + receiver.receive( + CultNetRudpPacket(CultNetRudpPacketType.ACCEPT, 123, 91, 0, 0, "control") + ) + + first = sender.send("schema", b"first", CultNetRudpSendOptions(reliable=True, ordered=True)) + second = sender.send("schema", b"second", CultNetRudpSendOptions(reliable=True, ordered=True)) + third = sender.send("schema", b"third", CultNetRudpSendOptions(reliable=True, ordered=True)) + + self.assertEqual([frame.payload.decode("utf-8") for frame in receiver.receive(first).delivered], ["first"]) + self.assertEqual(receiver.receive(third).delivered, ()) + self.assertEqual(receiver.receive(first).delivered, ()) + self.assertEqual( + [frame.payload.decode("utf-8") for frame in receiver.receive(second).delivered], + ["second", "third"], + ) + def test_cultnet_database_subscription_helpers_match_schema_v0_shape(self) -> None: subscribe = database_subscribe( message_id="sub-message",