From 91d9ecda931e345385016e7d6022b2d5a0ff6cc4 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 17 Jun 2026 08:05:41 +0100 Subject: [PATCH] Add Python interop RUDP serve dial path --- docs/cultnet-transport-parity.md | 5 +- docs/runtime-parity-scope.md | 2 +- packages/cultcache-py/PARITY.md | 5 +- .../src/cultnet_py/interop_peer.py | 198 ++++++++++++++++-- .../test/interop/cultnet-interop.test.ts | 20 ++ 5 files changed, 209 insertions(+), 21 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index b1f0f099..848cee3a 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -321,7 +321,10 @@ Current progress: 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. + `CultNetRudpSession` state. The Python interop peer now also serves and dials + schema-v0 over `interop-rudp`, advertises TCP-framed and RUDP in discovery and + hello, and caps interop RUDP schema frames into 1024-byte fragments so catalog + responses stay inside UDP datagram limits while remaining session-owned. - 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 cb6cd67d..bd8548ce 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 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 | +| 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, Python interop RUDP serve/dial path, 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 interop serve/dial path, 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/PARITY.md b/packages/cultcache-py/PARITY.md index 2aa7a308..18607d87 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -301,8 +301,9 @@ node --test packages\cultnet-ts\dist-test\test\interop\cultnet-interop.test.js owns the shared security/session primitives for helper parity, while the live transport harness still speaks the schema-v0 interop lane. - Python does not make RUDP the only public pipe yet. The local CultMesh server, - raw client, and interop peer still use the TCP-framed schema lane for many - service and harness flows, while `docs/cultnet-transport-parity.md` tracks the + raw client, and interop peer still keep the TCP-framed schema lane for + compatibility, while the Python interop peer can now serve and dial schema-v0 + over fragmented RUDP. `docs/cultnet-transport-parity.md` tracks the remaining work to put broader server-side bodies behind the shared transport port instead of cloning LiteNetLib. The raw client request/subscription layer now routes through a schema-transport factory, so the caller-facing semantics diff --git a/packages/cultcache-py/src/cultnet_py/interop_peer.py b/packages/cultcache-py/src/cultnet_py/interop_peer.py index 8c5c1bf3..d26f1b70 100644 --- a/packages/cultcache-py/src/cultnet_py/interop_peer.py +++ b/packages/cultcache-py/src/cultnet_py/interop_peer.py @@ -38,7 +38,19 @@ from cultmesh_py import ( ) from .schema_catalog import INTEROP_WIRE_CONTRACT, wire_message_schema_descriptors -from .transport import TcpFramedTransportConnection, create_tcp_framed_transport_profile +from .client import create_rudp_schema_transport +from .transport import ( + CultNetRudpPacket, + CultNetRudpPacketType, + CultNetRudpSendOptions, + CultNetRudpSession, + CultNetRudpSessionOptions, + TcpFramedTransportConnection, + create_rudp_transport_profile, + create_tcp_framed_transport_profile, + decode_rudp_packet, + encode_rudp_packet, +) INTEROP_DOCUMENT_TYPE = "cultnet.interop-note" INTEROP_SCHEMA_VERSION = "cultnet.interop_note.v0" @@ -60,6 +72,8 @@ WITNESS_ARTIFACT_BUNDLE_SCHEMA_VERSION = "cultnet.witness_artifact_bundle.v0" SIMULATION_FACT_SCHEMA_ID = simulation_fact_document.catalog_entry().schema_id DISCOVERY_PROBE_SCHEMA_VERSION = "cultnet.discovery_probe.v0" DISCOVERY_ANNOUNCE_SCHEMA_VERSION = "cultnet.discovery_announce.v0" +RUDP_CONNECTION_ID = 0x43554C54 +RUDP_INTEROP_MAX_FRAGMENT_BYTES = 1024 @dataclass(frozen=True) @@ -87,6 +101,7 @@ class PeerState: shard_id: str shard_epoch: int shard_endpoint: str + transport_profiles: list[dict[str, Any]] shard_log: list[dict[str, Any]] shard_log_lock: threading.Lock observations: dict[tuple[str, int, str, str], dict[str, dict[str, Any]]] @@ -117,6 +132,7 @@ def main(argv: list[str] | None = None) -> int: serve_parser.add_argument("--bind-host", default="127.0.0.1") serve_parser.add_argument("--advertise-host", required=True) serve_parser.add_argument("--tcp-port", type=int, required=True) + serve_parser.add_argument("--rudp-port", type=int) serve_parser.add_argument("--discovery-port", type=int, required=True) serve_parser.add_argument("--discovery-group", required=True) serve_parser.add_argument("--schema-path", required=True) @@ -130,7 +146,8 @@ def main(argv: list[str] | None = None) -> int: dial_parser = sub.add_parser("dial") add_common_runtime_args(dial_parser) dial_parser.add_argument("--target-host", required=True) - dial_parser.add_argument("--target-port", type=int, required=True) + dial_parser.add_argument("--target-port", type=int) + dial_parser.add_argument("--target-rudp-port", type=int) dial_parser.add_argument("--schema-path", required=True) dial_parser.add_argument("--timeout-ms", type=int, default=4000) @@ -154,6 +171,7 @@ def add_common_runtime_args(parser: argparse.ArgumentParser) -> None: def serve(args: argparse.Namespace) -> None: + rudp_port = int(args.rudp_port or args.tcp_port) state = build_state( args.runtime_id, args.runtime_kind, @@ -161,6 +179,7 @@ def serve(args: argparse.Namespace) -> None: args.agent_id, args.schema_path, shard_endpoint=f"cultnet://{args.advertise_host}:{args.tcp_port}", + transport_profiles=interop_transport_profiles(args.runtime_id, args.advertise_host, args.tcp_port, rudp_port), ) state.cache.put(state.bindings["note"].document, f"note:{args.runtime_id}", build_interop_note(args.runtime_id, args.display_name)) @@ -171,6 +190,11 @@ def serve(args: argparse.Namespace) -> None: tcp_server.listen() tcp_server.settimeout(0.2) + rudp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + rudp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + rudp_socket.bind((args.bind_host, rudp_port)) + rudp_socket.settimeout(0.02) + udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) udp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) udp_socket.bind(("", args.discovery_port)) @@ -182,7 +206,8 @@ def serve(args: argparse.Namespace) -> None: threads = [ threading.Thread(target=accept_loop, args=(tcp_server, state, stop), daemon=True), - threading.Thread(target=discovery_loop, args=(udp_socket, args.advertise_host, args.tcp_port, state, stop), daemon=True), + threading.Thread(target=rudp_loop, args=(rudp_socket, args.advertise_host, args.tcp_port, rudp_port, state, stop), daemon=True), + threading.Thread(target=discovery_loop, args=(udp_socket, args.advertise_host, args.tcp_port, rudp_port, state, stop), daemon=True), ] for thread in threads: thread.start() @@ -192,10 +217,11 @@ def serve(args: argparse.Namespace) -> None: signal.signal(signal.SIGTERM, request_stop) signal.signal(signal.SIGINT, request_stop) - write_json({"status": "ready", "mode": "serve", "runtimeId": args.runtime_id, "runtimeKind": args.runtime_kind, "tcpPort": args.tcp_port, "discoveryPort": args.discovery_port, "discoveryGroup": args.discovery_group}) + write_json({"status": "ready", "mode": "serve", "runtimeId": args.runtime_id, "runtimeKind": args.runtime_kind, "tcpPort": args.tcp_port, "rudpPort": rudp_port, "discoveryPort": args.discovery_port, "discoveryGroup": args.discovery_group}) while not stop.is_set(): time.sleep(0.1) tcp_server.close() + rudp_socket.close() udp_socket.close() @@ -231,6 +257,104 @@ def handle_connection(client: socket.socket, state: PeerState) -> None: write_message(transport, response) +@dataclass +class RudpPeerConnection: + session: CultNetRudpSession + subscriptions: dict[str, DatabaseSubscription] + + +def rudp_loop( + sock: socket.socket, + advertise_host: str, + tcp_port: int, + rudp_port: int, + state: PeerState, + stop: threading.Event, +) -> None: + peers: dict[tuple[str, int], RudpPeerConnection] = {} + while not stop.is_set(): + try: + wire, remote = sock.recvfrom(65535) + except TimeoutError: + poll_rudp_resends(sock, peers) + continue + except OSError: + break + try: + packet = decode_rudp_packet(wire) + except ValueError: + continue + if packet.connection_id != RUDP_CONNECTION_ID: + continue + peer = peers.get(remote) + if packet.packet_type == CultNetRudpPacketType.CONNECT: + peer = RudpPeerConnection( + session=CultNetRudpSession( + CultNetRudpSessionOptions( + connection_id=RUDP_CONNECTION_ID, + resend_delay_ms=25, + ) + ), + subscriptions={}, + ) + peers[remote] = peer + send_rudp_packet(sock, remote, peer.session.accept_connect(packet, now_ms(), b"cultnet-interop-rudp")) + continue + if peer is None: + continue + try: + result = peer.session.receive(packet, now_ms()) + if result.reply is not None: + send_rudp_packet(sock, remote, result.reply) + if result.disconnected: + peers.pop(remote, None) + continue + for frame in result.delivered: + if frame.channel_id != "schema": + continue + message = msgpack.unpackb(frame.payload, raw=False) + if not isinstance(message, dict): + continue + for response in handle_server_message(state, message, peer.subscriptions): + send_rudp_schema_frame(sock, remote, peer.session, response) + if packet.packet_type == CultNetRudpPacketType.DATA or result.delivered: + send_rudp_packet(sock, remote, peer.session.create_ack()) + except Exception as error: + sys.stderr.write(json.dumps({ + "event": "rudpMessageError", + "runtimeId": state.runtime_id, + "error": str(error), + }) + "\n") + sys.stderr.flush() + + +def poll_rudp_resends(sock: socket.socket, peers: dict[tuple[str, int], RudpPeerConnection]) -> None: + current = now_ms() + for remote, peer in list(peers.items()): + for packet in peer.session.due_resends(current): + send_rudp_packet(sock, remote, packet) + + +def send_rudp_schema_frame( + sock: socket.socket, + remote: tuple[str, int], + session: CultNetRudpSession, + message: dict[str, Any], +) -> None: + payload = msgpack.packb(message, use_bin_type=True) + for packet in session.send_many( + "schema", + payload, + CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=now_ms()), + max_fragment_bytes=RUDP_INTEROP_MAX_FRAGMENT_BYTES, + ): + send_rudp_packet(sock, remote, packet) + + +def send_rudp_packet(sock: socket.socket, remote: tuple[str, int], packet: CultNetRudpPacket) -> None: + sock.sendto(encode_rudp_packet(packet), remote) + + def handle_server_message(state: PeerState, message: dict[str, Any], subscriptions: dict[str, DatabaseSubscription]) -> list[dict[str, Any]]: schema_version = message.get("schemaVersion") if schema_version == "cultnet.hello.v0": @@ -549,7 +673,7 @@ def database_delete_notifications(message: dict[str, Any], subscriptions: dict[s return notifications -def discovery_loop(sock: socket.socket, advertise_host: str, tcp_port: int, state: PeerState, stop: threading.Event) -> None: +def discovery_loop(sock: socket.socket, advertise_host: str, tcp_port: int, rudp_port: int, state: PeerState, stop: threading.Event) -> None: while not stop.is_set(): try: packet, remote = sock.recvfrom(65536) @@ -573,14 +697,7 @@ def discovery_loop(sock: socket.socket, advertise_host: str, tcp_port: int, stat "tcpHost": advertise_host, "tcpPort": tcp_port, "wireContract": INTEROP_WIRE_CONTRACT, - "transportProfiles": [ - create_tcp_framed_transport_profile( - state.runtime_id, - transport_id="interop-tcp", - host=advertise_host, - port=tcp_port, - ) - ], + "transportProfiles": interop_transport_profiles(state.runtime_id, advertise_host, tcp_port, rudp_port), "supportedDocumentTypes": [INTEROP_DOCUMENT_TYPE], "supportsSchemaCatalog": True, } @@ -618,9 +735,23 @@ def probe(args: argparse.Namespace) -> None: def dial(args: argparse.Namespace) -> None: + if args.target_port is None and args.target_rudp_port is None: + raise ValueError("dial mode requires --target-port or --target-rudp-port") state = build_state(args.runtime_id, args.runtime_kind, args.display_name, args.agent_id, args.schema_path, store_suffix="-dial") - with socket.create_connection((args.target_host, args.target_port), timeout=args.timeout_ms / 1000) as client: - stream = client.makefile("rwb", buffering=0) + tcp_client: socket.socket | None = None + if args.target_rudp_port is not None: + transport = create_rudp_schema_transport( + host=args.target_host, + port=args.target_rudp_port, + connection_id=RUDP_CONNECTION_ID, + timeout_seconds=args.timeout_ms / 1000, + runtime_id=f"{args.runtime_id}-interop-rudp-dial", + transport_id="interop-rudp", + max_fragment_bytes=RUDP_INTEROP_MAX_FRAGMENT_BYTES, + ) + else: + tcp_client = socket.create_connection((args.target_host, args.target_port), timeout=args.timeout_ms / 1000) + stream = tcp_client.makefile("rwb", buffering=0) transport = TcpFramedTransportConnection( stream, profile=create_tcp_framed_transport_profile( @@ -629,6 +760,8 @@ def dial(args: argparse.Namespace) -> None: port=args.target_port, ), ) + try: + target_port = args.target_rudp_port if args.target_rudp_port is not None else args.target_port write_message(transport, hello_message(state)) remote_hello = read_until(transport, lambda message: message.get("schemaVersion") == "cultnet.hello.v0", args.timeout_ms) @@ -669,12 +802,17 @@ def dial(args: argparse.Namespace) -> None: write_message(transport, raw_document_put(state.bindings_by_schema_id[INTEROP_FIRE_COMMAND_SCHEMA_ID], f"{args.runtime_id}-fire-put", command["commandId"], command, state)) fire_receipt_message = read_until(transport, lambda candidate: candidate.get("schemaVersion") == "cultnet.document_put_raw.v0" and candidate.get("document", {}).get("schemaId") == INTEROP_FIRE_RECEIPT_SCHEMA_ID, args.timeout_ms) fire_receipt = apply_raw_document_put(state, fire_receipt_message["document"]) + finally: + transport.close() + if tcp_client is not None: + tcp_client.close() write_json({ "mode": "dial", "runtimeId": args.runtime_id, "targetHost": args.target_host, - "targetPort": args.target_port, + "targetPort": target_port, + "transport": "rudp" if args.target_rudp_port is not None else "tcp_framed", "remoteHello": remote_hello, "hasInteropSchema": has_schema, "retrievedNote": note, @@ -692,6 +830,7 @@ def build_state( schema_path: str, store_suffix: str = "", shard_endpoint: str | None = None, + transport_profiles: list[dict[str, Any]] | None = None, ) -> PeerState: schema_json = Path(schema_path).read_text(encoding="utf-8") schema = json.loads(schema_json) @@ -718,6 +857,7 @@ def build_state( shard_id="interop", shard_epoch=1, shard_endpoint=shard_endpoint or f"cultnet://{runtime_id}", + transport_profiles=transport_profiles or [], shard_log=[], shard_log_lock=threading.Lock(), observations={}, @@ -810,7 +950,7 @@ def binding( def hello_message(state: PeerState) -> dict[str, Any]: - return { + message = { "schemaVersion": "cultnet.hello.v0", "runtimeId": state.runtime_id, "runtimeKind": state.runtime_kind, @@ -828,6 +968,26 @@ def hello_message(state: PeerState) -> dict[str, Any]: "supportedMessageVersions": [INTEROP_SCHEMA_VERSION], "supportsSchemaCatalog": True, } + if state.transport_profiles: + message["transportProfiles"] = state.transport_profiles + return message + + +def interop_transport_profiles(runtime_id: str, host: str, tcp_port: int, rudp_port: int) -> list[dict[str, Any]]: + return [ + create_tcp_framed_transport_profile( + runtime_id, + transport_id="interop-tcp", + host=host, + port=tcp_port, + ), + create_rudp_transport_profile( + runtime_id, + transport_id="interop-rudp", + host=host, + port=rudp_port, + ), + ] def catalog_response(state: PeerState, request: dict[str, Any]) -> dict[str, Any]: @@ -1001,6 +1161,10 @@ def read_until(transport: TcpFramedTransportConnection, predicate: Callable[[dic raise TimeoutError(f"Timed out waiting for CultNet message after {timeout_ms}ms") +def now_ms() -> int: + return int(time.time() * 1000) + + def build_interop_note(runtime_id: str, display_name: str) -> dict[str, Any]: return { "schemaVersion": INTEROP_SCHEMA_VERSION, diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts index 3f680d36..3f982239 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts @@ -403,6 +403,7 @@ test("CultNet TS/Rust/C#/Python peers discover each other and exchange raw state "ts-csharp-dial", "python-peer", "python-client-dial", + "python-rudp-client-dial", ]); const discoveryPort = await getFreePort(); @@ -592,6 +593,25 @@ test("CultNet TS/Rust/C#/Python peers discover each other and exchange raw state assert.ok(pythonDial.mutatedNote.tags.includes("decorated:python-client")); assert.equal(pythonDial.fireReceipt.accepted, true); + logInteropPhase("schema-v0", "python dials python over rudp"); + const pythonRudpDial = await runJsonCommand("python-rudp-dial", pythonCommand, [ + "-m", "cultnet_py.interop_peer", + "dial", + "--runtime-id", "python-rudp-client", + "--runtime-kind", "python", + "--display-name", "Python RUDP Dialer", + "--agent-id", "python-rudp-client-agent", + "--target-host", "127.0.0.1", + "--target-rudp-port", String(pythonPort), + "--schema-path", interopSchemaPath, + ], cultcachePyRoot, { PYTHONPATH: cultcachePySrc }); + assert.equal(pythonRudpDial.transport, "rudp"); + assert.equal(pythonRudpDial.remoteHello.runtimeId, "python-peer"); + assert.equal(pythonRudpDial.hasInteropSchema, true); + assert.equal(pythonRudpDial.retrievedNote.authorRuntimeId, "python-peer"); + assert.ok(pythonRudpDial.mutatedNote.tags.includes("decorated:python-rudp-client")); + assert.equal(pythonRudpDial.fireReceipt.accepted, true); + logInteropPhase("schema-v0", "typescript dials python"); const tsDialPython = await runJsonCommand("ts-dial-python", process.execPath, [ tsPeerScript,