diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 6c2c5513..9789a8cd 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -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 diff --git a/packages/cultcache-py/src/cultcache_py/verify.py b/packages/cultcache-py/src/cultcache_py/verify.py index 7772b428..b6a85430 100644 --- a/packages/cultcache-py/src/cultcache_py/verify.py +++ b/packages/cultcache-py/src/cultcache_py/verify.py @@ -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", diff --git a/packages/cultcache-py/src/cultmesh_py/server.py b/packages/cultcache-py/src/cultmesh_py/server.py index dab652df..a98f4376 100644 --- a/packages/cultcache-py/src/cultmesh_py/server.py +++ b/packages/cultcache-py/src/cultmesh_py/server.py @@ -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, } diff --git a/packages/cultcache-py/src/cultnet_py/__init__.py b/packages/cultcache-py/src/cultnet_py/__init__.py index bfc3add7..4b33e74e 100644 --- a/packages/cultcache-py/src/cultnet_py/__init__.py +++ b/packages/cultcache-py/src/cultnet_py/__init__.py @@ -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", diff --git a/packages/cultcache-py/src/cultnet_py/client.py b/packages/cultcache-py/src/cultnet_py/client.py index 1fe54e50..9fafcc5c 100644 --- a/packages/cultcache-py/src/cultnet_py/client.py +++ b/packages/cultcache-py/src/cultnet_py/client.py @@ -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") diff --git a/packages/cultcache-py/src/cultnet_py/interop_peer.py b/packages/cultcache-py/src/cultnet_py/interop_peer.py index 36924558..7f093938 100644 --- a/packages/cultcache-py/src/cultnet_py/interop_peer.py +++ b/packages/cultcache-py/src/cultnet_py/interop_peer.py @@ -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, diff --git a/packages/cultcache-py/src/cultnet_py/transport.py b/packages/cultcache-py/src/cultnet_py/transport.py new file mode 100644 index 00000000..3053825e --- /dev/null +++ b/packages/cultcache-py/src/cultnet_py/transport.py @@ -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() diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index d170dca2..0d2c27ca 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -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",