Keep reconnecting after stalls and unexpected errors

src/aiopoolside/client.py

@@ -116,14 +116,24 @@ async def async_disconnect(self) -> None:
 
     async def _async_connect_once(self) -> None:
         """Perform a single connection attempt: handshake, verify, wait for ready."""
+        # A controller that accepts the socket but never answers would
+        # otherwise stall this attempt - and with it the reconnect loop -
+        # forever, so both steps are bounded.
         try:
-            ws = await self._session.ws_connect(f"ws://{self._host}:{self._port}/")
+            async with asyncio.timeout(PING_TIMEOUT):
+                ws = await self._session.ws_connect(f"ws://{self._host}:{self._port}/")
+        except TimeoutError as err:
+            raise PoolsideConnectionError("Timed out connecting") from err
         except aiohttp.ClientError as err:
             raise PoolsideConnectionError(str(err)) from err
 
         noise = NoiseSession(self._client_private_key)
         try:
-            remote_static = await noise.handshake(ws)
+            async with asyncio.timeout(PING_TIMEOUT):
+                remote_static = await noise.handshake(ws)
+        except TimeoutError as err:
+            await ws.close()
+            raise PoolsideConnectionError("Timed out during handshake") from err
         except (NoiseTransportError, aiohttp.ClientError) as err:
             await ws.close()
             raise PoolsideConnectionError(str(err)) from err
@@ -175,10 +185,17 @@ async def _async_reconnect_loop(self) -> None:
         delay = RECONNECT_INITIAL_DELAY
         while not self._closing:
             if self._receive_task is not None:
-                await self._receive_task
+                # An unexpected error here must not end the loop, or the
+                # client would stay disconnected for good without a trace.
+                try:
+                    await self._receive_task
+                except Exception:
+                    LOGGER.exception("Receive loop failed")
             if self._closing:
                 return
             self._set_connected(False)
+            if self._ws is not None and not self._ws.closed:
+                await self._ws.close()
             await asyncio.sleep(delay + random.uniform(0, delay))
             delay = min(delay * 2, RECONNECT_MAX_DELAY)
             try:
@@ -190,6 +207,9 @@ async def _async_reconnect_loop(self) -> None:
             except PoolsideConnectionError as err:
                 LOGGER.debug("Reconnect attempt failed: %s", err)
                 continue
+            except Exception:
+                LOGGER.exception("Unexpected error while reconnecting")
+                continue
             delay = RECONNECT_INITIAL_DELAY
 
     async def _async_ping_loop(self) -> None:
@@ -371,8 +391,11 @@ async def async_refresh_status(self) -> None:
         until someone touches them from the HA side.
         """
         try:
-            items = await self.async_send_request("Device.getStatus", {})
-        except PoolsideConnectionError, PoolsideCommandError:
+            # Bounded so an unanswered request can't stall a (re)connect; a
+            # dead connection is then caught by the ping loop.
+            async with asyncio.timeout(PING_TIMEOUT):
+                items = await self.async_send_request("Device.getStatus", {})
+        except TimeoutError, PoolsideConnectionError, PoolsideCommandError:
             LOGGER.exception("Failed to fetch the initial status snapshot")
             return
         self._handle_status_push(items or [])

tests/test_client.py

@@ -1,5 +1,6 @@
 """Tests for PoolsideClient's optimistic desired-state tracking."""
 
+import asyncio
 from unittest.mock import AsyncMock, MagicMock
 
 import aiohttp
@@ -101,3 +102,77 @@ async def test_refresh_status_swallows_errors(
     send_request.side_effect = error
 
     await client.async_refresh_status()
+
+
+async def test_refresh_status_times_out(
+    client: PoolsideClient, send_request: AsyncMock, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """An unanswered getStatus gives up instead of stalling connect() forever."""
+    monkeypatch.setattr("aiopoolside.client.PING_TIMEOUT", 0.01)
+
+    async def never_answers(*_args: object) -> None:
+        await asyncio.Event().wait()
+
+    send_request.side_effect = never_answers
+
+    await asyncio.wait_for(client.async_refresh_status(), 1)
+
+
+async def test_reconnect_loop_survives_unexpected_error(
+    client: PoolsideClient,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """An unexpected error from a reconnect attempt is logged and retried."""
+    monkeypatch.setattr("aiopoolside.client.RECONNECT_INITIAL_DELAY", 0)
+    calls = 0
+
+    async def connect_once() -> None:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise RuntimeError("boom")
+        client._closing = True
+        client._receive_task = _finished_task()
+
+    client._async_connect_once = connect_once  # type: ignore[method-assign]
+    client._receive_task = _finished_task()
+
+    await asyncio.wait_for(client._async_reconnect_loop(), 1)
+
+    assert calls == 2
+    assert "Unexpected error while reconnecting" in caplog.text
+
+
+async def test_reconnect_loop_survives_failed_receive_loop(
+    client: PoolsideClient,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A receive loop that dies with an error still leads to a reconnect."""
+    monkeypatch.setattr("aiopoolside.client.RECONNECT_INITIAL_DELAY", 0)
+    calls = 0
+
+    async def failing_receive_loop() -> None:
+        raise RuntimeError("bad message")
+
+    async def connect_once() -> None:
+        nonlocal calls
+        calls += 1
+        client._closing = True
+        client._receive_task = _finished_task()
+
+    client._async_connect_once = connect_once  # type: ignore[method-assign]
+    client._receive_task = asyncio.ensure_future(failing_receive_loop())
+
+    await asyncio.wait_for(client._async_reconnect_loop(), 1)
+
+    assert calls == 1
+    assert "Receive loop failed" in caplog.text
+
+
+def _finished_task() -> asyncio.Future[None]:
+    """Return an already-completed future standing in for a receive task."""
+    future: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+    future.set_result(None)
+    return future

tests/test_client_connect.py

@@ -6,14 +6,16 @@
 exercised together rather than mocked piece by piece.
 """
 
+import asyncio
 import json
 from typing import Any
 
 from aiohttp import web
 from aiohttp.pytest_plugin import AiohttpClient
 from noise.connection import Keypair, NoiseConnection
+import pytest
 
-from aiopoolside.client import PoolsideClient
+from aiopoolside.client import PoolsideClient, PoolsideConnectionError
 from aiopoolside.const import NOISE_PROLOGUE, NOISE_PROTOCOL_NAME
 from aiopoolside.noise_transport import generate_keypair
 
@@ -125,3 +127,39 @@ async def test_async_connect_fetches_full_status_snapshot(
         assert client.get_status("control-1", "PowerState") == "ON"
     finally:
         await client.async_disconnect()
+
+
+async def test_async_connect_times_out_on_silent_controller(
+    aiohttp_client: AiohttpClient, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A controller that accepts the socket but never answers fails the attempt.
+
+    Without a bound, the handshake would wait forever - and with it the
+    reconnect loop, leaving the client disconnected for good.
+    """
+    monkeypatch.setattr("aiopoolside.client.PING_TIMEOUT", 0.1)
+
+    async def silent_handler(request: web.Request) -> web.WebSocketResponse:
+        ws = web.WebSocketResponse()
+        await ws.prepare(request)
+        await ws.receive()  # read the first handshake message, never reply
+        await ws.receive()  # stay open until the client gives up
+        return ws
+
+    app = web.Application()
+    app.router.add_get("/", silent_handler)
+    test_client = await aiohttp_client(app)
+    assert test_client.port is not None
+    client_private, _client_public = generate_keypair()
+    _server_private, server_public = generate_keypair()
+
+    client = PoolsideClient(
+        session=test_client.session,
+        host=test_client.host,
+        port=test_client.port,
+        client_private_key=client_private,
+        controller_public_key=server_public,
+        controller_uuid=CONTROLLER_UUID,
+    )
+    with pytest.raises(PoolsideConnectionError, match="handshake"):
+        await asyncio.wait_for(client.async_connect(), 5)