Reconnect the live update stream after a transient failure

.gitignore

@@ -11,6 +11,7 @@ __pycache__
 coverage.xml
 .ruff_cache
 htmlcov/
+.superpowers/
 
 # Home Assistant configuration
 config/*

custom_components/zaptec/const.py

@@ -36,6 +36,18 @@
 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)."""
+
 # 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,19 +3,30 @@
 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,
+)
 from .coordinator import ZaptecUpdateCoordinator
 from .entity import KeyUnavailableError, ZaptecBaseEntity
 from .zaptec import Charger, Installation, Zaptec, ZaptecBase
@@ -32,6 +43,45 @@ 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
+    warned = False
+    while True:
+        connected_at = time.monotonic()
+        try:
+            await install.stream_main(cb=cb, ssl_context=ssl_context)
+        except Exception:
+            if time.monotonic() - connected_at >= STREAM_RECONNECT_MAX_DELAY:
+                # The previous connection lived long enough to count this
+                # as a fresh outage rather than a continuation of the last.
+                delay = STREAM_RECONNECT_INIT_DELAY
+                warned = False
+            if not warned:
+                _LOGGER.warning(
+                    "Stream for %s disconnected, reconnecting", install.qual_id, exc_info=True
+                )
+                warned = True
+            else:
+                _LOGGER.debug("Stream for %s still reconnecting", install.qual_id, exc_info=True)
+            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 +246,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/api.py

@@ -471,10 +471,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

tests/test_manager.py

@@ -0,0 +1,129 @@
+"""Tests for custom_components.zaptec.manager."""
+
+import asyncio
+import logging
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from custom_components.zaptec import manager as manager_module
+from custom_components.zaptec.manager import (
+    STREAM_RECONNECT_INIT_DELAY,
+    STREAM_RECONNECT_MAX_DELAY,
+    _stream_supervisor,
+)
+
+
+def _fake_install() -> SimpleNamespace:
+    """Return a stand-in for Installation carrying only what _stream_supervisor uses."""
+    return SimpleNamespace(qual_id="Installation[nst-1]", stream_main=AsyncMock())
+
+
+@pytest.mark.asyncio
+async def test_stream_supervisor_stops_when_stream_main_returns_normally() -> None:
+    """stream_main() returning None (e.g. 403/Forbidden) is a permanent stop."""
+    install = _fake_install()
+    install.stream_main.return_value = None
+
+    await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None)
+
+    install.stream_main.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_stream_supervisor_retries_on_exception(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A raised exception is retried, not left dead, using the initial backoff delay."""
+    install = _fake_install()
+    install.stream_main.side_effect = [ConnectionError("boom"), None]
+    sleep_mock = AsyncMock()
+    monkeypatch.setattr(asyncio, "sleep", sleep_mock)
+    # Neutralize jitter so the first delay is deterministically the init delay.
+    monkeypatch.setattr(manager_module.random, "normalvariate", lambda mu, sigma: mu)  # noqa: ARG005
+
+    await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None)
+
+    assert install.stream_main.await_count == 2  # noqa: PLR2004
+    sleep_mock.assert_awaited_once()
+    (delay,), _ = sleep_mock.await_args
+    assert delay == pytest.approx(STREAM_RECONNECT_INIT_DELAY)
+
+
+@pytest.mark.asyncio
+async def test_stream_supervisor_backoff_never_exceeds_max_delay(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Across many consecutive failures, every sleep delay stays within the cap."""
+    install = _fake_install()
+    install.stream_main.side_effect = [ConnectionError("boom")] * 10 + [None]
+    sleep_mock = AsyncMock()
+    monkeypatch.setattr(asyncio, "sleep", sleep_mock)
+
+    await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None)
+
+    assert install.stream_main.await_count == 11  # noqa: PLR2004
+    assert sleep_mock.await_count == 10  # noqa: PLR2004
+    for (delay,), _ in sleep_mock.await_args_list:
+        assert delay <= STREAM_RECONNECT_MAX_DELAY
+
+
+@pytest.mark.asyncio
+async def test_stream_supervisor_propagates_cancelled_error_without_retrying(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Task cancellation (integration unload/reload) is not treated as a retryable failure."""
+    install = _fake_install()
+    install.stream_main.side_effect = asyncio.CancelledError()
+    monkeypatch.setattr(asyncio, "sleep", AsyncMock())
+
+    with pytest.raises(asyncio.CancelledError):
+        await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None)
+
+    install.stream_main.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_stream_supervisor_logs_warning_once_then_debug(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Only the first failure of an outage logs at warning; the rest log at debug."""
+    install = _fake_install()
+    install.stream_main.side_effect = [ConnectionError("1"), ConnectionError("2"), None]
+    monkeypatch.setattr(asyncio, "sleep", AsyncMock())
+
+    with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"):
+        await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None)
+
+    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
+    debugs = [r for r in caplog.records if r.levelno == logging.DEBUG]
+    assert len(warnings) == 1
+    assert len(debugs) == 1
+
+
+@pytest.mark.asyncio
+async def test_stream_supervisor_resets_backoff_after_long_lived_connection(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A connection that outlived the max backoff delay counts as a fresh outage.
+
+    Verified via logging: a failure treated as a new outage warns again (see
+    test_stream_supervisor_logs_warning_once_then_debug for the same-outage
+    case, which stays at DEBUG).
+    """
+    install = _fake_install()
+    install.stream_main.side_effect = [ConnectionError("1"), ConnectionError("2"), None]
+    monkeypatch.setattr(asyncio, "sleep", AsyncMock())
+    monkeypatch.setattr(manager_module, "STREAM_RECONNECT_MAX_DELAY", 100.0)
+    # 5 monotonic() calls: connected_at + failure-check per failed attempt (x2),
+    # then connected_at for the successful 3rd. Attempt 2's gap (0.0 -> 200.0)
+    # exceeds MAX_DELAY (100.0), so it counts as a new outage.
+    # Fallback (not bare next(clock)): Windows' ProactorEventLoop calls
+    # monotonic() more times during teardown, after the coroutine has returned.
+    clock = iter([0.0, 0.0, 0.0, 200.0, 500.0])
+    monkeypatch.setattr(manager_module.time, "monotonic", lambda: next(clock, 500.0))
+
+    with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"):
+        await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None)
+
+    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
+    assert len(warnings) == 2  # noqa: PLR2004  # both failures counted as separate outages

tests/zaptec/test_api.py

@@ -546,6 +546,33 @@ def test_stream_update_zero_guid_is_ignored() -> None:
     charger.set_attributes.assert_not_called()
 
 
+# ---------------------------------------------------------------------------
+#   Installation.stream_main error propagation
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_stream_main_propagates_non_forbidden_error() -> None:
+    """A non-403 error fetching stream connection details now propagates (issue #417)."""
+    inst = Installation({"Id": "inst-1"}, _fake_owner())
+    inst.live_stream_connection_details = AsyncMock(  # type: ignore[method-assign]
+        side_effect=RequestError("server error", HTTPStatus.BAD_GATEWAY)
+    )
+    with pytest.raises(RequestError):
+        await inst.stream_main()
+
+
+@pytest.mark.asyncio
+async def test_stream_main_forbidden_returns_none() -> None:
+    """A 403 fetching stream connection details still returns cleanly."""
+    inst = Installation({"Id": "inst-1"}, _fake_owner())
+    inst.live_stream_connection_details = AsyncMock(  # type: ignore[method-assign]
+        side_effect=RequestError("no access", HTTPStatus.FORBIDDEN)
+    )
+    result = await inst.stream_main()
+    assert result is None
+
+
 # ---------------------------------------------------------------------------
 #   Zaptec mapping / registry + poll dispatch
 # ---------------------------------------------------------------------------