From 699432128b7291de4f913d6500937c7df0886819 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 19:38:20 +0200 Subject: [PATCH] cultmesh-py: a put the store cannot take is refused to its peer, never raised A well-formed put still killed the RUDP thread: _handle_raw_put called apply_raw_put_message unguarded, and cache.put_envelope raises for a blank recordKey, a payload that is not MessagePack or does not decode to its schema, a global document under another key, or anything a document's own decoder or validator raises. Over TCP it ended the connection. The apply site now refuses any put the store cannot take to the peer that sent it, as malformed_document_put, over RUDP and TCP alike, and logs the error's type. Neither the refusal nor the log quotes the error, which can quote the put. No new shape checks: the store decides what it can take. The test sends each shape over RUDP and over TCP and shows the RUDP thread alive, the TCP connection serving its next message, and other peers served. Co-Authored-By: Claude Opus 5.5 --- .../cultmesh-py/src/cultmesh_py/server.py | 17 ++++- packages/cultmesh-py/tests/test_cultmesh.py | 63 +++++++++++++++++++ 2 files changed, 79 insertions(+), 1 deletion(-) diff --git a/packages/cultmesh-py/src/cultmesh_py/server.py b/packages/cultmesh-py/src/cultmesh_py/server.py index 591a6f46..25deefd7 100644 --- a/packages/cultmesh-py/src/cultmesh_py/server.py +++ b/packages/cultmesh-py/src/cultmesh_py/server.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging import socket import threading import time @@ -41,6 +42,8 @@ from cultnet_py.cultmesh_contracts import ( CultMeshVerseCatalog, ) +_LOGGER = logging.getLogger(__name__) + @dataclass(frozen=True) class _DatabaseSubscription: @@ -430,7 +433,19 @@ class CultMeshLocalServer: unservable = self._unservable_put(message, over_rudp=over_rudp) if unservable is not None: return [unservable] - change = self.node.database.apply_raw_put_message(message) + try: + change = self.node.database.apply_raw_put_message(message) + except Exception as error: # noqa: BLE001 - a put the store cannot take is that peer's refusal + # The store refused the put: a blank key, a payload its schema cannot decode, a global + # document under another key, or anything a document's own decoder or validator + # raises. It is refused to the peer that sent it; no thread and no other peer is + # affected. Neither the refusal nor the log quotes the error, which can quote the put. + _LOGGER.warning("CultMesh server refused a raw put the store could not take (%s).", type(error).__name__) + return [self._error_response( + "Raw put document could not be stored.", + message_id=str(message.get("messageId") or ""), + code="malformed_document_put", + )] if change is None: return [self._error_response( "Raw put message did not apply to a registered document.", diff --git a/packages/cultmesh-py/tests/test_cultmesh.py b/packages/cultmesh-py/tests/test_cultmesh.py index ba1cc635..163e08be 100644 --- a/packages/cultmesh-py/tests/test_cultmesh.py +++ b/packages/cultmesh-py/tests/test_cultmesh.py @@ -4613,6 +4613,69 @@ class CultMeshTests(unittest.TestCase): self.assertEqual(after["documents"], []) self.assertTrue(alive) + def test_cultmesh_local_server_refuses_a_put_its_store_cannot_take_and_serves_on(self) -> None: + # Well-formed puts the store refuses: a blank key, a payload that is not MessagePack, a + # payload its schema cannot decode, a global document under another key. Each is refused + # to the peer that sent it, over RUDP and over TCP; the RUDP thread lives, the TCP + # connection serves its next message, and other peers are served. + document = self._rudp_bound_note() + global_document = define_database_entry_type( + "mesh.rudp_global_note", + [("schema_version", 0), ("body", 1)], + schema_id="sha256:mesh-rudp-global", + schema_name="mesh.rudp_global_note", + schema_version="mesh.rudp_global_note.v1", + global_document=True, + ) + + def cases() -> list[dict[str, Any]]: + puts = [] + for name, field_name, value in ( + ("empty-key", "recordKey", ""), + ("blank-key", "recordKey", " "), + ("garbage-payload", "payload", bytes([0xC1])), + ("int-payload", "payload", msgpack.packb(5)), + ): + put = self._rudp_note_put(document, name, "note:ok", 10) + put["document"][field_name] = value + puts.append(put) + puts.append(self._rudp_note_put(global_document, "global-wrong-key", "not-global", 10)) + return puts + + node = CultMesh.create_node(runtime_id="mesh-unstorable-put") + node.database.register_document(document) + node.database.register_document(global_document) + server = CultMesh.serve_node(node) + try: + rudp_refusals = [(put["messageId"], self._rudp_exchange(server, put)[0]) for put in cases()] + rudp_after, _ = self._rudp_exchange(server, hello(runtime_id="rudp-after").to_wire()) + rudp_alive = server._rudp_thread is not None and server._rudp_thread.is_alive() + + tcp_refusals = [] + with CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=4.0).open_transport() as transport: + for put in cases(): + transport.send("schema", msgpack.packb(put, use_bin_type=True)) + tcp_refusals.append((put["messageId"], msgpack.unpackb(transport.receive().payload, raw=False))) + transport.send("schema", msgpack.packb(hello(runtime_id="same-connection").to_wire(), use_bin_type=True)) + same_connection = msgpack.unpackb(transport.receive().payload, raw=False) + other_connection = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=4.0).request( + hello(runtime_id="other-connection").to_wire(), expected_schema_version="cultnet.hello.v0" + ) + finally: + server.stop() + self.assertEqual(len(rudp_refusals), 5) + for message_id, refusal in rudp_refusals + tcp_refusals: + self.assertEqual(refusal["schemaVersion"], "cultnet.error.v0", message_id) + self.assertEqual(refusal["messageId"], message_id) + self.assertEqual(refusal["code"], "malformed_document_put", message_id) + self.assertNotIn("note:ok", refusal["error"], message_id) + self.assertEqual(len(tcp_refusals), 5) + self.assertEqual(rudp_after["schemaVersion"], "cultnet.hello.v0") + self.assertTrue(rudp_alive) + self.assertEqual(same_connection["schemaVersion"], "cultnet.hello.v0") + self.assertEqual(other_connection["schemaVersion"], "cultnet.hello.v0") + self.assertIsNone(node.database.get(document, "note:ok")) + def test_cultmesh_local_server_refuses_options_that_could_answer_nothing(self) -> None: # The Rust document server's option rules: every limit is greater than zero, and the limits # leave room for an empty snapshot, the smallest response the server sends. One-byte