From fd394d54dba92c8b8ebfed1946a9e17f0b46acc7 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 13:41:34 +0200 Subject: [PATCH] cultmesh-py: refuse a put the local server could never serve With max_snapshot_bytes set, CultMeshLocalServer stored any raw put while refusing every snapshot response over that size, so it could hold a document no request would ever return. _handle_raw_put now asks the database for the snapshot response that would serve the document alone, built from the envelope the put would store (the resolved local schema id, not the put's alias) under an empty message id, the shortest this server answers. It is measured with the snapshot path's own encoder. Over the bound the put is refused before it is stored, as cultnet.error.v0 code document_unservable with responseBytes and maxSnapshotBytes. At the bound it is stored and served at exactly that size. The database's put and snapshot paths share one envelope builder and one served-record builder, so the preview and the stored record cannot drift. Co-Authored-By: Claude Opus 5.5 --- packages/cultmesh-py/src/cultmesh_py/node.py | 64 +++++++++++----- .../cultmesh-py/src/cultmesh_py/server.py | 24 +++++- packages/cultmesh-py/tests/test_cultmesh.py | 76 +++++++++++++++++++ 3 files changed, 144 insertions(+), 20 deletions(-) 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",