From 8da99394cc74624d21afcec85fbbfaed00126994 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 17 Jun 2026 06:34:30 +0100 Subject: [PATCH] Add Python CultMesh RUDP server listener --- docs/cultnet-transport-parity.md | 6 +- docs/runtime-parity-scope.md | 2 +- packages/cultcache-py/README.md | 5 +- .../cultcache-py/src/cultmesh_py/facade.py | 6 + .../cultcache-py/src/cultmesh_py/server.py | 155 +++++++++++++++++- packages/cultcache-py/tests/test_cultcache.py | 23 +++ 6 files changed, 192 insertions(+), 5 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index da2c4f2f..92cb27e5 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -317,7 +317,11 @@ Current progress: endpoint or peer-card contact hints while leaving packet/session semantics in `cultnet_py`. Python peer catalogs also expose authorized-peer lookup, and the branded facade can create a RUDP client from the first peer authorized by - the authority lease catalog for a Verse role. + the authority lease catalog for a Verse role. `CultMeshLocalServer` now also + starts a RUDP listener beside the TCP-framed listener by default and advertises + both transport profiles in `cultnet.hello.v0`; the RUDP listener keeps schema + request handling in the same server authority while demultiplexing per-remote + `CultNetRudpSession` state. - Rust now has a poll-driven `CultNetRudpReconnectLoop` for substrate callers: service loops report closure, ask `reconnect_if_due(now_ms)` inside their own reactor, and the shared controller owns retry delay/exhaustion state while the diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index d57f35ff..74393f60 100644 --- a/docs/runtime-parity-scope.md +++ b/docs/runtime-parity-scope.md @@ -31,7 +31,7 @@ CultMesh. | Runtime | Role | Claimed parity | Not claimed | | --- | --- | --- | --- | | C# | Reference implementation and production game/server substrate | CultCache typed documents and SoA, CultNet schema-v0, LiteNetLib production path with adapter profile, shared transport profiles, shared RUDP socket transport, connected RUDP CultMesh client helpers, RUDP reconnect policy/controller and reconnect-loop helper, LiteNetLib client reconnect-controller adoption, schema-client transport port for service request/response clients, CultMesh node/session/catalog/stream/simulation/RUDP surfaces, authorized peer lookup | No need to abandon LiteNetLib before the CultNet adapter fully owns production service adoption | -| Python | Package runtime, local daemon endpoint, raw interop peer, typed state participant | CultCache/CultNet/CultMesh package surface, branded CultMesh node/schema/shard/Verse/peer/authority/stream/RUDP ergonomics, authorized peer lookup, shard catalogs/logs, schema-transport factory for raw client request/subscription flows, local TCP CultMesh server, session/prediction/simulation fact helpers, RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, benchmark evidence against C# cache paths | Universal C# throughput, full C# daemon/server stack, and RUDP as the only public service pipe | +| Python | Package runtime, local daemon endpoint, raw interop peer, typed state participant | CultCache/CultNet/CultMesh package surface, branded CultMesh node/schema/shard/Verse/peer/authority/stream/RUDP ergonomics, authorized peer lookup, shard catalogs/logs, schema-transport factory for raw client request/subscription flows, local TCP and RUDP CultMesh server listeners, session/prediction/simulation fact helpers, RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, benchmark evidence against C# cache paths | Universal C# throughput, full C# daemon/server stack, and RUDP as the only public service pipe | | TypeScript | Node/browser-adjacent projection and interop runtime | CultCache parity tests, CultNet schema-v0 peer, advertised schema/shard catalog wire contracts plus request/sync helpers, transport profiles, TCP-framed transport port and peer helper, shared RUDP socket transport, RUDP reconnect policy/controller, RUDP reconnect-loop helper, branded CultMesh node/schema/shard/Verse/peer/authority/stream/RUDP ergonomics, authorized peer lookup | Full C# game session/server stack | | Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, schema and shard catalog helpers, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport, connected RUDP CultMesh client helpers, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, small CultMesh facade for schema registry, shard catalog, peer cards, authority leases, authorized peer lookup, RUDP endpoint parsing, and peer-card/authorized-peer RUDP client construction | Full CultMesh game-session/server facade, stream catalog, or production game-session server | | Kotlin | JVM/Android client substrate | Typed MessagePack cache, built-in schema catalog advertisement, schema/shard catalogs with shared schema-message transport fetch/sync helpers, authority leases, authorized peer lookup, stream catalog, branded `CultMesh` facade, WebSocket transport profile/connection helper, RUDP client/server sugar, connected RUDP CultMesh client helpers, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence | C# SoA store, full C# server daemon, or Unity hot-loop memory layout | diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index 9f53eaf2..265eecb7 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -447,8 +447,9 @@ Current receipts: Verse catalog, peer exchange, authority lease, stream negotiation, CultNet helper shapes, raw snapshot/shard-log application, simulation claim hashing, consensus aggregation, game-session fact commits, local prediction - reconciliation, committed simulation fact payload slots, and witness artifact - bundle payload slots. + reconciliation, committed simulation fact payload slots, the local CultMesh + server's TCP and RUDP schema request paths, and witness artifact bundle + payload slots. - `cultcache_py`, `cultnet_py`, and `cultmesh_py` ship `py.typed` markers so downstream type checkers can inspect the package surface instead of treating the runtime as an untyped xenos swamp. diff --git a/packages/cultcache-py/src/cultmesh_py/facade.py b/packages/cultcache-py/src/cultmesh_py/facade.py index fe8562d7..a7900a44 100644 --- a/packages/cultcache-py/src/cultmesh_py/facade.py +++ b/packages/cultcache-py/src/cultmesh_py/facade.py @@ -292,6 +292,9 @@ class CultMesh: display_name: str | None = None, max_snapshot_documents: int | None = None, max_snapshot_bytes: int | None = None, + enable_rudp: bool = True, + rudp_connection_id: int = 0x43554C54, + rudp_resend_delay_ms: int = 25, ) -> CultMeshLocalServer: return CultMeshLocalServer( node=node, @@ -303,6 +306,9 @@ class CultMesh: display_name=display_name, max_snapshot_documents=max_snapshot_documents, max_snapshot_bytes=max_snapshot_bytes, + enable_rudp=enable_rudp, + rudp_connection_id=rudp_connection_id, + rudp_resend_delay_ms=rudp_resend_delay_ms, ).start() @staticmethod diff --git a/packages/cultcache-py/src/cultmesh_py/server.py b/packages/cultcache-py/src/cultmesh_py/server.py index 73182224..986460ff 100644 --- a/packages/cultcache-py/src/cultmesh_py/server.py +++ b/packages/cultcache-py/src/cultmesh_py/server.py @@ -2,6 +2,7 @@ from __future__ import annotations import socket import threading +import time from dataclasses import dataclass, field from typing import Any @@ -9,8 +10,16 @@ import msgpack from cultnet_py import ( CultNetSimulationObservationHub, + CultNetRudpPacket, + CultNetRudpPacketType, + CultNetRudpSendOptions, + CultNetRudpSession, + CultNetRudpSessionOptions, TcpFramedTransportConnection, + create_rudp_transport_profile, create_tcp_framed_transport_profile, + decode_rudp_packet, + encode_rudp_packet, wire_message_schema_descriptors, ) @@ -38,6 +47,12 @@ class _DatabaseSubscription: ) +@dataclass +class _RudpPeerConnection: + session: CultNetRudpSession + subscriptions: dict[str, _DatabaseSubscription] = field(default_factory=dict) + + @dataclass class CultMeshLocalServer: node: CultMeshNode @@ -50,9 +65,14 @@ class CultMeshLocalServer: runtime_kind: str = "python" max_snapshot_documents: int | None = None max_snapshot_bytes: int | None = None + enable_rudp: bool = True + rudp_connection_id: int = 0x43554C54 + rudp_resend_delay_ms: int = 25 _socket: socket.socket | None = field(default=None, init=False, repr=False) + _rudp_socket: socket.socket | None = field(default=None, init=False, repr=False) _stop: threading.Event = field(default_factory=threading.Event, init=False, repr=False) _thread: threading.Thread | None = field(default=None, init=False, repr=False) + _rudp_thread: threading.Thread | None = field(default=None, init=False, repr=False) def __post_init__(self) -> None: if self.max_snapshot_documents is not None and self.max_snapshot_documents < 0: @@ -73,10 +93,24 @@ class CultMeshLocalServer: self._stop.clear() self._thread = threading.Thread(target=self._accept_loop, daemon=True) self._thread.start() + if self.enable_rudp: + rudp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + rudp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + rudp_socket.bind((self.host, self.port)) + rudp_socket.settimeout(0.02) + self._rudp_socket = rudp_socket + self._rudp_thread = threading.Thread(target=self._rudp_loop, daemon=True) + self._rudp_thread.start() return self def stop(self) -> None: self._stop.set() + if self._rudp_socket is not None: + try: + self._rudp_socket.close() + except OSError: + pass + self._rudp_socket = None if self._socket is not None: try: self._socket.close() @@ -86,6 +120,9 @@ class CultMeshLocalServer: if self._thread is not None: self._thread.join(timeout=2.0) self._thread = None + if self._rudp_thread is not None: + self._rudp_thread.join(timeout=2.0) + self._rudp_thread = None def __enter__(self) -> "CultMeshLocalServer": return self.start() @@ -163,6 +200,106 @@ class CultMeshLocalServer: for response in responses: transport.send("schema", msgpack.packb(response, use_bin_type=True)) + def _rudp_loop(self) -> None: + rudp_socket = self._rudp_socket + if rudp_socket is None: + return + peers: dict[tuple[str, int], _RudpPeerConnection] = {} + while not self._stop.is_set(): + try: + wire, remote_addr = rudp_socket.recvfrom(65535) + except TimeoutError: + self._poll_rudp_resends(rudp_socket, peers) + continue + except OSError: + return + + try: + packet = decode_rudp_packet(wire) + except ValueError: + continue + if packet.connection_id != self.rudp_connection_id: + continue + + now_ms = _now_ms() + peer = peers.get(remote_addr) + if packet.packet_type == CultNetRudpPacketType.CONNECT: + peer = _RudpPeerConnection( + CultNetRudpSession( + CultNetRudpSessionOptions( + connection_id=self.rudp_connection_id, + resend_delay_ms=self.rudp_resend_delay_ms, + ) + ) + ) + peers[remote_addr] = peer + self._send_rudp_packet( + rudp_socket, + remote_addr, + peer.session.accept_connect(packet, now_ms, b"cultmesh-local-rudp"), + ) + continue + if peer is None: + continue + + result = peer.session.receive(packet, now_ms) + if result.reply is not None: + self._send_rudp_packet(rudp_socket, remote_addr, result.reply) + if result.disconnected: + peers.pop(remote_addr, None) + continue + for delivered in result.delivered: + if delivered.channel_id != "schema": + continue + try: + message = msgpack.unpackb(delivered.payload, raw=False) + except (msgpack.ExtraData, msgpack.FormatError, msgpack.StackError, ValueError): + continue + if not isinstance(message, dict): + continue + responses = self._handle_connection_message(message, peer.subscriptions) + for response in responses: + self._send_rudp_schema_frame( + rudp_socket, + remote_addr, + peer.session, + msgpack.packb(response, use_bin_type=True), + ) + if packet.packet_type == CultNetRudpPacketType.DATA or result.delivered: + self._send_rudp_packet(rudp_socket, remote_addr, peer.session.create_ack()) + + def _poll_rudp_resends( + self, + rudp_socket: socket.socket, + peers: dict[tuple[str, int], _RudpPeerConnection], + ) -> None: + now_ms = _now_ms() + for remote_addr, peer in list(peers.items()): + for packet in peer.session.due_resends(now_ms): + self._send_rudp_packet(rudp_socket, remote_addr, packet) + + def _send_rudp_schema_frame( + self, + rudp_socket: socket.socket, + remote_addr: tuple[str, int], + session: CultNetRudpSession, + payload: bytes, + ) -> None: + for packet in session.send_many( + "schema", + payload, + CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=_now_ms()), + ): + self._send_rudp_packet(rudp_socket, remote_addr, packet) + + @staticmethod + def _send_rudp_packet( + rudp_socket: socket.socket, + remote_addr: tuple[str, int], + packet: CultNetRudpPacket, + ) -> None: + rudp_socket.sendto(encode_rudp_packet(packet), remote_addr) + def _handle_connection_message( self, message: dict[str, Any], @@ -328,7 +465,19 @@ class CultMeshLocalServer: transport_id="cultmesh-local-tcp", host=self.host, port=self.port, - ) + ), + *( + [ + create_rudp_transport_profile( + self.node.runtime_id, + transport_id="cultmesh-local-rudp", + host=self.host, + port=self.port, + ) + ] + if self.enable_rudp + else [] + ), ], "supportsSchemaCatalog": True, } @@ -467,3 +616,7 @@ class CultMeshLocalServer: "messageId": request.get("messageId", ""), "shards": shards, } + + +def _now_ms() -> int: + return int(time.time() * 1000) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index ea1ba26e..16d72d98 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -4853,6 +4853,20 @@ class CultCacheTests(unittest.TestCase): raw_client = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=2.0) hello_response = raw_client.request(hello(runtime_id="probe"), expected_schema_version="cultnet.hello.v0") schema_response = raw_client.fetch_schema_catalog(schema_ids=["mesh.server_note.v1"], include_schema_json=True) + rudp_client = CultMesh.create_client( + endpoint=f"rudp://127.0.0.1:{server.port}", + timeout_seconds=2.0, + connection_id=server.rudp_connection_id, + runtime_id="rudp-probe", + ) + rudp_hello_response = rudp_client.request( + hello(runtime_id="rudp-probe"), + expected_schema_version="cultnet.hello.v0", + ) + rudp_schema_response = rudp_client.fetch_schema_catalog( + schema_ids=["mesh.server_note.v1"], + include_schema_json=True, + ) wire_schema_response = raw_client.fetch_schema_catalog(kinds=["wire_message"], include_schema_json=True) synced_schema_catalog = CultNetSchemaCatalog() synced_wire_descriptors = raw_client.sync_schema_catalog( @@ -4934,6 +4948,15 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(hello_response["supportedMutationContracts"][0]["documentType"], "mesh.server_note") self.assertIn("documentDelete", hello_response["supportedMutationContracts"][0]["operations"]) self.assertIn("shardLog", hello_response["supportedMutationContracts"][0]["operations"]) + advertised_transports = { + transport["protocol"]: transport + for profile in hello_response["transportProfiles"] + for transport in profile["transports"] + } + self.assertEqual(advertised_transports["tcp_framed"]["port"], server.port) + self.assertEqual(advertised_transports["rudp"]["port"], server.port) + self.assertEqual(rudp_hello_response["runtimeId"], "mesh-server") + self.assertEqual(rudp_schema_response["schemas"][0]["schemaId"], "mesh.server_note.v1") self.assertEqual(schema_response["schemas"][0]["schemaId"], "mesh.server_note.v1") self.assertIn("schemaJson", schema_response["schemas"][0]) self.assertIn("cultnet.document_delete.v0", schema_response["schemas"][0]["wireContracts"])