Detect dead connections via a read timeout and reconnect

.github/workflows/lint.yml

@@ -15,6 +15,7 @@ jobs:
           - "3.11"
           - "3.12"
           - "3.13"
+          - "3.14"
 
     steps:
       - uses: actions/checkout@v6

examples/loads/control_load.py

@@ -6,7 +6,7 @@
 import sys
 import termios
 import tty
-from collections.abc import Iterator
+from collections.abc import Generator
 from contextlib import contextmanager, suppress
 
 from aiovantage import Vantage
@@ -36,7 +36,7 @@ async def parse_keypress() -> str | None:
 
 
 @contextmanager
-def cbreak_mode(descriptor: int) -> Iterator[None]:
+def cbreak_mode(descriptor: int) -> Generator[None, None, None]:
     """Context manager to read terminal input character by character."""
     old_attrs = termios.tcgetattr(descriptor)
     try:

src/aiovantage/_command_client/client.py

@@ -9,7 +9,11 @@
 from typing_extensions import Self
 
 from aiovantage._logger import logger
-from aiovantage.errors import CommandError, raise_command_error
+from aiovantage.errors import (
+    ClientConnectionError,
+    CommandError,
+    raise_command_error,
+)
 
 from .connection import CommandConnection
 from .converter import Converter
@@ -139,34 +143,42 @@ async def raw_request(self, request: str) -> list[str]:
 
         # Send the command
         async with self._command_lock:
-            logger.debug("Sending command: %s", request)
-            await conn.write(f"{request}\n")
-
-            # Read all lines of the response
-            response_lines: list[str] = []
-            while True:
-                response_line = await conn.readuntil(b"\r\n", self._read_timeout)
-                response_line = response_line.rstrip()
-
-                # Handle command errors
-                if response_line.startswith("R:ERROR"):
-                    # Parse a command error from a message.
-                    match = re.match(r"R:ERROR:(\d+) (.+)", response_line)
-                    if not match:
-                        raise CommandError(response_line)
-
-                    # Convert the error code to a specific exception, if possible
-                    raise_command_error(int(match.group(1)), match.group(2))
-
-                # Ignore potentially interleaved "event" messages
-                if response_line.startswith(("S:", "L:", "EL:")):
-                    logger.debug("Ignoring event message: %s", response_line)
-                    continue
-
-                # Return the response once we see the response line
-                response_lines.append(response_line)
-                if response_line.startswith("R:"):
-                    break
+            try:
+                logger.debug("Sending command: %s", request)
+                await conn.write(f"{request}\n")
+
+                # Read all lines of the response
+                response_lines: list[str] = []
+                while True:
+                    response_line = await conn.readuntil(b"\r\n", self._read_timeout)
+                    response_line = response_line.rstrip()
+
+                    # Handle command errors
+                    if response_line.startswith("R:ERROR"):
+                        # Parse a command error from a message.
+                        match = re.match(r"R:ERROR:(\d+) (.+)", response_line)
+                        if not match:
+                            raise CommandError(response_line)
+
+                        # Convert the error code to a specific exception, if possible
+                        raise_command_error(int(match.group(1)), match.group(2))
+
+                    # Ignore potentially interleaved "event" messages
+                    if response_line.startswith(("S:", "L:", "EL:")):
+                        logger.debug("Ignoring event message: %s", response_line)
+                        continue
+
+                    # Return the response once we see the response line
+                    response_lines.append(response_line)
+                    if response_line.startswith("R:"):
+                        break
+            except ClientConnectionError:
+                # A read timeout or connection error leaves the socket in an unknown
+                # state; close it so the next request reconnects with a fresh socket
+                # instead of reusing a dead one (which would keep timing out until the
+                # OS finally gives up on the TCP connection, ~15 min later).
+                self._connection.close()
+                raise
 
         logger.debug("Received response: %s", "\n".join(response_lines))
 

src/aiovantage/_command_client/events.py

@@ -27,6 +27,15 @@
 # The interval between keepalive messages, in seconds.
 KEEPALIVE_INTERVAL = 60
 
+# The maximum time to wait for a message before assuming the connection is dead.
+# Must be greater than KEEPALIVE_INTERVAL: on a healthy connection the periodic ECHO
+# keepalive produces traffic that resets this timeout, while on a dead connection no
+# data arrives and the read times out, surfacing the disconnect promptly. Without a
+# timeout, a hard power-off (no TCP RST) leaves the read blocked until the OS gives up
+# on the keepalive writes (~15 min), so the Disconnected event never fires and the
+# stream never reconnects in time.
+READ_TIMEOUT = 2 * KEEPALIVE_INTERVAL
+
 
 class EventStream(EventDispatcher):
     """Client to subscribe to events from the Vantage Host Command (HC) service.
@@ -216,15 +225,24 @@ async def _message_handler(self) -> None:
                     self._resubscribe()
                 connect_attempts = 1
 
-                # Wait for new messages
+                # Wait for new messages. A read timeout (longer than the keepalive
+                # interval) ensures a dead connection is detected promptly: the ECHO
+                # keepalive keeps a healthy link's read fed, so a timeout means the
+                # connection is gone. ClientTimeoutError is a ClientConnectionError,
+                # so it falls through to the reconnect logic below.
                 while True:
-                    message = await conn.readuntil(b"\r\n")
+                    message = await conn.readuntil(b"\r\n", READ_TIMEOUT)
                     message = message.rstrip()
                     logger.debug("Received message: %s", message)
                     self._parse_message(message)
 
             except ClientConnectionError:
-                pass  # Pass through to retry logic below
+                # The connection was lost, or a read timed out (ClientTimeoutError is a
+                # ClientConnectionError) indicating a dead link. A timeout does not close
+                # the socket, so close it explicitly here - otherwise _get_connection would
+                # reuse the stale connection on the next iteration, re-emitting Reconnected
+                # on a dead socket and flapping every READ_TIMEOUT instead of recovering.
+                self._connection.close()
 
             # If we get here, the connection was lost
             if connect_attempts == 1: