From 5699ea68cf596a795c2802e6428fd8abc44ead65 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 14 Jun 2026 14:23:03 +0100 Subject: [PATCH] Return Python CultMesh peer errors --- packages/cultcache-py/PARITY.md | 3 + .../cultcache-py/src/cultmesh_py/server.py | 57 ++++++++++++++++--- packages/cultcache-py/tests/test_cultcache.py | 39 +++++++++++++ 3 files changed, 90 insertions(+), 9 deletions(-) diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index 9a5cb329..2db46c60 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -116,6 +116,9 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti. shared wire-message descriptors for the CultNet/CultMesh messages it handles. Hello and payload schema descriptors advertise the raw snapshot, put, delete, and shard-log mutation surfaces the server actually serves. + Unsupported request schemas, malformed raw put/delete messages, and snapshot + limit violations return structured `cultnet.error.v0` documents instead of + silent stream stalls. Its shard catalog advertises the database-owned shard ids, schema ids, and latest shard epochs so discovery and shard-log catch-up agree on the same authority. diff --git a/packages/cultcache-py/src/cultmesh_py/server.py b/packages/cultcache-py/src/cultmesh_py/server.py index 086eeb71..cf302baa 100644 --- a/packages/cultcache-py/src/cultmesh_py/server.py +++ b/packages/cultcache-py/src/cultmesh_py/server.py @@ -113,7 +113,12 @@ class CultMeshLocalServer: if schema_version == "cultnet.simulation_observation.v0": candidates = self._candidate_responses(message) return candidates[0] if candidates else None - return None + return self._error_response( + f"Unsupported CultNet message schema: {schema_version!r}.", + message_id=str(message.get("messageId") or ""), + code="unsupported_schema_version", + details={"schemaVersion": schema_version}, + ) def _accept_loop(self) -> None: while not self._stop.is_set(): @@ -186,10 +191,18 @@ class CultMeshLocalServer: ) -> list[dict[str, Any]]: document_record = message.get("document") if not isinstance(document_record, dict): - return [] + return [self._error_response( + "Raw put messages must contain a document map.", + message_id=str(message.get("messageId") or ""), + code="malformed_document_put", + )] change = self.node.database.apply_raw_put_message(message) if change is None: - return [] + return [self._error_response( + "Raw put message did not apply to a registered document.", + message_id=str(message.get("messageId") or ""), + code="unregistered_document_put", + )] return self._database_change_notifications(message, document_record, change.change_kind, subscriptions) def _handle_raw_delete( @@ -200,10 +213,18 @@ class CultMeshLocalServer: schema_id = str(message.get("schemaId") or "") record_key = str(message.get("recordKey") or "") if not schema_id or not record_key: - return [] + return [self._error_response( + "Raw delete messages must contain schemaId and recordKey.", + message_id=str(message.get("messageId") or ""), + code="malformed_document_delete", + )] change = self.node.database.apply_raw_delete_message(message) if change is None: - return [] + return [self._error_response( + "Raw delete message did not apply to a registered document.", + message_id=str(message.get("messageId") or ""), + code="unregistered_document_delete", + )] return self._database_delete_notifications(message, subscriptions) def _database_change_notifications( @@ -299,22 +320,40 @@ class CultMeshLocalServer: document_count = len(response.get("documents") or []) if self.max_snapshot_documents is not None and document_count > self.max_snapshot_documents: return self._error_response( - f"Snapshot document limit exceeded: {document_count} > {self.max_snapshot_documents}." + f"Snapshot document limit exceeded: {document_count} > {self.max_snapshot_documents}.", + message_id=str(response.get("messageId") or ""), + code="snapshot_document_limit_exceeded", + details={"documentCount": document_count, "maxSnapshotDocuments": self.max_snapshot_documents}, ) if self.max_snapshot_bytes is not None: response_bytes = len(msgpack.packb(response, use_bin_type=True)) if response_bytes > self.max_snapshot_bytes: return self._error_response( - f"Snapshot byte limit exceeded: {response_bytes} > {self.max_snapshot_bytes}." + f"Snapshot byte limit exceeded: {response_bytes} > {self.max_snapshot_bytes}.", + message_id=str(response.get("messageId") or ""), + code="snapshot_byte_limit_exceeded", + details={"responseBytes": response_bytes, "maxSnapshotBytes": self.max_snapshot_bytes}, ) return response @staticmethod - def _error_response(error: str) -> dict[str, Any]: - return { + def _error_response( + error: str, + *, + message_id: str = "", + code: str | None = None, + details: dict[str, Any] | None = None, + ) -> dict[str, Any]: + response: dict[str, Any] = { "schemaVersion": "cultnet.error.v0", + "messageId": message_id, "error": error, } + if code is not None: + response["code"] = code + if details is not None: + response["details"] = details + return response def _candidate_responses(self, message: dict[str, Any]) -> list[dict[str, Any]]: if self.observation_hub is None: diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index cf1340cd..e4724e8e 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -3168,6 +3168,45 @@ class CultCacheTests(unittest.TestCase): self.assertEqual([document["recordKey"] for document in filtered_snapshot["documents"]], ["note:1"]) self.assertEqual(error["schemaVersion"], "cultnet.error.v0") self.assertIn("Snapshot document limit exceeded", error["error"]) + self.assertEqual(error["code"], "snapshot_document_limit_exceeded") + self.assertEqual(error["details"]["documentCount"], 2) + self.assertEqual(error["details"]["maxSnapshotDocuments"], 1) + + def test_cultmesh_local_server_returns_errors_for_bad_requests(self) -> None: + document = define_database_entry_type( + "mesh.error_note", + [("body", 0)], + schema_id="mesh.error_note.v1", + ) + node = CultMesh.create_node(runtime_id="mesh-error-server") + node.database.register_document(document) + + server = CultMesh.serve_node(node) + try: + raw_client = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=2.0) + unsupported = raw_client.request( + {"schemaVersion": "cultnet.nope.v0", "messageId": "bad-schema"}, + expected_schema_version="cultnet.error.v0", + ) + malformed_put = raw_client.request( + {"schemaVersion": "cultnet.document_put_raw.v0", "messageId": "bad-put"}, + expected_schema_version="cultnet.error.v0", + ) + malformed_delete = raw_client.request( + {"schemaVersion": "cultnet.document_delete.v0", "messageId": "bad-delete", "schemaId": "mesh.error_note.v1"}, + expected_schema_version="cultnet.error.v0", + ) + finally: + server.stop() + + self.assertEqual(unsupported["messageId"], "bad-schema") + self.assertEqual(unsupported["code"], "unsupported_schema_version") + self.assertEqual(unsupported["details"]["schemaVersion"], "cultnet.nope.v0") + self.assertIn("Unsupported CultNet message schema", unsupported["error"]) + self.assertEqual(malformed_put["messageId"], "bad-put") + self.assertEqual(malformed_put["code"], "malformed_document_put") + self.assertEqual(malformed_delete["messageId"], "bad-delete") + self.assertEqual(malformed_delete["code"], "malformed_document_delete") def test_cultmesh_authority_lease_requires_live_matching_lease(self) -> None: peer = CultMeshPeerCard(