Merge pull request 'cultmesh-py: answer an unreadable RUDP frame and end the session after 8 consecutive refusals' (#45) from libby/py-refused-frames-cap into main

packages/cultmesh-py/src/cultmesh_py/server.py

@@ -59,6 +59,16 @@ from cultnet_py.cultmesh_contracts import (
 )
 
 _LOGGER = logging.getLogger(__name__)
+
+# How many consecutive refused frames of one kind a RUDP session may send before it is ended: undecodable
+# frames (reset by any frame that decodes) and deletes the write authority refuses (reset by any decoded
+# frame that is not a refused delete). Fewer are answered with a typed refusal and the session is kept; the
+# frame that reaches this number is answered too, then the session ends. The same number is the cap in every
+# runtime.
+MAX_CONSECUTIVE_REFUSED_FRAMES = 8
+
+# How long a session whose capped frame was answered waits for the peer to acknowledge the answer before it ends anyway.
+_RUDP_CAPPED_ANSWER_GRACE_MS = 2000
 _EPHEMERAL_PAIR_ATTEMPTS = 64
 
 
@@ -90,6 +100,10 @@ class _DatabaseSubscription:
 class _RudpPeerConnection:
     session: CultNetRudpSession
     subscriptions: dict[str, _DatabaseSubscription] = field(default_factory=dict)
+    undecodable_frames_in_a_row: int = 0
+    refused_deletes_in_a_row: int = 0
+    # Set once the peer's capped frame has been answered: the session ends when the answers are acknowledged.
+    ending_deadline_ms: int | None = None
 
 
 @dataclass
@@ -346,6 +360,7 @@ class CultMeshLocalServer:
             if packet.connection_id != self.rudp_connection_id:
                 continue
 
+            self._end_answered_peers(rudp_socket, peers)
             now_ms = _now_ms()
             peer = peers.get(remote_addr)
             if packet.packet_type == CultNetRudpPacketType.CONNECT:
@@ -394,18 +409,31 @@ class CultMeshLocalServer:
                 for delivered in result.delivered:
                     if delivered.channel_id != "schema":
                         continue
+                    if peer.ending_deadline_ms is not None:
+                        # The capped frame was answered; nothing after it is served while the answer is delivered.
+                        break
                     try:
                         message = msgpack.unpackb(delivered.payload, raw=False)
                     except (msgpack.ExtraData, msgpack.FormatError, msgpack.StackError, ValueError):
-                        continue
-                    if not isinstance(message, dict):
-                        continue
-                    responses = self._handle_connection_message(
-                        message,
-                        peer.subscriptions,
-                        over_rudp=True,
-                        peer=CultMeshServerPeer(f"{remote_addr[0]}:{remote_addr[1]}", "rudp"),
-                    )
+                        message = None
+                    if isinstance(message, dict):
+                        peer.undecodable_frames_in_a_row = 0
+                        responses = self._handle_connection_message(
+                            message,
+                            peer.subscriptions,
+                            over_rudp=True,
+                            peer=CultMeshServerPeer(f"{remote_addr[0]}:{remote_addr[1]}", "rudp"),
+                        )
+                        refused_delete = message.get("schemaVersion") == "cultnet.document_delete.v0" and any(
+                            response.get("code") == "write_not_authorized" for response in responses
+                        )
+                        peer.refused_deletes_in_a_row = peer.refused_deletes_in_a_row + 1 if refused_delete else 0
+                        refused, run = refused_delete, peer.refused_deletes_in_a_row
+                    else:
+                        # A frame that is no CultNet message is answered, not dropped, and the session kept.
+                        peer.undecodable_frames_in_a_row += 1
+                        refused, run = True, peer.undecodable_frames_in_a_row
+                        responses = [self._message_invalid({}, "message", "CultMesh RUDP server could not read the message.")]
                     for response in responses:
                         if unsendable is None:
                             unsendable = self._send_rudp_schema_frame(
@@ -415,6 +443,11 @@ class CultMeshLocalServer:
                                 msgpack.packb(response, use_bin_type=True),
                                 message_id=str(response.get("messageId") or ""),
                             )
+                    if refused and run >= MAX_CONSECUTIVE_REFUSED_FRAMES:
+                        # The capped frame is answered like any other, reliably. The session ends once the peer
+                        # has acknowledged every answer (or the grace runs out): see _end_answered_peers.
+                        peer.ending_deadline_ms = now_ms + _RUDP_CAPPED_ANSWER_GRACE_MS
+                        break
             except Exception as error:  # noqa: BLE001 - one peer's message never ends the thread
                 # Whatever a peer's message makes the handling raise is that peer's failure.
                 # The session recorded the frame's sequence, so it cannot be kept: a retransmit
@@ -445,11 +478,26 @@ class CultMeshLocalServer:
             if unsendable is not None:
                 self._end_unsendable_peer(rudp_socket, peers, remote_addr, unsendable)
 
+    def _end_answered_peers(
+        self,
+        rudp_socket: socket.socket,
+        peers: dict[tuple[str, int], _RudpPeerConnection],
+    ) -> None:
+        """Ends each session whose capped frame was answered once the peer has acknowledged every answer (a lost
+        answer is resent first), or when the grace has run out."""
+        now_ms = _now_ms()
+        for remote_addr, peer in list(peers.items()):
+            if peer.ending_deadline_ms is None:
+                continue
+            if peer.session.outstanding_reliable_packet_count == 0 or now_ms >= peer.ending_deadline_ms:
+                self._end_refused_peer(rudp_socket, peers, remote_addr)
+
     def _poll_rudp_resends(
         self,
         rudp_socket: socket.socket,
         peers: dict[tuple[str, int], _RudpPeerConnection],
     ) -> None:
+        self._end_answered_peers(rudp_socket, peers)
         now_ms = _now_ms()
         for remote_addr, peer in list(peers.items()):
             for packet in peer.session.due_resends(now_ms):

packages/cultmesh-py/tests/test_cultmesh.py

@@ -24,7 +24,9 @@ from cultcache_py import (
     define_document_type,
 )
 from cultcache_py.interop import INTEROP_SCHEMA_VERSION, interop_note_document
+from cultmesh_py.server import MAX_CONSECUTIVE_REFUSED_FRAMES
 from cultmesh_py.verify import verify
+from cultmesh_py.write_authority import CultMeshRemoteWriteDecision
 from cultnet_py import (
     compute_simulation_claim_hash,
     CultNetClientAuthorityScope,
@@ -4889,6 +4891,140 @@ class CultMeshTests(unittest.TestCase):
         self.assertEqual(after["documents"], [])
         self.assertTrue(alive)
 
+    def _rudp_open(self, server: Any) -> Any:
+        return 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,
+        )
+
+    def _rudp_answer(self, transport: Any, payload: bytes) -> dict[str, Any]:
+        transport.send("schema", payload)
+        return msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False)
+
+    def _assert_rudp_session_ended(self, transport: Any, server: Any) -> None:
+        transport.send("schema", msgpack.packb(snapshot_request(message_id="after-cap").to_wire(), use_bin_type=True))
+        with self.assertRaises(Exception):
+            transport.receive(timeout_seconds=1.0)
+        self.assertTrue(server._rudp_thread.is_alive())
+
+    def test_cultmesh_local_server_answers_unreadable_frames_and_ends_the_session_at_eight_in_a_row(self) -> None:
+        # An undecodable frame (not a msgpack map) is answered message_invalid and the session kept;
+        # eight in a row end it (the eighth is answered too); any decoded frame resets the run.
+        self.assertEqual(MAX_CONSECUTIVE_REFUSED_FRAMES, 8)
+        node = CultMesh.create_node(runtime_id="mesh-rudp-unreadable")
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites())
+        garbage = b"\xc1"
+        snapshot = msgpack.packb(snapshot_request(message_id="ok").to_wire(), use_bin_type=True)
+        transport = self._rudp_open(server)
+        try:
+            for _ in range(2):
+                for _ in range(MAX_CONSECUTIVE_REFUSED_FRAMES - 1):
+                    reply = self._rudp_answer(transport, garbage)
+                    self.assertEqual(reply["code"], "message_invalid")
+                    self.assertEqual(reply["details"]["field"], "message")
+                self.assertEqual(self._rudp_answer(transport, snapshot)["messageId"], "ok")
+            for _ in range(MAX_CONSECUTIVE_REFUSED_FRAMES - 1):
+                self._rudp_answer(transport, garbage)
+            transport.send("schema", garbage)
+            eighth = msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False)
+            self.assertEqual(eighth["code"], "message_invalid")
+            self._assert_rudp_session_ended(transport, server)
+        finally:
+            transport.close()
+            server.stop()
+
+    def test_cultmesh_local_server_resends_a_lost_capped_answer_before_it_ends_the_session(self) -> None:
+        # The eighth refusal is answered reliably: the datagram carrying it is lost on the wire, and the
+        # peer still gets the answer (by resend) before the goodbye. An unreliable final answer, or a
+        # goodbye sent at once, would leave the peer without it.
+        node = CultMesh.create_node(runtime_id="mesh-rudp-lost-answer")
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites())
+        send = server._send_rudp_packet
+        dropped: list[Any] = []
+        armed = threading.Event()
+
+        def lossy(rudp_socket: Any, remote_addr: Any, packet: Any) -> Any:
+            if armed.is_set() and not dropped and packet.packet_type == CultNetRudpPacketType.DATA and b"message_invalid" in (packet.payload or b""):
+                dropped.append(packet)
+                return None
+            return send(rudp_socket, remote_addr, packet)
+
+        server._send_rudp_packet = lossy
+        transport = self._rudp_open(server)
+        try:
+            for _ in range(MAX_CONSECUTIVE_REFUSED_FRAMES - 1):
+                self._rudp_answer(transport, b"\xc1")
+            armed.set()
+            reply = self._rudp_answer(transport, b"\xc1")
+            self.assertEqual(len(dropped), 1, "fixture: the first transmission of the last answer was lost")
+            self.assertEqual(reply["code"], "message_invalid")
+            self._assert_rudp_session_ended(transport, server)
+        finally:
+            transport.close()
+            server.stop()
+
+    def test_cultmesh_local_server_serves_nothing_after_the_capped_frame(self) -> None:
+        node = CultMesh.create_node(runtime_id="mesh-rudp-after-cap")
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites())
+        transport = self._rudp_open(server)
+        try:
+            for _ in range(MAX_CONSECUTIVE_REFUSED_FRAMES - 1):
+                self._rudp_answer(transport, b"\xc1")
+            transport.send("schema", b"\xc1")
+            transport.send("schema", msgpack.packb(snapshot_request(message_id="behind-the-cap").to_wire(), use_bin_type=True))
+            first = msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False)
+            self.assertEqual(first["code"], "message_invalid")
+            with self.assertRaises(Exception):
+                transport.receive(timeout_seconds=1.0)
+        finally:
+            transport.close()
+            server.stop()
+
+    def test_cultmesh_local_server_ends_the_session_at_eight_refused_deletes_in_a_row(self) -> None:
+        class GrantOnlyKeyB:
+            def decide(self, request: Any) -> Any:
+                return CultMeshRemoteWriteDecision.grant() if request.record_key == "note:b" else CultMeshRemoteWriteDecision.refuse("no")
+
+        document = self._rudp_bound_note()
+        node = CultMesh.create_node(runtime_id="mesh-rudp-deletes")
+        node.database.register_document(document)
+        server = CultMesh.serve_node(node, write_authority=GrantOnlyKeyB())
+
+        def delete(key: str) -> bytes:
+            return msgpack.packb({
+                "schemaVersion": "cultnet.document_delete.v0",
+                "messageId": f"del-{key}",
+                "schemaId": document.schema_id,
+                "recordKey": key,
+            }, use_bin_type=True)
+
+        snapshot = msgpack.packb(snapshot_request(message_id="ok").to_wire(), use_bin_type=True)
+        transport = self._rudp_open(server)
+        try:
+            def refused_run(count: int) -> None:
+                for _ in range(count):
+                    reply = self._rudp_answer(transport, delete("note:a"))
+                    self.assertEqual(reply["code"], "write_not_authorized")
+                    self.assertEqual(reply["details"]["value"], "note:a")
+
+            refused_run(MAX_CONSECUTIVE_REFUSED_FRAMES - 1)
+            # A granted delete resets the run (nothing is stored under it, so the store answers that).
+            self.assertEqual(self._rudp_answer(transport, delete("note:b"))["code"], "unregistered_document_delete")
+            refused_run(MAX_CONSECUTIVE_REFUSED_FRAMES - 1)
+            # So does any decoded non-delete frame.
+            self.assertEqual(self._rudp_answer(transport, snapshot)["messageId"], "ok")
+            refused_run(MAX_CONSECUTIVE_REFUSED_FRAMES - 1)
+            transport.send("schema", delete("note:a"))
+            eighth = msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False)
+            self.assertEqual(eighth["code"], "write_not_authorized")
+            self._assert_rudp_session_ended(transport, server)
+        finally:
+            transport.close()
+            server.stop()
+
     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