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 schema transport client port

Browse source
This commit is contained in:
Meta Vens 2026-06-16 16:13:56 +01:00
commit e65c0d0fc6
7 changed files with 144 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -291,6 +291,11 @@ Current progress:
callers: the caller-owned receive loop reports closure, the shared controller
owns retry delay/exhaustion state, and the caller-provided factory opens the
next `CultNetRudpSocketTransportConnection`.
- Python raw CultNet clients now consume a caller-provided
`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.
- 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

2
docs/runtime-parity-scope.md
View file

@ -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, 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 surfaces | 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, schema catalogs, shard catalogs/logs, local TCP CultMesh server, session/prediction/simulation fact helpers, authority leases, streams, 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, schema catalogs, shard catalogs/logs, schema-transport factory for raw client request/subscription flows, local TCP CultMesh server, session/prediction/simulation fact helpers, authority leases, streams, 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, local CultMesh node/catalog/authority/stream ergonomics | 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, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, small CultMesh facade for peer cards, authority leases, RUDP endpoint parsing, and peer-card 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, stream catalog, branded `CultMesh` facade, WebSocket transport profile/connection helper, RUDP client/server sugar, 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 |

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

@ -18,7 +18,10 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
database subscription changes. Request-response reads and database
subscription streams raise typed `CultNetPeerError` values for
`cultnet.error.v0` peer responses while preserving the raw error document for
callers that need the wire evidence.
callers that need the wire evidence. `CultNetRawClient` and database
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.
- Python CultNet raw snapshot:
Provides typed `CultNetRawSnapshotResponse` and `CultNetRawDocumentRecord`
helpers plus `CultNetRawClient.fetch_snapshot_response(...)` for inspecting
@ -289,11 +292,13 @@ node --test packages\cultnet-ts\dist-test\test\interop\cultnet-interop.test.js
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
remaining work to put broader server-side bodies behind the shared transport
port instead of cloning LiteNetLib. 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.
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.
- 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.

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

@ -1,4 +1,11 @@
from .client import CultNetDatabaseSubscription, CultNetPeerError, CultNetRawClient
from .client import (
CultNetDatabaseSubscription,
CultNetPeerError,
CultNetRawClient,
CultNetSchemaTransport,
CultNetSchemaTransportFactory,
create_tcp_framed_schema_transport,
)
from .framing import decode_frame, encode_frame, read_frame, write_frame
from .messages import (
CultNetMessage,
@ -105,6 +112,8 @@ __all__ = [
"CultNetDatabaseChange",
"CultNetPeerError",
"CultNetRawClient",
"CultNetSchemaTransport",
"CultNetSchemaTransportFactory",
"CultNetClientSecurityOptions",
"CultNetSecret",
"CultNetServerSecurityOptions",
@ -161,6 +170,7 @@ __all__ = [
"compute_simulation_claim_hash",
"compute_reconnect_delay_ms",
"create_reconnect_policy",
"create_tcp_framed_schema_transport",
"create_rudp_transport_profile",
"create_tcp_framed_transport_profile",
"database_subscribe",

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

@ -1,8 +1,9 @@
from __future__ import annotations
import socket
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from typing import Any, Protocol
import msgpack
@ -23,6 +24,17 @@ from .subscription import CultNetDatabaseChange
from .transport import TcpFramedTransportConnection, create_tcp_framed_transport_profile
class CultNetSchemaTransport(Protocol):
def __enter__(self) -> "CultNetSchemaTransport": ...
def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None: ...
def send(self, channel_id: str, payload: bytes) -> None: ...
def receive(self) -> Any: ...
def close(self) -> None: ...
CultNetSchemaTransportFactory = Callable[[], CultNetSchemaTransport]
class CultNetPeerError(RuntimeError):
def __init__(self, response: dict[str, Any]) -> None:
self.response = dict(response)
@ -35,27 +47,16 @@ class CultNetRawClient:
host: str
port: int
timeout_seconds: float = 4.0
create_transport: CultNetSchemaTransportFactory | None = None
def send(self, message: CultNetMessage | dict[str, Any]) -> None:
wire = message.to_wire() if isinstance(message, CultNetMessage) else message
with socket.create_connection((self.host, self.port), timeout=self.timeout_seconds) as connection:
connection.settimeout(self.timeout_seconds)
stream = connection.makefile("rwb")
transport = TcpFramedTransportConnection(
stream,
profile=create_tcp_framed_transport_profile("cultnet-python-client", host=self.host, port=self.port),
)
with self.open_transport() as transport:
transport.send("schema", msgpack.packb(wire, use_bin_type=True))
def request(self, message: CultNetMessage | dict[str, Any], *, expected_schema_version: str) -> dict[str, Any]:
wire = message.to_wire() if isinstance(message, CultNetMessage) else message
with socket.create_connection((self.host, self.port), timeout=self.timeout_seconds) as connection:
connection.settimeout(self.timeout_seconds)
stream = connection.makefile("rwb")
transport = TcpFramedTransportConnection(
stream,
profile=create_tcp_framed_transport_profile("cultnet-python-client", host=self.host, port=self.port),
)
with self.open_transport() as transport:
transport.send("schema", msgpack.packb(wire, use_bin_type=True))
response = msgpack.unpackb(transport.receive().payload, raw=False)
if not isinstance(response, dict):
@ -66,6 +67,16 @@ class CultNetRawClient:
raise ValueError(f"Expected {expected_schema_version}, received {schema_version!r}")
return response
def open_transport(self) -> CultNetSchemaTransport:
if self.create_transport is not None:
return self.create_transport()
return create_tcp_framed_schema_transport(
host=self.host,
port=self.port,
timeout_seconds=self.timeout_seconds,
runtime_id="cultnet-python-client",
)
def fetch_schema_catalog(
self,
*,
@ -245,6 +256,7 @@ class CultNetRawClient:
host=self.host,
port=self.port,
timeout_seconds=self.timeout_seconds,
create_transport=self.create_transport,
subscription_id=subscription_id,
message_id=message_id,
schema_ids=schema_ids,
@ -263,17 +275,19 @@ class CultNetDatabaseSubscription:
schema_ids: list[str] | None = None
record_keys: list[str] | None = None
include_snapshot: bool = True
_connection: socket.socket | None = None
_stream: Any | None = None
_transport: TcpFramedTransportConnection | None = None
create_transport: CultNetSchemaTransportFactory | None = None
_transport: CultNetSchemaTransport | None = None
def __enter__(self) -> "CultNetDatabaseSubscription":
self._connection = socket.create_connection((self.host, self.port), timeout=self.timeout_seconds)
self._connection.settimeout(self.timeout_seconds)
self._stream = self._connection.makefile("rwb")
self._transport = TcpFramedTransportConnection(
self._stream,
profile=create_tcp_framed_transport_profile("cultnet-python-subscription", host=self.host, port=self.port),
self._transport = (
self.create_transport()
if self.create_transport is not None
else create_tcp_framed_schema_transport(
host=self.host,
port=self.port,
timeout_seconds=self.timeout_seconds,
runtime_id="cultnet-python-subscription",
)
)
self.send(database_subscribe(
message_id=self.message_id,
@ -286,7 +300,7 @@ class CultNetDatabaseSubscription:
def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None:
try:
if self._stream is not None:
if self._transport is not None:
self.send(database_unsubscribe(
message_id=f"{self.message_id}-unsubscribe",
subscription_id=self.subscription_id,
@ -295,11 +309,6 @@ class CultNetDatabaseSubscription:
if self._transport is not None:
self._transport.close()
self._transport = None
if self._stream is not None:
self._stream = None
if self._connection is not None:
self._connection.close()
self._connection = None
def send(self, message: CultNetMessage | dict[str, Any]) -> None:
if self._transport is None:
@ -349,3 +358,32 @@ class CultNetDatabaseSubscription:
def _raise_peer_error(response: dict[str, Any]) -> None:
if response.get("schemaVersion") == "cultnet.error.v0":
raise CultNetPeerError(response)
def create_tcp_framed_schema_transport(
*,
host: str,
port: int,
timeout_seconds: float = 4.0,
runtime_id: str = "cultnet-python-client",
) -> TcpFramedTransportConnection:
connection = socket.create_connection((host, port), timeout=timeout_seconds)
connection.settimeout(timeout_seconds)
stream = connection.makefile("rwb")
return _OwnedTcpFramedTransportConnection(
stream,
connection,
profile=create_tcp_framed_transport_profile(runtime_id, host=host, port=port),
)
class _OwnedTcpFramedTransportConnection(TcpFramedTransportConnection):
def __init__(self, stream: Any, connection: socket.socket, *, profile: dict[str, Any]) -> None:
super().__init__(stream, profile=profile)
self._connection = connection
def close(self) -> None:
try:
super().close()
finally:
self._connection.close()

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

@ -1026,6 +1026,12 @@ class TcpFramedTransportConnection:
self._frames_received = 0
self._frames_sent = 0
def __enter__(self) -> "TcpFramedTransportConnection":
return self
def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None:
self.close()
@property
def stats(self) -> CultNetTransportStats:
return CultNetTransportStats(

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

@ -626,6 +626,48 @@ class CultCacheTests(unittest.TestCase):
self.assertEqual(receiver.stats.bytes_received, len(payload) + 4)
self.assertEqual(receiver.profile["transports"][0]["protocol"], "tcp_framed")
def test_cultnet_raw_client_uses_schema_transport_factory(self) -> None:
import msgpack # type: ignore
sent_messages: list[dict[str, object]] = []
test_case = self
class FakeSchemaTransport:
profile = create_tcp_framed_transport_profile("fake-client")
def __enter__(self) -> "FakeSchemaTransport":
return self
def __exit__(self, exc_type: object, exc: object, traceback: object) -> None:
self.close()
def send(self, channel_id: str, payload: bytes) -> None:
test_case.assertEqual(channel_id, "schema")
sent_messages.append(msgpack.unpackb(payload, raw=False))
def receive(self) -> object:
response = {
"schemaVersion": "cultnet.schema_catalog_response.v0",
"messageId": "transport-factory",
"schemas": [],
}
return type("Frame", (), {"channel_id": "schema", "payload": msgpack.packb(response, use_bin_type=True)})()
def close(self) -> None:
pass
client = CultNetRawClient(
"unused.example.test",
1,
create_transport=FakeSchemaTransport,
)
response = client.fetch_schema_catalog(message_id="transport-factory")
self.assertEqual(response["schemaVersion"], "cultnet.schema_catalog_response.v0")
self.assertEqual(sent_messages[0]["schemaVersion"], "cultnet.schema_catalog_request.v0")
self.assertEqual(sent_messages[0]["messageId"], "transport-factory")
def test_cultnet_rudp_packet_codec_uses_deterministic_reliable_ordered_fixture(self) -> None:
encoded = encode_rudp_packet(
CultNetRudpPacket(