Reconnect the live update stream after a transient failure

custom_components/zaptec/const.py

@@ -36,6 +36,21 @@
 ZAPTEC_POLL_INSTALLATION_TRIGGER_DELAYS = [2, 7]
 """Delays in seconds for installation state updates after a change."""
 
+STREAM_RECONNECT_INIT_DELAY = 1.0
+"""Initial delay in seconds before the first stream reconnect attempt."""
+
+STREAM_RECONNECT_FACTOR = 2.0
+"""Exponential backoff multiplier applied between stream reconnect attempts."""
+
+STREAM_RECONNECT_JITTER = 0.1
+"""Relative jitter applied to the stream reconnect backoff delay."""
+
+STREAM_RECONNECT_MAX_DELAY = 300.0
+"""Maximum delay in seconds between stream reconnect attempts (5 minutes)."""
+
+STREAM_RECONNECT_STABLE_TIME = 300.0
+"""Uptime in seconds after which a stream connection counts as healthy again."""
+
 # This sets the delay after doing actions and the poll of updated values.
 # It was 0.3 and evidently that is a bit too fast for Zaptec cloud to handle.
 REQUEST_REFRESH_DELAY = 1

custom_components/zaptec/manager.py

@@ -3,22 +3,34 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Iterable
+from collections.abc import Awaitable, Callable, Iterable
 import contextlib
 from copy import copy
 from dataclasses import dataclass
 import logging
+import random
+import ssl
+import time
 from typing import Any
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity import DeviceInfo, EntityDescription
 from homeassistant.util.ssl import get_default_context
 
-from .const import DOMAIN, KEYS_TO_SKIP_ENTITY_AVAILABILITY_CHECK, MANUFACTURER
+from .const import (
+    DOMAIN,
+    KEYS_TO_SKIP_ENTITY_AVAILABILITY_CHECK,
+    MANUFACTURER,
+    STREAM_RECONNECT_FACTOR,
+    STREAM_RECONNECT_INIT_DELAY,
+    STREAM_RECONNECT_JITTER,
+    STREAM_RECONNECT_MAX_DELAY,
+    STREAM_RECONNECT_STABLE_TIME,
+)
 from .coordinator import ZaptecUpdateCoordinator
 from .entity import KeyUnavailableError, ZaptecBaseEntity
-from .zaptec import Charger, Installation, Zaptec, ZaptecBase
+from .zaptec import STREAM_TRANSIENT_ERRORS, Charger, Installation, Zaptec, ZaptecBase
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -32,6 +44,65 @@ class ZaptecEntityDescription(EntityDescription):
     cls: type[ZaptecBaseEntity[Any]]
 
 
+async def _stream_supervisor(
+    install: Installation,
+    cb: Callable[[dict], Awaitable[None]],
+    ssl_context: ssl.SSLContext | None,
+) -> None:
+    """Run install.stream_main(), reconnecting after a transient failure.
+
+    stream_main() returning normally means a permanent stop; raising means
+    a transient failure to retry with backoff. asyncio.CancelledError is a
+    BaseException, not an Exception, so it's never caught here -- task
+    cancellation still stops this immediately.
+    """
+    delay = STREAM_RECONNECT_INIT_DELAY
+    reconnects = 0
+    connected_at: float | None = None
+
+    def on_connect() -> None:
+        nonlocal connected_at
+        connected_at = time.monotonic()
+        if reconnects:
+            _LOGGER.info(
+                "Stream for %s connected after %s reconnect attempt(s)",
+                install.qual_id,
+                reconnects,
+            )
+
+    while True:
+        connected_at = None
+        try:
+            await install.stream_main(cb=cb, ssl_context=ssl_context, on_connect=on_connect)
+        except Exception as err:
+            # A connection that stayed up counts as recovered, so the next
+            # failure starts a fresh outage instead of continuing the last.
+            if connected_at is not None and (
+                time.monotonic() - connected_at >= STREAM_RECONNECT_STABLE_TIME
+            ):
+                delay = STREAM_RECONNECT_INIT_DELAY
+                reconnects = 0
+            unexpected = not isinstance(err, STREAM_TRANSIENT_ERRORS)
+            if reconnects:
+                _LOGGER.debug(
+                    "Stream for %s still reconnecting (%r)", install.qual_id, err, exc_info=True
+                )
+            else:
+                _LOGGER.warning(
+                    "Stream for %s disconnected (%r), reconnecting",
+                    install.qual_id,
+                    err,
+                    exc_info=unexpected,
+                )
+            reconnects += 1
+            await asyncio.sleep(delay)
+            delay = min(delay * STREAM_RECONNECT_FACTOR, STREAM_RECONNECT_MAX_DELAY)
+            delay = random.normalvariate(delay, delay * STREAM_RECONNECT_JITTER)
+            delay = min(delay, STREAM_RECONNECT_MAX_DELAY)
+        else:
+            return
+
+
 class ZaptecManager:
     """Manager for Zaptec data."""
 
@@ -196,7 +267,8 @@ def create_streams(self) -> None:
             if install.id in self.zaptec:
                 task = self.config_entry.async_create_background_task(
                     self.hass,
-                    install.stream_main(
+                    _stream_supervisor(
+                        install,
                         cb=self.stream_callback,
                         ssl_context=get_default_context(),
                     ),

custom_components/zaptec/zaptec/__init__.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from .api import Charger, Installation, Zaptec, ZaptecBase
+from .api import STREAM_TRANSIENT_ERRORS, Charger, Installation, Zaptec, ZaptecBase
 from .const import MISSING, RETRYABLE_HTTP_STATUSES, Missing
 from .exceptions import (
     AuthenticationError,
@@ -20,6 +20,7 @@
 __all__ = [
     "MISSING",
     "RETRYABLE_HTTP_STATUSES",
+    "STREAM_TRANSIENT_ERRORS",
     "ZCONST",
     "AuthenticationError",
     "Charger",

custom_components/zaptec/zaptec/api.py

@@ -18,7 +18,7 @@
 import aiohttp
 from aiolimiter import AsyncLimiter
 from azure.servicebus.aio import ServiceBusClient
-from azure.servicebus.exceptions import ServiceBusError
+from azure.servicebus.exceptions import MessageAlreadySettled, ServiceBusError
 import pydantic
 
 from .const import (
@@ -46,6 +46,7 @@
     RequestError,
     RequestRetryError,
     RequestTimeoutError,
+    ZaptecApiError,
 )
 from .redact import Redactor
 from .utils import mc_nbfx_decoder, to_under
@@ -64,6 +65,16 @@
 TDict = dict[str, TValue]
 StreamCallback = Callable[[dict], Awaitable[None]]
 
+STREAM_TRANSIENT_ERRORS: tuple[type[Exception], ...] = (
+    ServiceBusError,
+    MessageAlreadySettled,  # a ValueError, not a ServiceBusError
+    OSError,  # includes ConnectionError
+    TimeoutError,
+    aiohttp.ClientError,
+    ZaptecApiError,
+)
+"""Errors a dropped stream is expected to fail with, e.g. a network outage."""
+
 
 class TLogExc(Protocol):
     """Protocol for logging exceptions."""
@@ -386,7 +397,10 @@ def _stream_log(self, data: dict[str, Any]) -> None:
         _LOGGER.debug("@@@  EVENT %s", self.zaptec.redact(data))
 
     async def stream_main(
-        self, cb: StreamCallback | None = None, ssl_context: ssl.SSLContext | None = None
+        self,
+        cb: StreamCallback | None = None,
+        ssl_context: ssl.SSLContext | None = None,
+        on_connect: Callable[[], None] | None = None,
     ) -> None:
         """Main stream handler."""
         # Already running?
@@ -435,6 +449,8 @@ async def stream_main(
                 # Store the receiver in order to close it and cancel this stream
                 self._stream_receiver = receiver
                 async with receiver:
+                    if on_connect:
+                        on_connect()
                     async for msg in receiver:
                         # For the exception in case it fails before setting the value
                         binmsg = "<unknown>"
@@ -471,10 +487,6 @@ async def stream_main(
                         # remove the msg from the "queue"
                         await receiver.complete_message(msg)
 
-        except Exception:
-            # Do this in order to show the error in the log.
-            _LOGGER.exception("Stream failed")
-
         finally:
             # Cleanup
             self._stream_receiver = None