Add Python CultMesh RUDP server listener
This commit is contained in:
Meta Vens
parent
b72bc7803b
commit
8da99394cc
6 changed files with 192 additions and 5 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 |
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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"])
|
||||
|
|
|
|||
Loading…
Reference in a new issue