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

Type Python CultNet peer errors

Browse source
This commit is contained in:
Meta Vens 2026-06-14 14:08:19 +01:00
commit 28dbb37dc6
4 changed files with 95 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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

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

@ -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",

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

@ -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)

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

@ -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()