GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Add Python RUDP schema client transport

Browse source
This commit is contained in:
Meta Vens 2026-06-16 23:37:59 +01:00
commit ce9d709dae
8 changed files with 202 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

6
docs/cultnet-transport-parity.md
View file

@ -301,7 +301,11 @@ Current progress:
`CultNetSchemaTransport` factory for request/response and database
subscription flows. The default factory opens the existing TCP-framed schema
transport, but the raw client no longer owns inline socket setup as its
request truth.
request truth. Python now also exposes `create_rudp_schema_transport(...)`,
which handshakes a socket-backed RUDP client and gives `CultNetRawClient` the
same reliable ordered `schema` request path over `cultnet.transport.rudp.v0`;
`CultMesh.create_client(endpoint="rudp://...")` is the branded ergonomic
helper for that path.
- Python's branded `CultMesh` facade now creates schema catalogs, built-in
schema catalogs, and shard catalogs by delegating to the existing
`cultnet_py` owners, matching the TypeScript/Kotlin/Rust entrypoint pattern

14
packages/cultcache-py/PARITY.md
View file

@ -22,6 +22,10 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
subscriptions now consume a caller-provided `CultNetSchemaTransport` factory,
with `create_tcp_framed_schema_transport(...)` as the default, so request and
subscription semantics are no longer owned by inline TCP socket setup.
`create_rudp_schema_transport(...)` opens the same raw-client request path
over a socket-backed RUDP reliable ordered `schema` channel, and the branded
`CultMesh.create_client(endpoint="rudp://...")` helper selects that transport
from a RUDP endpoint without making callers assemble a factory closure.
- Python CultNet raw snapshot:
Provides typed `CultNetRawSnapshotResponse` and `CultNetRawDocumentRecord`
helpers plus `CultNetRawClient.fetch_snapshot_response(...)` for inspecting
@ -303,10 +307,12 @@ node --test packages\cultnet-ts\dist-test\test\interop\cultnet-interop.test.js
port instead of cloning LiteNetLib. The raw client request/subscription layer
now routes through a schema-transport factory, so the caller-facing semantics
are transport-port-shaped even when the default factory opens TCP-framed
connections. The reconnect policy/controller primitive is portable and
advertised by RUDP profiles, and `CultNetRudpReconnectLoop` gives Python
callers the same RUDP retry-scheduling helper as TypeScript/Kotlin.
Automatic reconnect use by Python service bodies is not yet claimed.
connections. Python raw clients can now use a RUDP schema transport directly
through `create_rudp_schema_transport(...)` or the `CultMesh.create_client`
RUDP endpoint helper. The reconnect policy/controller primitive is portable
and advertised by RUDP profiles, and `CultNetRudpReconnectLoop` gives Python
callers the same RUDP retry-scheduling helper as TypeScript/Kotlin. Automatic
reconnect use by Python service bodies is not yet claimed.
- Python does not implement the full C# daemon/server stack. Its current role is
package runtime, raw interop peer, launchable local CultMesh endpoint, local
CultMesh session facade, and typed state participant.

10
packages/cultcache-py/README.md
View file

@ -169,6 +169,7 @@ from cultnet_py import (
simulation_observation,
witness_artifact_bundle,
)
from cultmesh_py import CultMesh
payload = hello(runtime_id="python-runtime").to_bytes()
message = parse_message(decode_frame(encode_frame(payload)))
@ -186,6 +187,15 @@ shard_log = client.fetch_shard_log_response(shard_id="interop", shard_epoch=1, a
# log_changes = apply_shard_log_response(cache, [note_doc], shard_log)
with client.subscribe_database(subscription_id="ui", schema_ids=["cultnet.interop-note"]) as live:
initial_snapshot = live.read_next_snapshot_response()
# The same raw-client request surface can ride RUDP when the peer advertises a
# rudp:// endpoint and shares the connection id for that lane.
rudp_client = CultMesh.create_client(
endpoint="rudp://127.0.0.1:3076",
connection_id=0x43554C54,
)
rudp_catalog = rudp_client.fetch_schema_catalog(kinds=["wire_message"])
claim_hash = compute_simulation_claim_hash("frame:42", "subject:player-1", "hit")
observation = simulation_observation(
message_id="obs-1",

46
packages/cultcache-py/src/cultmesh_py/facade.py
View file

@ -16,6 +16,7 @@ from cultnet_py import (
CultNetSchemaCatalog,
CultNetShardCatalog,
CultNetSimulationObservationHub,
create_rudp_schema_transport,
wire_message_schema_catalog,
)
@ -310,8 +311,35 @@ class CultMesh:
port: int = 3075,
*,
timeout_seconds: float = 4.0,
endpoint: str | CultMeshRudpEndpoint | None = None,
connection_id: int = 0x43554C54,
runtime_id: str = "cultmesh-python-rudp-client",
bind_host: str = "127.0.0.1",
bind_port: int = 0,
) -> CultNetRawClient:
return CultNetRawClient(host, port, timeout_seconds)
if endpoint is None and host.lower().startswith("rudp://"):
endpoint = host
if endpoint is None:
return CultNetRawClient(host, port, timeout_seconds)
parsed_endpoint = (
CultMesh.parse_rudp_endpoint(endpoint)
if isinstance(endpoint, str)
else endpoint
)
return CultNetRawClient(
parsed_endpoint.host,
parsed_endpoint.port,
timeout_seconds,
create_transport=lambda: create_rudp_schema_transport(
host=parsed_endpoint.host,
port=parsed_endpoint.port,
connection_id=connection_id,
timeout_seconds=timeout_seconds,
runtime_id=runtime_id,
bind_host=bind_host,
bind_port=bind_port,
),
)
@staticmethod
def connect_client(
@ -319,8 +347,22 @@ class CultMesh:
port: int = 3075,
*,
timeout_seconds: float = 4.0,
endpoint: str | CultMeshRudpEndpoint | None = None,
connection_id: int = 0x43554C54,
runtime_id: str = "cultmesh-python-rudp-client",
bind_host: str = "127.0.0.1",
bind_port: int = 0,
) -> CultNetRawClient:
return CultNetRawClient(host, port, timeout_seconds)
return CultMesh.create_client(
host,
port,
timeout_seconds=timeout_seconds,
endpoint=endpoint,
connection_id=connection_id,
runtime_id=runtime_id,
bind_host=bind_host,
bind_port=bind_port,
)
def _bind_rudp_socket(bind_host: str, bind_port: int) -> socket_module.socket:

2
packages/cultcache-py/src/cultnet_py/__init__.py
View file

@ -4,6 +4,7 @@ from .client import (
CultNetRawClient,
CultNetSchemaTransport,
CultNetSchemaTransportFactory,
create_rudp_schema_transport,
create_tcp_framed_schema_transport,
)
from .framing import decode_frame, encode_frame, read_frame, write_frame
@ -170,6 +171,7 @@ __all__ = [
"compute_simulation_claim_hash",
"compute_reconnect_delay_ms",
"create_reconnect_policy",
"create_rudp_schema_transport",
"create_tcp_framed_schema_transport",
"create_rudp_transport_profile",
"create_tcp_framed_transport_profile",

58
packages/cultcache-py/src/cultnet_py/client.py
View file

@ -1,6 +1,7 @@
from __future__ import annotations
import socket
import time
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any, Protocol
@ -21,7 +22,13 @@ from .shard_catalog import CultNetShardCatalog, CultNetShardDescriptor
from .shard_log import CultNetShardLogResponse
from .snapshot import CultNetRawSnapshotResponse
from .subscription import CultNetDatabaseChange
from .transport import TcpFramedTransportConnection, create_tcp_framed_transport_profile
from .transport import (
CultNetRudpSocketMode,
CultNetRudpSocketTransportConnection,
CultNetRudpSocketTransportOptions,
TcpFramedTransportConnection,
create_tcp_framed_transport_profile,
)
class CultNetSchemaTransport(Protocol):
@ -377,6 +384,55 @@ def create_tcp_framed_schema_transport(
)
def create_rudp_schema_transport(
*,
host: str,
port: int,
connection_id: int,
timeout_seconds: float = 4.0,
runtime_id: str = "cultnet-python-rudp-client",
bind_host: str = "127.0.0.1",
bind_port: int = 0,
initial_sequence: int = 1,
resend_delay_ms: int = 25,
transport_id: str = "rudp",
max_payload_bytes: int | None = None,
max_fragment_bytes: int | None = None,
max_pending_reliable_packets: int | None = None,
) -> CultNetRudpSocketTransportConnection:
transport_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
transport_socket.bind((bind_host, bind_port))
transport_socket.settimeout(min(timeout_seconds, 0.02))
transport = CultNetRudpSocketTransportConnection(
CultNetRudpSocketTransportOptions(
runtime_id=runtime_id,
socket=transport_socket,
mode=CultNetRudpSocketMode.CLIENT,
remote_addr=(host, port),
connection_id=connection_id,
initial_sequence=initial_sequence,
resend_delay_ms=resend_delay_ms,
transport_id=transport_id,
max_payload_bytes=max_payload_bytes,
max_fragment_bytes=max_fragment_bytes,
max_pending_reliable_packets=max_pending_reliable_packets,
)
)
try:
transport.connect(b"cultnet-schema")
deadline = time.monotonic() + timeout_seconds
while not transport.connected:
transport.receive_once()
transport.poll_resends()
if time.monotonic() >= deadline:
raise TimeoutError("Timed out waiting for RUDP schema transport handshake")
return transport
except Exception:
transport.close()
raise
class _OwnedTcpFramedTransportConnection(TcpFramedTransportConnection):
def __init__(self, stream: Any, connection: socket.socket, *, profile: dict[str, Any]) -> None:
super().__init__(stream, profile=profile)

16
packages/cultcache-py/src/cultnet_py/transport.py
View file

@ -819,6 +819,12 @@ class CultNetRudpSocketTransportConnection:
self.disconnect_reason: bytes | None = None
self.pong_payloads: deque[bytes] = deque()
def __enter__(self) -> "CultNetRudpSocketTransportConnection":
return self
def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None:
self.close()
@property
def connected(self) -> bool:
return self.session.connected
@ -890,6 +896,16 @@ class CultNetRudpSocketTransportConnection:
self._send_packet(self.session.create_ack())
return frame
def receive(self, timeout_seconds: float | None = None) -> CultNetTransportFrame:
deadline = None if timeout_seconds is None else time.monotonic() + timeout_seconds
while True:
frame = self.receive_once()
if frame is not None:
return frame
self.poll_resends()
if deadline is not None and time.monotonic() >= deadline:
raise TimeoutError("Timed out waiting for RUDP schema frame")
def disconnect(self, reason: bytes = b"") -> None:
self._send_packet(self.session.create_disconnect(reason))

58
packages/cultcache-py/tests/test_cultcache.py
View file

@ -668,6 +668,64 @@ class CultCacheTests(unittest.TestCase):
self.assertEqual(sent_messages[0]["schemaVersion"], "cultnet.schema_catalog_request.v0")
self.assertEqual(sent_messages[0]["messageId"], "transport-factory")
def test_cultmesh_client_uses_rudp_schema_transport_endpoint(self) -> None:
import msgpack # type: ignore
server_socket = bind_udp_socket()
server_host, server_port = server_socket.getsockname()[:2]
connection_id = 0x43554C54
server = CultNetRudpSocketTransportConnection(
CultNetRudpSocketTransportOptions(
runtime_id="python-rudp-schema-server",
socket=server_socket,
mode=CultNetRudpSocketMode.SERVER,
connection_id=connection_id,
initial_sequence=100,
resend_delay_ms=25,
)
)
received_messages: list[dict[str, object]] = []
def serve_once() -> None:
deadline = time.monotonic() + 2.0
try:
while time.monotonic() < deadline:
frame = server.receive_once()
server.poll_resends()
if frame is None:
time.sleep(0.005)
continue
received = msgpack.unpackb(frame.payload, raw=False)
received_messages.append(received)
response = {
"schemaVersion": "cultnet.schema_catalog_response.v0",
"messageId": received["messageId"],
"schemas": [],
}
server.send("schema", msgpack.packb(response, use_bin_type=True))
return
raise AssertionError("RUDP schema server did not receive a request")
finally:
server.close()
server_thread = threading.Thread(target=serve_once, daemon=True)
server_thread.start()
client = CultMesh.create_client(
endpoint=f"rudp://{server_host}:{server_port}",
timeout_seconds=2.0,
connection_id=connection_id,
runtime_id="python-rudp-schema-client",
)
response = client.fetch_schema_catalog(message_id="rudp-transport-factory")
server_thread.join(timeout=2.0)
self.assertFalse(server_thread.is_alive())
self.assertEqual(response["schemaVersion"], "cultnet.schema_catalog_response.v0")
self.assertEqual(response["messageId"], "rudp-transport-factory")
self.assertEqual(received_messages[0]["schemaVersion"], "cultnet.schema_catalog_request.v0")
self.assertEqual(received_messages[0]["messageId"], "rudp-transport-factory")
def test_cultnet_rudp_packet_codec_uses_deterministic_reliable_ordered_fixture(self) -> None:
encoded = encode_rudp_packet(
CultNetRudpPacket(