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

Add Python CultNet transport port

Browse source
This commit is contained in:
Meta Vens 2026-06-14 21:38:59 +01:00
commit 3fa04cc014
8 changed files with 191 additions and 56 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -142,9 +142,14 @@ Current progress:
frame/close/error events, `send(channel, payload)`, `close`, and transfer
stats, and `CultNetPeer` can now run over either the legacy `Duplex` stream
path or the transport connection path.
- The remaining parity work is to add equivalent ports to C#, Rust, Python, and
Kotlin, then move each runtime's existing TCP/LiteNetLib/WebSocket bodies
behind those ports before implementing `rudp`.
- Python now has the same `tcp_framed` transport profile helper plus a
synchronous `TcpFramedTransportConnection` with `send`, `receive`, `close`,
and stats. `CultNetRawClient` and database subscriptions send/read through
that port instead of owning frame I/O directly.
- The remaining parity work is to add equivalent ports to C#, Rust, and Kotlin,
deepen Python's server-side use of its port, and move each runtime's existing
TCP/LiteNetLib/WebSocket bodies behind those ports before implementing
`rudp`.
## Reliable UDP Semantics To Lift From LiteNetLib

4
packages/cultcache-py/src/cultcache_py/verify.py
View file

@ -65,12 +65,16 @@ EXPECTED_EXPORTS: dict[str, tuple[str, ...]] = {
"CultNetSimulationConsensusOptions",
"CultNetSimulationObservation",
"CultNetSimulationObservationHub",
"CultNetTransportFrame",
"CultNetTransportStats",
"TcpFramedTransportConnection",
"INTEROP_WIRE_CONTRACT",
"ValidatedCultNetSessionToken",
"apply_raw_document_record",
"apply_raw_snapshot",
"apply_shard_log_response",
"compute_simulation_claim_hash",
"create_tcp_framed_transport_profile",
"database_subscribe",
"database_unsubscribe",
"decode_frame",

34
packages/cultcache-py/src/cultmesh_py/server.py
View file

@ -7,7 +7,13 @@ from typing import Any
import msgpack
from cultnet_py import CultNetSimulationObservationHub, read_frame, wire_message_schema_descriptors, write_frame
from cultnet_py import (
CultNetSimulationObservationHub,
create_tcp_framed_transport_profile,
read_frame,
wire_message_schema_descriptors,
write_frame,
)
from .node import CultMeshNode
from .wire import (
@ -309,26 +315,12 @@ class CultMeshLocalServer:
else []
),
"transportProfiles": [
{
"schemaVersion": "cultnet.transport_profile.v0",
"runtimeId": self.node.runtime_id,
"transports": [
{
"transportId": "cultmesh-local-tcp",
"protocol": "tcp_framed",
"host": self.host,
"port": self.port,
"wireContracts": ["cultnet.schema.v0"],
"channels": [
{
"channelId": "schema",
"delivery": "reliable",
"ordering": "ordered",
}
],
}
],
}
create_tcp_framed_transport_profile(
self.node.runtime_id,
transport_id="cultmesh-local-tcp",
host=self.host,
port=self.port,
)
],
"supportsSchemaCatalog": True,
}

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

@ -65,6 +65,12 @@ from .shard_replication import (
)
from .snapshot import CultNetRawDocumentRecord, CultNetRawSnapshotResponse
from .subscription import CultNetDatabaseChange
from .transport import (
CultNetTransportFrame,
CultNetTransportStats,
TcpFramedTransportConnection,
create_tcp_framed_transport_profile,
)
from .simulation import (
CultNetSimulationConsensus,
CultNetSimulationConsensusCandidate,
@ -112,11 +118,15 @@ __all__ = [
"CultNetSimulationConsensusOptions",
"CultNetSimulationObservation",
"CultNetSimulationObservationHub",
"CultNetTransportFrame",
"CultNetTransportStats",
"TcpFramedTransportConnection",
"ValidatedCultNetSessionToken",
"apply_raw_document_record",
"apply_raw_snapshot",
"apply_shard_log_response",
"compute_simulation_claim_hash",
"create_tcp_framed_transport_profile",
"database_subscribe",
"database_unsubscribe",
"decode_witness_artifact_bundle_payload",

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

@ -6,7 +6,6 @@ from typing import Any
import msgpack
from .framing import read_frame, write_frame
from .messages import (
CultNetMessage,
database_subscribe,
@ -21,6 +20,7 @@ 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
class CultNetPeerError(RuntimeError):
@ -41,17 +41,23 @@ class CultNetRawClient:
with socket.create_connection((self.host, self.port), timeout=self.timeout_seconds) as connection:
connection.settimeout(self.timeout_seconds)
stream = connection.makefile("rwb")
write_frame(stream, msgpack.packb(wire, use_bin_type=True))
stream.flush()
transport = TcpFramedTransportConnection(
stream,
profile=create_tcp_framed_transport_profile("cultnet-python-client", host=self.host, port=self.port),
)
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")
write_frame(stream, msgpack.packb(wire, use_bin_type=True))
stream.flush()
response = msgpack.unpackb(read_frame(stream), raw=False)
transport = TcpFramedTransportConnection(
stream,
profile=create_tcp_framed_transport_profile("cultnet-python-client", host=self.host, port=self.port),
)
transport.send("schema", msgpack.packb(wire, use_bin_type=True))
response = msgpack.unpackb(transport.receive().payload, raw=False)
if not isinstance(response, dict):
raise ValueError("CultNet response must be a MessagePack map")
schema_version = response.get("schemaVersion")
@ -259,11 +265,16 @@ class CultNetDatabaseSubscription:
include_snapshot: bool = True
_connection: socket.socket | None = None
_stream: Any | None = None
_transport: TcpFramedTransportConnection | 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.send(database_subscribe(
message_id=self.message_id,
subscription_id=self.subscription_id,
@ -281,24 +292,25 @@ class CultNetDatabaseSubscription:
subscription_id=self.subscription_id,
))
finally:
if self._transport is not None:
self._transport.close()
self._transport = None
if self._stream is not None:
self._stream.close()
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._stream is None:
if self._transport is None:
raise RuntimeError("CultNet database subscription is not open")
wire = message.to_wire() if isinstance(message, CultNetMessage) else message
write_frame(self._stream, msgpack.packb(wire, use_bin_type=True))
self._stream.flush()
self._transport.send("schema", msgpack.packb(wire, use_bin_type=True))
def read_next(self) -> dict[str, Any]:
if self._stream is None:
if self._transport is None:
raise RuntimeError("CultNet database subscription is not open")
response = msgpack.unpackb(read_frame(self._stream), raw=False)
response = msgpack.unpackb(self._transport.receive().payload, raw=False)
if not isinstance(response, dict):
raise ValueError("CultNet subscription response must be a MessagePack map")
schema_version = response.get("schemaVersion")

27
packages/cultcache-py/src/cultnet_py/interop_peer.py
View file

@ -39,6 +39,7 @@ from cultmesh_py import (
from .framing import read_frame, write_frame
from .schema_catalog import INTEROP_WIRE_CONTRACT, wire_message_schema_descriptors
from .transport import create_tcp_framed_transport_profile
INTEROP_DOCUMENT_TYPE = "cultnet.interop-note"
INTEROP_SCHEMA_VERSION = "cultnet.interop_note.v0"
@ -570,26 +571,12 @@ def discovery_loop(sock: socket.socket, advertise_host: str, tcp_port: int, stat
"tcpPort": tcp_port,
"wireContract": INTEROP_WIRE_CONTRACT,
"transportProfiles": [
{
"schemaVersion": "cultnet.transport_profile.v0",
"runtimeId": state.runtime_id,
"transports": [
{
"transportId": "interop-tcp",
"protocol": "tcp_framed",
"host": advertise_host,
"port": tcp_port,
"wireContracts": [INTEROP_WIRE_CONTRACT],
"channels": [
{
"channelId": "schema",
"delivery": "reliable",
"ordering": "ordered",
}
],
}
],
}
create_tcp_framed_transport_profile(
state.runtime_id,
transport_id="interop-tcp",
host=advertise_host,
port=tcp_port,
)
],
"supportedDocumentTypes": [INTEROP_DOCUMENT_TYPE],
"supportsSchemaCatalog": True,

98
packages/cultcache-py/src/cultnet_py/transport.py Normal file
View file

@ -0,0 +1,98 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, BinaryIO
from .framing import read_frame, write_frame
@dataclass(frozen=True)
class CultNetTransportStats:
bytes_received: int = 0
bytes_sent: int = 0
frames_received: int = 0
frames_sent: int = 0
@dataclass(frozen=True)
class CultNetTransportFrame:
channel_id: str
payload: bytes
def create_tcp_framed_transport_profile(
runtime_id: str,
*,
transport_id: str = "tcp-framed",
host: str | None = None,
port: int | None = None,
max_payload_bytes: int | None = None,
max_fragment_bytes: int | None = None,
) -> dict[str, Any]:
channel: dict[str, Any] = {
"channelId": "schema",
"delivery": "reliable",
"ordering": "ordered",
}
if max_payload_bytes is not None:
channel["maxPayloadBytes"] = max_payload_bytes
if max_fragment_bytes is not None:
channel["maxFragmentBytes"] = max_fragment_bytes
transport: dict[str, Any] = {
"transportId": transport_id,
"protocol": "tcp_framed",
"wireContracts": ["cultnet.schema.v0"],
"channels": [channel],
}
if host is not None:
transport["host"] = host
if port is not None:
transport["port"] = port
return {
"schemaVersion": "cultnet.transport_profile.v0",
"runtimeId": runtime_id,
"transports": [transport],
}
class TcpFramedTransportConnection:
def __init__(self, stream: BinaryIO, *, profile: dict[str, Any]) -> None:
self.stream = stream
self.profile = dict(profile)
self._bytes_received = 0
self._bytes_sent = 0
self._frames_received = 0
self._frames_sent = 0
@property
def stats(self) -> CultNetTransportStats:
return CultNetTransportStats(
bytes_received=self._bytes_received,
bytes_sent=self._bytes_sent,
frames_received=self._frames_received,
frames_sent=self._frames_sent,
)
def send(self, channel_id: str, payload: bytes) -> None:
if channel_id != "schema":
raise ValueError(f"tcp_framed transport only supports the schema channel, got {channel_id!r}")
before = len(payload) + 4
write_frame(self.stream, payload)
flush = getattr(self.stream, "flush", None)
if callable(flush):
flush()
self._bytes_sent += before
self._frames_sent += 1
def receive(self) -> CultNetTransportFrame:
payload = read_frame(self.stream)
self._bytes_received += len(payload) + 4
self._frames_received += 1
return CultNetTransportFrame(channel_id="schema", payload=payload)
def close(self) -> None:
close = getattr(self.stream, "close", None)
if callable(close):
close()

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

@ -11,6 +11,7 @@ import time
import unittest
import hashlib
import hmac
import io
from dataclasses import asdict, dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
@ -66,10 +67,12 @@ from cultnet_py import (
CultNetSimulationConsensusCandidate,
CultNetSimulationObservation,
CultNetSimulationObservationHub,
TcpFramedTransportConnection,
CultNetWitnessArtifactBundle,
apply_raw_document_record,
apply_raw_snapshot,
apply_shard_log_response,
create_tcp_framed_transport_profile,
database_subscribe,
database_unsubscribe,
decode_frame,
@ -553,6 +556,30 @@ class CultCacheTests(unittest.TestCase):
self.assertEqual(parsed.body["runtimeId"], "python-test")
self.assertEqual(parsed.body["transportProfiles"][0]["transports"][0]["protocol"], "tcp_framed")
def test_cultnet_tcp_framed_transport_carries_schema_payloads_with_stats(self) -> None:
payload = b"cultnet-payload"
output = io.BytesIO()
sender = TcpFramedTransportConnection(
output,
profile=create_tcp_framed_transport_profile("sender", transport_id="test-tcp"),
)
sender.send("schema", payload)
self.assertEqual(sender.stats.frames_sent, 1)
self.assertEqual(sender.stats.bytes_sent, len(payload) + 4)
with self.assertRaisesRegex(ValueError, "only supports the schema channel"):
sender.send("unreliable", b"")
receiver = TcpFramedTransportConnection(
io.BytesIO(output.getvalue()),
profile=create_tcp_framed_transport_profile("receiver", transport_id="test-tcp"),
)
frame = receiver.receive()
self.assertEqual(frame.channel_id, "schema")
self.assertEqual(frame.payload, payload)
self.assertEqual(receiver.stats.frames_received, 1)
self.assertEqual(receiver.stats.bytes_received, len(payload) + 4)
self.assertEqual(receiver.profile["transports"][0]["protocol"], "tcp_framed")
def test_cultnet_database_subscription_helpers_match_schema_v0_shape(self) -> None:
subscribe = database_subscribe(
message_id="sub-message",