From 28dbb37dc6b80d313ce996fe77b96bcf0a2cdac7 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 14 Jun 2026 14:08:19 +0100 Subject: [PATCH] Type Python CultNet peer errors --- packages/cultcache-py/PARITY.md | 5 +- .../cultcache-py/src/cultnet_py/__init__.py | 3 +- .../cultcache-py/src/cultnet_py/client.py | 16 +++- packages/cultcache-py/tests/test_cultcache.py | 76 ++++++++++++++++++- 4 files changed, 95 insertions(+), 5 deletions(-) diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index 75162801..2ec6c76a 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -15,7 +15,10 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti. shard-log membership before stamping shard cursor metadata. - Python CultNet raw client: Fetches schema catalogs, raw snapshots, shard catalogs, shard logs, and live - database subscription changes. + 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. - Python CultNet raw snapshot: Provides typed `CultNetRawSnapshotResponse` and `CultNetRawDocumentRecord` helpers plus `CultNetRawClient.fetch_snapshot_response(...)` for inspecting diff --git a/packages/cultcache-py/src/cultnet_py/__init__.py b/packages/cultcache-py/src/cultnet_py/__init__.py index 144c21a6..297a3e90 100644 --- a/packages/cultcache-py/src/cultnet_py/__init__.py +++ b/packages/cultcache-py/src/cultnet_py/__init__.py @@ -1,4 +1,4 @@ -from .client import CultNetDatabaseSubscription, CultNetRawClient +from .client import CultNetDatabaseSubscription, CultNetPeerError, CultNetRawClient from .framing import decode_frame, encode_frame, read_frame, write_frame from .messages import ( CultNetMessage, @@ -68,6 +68,7 @@ __all__ = [ "CultNetWitnessArtifactBundle", "CultNetDatabaseSubscription", "CultNetDatabaseChange", + "CultNetPeerError", "CultNetRawClient", "CultNetSchemaCatalog", "CultNetSchemaDescriptor", diff --git a/packages/cultcache-py/src/cultnet_py/client.py b/packages/cultcache-py/src/cultnet_py/client.py index 97fa99c7..1fe54e50 100644 --- a/packages/cultcache-py/src/cultnet_py/client.py +++ b/packages/cultcache-py/src/cultnet_py/client.py @@ -23,6 +23,13 @@ from .snapshot import CultNetRawSnapshotResponse from .subscription import CultNetDatabaseChange +class CultNetPeerError(RuntimeError): + def __init__(self, response: dict[str, Any]) -> None: + self.response = dict(response) + self.peer_error = str(response.get("error") or "CultNet peer returned an error") + super().__init__(self.peer_error) + + @dataclass(frozen=True) class CultNetRawClient: host: str @@ -49,8 +56,7 @@ class CultNetRawClient: raise ValueError("CultNet response must be a MessagePack map") schema_version = response.get("schemaVersion") if schema_version != expected_schema_version: - if schema_version == "cultnet.error.v0": - raise ValueError(str(response.get("error") or "CultNet peer returned an error")) + _raise_peer_error(response) raise ValueError(f"Expected {expected_schema_version}, received {schema_version!r}") return response @@ -296,6 +302,7 @@ class CultNetDatabaseSubscription: if not isinstance(response, dict): raise ValueError("CultNet subscription response must be a MessagePack map") schema_version = response.get("schemaVersion") + _raise_peer_error(response) if schema_version not in {"cultnet.snapshot_response_raw.v0", "cultnet.database_change_raw.v0"}: raise ValueError(f"Unexpected CultNet subscription message {schema_version!r}") return response @@ -325,3 +332,8 @@ class CultNetDatabaseSubscription: while max_messages is None or received < max_messages: yield self.read_next_change() received += 1 + + +def _raise_peer_error(response: dict[str, Any]) -> None: + if response.get("schemaVersion") == "cultnet.error.v0": + raise CultNetPeerError(response) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 9ade7e9c..46221345 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -49,6 +49,7 @@ from cultnet_py import ( CultNetSchemaWriteForwarder, CultNetSchemaShardLogFetcher, CultNetSchemaShardSnapshotFetcher, + CultNetPeerError, CultNetShardReplicator, CultNetShardReplicatorOptions, CultNetShardMutationLogStore, @@ -621,6 +622,43 @@ class CultCacheTests(unittest.TestCase): "cultnet.shard_log_request.v0", ]) + def test_cultnet_raw_client_raises_typed_peer_error_response(self) -> None: + import msgpack # type: ignore + from cultnet_py import read_frame, write_frame + + ready = threading.Event() + port_holder: list[int] = [] + + def serve_error() -> None: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server: + server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server.bind(("127.0.0.1", 0)) + port_holder.append(server.getsockname()[1]) + server.listen(1) + ready.set() + connection, _ = server.accept() + with connection: + stream = connection.makefile("rwb") + read_frame(stream) + write_frame(stream, msgpack.packb({ + "schemaVersion": "cultnet.error.v0", + "messageId": "error-1", + "error": "Snapshot document limit exceeded", + }, use_bin_type=True)) + stream.flush() + + thread = threading.Thread(target=serve_error, daemon=True) + thread.start() + self.assertTrue(ready.wait(2.0)) + + client = CultNetRawClient("127.0.0.1", port_holder[0], timeout_seconds=2.0) + with self.assertRaisesRegex(CultNetPeerError, "Snapshot document limit exceeded") as raised: + client.fetch_snapshot() + + self.assertEqual(raised.exception.peer_error, "Snapshot document limit exceeded") + self.assertEqual(raised.exception.response["schemaVersion"], "cultnet.error.v0") + thread.join(2.0) + def test_cultnet_database_subscription_reads_snapshot_and_change(self) -> None: import msgpack # type: ignore from cultnet_py import read_frame, write_frame @@ -713,6 +751,42 @@ class CultCacheTests(unittest.TestCase): "cultnet.database_unsubscribe.v0", ]) + def test_cultnet_database_subscription_raises_typed_peer_error(self) -> None: + import msgpack # type: ignore + from cultnet_py import read_frame, write_frame + + ready = threading.Event() + port_holder: list[int] = [] + + def serve_subscription_error() -> None: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server: + server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server.bind(("127.0.0.1", 0)) + port_holder.append(server.getsockname()[1]) + server.listen(1) + ready.set() + connection, _ = server.accept() + with connection: + stream = connection.makefile("rwb") + read_frame(stream) + write_frame(stream, msgpack.packb({ + "schemaVersion": "cultnet.error.v0", + "messageId": "subscription-error", + "error": "subscription rejected", + }, use_bin_type=True)) + stream.flush() + + thread = threading.Thread(target=serve_subscription_error, daemon=True) + thread.start() + self.assertTrue(ready.wait(2.0)) + + client = CultNetRawClient("127.0.0.1", port_holder[0], timeout_seconds=2.0) + with client.subscribe_database(subscription_id="sub-error") as subscription: + with self.assertRaisesRegex(CultNetPeerError, "subscription rejected") as raised: + subscription.read_next() + self.assertEqual(raised.exception.response["messageId"], "subscription-error") + thread.join(2.0) + def test_cultnet_replication_helpers_apply_raw_snapshot_and_shard_log(self) -> None: document = define_database_entry_type( "replica.item", @@ -3076,7 +3150,7 @@ class CultCacheTests(unittest.TestCase): snapshot_request(schema_ids=["mesh.snapshot_limit_note.v1"]), expected_schema_version="cultnet.error.v0", ) - with self.assertRaisesRegex(ValueError, "Snapshot document limit exceeded"): + with self.assertRaisesRegex(CultNetPeerError, "Snapshot document limit exceeded"): raw_client.fetch_snapshot(schema_ids=["mesh.snapshot_limit_note.v1"]) finally: server.stop()