diff --git a/custom_components/zaptec/const.py b/custom_components/zaptec/const.py index 739326f3..043f41e2 100644 --- a/custom_components/zaptec/const.py +++ b/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 diff --git a/custom_components/zaptec/manager.py b/custom_components/zaptec/manager.py index 79b9fc02..7e02ec9a 100644 --- a/custom_components/zaptec/manager.py +++ b/custom_components/zaptec/manager.py @@ -3,11 +3,14 @@ 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 @@ -15,10 +18,19 @@ 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(), ), diff --git a/custom_components/zaptec/zaptec/__init__.py b/custom_components/zaptec/zaptec/__init__.py index a71ebc9a..955b653f 100644 --- a/custom_components/zaptec/zaptec/__init__.py +++ b/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", diff --git a/custom_components/zaptec/zaptec/api.py b/custom_components/zaptec/zaptec/api.py index cd818dd4..f0563a14 100644 --- a/custom_components/zaptec/zaptec/api.py +++ b/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 = "" @@ -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 diff --git a/tests/test_manager.py b/tests/test_manager.py new file mode 100644 index 00000000..d66ead3a --- /dev/null +++ b/tests/test_manager.py @@ -0,0 +1,236 @@ +"""Tests for custom_components.zaptec.manager.""" + +import asyncio +from collections.abc import Callable +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_RECONNECT_STABLE_TIME, + _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_stable_connection( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """A connection that stayed up past the stable time 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() + now = 0.0 + # Fail, then hold a connection past the stable time before failing again. + attempts = iter([False, True]) + + async def stream_main(*, on_connect: Callable[[], None], **kwargs: object) -> None: + nonlocal now + connects = next(attempts, None) + if connects is None: + return # permanent stop, ends the supervisor loop + if connects: + on_connect() + now += STREAM_RECONNECT_STABLE_TIME + 1 + raise ConnectionError("boom") + + install.stream_main.side_effect = stream_main + sleep_mock = AsyncMock() + monkeypatch.setattr(asyncio, "sleep", sleep_mock) + monkeypatch.setattr(manager_module.time, "monotonic", lambda: now) + monkeypatch.setattr(manager_module.random, "normalvariate", lambda mu, sigma: mu) # noqa: ARG005 + + 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 + (last_delay,), _ = sleep_mock.await_args_list[-1] + assert last_delay == pytest.approx(STREAM_RECONNECT_INIT_DELAY) # backoff restarted too + + +@pytest.mark.asyncio +async def test_stream_supervisor_logs_reconnect_count_on_connect( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Reconnecting logs how many attempts it took; the first connect stays quiet.""" + install = _fake_install() + outcomes = iter([ConnectionError("1"), ConnectionError("2")]) + + async def stream_main(*, on_connect: Callable[[], None], **kwargs: object) -> None: + outcome = next(outcomes, None) + if isinstance(outcome, Exception): + raise outcome + on_connect() + + install.stream_main.side_effect = stream_main + 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) + + infos = [r for r in caplog.records if r.levelno == logging.INFO] + assert len(infos) == 1 + assert "after 2 reconnect attempt(s)" in infos[0].getMessage() + + +@pytest.mark.asyncio +async def test_stream_supervisor_logs_traceback_only_for_unexpected_errors( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """An expected disconnect warns without a traceback; anything else keeps one.""" + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + for error, has_traceback in ((ConnectionError("boom"), False), (ValueError("boom"), True)): + install = _fake_install() + install.stream_main.side_effect = [error, None] + caplog.clear() + + with caplog.at_level(logging.DEBUG, logger="custom_components.zaptec.manager"): + await _stream_supervisor(install, cb=AsyncMock(), ssl_context=None) + + (warning,) = [r for r in caplog.records if r.levelno == logging.WARNING] + assert bool(warning.exc_info) is has_traceback + + +@pytest.mark.asyncio +async def test_stream_supervisor_stays_quiet_when_the_first_connect_succeeds( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Connecting without ever having failed reports no reconnect count.""" + install = _fake_install() + + async def stream_main(*, on_connect: Callable[[], None], **kwargs: object) -> None: + on_connect() + + install.stream_main.side_effect = stream_main + 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) + + assert not [r for r in caplog.records if r.levelno == logging.INFO] + + +@pytest.mark.asyncio +async def test_stream_supervisor_does_not_reset_when_the_retry_never_connects( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """The uptime of an earlier connection must not carry over to a later attempt.""" + install = _fake_install() + now = 0.0 + # Connect and stay up past the stable time, then fail without connecting at all. + attempts = iter([True, False]) + + async def stream_main(*, on_connect: Callable[[], None], **kwargs: object) -> None: + nonlocal now + connects = next(attempts, None) + if connects is None: + return # permanent stop, ends the supervisor loop + if connects: + on_connect() + now += STREAM_RECONNECT_STABLE_TIME + 1 + raise ConnectionError("boom") + + install.stream_main.side_effect = stream_main + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + monkeypatch.setattr(manager_module.time, "monotonic", lambda: now) + + 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) == 1 # the second failure never connected, so it is the same outage diff --git a/tests/zaptec/test_api.py b/tests/zaptec/test_api.py index 825b05d9..1a43b779 100644 --- a/tests/zaptec/test_api.py +++ b/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 # ---------------------------------------------------------------------------