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

cultmesh-py: a refusal acknowledges nothing and names its own send failure; TCP handling refuses instead of dying; the daemon checks its port

Browse source 
Queue-full refusal (findings 2 and 6). The refusal was built in
_send_rudp_schema_frame before _end_refused_peer reset the session, so
its ack fields covered the refused request (ack=102 mask=0x3). A reply
the queue cannot take now raises _ReplyUnqueueable; the loop ends the
peer through _end_refused_peer, which builds the refusal before the
reset, gives it ack fields built after it (acknowledging nothing, like
the goodbye and like Rust's refusal), sends it unreliable and unordered,
then the goodbye. A refusal fragment that can never be sent is not
dropped: the goodbye names its failure ("packet could not be sent:
EMSGSIZE"), where it used to say the session refused a packet.

Queue full to the brim (item 1). The tests fill a queue of 8 exactly with
a reply the peer never acknowledges, then show the next reply is answered
with response_unqueueable, acknowledging nothing, followed by the
goodbye; and that a refusal whose fragment cannot be sent ends the
session with a goodbye naming EMSGSIZE.

TCP (finding 8). A message whose handling raised (snapshot recordKeys as
an int, a shard-log afterSequence that is not a number, subscribe,
schema-catalog and verse requests with int fields) killed the
connection's thread with a traceback that could quote the peer. The TCP
handler now refuses such a message with fixed text (message_unhandled),
logs the error type only, and serves the connection's next message.

Daemon (finding 10). --port 99999 is an argparse error (exit 2), not a
traceback.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 23:04:07 +02:00
commit d541e9eb35
3 changed files with 272 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

10
packages/cultmesh-py/src/cultmesh_py/daemon.py
View file

@ -24,10 +24,18 @@ from cultnet_py.cultmesh_contracts import (
READY_SCHEMA_VERSION = "cultmesh.daemon_ready.v0"
def _port(value: str) -> int:
"""A TCP/UDP port: 0 (any free port) through 65535."""
port = int(value)
if not 0 <= port <= 65535:
raise argparse.ArgumentTypeError("port must be between 0 and 65535")
return port
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="cultmesh-py-daemon")
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=3075)
parser.add_argument("--port", type=_port, default=3075)
parser.add_argument("--runtime-id", default="python-runtime")
parser.add_argument("--display-name")
parser.add_argument("--cache-file")

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

@ -4,7 +4,7 @@ import logging
import socket
import threading
import time
from dataclasses import dataclass, field
from dataclasses import dataclass, field, replace
from typing import Any
import msgpack
@ -39,6 +39,8 @@ CULTMESH_RUDP_DEFAULT_MAX_FRAGMENT_BYTES = 1200
# The answer to a request whose reply the peer's reliable queue cannot take. Fixed text: it
# never quotes the reply or the queue's error.
_REPLY_UNQUEUEABLE = "The reply could not be queued for this peer."
# The answer to a message the server could not handle. Fixed text: it never quotes the message.
_MESSAGE_UNHANDLED = "The message could not be handled."
from cultnet_py.cultmesh_contracts import (
PEER_EXCHANGE_REQUEST,
VERSE_CATALOG_REQUEST,
@ -49,6 +51,15 @@ from cultnet_py.cultmesh_contracts import (
_LOGGER = logging.getLogger(__name__)
class _ReplyUnqueueable(Exception):
"""A reply the peer's reliable queue cannot take. The peer's session ends, and the peer is
answered first with a small cultnet.error.v0 under the request's message id."""
def __init__(self, message_id: str) -> None:
super().__init__("reply could not be queued")
self.message_id = message_id
@dataclass(frozen=True)
class _DatabaseSubscription:
subscription_id: str
@ -259,7 +270,22 @@ class CultMeshLocalServer:
return
if not isinstance(message, dict):
continue
responses = self._handle_connection_message(message, subscriptions)
try:
responses = self._handle_connection_message(message, subscriptions)
except Exception as error: # noqa: BLE001 - one message never ends the connection
# Whatever a peer's message makes the handling raise is that peer's failure:
# it is refused with fixed text and the connection serves on. The log names
# the error type only, because the error's own message can quote the peer.
_LOGGER.warning(
"CultMesh server could not handle a peer's message (%s); it was refused.",
type(error).__name__,
)
message_id = message.get("messageId")
responses = [self._error_response(
_MESSAGE_UNHANDLED,
message_id=message_id if isinstance(message_id, str) else "",
code="message_unhandled",
)]
for response in responses:
transport.send("schema", msgpack.packb(response, use_bin_type=True))
@ -361,7 +387,14 @@ class CultMeshLocalServer:
"CultMesh RUDP server could not handle a peer's message (%s); that peer's session ended.",
type(error).__name__,
)
self._end_refused_peer(rudp_socket, peers, remote_addr)
refusal = None
if isinstance(error, _ReplyUnqueueable):
refusal = self._error_response(
_REPLY_UNQUEUEABLE,
message_id=error.message_id,
code="response_unqueueable",
)
self._end_refused_peer(rudp_socket, peers, remote_addr, refusal=refusal)
continue
if unsendable is None and (
packet.reliable or packet.packet_type == CultNetRudpPacketType.DATA or result.delivered
@ -405,23 +438,8 @@ class CultMeshLocalServer:
)
except ValueError:
# The peer's reliable queue cannot take this reply. The peer is answered, not left
# to time out: a small cultnet.error.v0 goes unreliable and unordered, so it needs
# no room in the queue that just refused. Then the error ends the session, as any
# message the handling cannot serve does.
refusal = self._error_response(
_REPLY_UNQUEUEABLE,
message_id=message_id,
code="response_unqueueable",
)
for packet in session.send_many(
"schema",
msgpack.packb(refusal, use_bin_type=True),
CultNetRudpSendOptions(reliable=False, ordered=False, now_ms=_now_ms()),
max_fragment_bytes=self.rudp_max_fragment_bytes,
):
if self._send_rudp_packet(rudp_socket, remote_addr, packet) is not None:
break
raise
# to time out, and its session ends: see _end_refused_peer.
raise _ReplyUnqueueable(message_id) from None
for packet in packets:
unsendable = self._send_rudp_packet(rudp_socket, remote_addr, packet)
if unsendable is not None:
@ -452,11 +470,36 @@ class CultMeshLocalServer:
rudp_socket: socket.socket,
peers: dict[tuple[str, int], _RudpPeerConnection],
remote_addr: tuple[str, int],
*,
refusal: dict[str, Any] | None = None,
) -> None:
"""Ends a peer's session that refused one of its messages, and says goodbye. A
refusal, when there is one, goes first, unreliable and unordered: the session ends
with it, so it is never resent and needs no room in the reliable queue. It is built
before the reset and carries acknowledgement fields built after it, which acknowledge
nothing, like the goodbye: the ack would otherwise cover the very frame refused. A
refusal fragment that can never be sent is not dropped: the goodbye names its failure."""
peer = peers.pop(remote_addr)
# The goodbye is built after the reset, or its ack field would acknowledge the
# very frame the session refused.
packets: list[CultNetRudpPacket] = []
if refusal is not None:
try:
packets = peer.session.send_many(
"schema",
msgpack.packb(refusal, use_bin_type=True),
CultNetRudpSendOptions(reliable=False, ordered=False, now_ms=_now_ms()),
max_fragment_bytes=self.rudp_max_fragment_bytes,
)
except ValueError:
packets = []
peer.session.reset_peer_state()
nothing = peer.session.create_ack()
for packet in packets:
unsendable = self._send_rudp_packet(
rudp_socket, remote_addr, replace(packet, ack=nothing.ack, ack_mask=nothing.ack_mask)
)
if unsendable is not None:
self._send_rudp_packet(rudp_socket, remote_addr, peer.session.end_unsendable_session(unsendable))
return
goodbye = peer.session.create_disconnect(b"session refused a packet")
self._send_rudp_packet(rudp_socket, remote_addr, goodbye)

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

@ -51,6 +51,9 @@ from cultnet_py import (
CultNetSimulationObservation,
CultNetSimulationObservationHub,
CultNetRudpPacketType,
CultNetRudpSendOptions,
CultNetRudpSession,
CultNetRudpSessionOptions,
CultNetRudpSocketMode,
CultNetRudpSocketTransportConnection,
CultNetRudpSocketTransportOptions,
@ -59,10 +62,12 @@ from cultnet_py import (
decode_rudp_packet,
document_delete,
document_put_raw,
encode_rudp_packet,
hello,
simulation_observation,
snapshot_request,
)
from cultnet_py.cultmesh_contracts import VERSE_CATALOG_REQUEST
from cultmesh_py import create_node
from cultmesh_py import (
CultMesh,
@ -3794,6 +3799,17 @@ class CultMeshTests(unittest.TestCase):
self.assertIn("error: max_snapshot_documents must be greater than zero", completed.stderr)
self.assertNotIn("Traceback", completed.stderr)
self.assertFalse(ready_path.exists())
# A port no socket can bind is a usage error too.
completed = subprocess.run(
[sys.executable, "-m", "cultmesh_py.daemon", "--port", "99999"],
capture_output=True,
text=True,
env=env,
timeout=30,
)
self.assertEqual(completed.returncode, 2, completed.stderr)
self.assertIn("port must be between 0 and 65535", completed.stderr)
self.assertNotIn("Traceback", completed.stderr)
def test_cultmesh_daemon_serves_opt_in_simulation_observations(self) -> None:
with tempfile.TemporaryDirectory() as temp:
@ -4547,6 +4563,188 @@ class CultMeshTests(unittest.TestCase):
self.assertEqual(after_reply["messageId"], "after-queue")
self.assertTrue(alive)
class _RawPeer:
"""A peer driven packet by packet, so a test decides what it acknowledges: nothing."""
def __init__(self, server: Any) -> None:
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.bind(("127.0.0.1", 0))
self.sock.settimeout(0.05)
self.address = self.sock.getsockname()[:2]
self.target = ("127.0.0.1", server.port)
self.session = CultNetRudpSession(
CultNetRudpSessionOptions(
connection_id=server.rudp_connection_id, initial_sequence=100, resend_delay_ms=100_000
)
)
self.sock.sendto(encode_rudp_packet(self.session.create_connect(0)), self.target)
accept = self.receive(1.0)
assert accept is not None and accept.packet_type == CultNetRudpPacketType.ACCEPT, accept
self.session.receive(accept, 1)
def receive(self, timeout: float = 0.3) -> Any:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
try:
wire, _ = self.sock.recvfrom(65535)
except (TimeoutError, socket.timeout):
continue
return decode_rudp_packet(wire)
return None
def drain(self, timeout: float = 0.5) -> list[Any]:
# Bounded overall: the server resends what the peer never acknowledges.
packets = []
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
packet = self.receive(0.05)
if packet is not None:
packets.append(packet)
return packets
def send(self, message: dict[str, Any]) -> list[int]:
packets = self.session.send_many(
"schema",
msgpack.packb(message, use_bin_type=True),
CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=2),
max_fragment_bytes=1200,
)
for packet in packets:
self.sock.sendto(encode_rudp_packet(packet), self.target)
return [packet.sequence for packet in packets]
def close(self) -> None:
self.sock.close()
@staticmethod
def _acknowledges(packet: Any, sequence: int) -> bool:
return packet.ack == sequence or any(
packet.ack_mask & (1 << bit) and packet.ack > bit and packet.ack - bit - 1 == sequence
for bit in range(32)
)
def _brim_server(self) -> tuple[Any, Any]:
"""A server whose peer's reliable queue (8) is filled exactly by an 8-fragment reply the
peer never acknowledges, and the peer."""
document = self._rudp_bound_note()
node = CultMesh.create_node(runtime_id="mesh-rudp-brim")
node.database.register_document(document)
server = CultMesh.serve_node(node, rudp_max_fragment_bytes=100, rudp_max_pending_reliable_packets=8)
def served(body: int) -> int:
put = self._rudp_note_put(document, "p", "note:q", body)
return len(msgpack.packb(node.database.raw_put_served_alone(put, message_id=""), use_bin_type=True))
body = 400 + 800 - served(400)
self.assertEqual(served(body), 800, "fixture: served size is linear in the body")
node.database.apply_raw_put_message(self._rudp_note_put(document, "p", "note:q", body))
peer = self._RawPeer(server)
peer.send(snapshot_request(message_id="", record_keys=["note:q"]).to_wire())
first = {p.sequence for p in peer.drain(0.3) if p.packet_type == CultNetRudpPacketType.DATA}
self.assertEqual(len(first), 8, "fixture: the first reply fills the queue exactly")
return server, peer
def test_cultmesh_local_server_answers_a_peer_whose_queue_is_full_to_the_brim(self) -> None:
# The next reply cannot be queued. The peer is answered with a cultnet.error.v0 that takes
# no room in the full queue and acknowledges nothing the peer sent, then its session ends.
server, peer = self._brim_server()
try:
refused = peer.send(snapshot_request(message_id="b", record_keys=["note:none"]).to_wire())
packets = peer.drain(0.5)
alive = server._rudp_thread is not None and server._rudp_thread.is_alive()
finally:
peer.close()
server.stop()
parts = sorted(
(
p for p in packets
if p.packet_type == CultNetRudpPacketType.DATA and p.channel_id == "schema" and not p.reliable
),
key=lambda p: p.fragment_index,
)
self.assertTrue(parts, "the refusal reaches the peer")
for part in parts:
self.assertFalse(
any(self._acknowledges(part, sequence) for sequence in refused),
f"the refusal acknowledges the refused request: ack={part.ack} mask={part.ack_mask:#x}",
)
refusal = msgpack.unpackb(b"".join(bytes(p.payload) for p in parts), raw=False)
self.assertEqual(refusal["schemaVersion"], "cultnet.error.v0")
self.assertEqual(refusal["messageId"], "b")
self.assertEqual(refusal["code"], "response_unqueueable")
goodbyes = [bytes(p.payload) for p in packets if p.packet_type == CultNetRudpPacketType.DISCONNECT]
self.assertEqual(goodbyes, [b"session refused a packet"], "then the session ends")
self.assertTrue(alive)
def test_cultmesh_local_server_names_a_refusal_that_can_never_be_sent_in_the_goodbye(self) -> None:
# The refusal's own fragment fails permanently: the failure is not dropped, the goodbye
# names it by its fixed code, never the error's text.
import errno
server, peer = self._brim_server()
original_sendto = socket.socket.sendto
def sendto(sock: socket.socket, data: bytes, *args: Any) -> int:
packet = decode_rudp_packet(data)
if (
args and tuple(args[-1]) == tuple(peer.address)
and packet.packet_type == CultNetRudpPacketType.DATA and not packet.reliable
):
raise OSError(errno.EMSGSIZE, "Message too long CANARY-7f3a")
return original_sendto(sock, data, *args)
try:
with patch.object(socket.socket, "sendto", sendto):
peer.send(snapshot_request(message_id="b", record_keys=["note:none"]).to_wire())
packets = peer.drain(0.5)
alive = server._rudp_thread is not None and server._rudp_thread.is_alive()
finally:
peer.close()
server.stop()
self.assertFalse(
[p for p in packets if p.packet_type == CultNetRudpPacketType.DATA and not p.reliable],
"the refusal could not be sent",
)
goodbyes = [bytes(p.payload) for p in packets if p.packet_type == CultNetRudpPacketType.DISCONNECT]
self.assertEqual(goodbyes, [b"packet could not be sent: EMSGSIZE"])
self.assertTrue(alive)
def test_cultmesh_local_server_refuses_a_tcp_message_it_cannot_handle_and_serves_on(self) -> None:
# Messages whose handling raises: the TCP peer is refused with fixed text, the log names
# the error type only, and the connection's thread serves its next message.
shapes = [
{"schemaVersion": "cultnet.snapshot_request.v0", "messageId": "snapshot", "recordKeys": 5},
{"schemaVersion": "cultnet.shard_log_request.v0", "messageId": "shard-log", "shardId": "a", "afterSequence": "CANARY-7f3a"},
{"schemaVersion": "cultnet.database_subscribe.v0", "messageId": "subscribe", "subscriptionId": "s", "schemaIds": 5},
{"schemaVersion": "cultnet.schema_catalog_request.v0", "messageId": "catalog", "schemaIds": 5},
{"schemaVersion": VERSE_CATALOG_REQUEST, "messageId": "verse", "verseIds": 5},
]
deaths: list[str] = []
original_hook = threading.excepthook
threading.excepthook = lambda args: deaths.append(f"{args.exc_type.__name__}")
server = CultMesh.serve_node(CultMesh.create_node(runtime_id="mesh-tcp-shapes"))
try:
with self.assertLogs("cultmesh_py.server", level="WARNING") as logs:
with CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=4.0).open_transport() as transport:
refusals = []
for shape in shapes:
transport.send("schema", msgpack.packb(shape, use_bin_type=True))
refusals.append((shape["messageId"], msgpack.unpackb(transport.receive().payload, raw=False)))
transport.send("schema", msgpack.packb(hello(runtime_id="same-connection").to_wire(), use_bin_type=True))
after = msgpack.unpackb(transport.receive().payload, raw=False)
finally:
server.stop()
threading.excepthook = original_hook
for message_id, refusal in refusals:
self.assertEqual(refusal["schemaVersion"], "cultnet.error.v0", message_id)
self.assertEqual(refusal["messageId"], message_id)
self.assertEqual(refusal["code"], "message_unhandled", message_id)
self.assertEqual(refusal["error"], "The message could not be handled.", message_id)
self.assertEqual(after["schemaVersion"], "cultnet.hello.v0")
self.assertEqual(deaths, [])
self.assertEqual(len(logs.output), len(shapes))
self.assertFalse([line for line in logs.output if "CANARY" in line])
def test_cultmesh_local_server_refuses_a_rudp_put_whose_reply_needs_more_than_65535_fragments(self) -> None:
# One message carries at most 65535 fragments, however deep the queue: with one-byte
# fragments a put served in 65536 bytes is refused, one served in 65535 is stored.