feat(runtime): add integration hardening and diagnostics

scripts/ws_protocol_client.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import argparse
 import base64
 import hashlib
+import http.client
 import json
 import os
 import socket
@@ -14,6 +15,8 @@ import threading
 import time
 from typing import Any
 
+MALFORMED_STRESS_MESSAGES = 80
+
 
 class WebSocket:
     def __init__(self, host: str, port: int = 80, timeout: float = 8.0):
@@ -153,6 +156,200 @@ def open_direction(view: dict[str, Any]) -> str:
     raise AssertionError("player cell has no open passage")
 
 
+def diagnostics(host: str) -> dict[str, Any]:
+    connection = http.client.HTTPConnection(host, 80, timeout=8.0)
+    try:
+        connection.request("GET", "/diagnostics")
+        response = connection.getresponse()
+        payload = response.read()
+        if response.status != 200:
+            raise RuntimeError(f"diagnostics returned HTTP {response.status}")
+        return json.loads(payload.decode("utf-8"))
+    finally:
+        connection.close()
+
+
+def verify_m010(host: str, stress_moves: int, repeat_matches: int) -> None:
+    clients: list[WebSocket] = []
+    try:
+        baseline = diagnostics(host)
+        player_1 = WebSocket(host, timeout=30.0)
+        player_2 = WebSocket(host, timeout=30.0)
+        clients.extend((player_1, player_2))
+        join(player_1, "player", "Нагрузочный Один")
+        session_2 = join(player_2, "player", "Нагрузочный Два")
+        observers = [WebSocket(host, timeout=30.0) for _ in range(4)]
+        clients.extend(observers)
+        for observer in observers:
+            join(observer, "observer")
+
+        observer_views: list[dict[str, Any] | None] = [None] * len(observers)
+        failures: list[BaseException] = []
+
+        def read_observer(index: int) -> None:
+            try:
+                observers[index].receive_type("active", attempts=50)
+                observer_views[index] = receive_view(observers[index])
+            except BaseException as error:  # Preserve worker failure for the main thread.
+                failures.append(error)
+
+        readers = [threading.Thread(target=read_observer, args=(index,))
+                   for index in range(len(observers))]
+        for reader in readers:
+            reader.start()
+        player_1.send("start", {}, "m010-start")
+        player_1.receive_type("active", attempts=50)
+        player_2.receive_type("active", attempts=50)
+        player_1_view = receive_view(player_1)
+        receive_view(player_2)
+        for reader in readers:
+            reader.join(timeout=40.0)
+        assert not failures and all(not reader.is_alive() for reader in readers)
+        assert all(view is not None and len(view["cells"]) == 8192
+                   for view in observer_views)
+
+        six_client_sample = diagnostics(host)
+        limits = six_client_sample["limits"]
+        assert six_client_sample["activeConnections"] == 6
+        assert six_client_sample["connectionHighWater"] == 6
+        assert six_client_sample["queueHighWater"] <= limits["queueEntries"]
+        print("M010 six-client load: PASS (2 players + 4 complete observers)")
+
+        observers[0].send("restart", {}, "m010-observer-restart")
+        observer_error = observers[0].receive_type("error", attempts=30)
+        assert observer_error["payload"]["code"] == "read_only"
+
+        heartbeat_latencies_ms: list[float] = []
+        for index, client in enumerate(clients):
+            started = time.monotonic()
+            client.send("heartbeat", {}, f"m010-heartbeat-{index}")
+            client.receive_type("ack", attempts=30)
+            heartbeat_latencies_ms.append((time.monotonic() - started) * 1000.0)
+        assert max(heartbeat_latencies_ms) <= limits["playerLatencyBudgetMs"]
+        print(f"M010 six-client responsiveness: PASS (max {max(heartbeat_latencies_ms):.1f} ms)")
+
+        slow_observer = observers[0]
+        slow_observer.socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024)
+        drain_stop = threading.Event()
+        drain_failures: list[BaseException] = []
+
+        def drain_fast_observer(observer: WebSocket) -> None:
+            observer.socket.settimeout(1.0)
+            while not drain_stop.is_set():
+                try:
+                    observer.receive()
+                except socket.timeout:
+                    continue
+                except BaseException as error:  # Preserve an unexpected disconnect.
+                    drain_failures.append(error)
+                    return
+
+        drainers = [threading.Thread(target=drain_fast_observer, args=(observer,))
+                    for observer in observers[1:]]
+        for drainer in drainers:
+            drainer.start()
+        opposite = {"north": "south", "east": "west", "south": "north", "west": "east"}
+        direction = open_direction(player_1_view)
+        command_latencies_ms: list[float] = []
+        for move_index in range(stress_moves):
+            time.sleep(0.16)
+            started = time.monotonic()
+            player_1.send("move", {"direction": direction}, f"m010-move-{move_index}")
+            player_1.receive_type("ack", attempts=50)
+            command_latencies_ms.append((time.monotonic() - started) * 1000.0)
+            player_1_view = receive_view(player_1)
+            receive_view(player_2)
+            direction = opposite[direction]
+
+        drain_stop.set()
+        for drainer in drainers:
+            drainer.join(timeout=3.0)
+        assert not drain_failures and all(not drainer.is_alive() for drainer in drainers)
+        for observer in observers[1:]:
+            observer.socket.settimeout(30.0)
+
+        abusive_observer = observers[1]
+
+        def malformed_flood() -> None:
+            try:
+                for _ in range(MALFORMED_STRESS_MESSAGES):
+                    abusive_observer.send_raw(b"{")
+            except OSError:
+                pass
+
+        flood = threading.Thread(target=malformed_flood)
+        flood.start()
+        stress_heartbeat_latencies_ms: list[float] = []
+        for index in range(10):
+            started = time.monotonic()
+            player_1.send("heartbeat", {}, f"m010-stress-heartbeat-{index}")
+            player_1.receive_type("ack", attempts=50)
+            stress_heartbeat_latencies_ms.append((time.monotonic() - started) * 1000.0)
+            time.sleep(0.02)
+        flood.join(timeout=10.0)
+        assert not flood.is_alive()
+        stressed = diagnostics(host)
+        maximum_latency_ms = max(command_latencies_ms + stress_heartbeat_latencies_ms)
+        assert maximum_latency_ms <= limits["playerLatencyBudgetMs"]
+        assert stressed["malformedMessages"] > baseline["malformedMessages"]
+        assert stressed["rateLimitedMessages"] > baseline["rateLimitedMessages"]
+        assert stressed["queueHighWater"] <= limits["queueEntries"]
+        assert stressed["minimumInternalFreeBytes"] > 32768
+        assert stressed["minimumInternalLargestBlockBytes"] > 16384
+        print(
+            "M010 slow/malformed stress: PASS "
+            f"(max player latency {maximum_latency_ms:.1f} ms, "
+            f"dropped {stressed['messagesDropped'] - baseline['messagesDropped']})"
+        )
+
+        for observer in observers:
+            observer.close()
+            if observer in clients:
+                clients.remove(observer)
+        observers.clear()
+        time.sleep(1.0)
+
+        reset_heap_samples: list[int] = []
+        for match_index in range(repeat_matches):
+            token_2 = session_2["token"]
+            player_2.close()
+            clients.remove(player_2)
+            player_1.receive_type("pause", attempts=100)
+            player_1.socket.settimeout(70.0)
+            result = player_1.receive_type("match_result", attempts=200)["payload"]
+            assert result["result"] == "draw"
+
+            player_2 = WebSocket(host, timeout=30.0)
+            clients.append(player_2)
+            player_2.send("restore", {"token": token_2}, f"m010-restore-{match_index}")
+            session_2 = player_2.receive_type("session", attempts=50)["payload"]
+            player_1.send("restart", {}, f"m010-restart-{match_index}")
+            player_1.receive_type("ack", attempts=100)
+            time.sleep(0.2)
+            sample = diagnostics(host)
+            reset_heap_samples.append(sample["internalFreeBytes"])
+            if match_index + 1 < repeat_matches:
+                player_1.send("start", {}, f"m010-start-{match_index + 1}")
+                player_1.receive_type("active", attempts=100)
+                player_2.receive_type("active", attempts=100)
+                receive_view(player_1)
+                receive_view(player_2)
+
+        final_sample = diagnostics(host)
+        assert final_sample["matchesCompleted"] - baseline["matchesCompleted"] >= repeat_matches
+        assert final_sample["matchesReset"] - baseline["matchesReset"] >= repeat_matches
+        if len(reset_heap_samples) > 1:
+            assert max(reset_heap_samples) - min(reset_heap_samples) <= 32768
+        print(
+            f"M010 repeated-match stability: PASS ({repeat_matches} matches, "
+            f"heap range {max(reset_heap_samples) - min(reset_heap_samples)} bytes)"
+        )
+        print(json.dumps(final_sample, ensure_ascii=False, sort_keys=True))
+    finally:
+        for client in clients:
+            client.close()
+
+
 def verify_m009(host: str) -> None:
     clients: list[WebSocket] = []
     try:
@@ -350,8 +547,17 @@ def main() -> None:
     parser.add_argument("host", help="maze32 IP address or hostname")
     parser.add_argument("--verify", action="store_true", help="run the Milestone 004 scenario")
     parser.add_argument("--verify-m009", action="store_true", help="run the Milestone 009 scenario")
+    parser.add_argument("--verify-m010", action="store_true", help="run the Milestone 010 load and stability scenario")
+    parser.add_argument("--stress-moves", type=int, default=80,
+                        help="accepted moves sent while one observer is not reading")
+    parser.add_argument("--repeat-matches", type=int, default=2,
+                        help="real 60-second timeout matches completed and reset")
     arguments = parser.parse_args()
-    if arguments.verify_m009:
+    if arguments.stress_moves < 1 or arguments.repeat_matches < 1:
+        parser.error("--stress-moves and --repeat-matches must be positive")
+    if arguments.verify_m010:
+        verify_m010(arguments.host, arguments.stress_moves, arguments.repeat_matches)
+    elif arguments.verify_m009:
         verify_m009(arguments.host)
     elif arguments.verify:
         verify(arguments.host)