diff --git a/packages/cultmesh-py/src/cultmesh_py/node.py b/packages/cultmesh-py/src/cultmesh_py/node.py index 09ac2012..bf092060 100644 --- a/packages/cultmesh-py/src/cultmesh_py/node.py +++ b/packages/cultmesh-py/src/cultmesh_py/node.py @@ -651,6 +651,38 @@ class CultMeshDatabase: return message def apply_raw_put_message(self, message: dict[str, Any]) -> CultMeshDatabaseChange | None: + prepared = self._raw_put_envelope(message) + if prepared is None: + return None + document, envelope = prepared + record_key = envelope.key + previous = self.cache.get(document, record_key) + value = self.cache.put_envelope(document, envelope) + change_kind = "added" if previous is None else "updated" + change = self._publish_local_change(document, record_key, change_kind, value, previous) + self._append_shard_log_put(CultNetMessage( + "cultnet.document_put_raw.v0", + {key: value for key, value in message.items() if key != "schemaVersion"}, + ), change_kind) + return change + + def raw_put_served_alone(self, message: dict[str, Any], *, message_id: str) -> dict[str, Any] | None: + """The snapshot response that would serve this put's document alone, as a wire map. + + It is built from the envelope the put would store, exactly as a snapshot builds it, so a + server can judge a put by the size it would be served at. None when the put would not apply. + """ + prepared = self._raw_put_envelope(message) + if prepared is None: + return None + return CultNetRawSnapshotResponse( + message_id=message_id, + documents=(self._served_record(prepared[1]),), + ).to_wire() + + def _raw_put_envelope( + self, message: dict[str, Any] + ) -> tuple[DocumentDefinition[Any], CultCacheEnvelope] | None: document_record = message.get("document") if not isinstance(document_record, dict): return None @@ -664,24 +696,24 @@ class CultMeshDatabase: ) except KeyError: return None - record_key = str(document_record["recordKey"]) - previous = self.cache.get(document, record_key) - envelope = CultCacheEnvelope( - key=record_key, + return document, CultCacheEnvelope( + key=str(document_record["recordKey"]), type=document.type, schema_id=resolved_schema_id, payload=bytes(document_record["payload"]), stored_at=str(document_record.get("storedAt") or ""), catalog_entry=document.catalog_entry(), ) - value = self.cache.put_envelope(document, envelope) - change_kind = "added" if previous is None else "updated" - change = self._publish_local_change(document, record_key, change_kind, value, previous) - self._append_shard_log_put(CultNetMessage( - "cultnet.document_put_raw.v0", - {key: value for key, value in message.items() if key != "schemaVersion"}, - ), change_kind) - return change + + @staticmethod + def _served_record(envelope: CultCacheEnvelope) -> CultNetRawDocumentRecord: + return CultNetRawDocumentRecord( + schema_id=envelope.schema_id or envelope.type, + record_key=envelope.key, + stored_at=envelope.stored_at, + payload_encoding="messagepack", + payload=envelope.payload, + ) def apply_raw_delete_message(self, message: dict[str, Any]) -> CultMeshDatabaseChange | None: schema_id = str(message.get("schemaId") or "") @@ -774,13 +806,7 @@ class CultMeshDatabase: continue if shard_record_keys is not None and (schema_id, envelope.key) not in shard_record_keys: continue - documents.append(CultNetRawDocumentRecord( - schema_id=schema_id, - record_key=envelope.key, - stored_at=envelope.stored_at, - payload_encoding="messagepack", - payload=envelope.payload, - )) + documents.append(self._served_record(envelope)) return CultNetRawSnapshotResponse( message_id=message_id, diff --git a/packages/cultmesh-py/src/cultmesh_py/server.py b/packages/cultmesh-py/src/cultmesh_py/server.py index fe4edd52..605b5a7e 100644 --- a/packages/cultmesh-py/src/cultmesh_py/server.py +++ b/packages/cultmesh-py/src/cultmesh_py/server.py @@ -24,6 +24,10 @@ from cultnet_py import ( ) from .node import CultMeshNode + +# The shortest message id this server answers a snapshot request under. It takes a +# request's messageId as given, an empty one included, so no response it sends is smaller. +_SHORTEST_SERVED_MESSAGE_ID = "" from cultnet_py.cultmesh_contracts import ( PEER_EXCHANGE_REQUEST, VERSE_CATALOG_REQUEST, @@ -357,6 +361,19 @@ class CultMeshLocalServer: message_id=str(message.get("messageId") or ""), code="malformed_document_put", )] + if self.max_snapshot_bytes is not None: + alone = self.node.database.raw_put_served_alone( + message, message_id=_SHORTEST_SERVED_MESSAGE_ID + ) + response_bytes = None if alone is None else _snapshot_response_bytes(alone) + if response_bytes is not None and response_bytes > self.max_snapshot_bytes: + return [self._error_response( + f"Document can never be served: its snapshot response is {response_bytes} bytes; " + f"limit is {self.max_snapshot_bytes}.", + message_id=str(message.get("messageId") or ""), + code="document_unservable", + details={"responseBytes": response_bytes, "maxSnapshotBytes": self.max_snapshot_bytes}, + )] change = self.node.database.apply_raw_put_message(message) if change is None: return [self._error_response( @@ -559,7 +576,7 @@ class CultMeshLocalServer: 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)) + response_bytes = _snapshot_response_bytes(response) if response_bytes > self.max_snapshot_bytes: return self._error_response( f"Snapshot byte limit exceeded: {response_bytes} > {self.max_snapshot_bytes}.", @@ -723,3 +740,8 @@ def _infer_schema_name(schema_id: str) -> str | None: return None version = schema_id[marker + 2:] return schema_id[:marker] if version.isdigit() else None + + +def _snapshot_response_bytes(response: dict[str, Any]) -> int: + """The one measure of a snapshot response, shared by the snapshot path and put admission.""" + return len(msgpack.packb(response, use_bin_type=True)) diff --git a/packages/cultmesh-py/tests/test_cultmesh.py b/packages/cultmesh-py/tests/test_cultmesh.py index 9a697370..ddd8e89a 100644 --- a/packages/cultmesh-py/tests/test_cultmesh.py +++ b/packages/cultmesh-py/tests/test_cultmesh.py @@ -14,6 +14,8 @@ from dataclasses import dataclass from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any, Callable + +import msgpack from cultcache_py import ( CultCache, SingleFileMessagePackBackingStore, @@ -4293,6 +4295,80 @@ class CultMeshTests(unittest.TestCase): self.assertEqual(error["details"]["documentCount"], 2) self.assertEqual(error["details"]["maxSnapshotDocuments"], 1) + def test_cultmesh_local_server_refuses_a_put_it_could_never_serve(self) -> None: + # A put is admitted only if some snapshot request could return it. The put arrives under a + # long foreign schema alias and is stored and served under the shorter local schema id, so + # the size that decides is the served one, not the put's own. + document = define_database_entry_type( + "mesh.served_bound_note", + [("schema_version", 0), ("body", 1)], + schema_id="sha256:mesh-served-bound", + schema_name="mesh.served_bound_note", + schema_version="mesh.served_bound_note.v1", + ) + + def put(message_id: str, key: str, body_length: int) -> dict[str, Any]: + return document_put_raw( + message_id=message_id, + key=key, + schema_id="runtime.generated.mesh.served_bound_note.ui.99", + stored_at="2026-09-30T00:00:00Z", + payload=document.encode_payload({ + "schema_version": "mesh.served_bound_note.v1", + "body": "x" * body_length, + }), + ).to_wire() + + def exchange(server: Any, *messages: dict[str, Any]) -> tuple[dict[str, Any], bytes]: + client = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=2.0) + with client.open_transport() as transport: + for message in messages: + transport.send("schema", msgpack.packb(message, use_bin_type=True)) + payload = transport.receive().payload + return msgpack.unpackb(payload, raw=False), payload + + empty_id_snapshot = snapshot_request(message_id="").to_wire() + + # The bound is what the server actually serves for the at-bound document alone, under the + # shortest message id it answers (an empty one), measured on an unbounded server. + def served_alone_bytes(key: str, body_length: int) -> int: + node = CultMesh.create_node(runtime_id="mesh-served-bound-probe") + node.database.register_document(document) + server = CultMesh.serve_node(node, enable_rudp=False) + try: + response, payload = exchange(server, put("probe", key, body_length), empty_id_snapshot) + finally: + server.stop() + self.assertEqual(response["schemaVersion"], "cultnet.snapshot_response_raw.v0") + self.assertEqual(len(response["documents"]), 1) + return len(payload) + + limit = served_alone_bytes("note:fits", 100) + self.assertEqual(served_alone_bytes("note:over", 101), limit + 1, "fixture: the documents must straddle the bound by one byte") + + node = CultMesh.create_node(runtime_id="mesh-served-bound") + node.database.register_document(document) + server = CultMesh.serve_node(node, max_snapshot_bytes=limit, enable_rudp=False) + try: + refused, _ = exchange(server, put("put-over", "note:over", 101)) + stored_after_refusal = node.database.get(document, "note:over") + served, served_payload = exchange(server, put("put-at", "note:fits", 100), empty_id_snapshot) + finally: + server.stop() + + self.assertEqual(refused["schemaVersion"], "cultnet.error.v0") + self.assertEqual(refused["messageId"], "put-over") + self.assertEqual(refused["code"], "document_unservable") + self.assertEqual(refused["details"], {"responseBytes": limit + 1, "maxSnapshotBytes": limit}) + self.assertIn(str(limit + 1), refused["error"]) + self.assertIn(str(limit), refused["error"]) + self.assertIsNone(stored_after_refusal) + + self.assertEqual(served["schemaVersion"], "cultnet.snapshot_response_raw.v0") + self.assertEqual([record["recordKey"] for record in served["documents"]], ["note:fits"]) + self.assertEqual(served["documents"][0]["schemaId"], document.catalog_entry().schema_id) + self.assertEqual(len(served_payload), limit) + def test_cultmesh_local_server_returns_errors_for_bad_requests(self) -> None: document = define_database_entry_type( "mesh.error_note",