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

cultmesh-py: a put the store cannot take is refused to its peer, never raised

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 19:38:20 +02:00
commit 699432128b
2 changed files with 79 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

17
packages/cultmesh-py/src/cultmesh_py/server.py
View file

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

63
packages/cultmesh-py/tests/test_cultmesh.py
View file

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