diff --git a/packages/cultmesh-py/src/cultmesh_py/__init__.py b/packages/cultmesh-py/src/cultmesh_py/__init__.py index af4da272..f2dc6b28 100644 --- a/packages/cultmesh-py/src/cultmesh_py/__init__.py +++ b/packages/cultmesh-py/src/cultmesh_py/__init__.py @@ -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", diff --git a/packages/cultmesh-py/src/cultmesh_py/facade.py b/packages/cultmesh-py/src/cultmesh_py/facade.py index 626b1dc7..c6b781c2 100644 --- a/packages/cultmesh-py/src/cultmesh_py/facade.py +++ b/packages/cultmesh-py/src/cultmesh_py/facade.py @@ -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 diff --git a/packages/cultmesh-py/src/cultmesh_py/server.py b/packages/cultmesh-py/src/cultmesh_py/server.py index 605b5a7e..daf0ed6c 100644 --- a/packages/cultmesh-py/src/cultmesh_py/server.py +++ b/packages/cultmesh-py/src/cultmesh_py/server.py @@ -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, diff --git a/packages/cultmesh-py/tests/test_cultmesh.py b/packages/cultmesh-py/tests/test_cultmesh.py index ddd8e89a..51398c89 100644 --- a/packages/cultmesh-py/tests/test_cultmesh.py +++ b/packages/cultmesh-py/tests/test_cultmesh.py @@ -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",