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

cultmesh-py: RUDP replies go in fragments, and a send failure ends no thread

Browse source 
The local server sent every RUDP reply as one datagram, so a snapshot of a
100 KB put raised EMSGSIZE inside _rudp_loop and ended the RUDP thread for
every peer. Replies are now fragmented by rudp_max_fragment_bytes (default
CULTMESH_RUDP_DEFAULT_MAX_FRAGMENT_BYTES, 1200, as in the Rust and
TypeScript servers), and each peer's reliable queue is bounded by
rudp_max_pending_reliable_packets; serve_node takes both.

A put arriving over RUDP whose served-alone reply would take more
fragments than one message (65535) or the peer's reliable queue can hold
is refused as document_unservable, with fragmentCount and
maxFragmentCount in its details. A datagram that cannot leave, or a reply
the peer's queue cannot take, is counted in rudp_send_failures and never
raised out of the loop.

This overlaps hands/ack-cutD (not yet on main), which classifies send
failures and ends only the affected peer's session; its counter has the
same name, and its classifier should replace the plain count here when it
merges.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 15:37:55 +02:00
commit 83be3bd90f
4 changed files with 252 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

3
packages/cultmesh-py/src/cultmesh_py/__init__.py
View file

@ -24,7 +24,7 @@ from .node import (
CultMeshReactiveDocumentReconciliation,
create_node,
)
from .server import CultMeshLocalServer
from .server import CULTMESH_RUDP_DEFAULT_MAX_FRAGMENT_BYTES, CultMeshLocalServer
from .session import (
CultMeshGameSession,
CultMeshGameSessionOptions,
@ -79,6 +79,7 @@ __all__ = [
"CultMeshReactiveDocumentReconciliation",
"CultMeshGameSession",
"CultMeshGameSessionOptions",
"CULTMESH_RUDP_DEFAULT_MAX_FRAGMENT_BYTES",
"CultMeshLocalServer",
"CultMeshPeerCard",
"CultMeshPeerCatalog",

6
packages/cultmesh-py/src/cultmesh_py/facade.py
View file

@ -31,7 +31,7 @@ from .node import (
CultMeshPublicationDocumentBinding,
create_node,
)
from .server import CultMeshLocalServer
from .server import CULTMESH_RUDP_DEFAULT_MAX_FRAGMENT_BYTES, CultMeshLocalServer
from .session import CultMeshGameSession, CultMeshGameSessionOptions
from .simulation import CultMeshSimulationFactCommitter
from cultnet_py.cultmesh_contracts import (
@ -404,6 +404,8 @@ class CultMesh:
enable_rudp: bool = True,
rudp_connection_id: int = 0x43554C54,
rudp_resend_delay_ms: int = 25,
rudp_max_fragment_bytes: int = CULTMESH_RUDP_DEFAULT_MAX_FRAGMENT_BYTES,
rudp_max_pending_reliable_packets: int | None = None,
) -> CultMeshLocalServer:
return CultMeshLocalServer(
node=node,
@ -418,6 +420,8 @@ class CultMesh:
enable_rudp=enable_rudp,
rudp_connection_id=rudp_connection_id,
rudp_resend_delay_ms=rudp_resend_delay_ms,
rudp_max_fragment_bytes=rudp_max_fragment_bytes,
rudp_max_pending_reliable_packets=rudp_max_pending_reliable_packets,
).start()
@staticmethod

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

@ -28,6 +28,11 @@ 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 = ""
# The most fragments one RUDP message can carry: its fragment count is a u16.
_MAX_FRAGMENTS_PER_MESSAGE = 0xFFFF
# Replies larger than one fragment are split into fragments of at most this many bytes,
# as the Rust and TypeScript document servers do by default.
CULTMESH_RUDP_DEFAULT_MAX_FRAGMENT_BYTES = 1200
from cultnet_py.cultmesh_contracts import (
PEER_EXCHANGE_REQUEST,
VERSE_CATALOG_REQUEST,
@ -72,17 +77,35 @@ class CultMeshLocalServer:
enable_rudp: bool = True
rudp_connection_id: int = 0x43554C54
rudp_resend_delay_ms: int = 25
# RUDP replies are fragmented by rudp_max_fragment_bytes, and each peer's reliable queue
# holds at most rudp_max_pending_reliable_packets (None: only the 65535 fragments one
# message can carry). A put arriving over RUDP whose served reply could not fit them is
# refused as document_unservable.
rudp_max_fragment_bytes: int = CULTMESH_RUDP_DEFAULT_MAX_FRAGMENT_BYTES
rudp_max_pending_reliable_packets: int | None = None
_socket: socket.socket | None = field(default=None, init=False, repr=False)
_rudp_socket: socket.socket | None = field(default=None, init=False, repr=False)
_stop: threading.Event = field(default_factory=threading.Event, init=False, repr=False)
_thread: threading.Thread | None = field(default=None, init=False, repr=False)
_rudp_thread: threading.Thread | None = field(default=None, init=False, repr=False)
_rudp_send_failures: int = field(default=0, init=False, repr=False)
@property
def rudp_send_failures(self) -> int:
"""RUDP replies and datagrams that could not be sent to a peer. Each is that peer's
loss: a reliable packet already queued stays pending and is resent, and no other
peer is affected."""
return self._rudp_send_failures
def __post_init__(self) -> None:
if self.max_snapshot_documents is not None and self.max_snapshot_documents < 0:
raise ValueError("max_snapshot_documents must be non-negative")
if self.max_snapshot_bytes is not None and self.max_snapshot_bytes < 0:
raise ValueError("max_snapshot_bytes must be non-negative")
if self.rudp_max_fragment_bytes <= 0:
raise ValueError("rudp_max_fragment_bytes must be greater than zero")
if self.rudp_max_pending_reliable_packets is not None and self.rudp_max_pending_reliable_packets <= 0:
raise ValueError("rudp_max_pending_reliable_packets must be greater than zero")
def start(self) -> "CultMeshLocalServer":
if self._socket is not None:
@ -237,6 +260,7 @@ class CultMeshLocalServer:
CultNetRudpSessionOptions(
connection_id=self.rudp_connection_id,
resend_delay_ms=self.rudp_resend_delay_ms,
max_pending_reliable_packets=self.rudp_max_pending_reliable_packets,
)
)
)
@ -267,7 +291,7 @@ class CultMeshLocalServer:
continue
if not isinstance(message, dict):
continue
responses = self._handle_connection_message(message, peer.subscriptions)
responses = self._handle_connection_message(message, peer.subscriptions, over_rudp=True)
for response in responses:
self._send_rudp_schema_frame(
rudp_socket,
@ -299,25 +323,40 @@ class CultMeshLocalServer:
session: CultNetRudpSession,
payload: bytes,
) -> None:
for packet in session.send_many(
"schema",
payload,
CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=_now_ms()),
):
try:
packets = session.send_many(
"schema",
payload,
CultNetRudpSendOptions(reliable=True, ordered=True, now_ms=_now_ms()),
max_fragment_bytes=self.rudp_max_fragment_bytes,
)
except ValueError:
# The peer's reliable queue cannot take this reply now. It is that peer's loss,
# never the thread's.
self._rudp_send_failures += 1
return
for packet in packets:
self._send_rudp_packet(rudp_socket, remote_addr, packet)
@staticmethod
def _send_rudp_packet(
self,
rudp_socket: socket.socket,
remote_addr: tuple[str, int],
packet: CultNetRudpPacket,
) -> None:
rudp_socket.sendto(encode_rudp_packet(packet), remote_addr)
# A datagram that cannot leave is that peer's lost datagram: counted, never raised
# (raising here ended the RUDP thread for every peer). A reliable one is resent.
try:
rudp_socket.sendto(encode_rudp_packet(packet), remote_addr)
except OSError:
self._rudp_send_failures += 1
def _handle_connection_message(
self,
message: dict[str, Any],
subscriptions: dict[str, _DatabaseSubscription],
*,
over_rudp: bool = False,
) -> list[dict[str, Any]]:
schema_version = message.get("schemaVersion")
if schema_version == "cultnet.database_subscribe.v0":
@ -341,7 +380,7 @@ class CultMeshLocalServer:
subscriptions.pop(str(message.get("subscriptionId") or ""), None)
return []
if schema_version == "cultnet.document_put_raw.v0":
return self._handle_raw_put(message, subscriptions)
return self._handle_raw_put(message, subscriptions, over_rudp=over_rudp)
if schema_version == "cultnet.document_delete.v0":
return self._handle_raw_delete(message, subscriptions)
if schema_version == "cultnet.simulation_observation.v0":
@ -353,6 +392,8 @@ class CultMeshLocalServer:
self,
message: dict[str, Any],
subscriptions: dict[str, _DatabaseSubscription],
*,
over_rudp: bool = False,
) -> list[dict[str, Any]]:
document_record = message.get("document")
if not isinstance(document_record, dict):
@ -361,19 +402,9 @@ 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},
)]
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)
if change is None:
return [self._error_response(
@ -586,6 +617,42 @@ class CultMeshLocalServer:
)
return response
def _unservable_put(self, message: dict[str, Any], *, over_rudp: bool) -> dict[str, Any] | None:
"""The refusal for a put this server could never serve, or None. The document alone,
as a snapshot would serve it under the shortest message id, must fit max_snapshot_bytes;
a put arriving over RUDP must also fit, as fragments of rudp_max_fragment_bytes, in one
message and the peer's reliable queue."""
if self.max_snapshot_bytes is None and not over_rudp:
return None
alone = self.node.database.raw_put_served_alone(message, message_id=_SHORTEST_SERVED_MESSAGE_ID)
if alone is None:
return None
response_bytes = _snapshot_response_bytes(alone)
details: dict[str, Any] = {"responseBytes": response_bytes, "maxSnapshotBytes": self.max_snapshot_bytes}
fits = self.max_snapshot_bytes is None or response_bytes <= self.max_snapshot_bytes
limits = f"limit is {self.max_snapshot_bytes}"
if over_rudp:
fragment_count = max(1, -(-response_bytes // self.rudp_max_fragment_bytes))
max_fragment_count = min(
_MAX_FRAGMENTS_PER_MESSAGE,
self.rudp_max_pending_reliable_packets or _MAX_FRAGMENTS_PER_MESSAGE,
)
details |= {"fragmentCount": fragment_count, "maxFragmentCount": max_fragment_count}
fits = fits and fragment_count <= max_fragment_count
limits = (
f"it takes {fragment_count} fragments; limits are "
f"{'unbounded' if self.max_snapshot_bytes is None else self.max_snapshot_bytes} bytes "
f"and {max_fragment_count} fragments"
)
if fits:
return None
return self._error_response(
f"Document can never be served: its snapshot response is {response_bytes} bytes; {limits}.",
message_id=str(message.get("messageId") or ""),
code="document_unservable",
details=details,
)
@staticmethod
def _error_response(
error: str,

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

@ -53,6 +53,7 @@ from cultnet_py import (
CultNetRudpSocketTransportConnection,
CultNetRudpSocketTransportOptions,
apply_shard_log_response,
create_rudp_schema_transport,
document_delete,
document_put_raw,
hello,
@ -4369,6 +4370,161 @@ class CultMeshTests(unittest.TestCase):
self.assertEqual(served["documents"][0]["schemaId"], document.catalog_entry().schema_id)
self.assertEqual(len(served_payload), limit)
def _rudp_bound_note(self) -> Any:
return define_database_entry_type(
"mesh.rudp_bound_note",
[("schema_version", 0), ("body", 1)],
schema_id="sha256:mesh-rudp-bound",
schema_name="mesh.rudp_bound_note",
schema_version="mesh.rudp_bound_note.v1",
)
@staticmethod
def _rudp_note_put(document: Any, message_id: str, key: str, body_length: int) -> dict[str, Any]:
return document_put_raw(
message_id=message_id,
key=key,
schema_id=document.catalog_entry().schema_id,
stored_at="2026-09-30T00:00:00Z",
payload=document.encode_payload({"schema_version": "mesh.rudp_bound_note.v1", "body": "x" * body_length}),
).to_wire()
@staticmethod
def _rudp_exchange(server: Any, *messages: dict[str, Any]) -> tuple[dict[str, Any], bytes]:
"""Sends each message over RUDP in 1200-byte fragments; returns the first reply."""
transport = create_rudp_schema_transport(
host="127.0.0.1",
port=server.port,
connection_id=server.rudp_connection_id,
timeout_seconds=4.0,
max_fragment_bytes=1200,
)
try:
for message in messages:
transport.send("schema", msgpack.packb(message, use_bin_type=True))
payload = transport.receive(timeout_seconds=4.0).payload
finally:
transport.close()
return msgpack.unpackb(payload, raw=False), payload
def test_cultmesh_local_server_serves_a_100_kb_put_over_rudp_in_fragments(self) -> None:
# The reply is larger than any one UDP datagram: it arrives only because it is sent
# in fragments, and the RUDP thread serves on afterwards.
document = self._rudp_bound_note()
node = CultMesh.create_node(runtime_id="mesh-rudp-large")
node.database.register_document(document)
server = CultMesh.serve_node(node)
try:
served, payload = self._rudp_exchange(
server,
self._rudp_note_put(document, "put-large", "note:large", 100_000),
snapshot_request(message_id="large").to_wire(),
)
again, _ = self._rudp_exchange(server, snapshot_request(message_id="again").to_wire())
alive = server._rudp_thread is not None and server._rudp_thread.is_alive()
finally:
server.stop()
self.assertEqual(served["schemaVersion"], "cultnet.snapshot_response_raw.v0")
self.assertEqual([record["recordKey"] for record in served["documents"]], ["note:large"])
self.assertGreater(len(served["documents"][0]["payload"]), 100_000)
self.assertGreater(len(payload), 65_507, "fixture: the reply cannot be one UDP datagram")
self.assertEqual(again["messageId"], "again")
self.assertTrue(alive)
self.assertEqual(server.rudp_send_failures, 0)
def test_cultmesh_local_server_refuses_a_rudp_put_whose_reply_would_overflow_the_reliable_queue(self) -> None:
# Replies go in fragments of 100 bytes and a peer's reliable queue holds 8: a put over RUDP
# whose served reply takes 8 fragments is admitted and served, 9 or 20 are refused.
document = self._rudp_bound_note()
# The served size of a document alone under the empty message id, measured on an
# unbounded server: the bytes a snapshot actually sends.
def served_alone_bytes(body_length: int) -> int:
node = CultMesh.create_node(runtime_id="mesh-rudp-bound-probe")
node.database.register_document(document)
server = CultMesh.serve_node(node)
try:
_, payload = self._rudp_exchange(
server, self._rudp_note_put(document, "probe", "note:q", body_length),
snapshot_request(message_id="").to_wire(),
)
finally:
server.stop()
return len(payload)
probe = served_alone_bytes(500)
def body_for(served_bytes: int) -> int:
body_length = 500 + served_bytes - probe
self.assertEqual(served_alone_bytes(body_length), served_bytes, "fixture: served size is linear in the body")
return body_length
node = CultMesh.create_node(runtime_id="mesh-rudp-bound")
node.database.register_document(document)
server = CultMesh.serve_node(node, rudp_max_fragment_bytes=100, rudp_max_pending_reliable_packets=8)
try:
refusals = [
self._rudp_exchange(server, self._rudp_note_put(document, f"put-{served}", "note:q", body_for(served)))[0]
for served in (2000, 801)
]
stored_after_refusals = node.database.get(document, "note:q")
served, served_payload = self._rudp_exchange(
server,
self._rudp_note_put(document, "put-at", "note:q", body_for(800)),
snapshot_request(message_id="").to_wire(),
)
alive = server._rudp_thread is not None and server._rudp_thread.is_alive()
finally:
server.stop()
for refusal, (served_bytes, fragments) in zip(refusals, ((2000, 20), (801, 9))):
self.assertEqual(refusal["schemaVersion"], "cultnet.error.v0")
self.assertEqual(refusal["messageId"], f"put-{served_bytes}")
self.assertEqual(refusal["code"], "document_unservable")
self.assertEqual(refusal["details"], {
"responseBytes": served_bytes,
"maxSnapshotBytes": None,
"fragmentCount": fragments,
"maxFragmentCount": 8,
})
self.assertIn(f"{served_bytes} bytes; it takes {fragments} fragments", refusal["error"])
self.assertNotIn("note:q", refusal["error"])
self.assertIsNone(stored_after_refusals)
self.assertEqual(served["schemaVersion"], "cultnet.snapshot_response_raw.v0")
self.assertEqual([record["recordKey"] for record in served["documents"]], ["note:q"])
self.assertEqual(len(served_payload), 800)
self.assertTrue(alive)
def test_cultmesh_local_server_rudp_thread_survives_a_datagram_that_cannot_leave(self) -> None:
# Fragments larger than any UDP datagram: the reply's one packet cannot be sent. That
# peer's loss is counted; the thread serves the next peer.
document = self._rudp_bound_note()
node = CultMesh.create_node(runtime_id="mesh-rudp-emsgsize")
node.database.register_document(document)
server = CultMesh.serve_node(node, rudp_max_fragment_bytes=200_000)
try:
transport = create_rudp_schema_transport(
host="127.0.0.1", port=server.port, connection_id=server.rudp_connection_id,
timeout_seconds=4.0, max_fragment_bytes=1200,
)
try:
transport.send("schema", msgpack.packb(self._rudp_note_put(document, "put-large", "note:big", 80_000), use_bin_type=True))
transport.send("schema", msgpack.packb(snapshot_request(message_id="big").to_wire(), use_bin_type=True))
deadline = time.monotonic() + 4.0
while server.rudp_send_failures == 0 and time.monotonic() < deadline:
transport.receive_once()
time.sleep(0.005)
finally:
transport.close()
failures = server.rudp_send_failures
hello_reply, _ = self._rudp_exchange(server, hello(runtime_id="after").to_wire())
alive = server._rudp_thread is not None and server._rudp_thread.is_alive()
finally:
server.stop()
self.assertGreaterEqual(failures, 1)
self.assertEqual(hello_reply["schemaVersion"], "cultnet.hello.v0")
self.assertTrue(alive)
def test_cultmesh_local_server_returns_errors_for_bad_requests(self) -> None:
document = define_database_entry_type(
"mesh.error_note",