fix(gateway): harden websocket streaming under client backpressure

scripts/ws_protocol_client.py

@@ -178,7 +178,7 @@ def verify_m010(host: str, stress_moves: int, repeat_matches: int) -> None:
         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)]
+        observers = [WebSocket(host, timeout=120.0) for _ in range(4)]
         clients.extend(observers)
         for observer in observers:
             join(observer, "observer")
@@ -197,14 +197,35 @@ def verify_m010(host: str, stress_moves: int, repeat_matches: int) -> None:
                    for index in range(len(observers))]
         for reader in readers:
             reader.start()
+        player_views: list[dict[str, Any] | None] = [None, None]
+        player_failures: list[BaseException] = []
+
+        def read_player(index: int, client: WebSocket) -> None:
+            try:
+                client.receive_type("active", attempts=50)
+                player_views[index] = receive_view(client)
+            except BaseException as error:  # Preserve worker failure for the main thread.
+                player_failures.append(error)
+
+        player_readers = [
+            threading.Thread(target=read_player, args=(0, player_1)),
+            threading.Thread(target=read_player, args=(1, player_2)),
+        ]
+        for reader in player_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 player_readers:
+            reader.join(timeout=180.0)
         for reader in readers:
-            reader.join(timeout=40.0)
-        assert not failures and all(not reader.is_alive() for reader in readers)
+            reader.join(timeout=180.0)
+        assert not player_failures and all(not reader.is_alive() for reader in player_readers), (
+            f"player read failures: {player_failures!r}"
+        )
+        assert not failures and all(not reader.is_alive() for reader in readers), (
+            f"observer read failures: {failures!r}"
+        )
+        player_1_view = player_views[0]
+        assert player_1_view is not None and player_views[1] is not None
         assert all(view is not None and len(view["cells"]) == 8192
                    for view in observer_views)
 
@@ -248,6 +269,7 @@ def verify_m010(host: str, stress_moves: int, repeat_matches: int) -> None:
                     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] = []
@@ -264,7 +286,9 @@ def verify_m010(host: str, stress_moves: int, repeat_matches: int) -> None:
         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)
+        assert not drain_failures and all(not drainer.is_alive() for drainer in drainers), (
+            f"observer drain failures: {drain_failures!r}"
+        )
         for observer in observers[1:]:
             observer.socket.settimeout(30.0)
 
@@ -290,7 +314,9 @@ def verify_m010(host: str, stress_moves: int, repeat_matches: int) -> None:
         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 maximum_latency_ms <= limits["playerLatencyBudgetMs"], (
+            f"maximum player latency {maximum_latency_ms:.1f} ms"
+        )
         assert stressed["malformedMessages"] > baseline["malformedMessages"]
         assert stressed["rateLimitedMessages"] > baseline["rateLimitedMessages"]
         assert stressed["queueHighWater"] <= limits["queueEntries"]
@@ -329,11 +355,20 @@ def verify_m010(host: str, stress_moves: int, repeat_matches: int) -> None:
             sample = diagnostics(host)
             reset_heap_samples.append(sample["internalFreeBytes"])
             if match_index + 1 < repeat_matches:
+                player_views = [None, None]
+                player_failures = []
+                player_readers = [
+                    threading.Thread(target=read_player, args=(0, player_1)),
+                    threading.Thread(target=read_player, args=(1, player_2)),
+                ]
+                for reader in player_readers:
+                    reader.start()
                 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)
+                for reader in player_readers:
+                    reader.join(timeout=40.0)
+                assert not player_failures and all(
+                    not reader.is_alive() for reader in player_readers
+                )
 
         final_sample = diagnostics(host)
         assert final_sample["matchesCompleted"] - baseline["matchesCompleted"] >= repeat_matches