fix(haos): bound supervisor websocket sends

Author: kingpanther13

tests/haos_image_build/build_image.py

@@ -25,6 +25,7 @@
 import subprocess
 import sys
 import tempfile
+import threading
 import time
 import urllib.error
 import urllib.parse
@@ -762,6 +763,45 @@ def _wait_supervisor_api_ready(self, timeout: float = 60.0) -> None:
             f"within {timeout:.0f}s after Core restart (attempts={attempts})"
         ) from last_error
 
+    def _send_with_deadline(
+        self,
+        message: str,
+        *,
+        deadline: float,
+        operation: str,
+    ) -> None:
+        """Send one frame within a deadline, cancelling a stalled socket write."""
+        connection = self._ws
+        if connection is None:
+            raise ConnectionError(f"WebSocket is not connected for {operation}")
+        _remaining_deadline_budget(deadline, operation)
+        send_errors: list[Exception] = []
+
+        def send() -> None:
+            try:
+                _remaining_deadline_budget(deadline, operation)
+                connection.send(message)
+            except Exception as exc:
+                send_errors.append(exc)
+
+        worker = threading.Thread(target=send, name="haos-ws-send", daemon=True)
+        worker.start()
+        worker.join(max(0.0, deadline - time.monotonic()))
+        if worker.is_alive():
+            try:
+                connection.close_socket()
+            except (OSError, RuntimeError) as exc:
+                LOG.debug("WS close error after stalled send: %r", exc)
+            finally:
+                if self._ws is connection:
+                    self._ws = None
+            raise TimeoutError(
+                f"{operation} exceeded its deadline after dispatch; "
+                "command outcome is unknown"
+            )
+        if send_errors:
+            raise send_errors[0]
+
     def supervisor_api(
         self,
         endpoint: str,
@@ -790,11 +830,12 @@ def supervisor_api(
         }
         if data is not None:
             msg["data"] = data
-        _remaining_deadline_budget(
-            deadline,
-            f"supervisor/api {method} {endpoint} send",
+        message = json.dumps(msg)
+        self._send_with_deadline(
+            message,
+            deadline=deadline,
+            operation=f"supervisor/api {method} {endpoint} send",
         )
-        self._ws.send(json.dumps(msg))
         # Skip any out-of-band messages (events on subscriptions etc.) and
         # match by id.
         while True:

tests/src/e2e/haos_only/test_zz_manage_app_tls.py

@@ -157,6 +157,17 @@ async def test_manage_app_reproduces_legacy_tls_failure_then_uses_fix(
             MCPAssertions(mcp) as assertions,
         ):
             slug = await _resolve_slug(mcp, NODERED_NAME)
+            initial_detail = (
+                await assertions.call_tool_success("ha_get_app", {"slug": slug})
+            ).get("addon") or {}
+            if initial_detail.get("state") != "started":
+                # Core's protocol restart can terminate Node-RED while its
+                # Supervisor WebSocket proxy reconnects. Start it again so the
+                # TLS assertions begin from a settled app state.
+                await assertions.call_tool_success(
+                    "ha_manage_app",
+                    {"slug": slug, "action": "start"},
+                )
             await _wait_addon_running(mcp, slug)
             detail = (
                 await assertions.call_tool_success("ha_get_app", {"slug": slug})

tests/src/unit/test_haos_supervisor_wait.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import json
+from threading import Event
 from typing import Any
 from unittest.mock import Mock, call, patch
 
@@ -58,15 +59,15 @@ def test_supervisor_api_shares_receive_deadline_across_frames() -> None:
 
     with patch(
         "tests.haos_image_build.build_image.time.monotonic",
-        side_effect=[10.0, 11.0, 12.5],
+        side_effect=[10.0, 10.5, 10.5, 10.5, 11.0, 12.5],
     ):
         assert ws.supervisor_api("/supervisor/info", timeout=5.0) == {}
 
     assert socket.recv.call_args_list == [call(timeout=4.0), call(timeout=2.5)]
 
 
-def test_supervisor_api_expired_deadline_does_not_dispatch() -> None:
-    """An expired command budget cannot produce an uncertain write outcome."""
+def test_supervisor_api_expiry_during_serialization_does_not_dispatch() -> None:
+    """Serialization time is charged before a command can be dispatched."""
     ws = HAWebSocket(
         "http://127.0.0.1:18123",
         OAuthCredentials(access_token="access", refresh_token="refresh"),
@@ -77,19 +78,58 @@ def test_supervisor_api_expired_deadline_does_not_dispatch() -> None:
     with (
         patch(
             "tests.haos_image_build.build_image.time.monotonic",
-            side_effect=[10.0, 10.0],
+            side_effect=[10.0, 12.0],
         ),
+        patch(
+            "tests.haos_image_build.build_image.json.dumps",
+            return_value='{"serialized": true}',
+        ) as serialize,
         pytest.raises(
             TimeoutError,
             match=r"supervisor/api post /core/update send exceeded its deadline",
         ),
     ):
-        ws.supervisor_api("/core/update", method="post", data={}, timeout=0.0)
+        ws.supervisor_api("/core/update", method="post", data={}, timeout=1.0)
 
+    serialize.assert_called_once()
     socket.send.assert_not_called()
     socket.recv.assert_not_called()
 
 
+def test_supervisor_api_stalled_send_is_bounded() -> None:
+    """A stalled socket write is cancelled within the command's deadline."""
+    ws = HAWebSocket(
+        "http://127.0.0.1:18123",
+        OAuthCredentials(access_token="access", refresh_token="refresh"),
+    )
+    socket = Mock()
+    release_send = Event()
+
+    def stalled_send(_: str) -> None:
+        release_send.wait()
+
+    socket.send.side_effect = stalled_send
+    socket.close_socket.side_effect = release_send.set
+    ws._ws = socket
+
+    with (
+        patch(
+            "tests.haos_image_build.build_image.time.monotonic",
+            return_value=10.0,
+        ),
+        pytest.raises(
+            TimeoutError,
+            match="command outcome is unknown",
+        ),
+    ):
+        ws.supervisor_api("/core/update", method="post", data={}, timeout=0.01)
+
+    socket.send.assert_called_once()
+    socket.close_socket.assert_called_once()
+    socket.recv.assert_not_called()
+    assert ws._ws is None
+
+
 @pytest.mark.parametrize(
     ("error", "expected_code", "expected_message"),
     [