diff --git a/OWNd/connection.py b/OWNd/connection.py index 2304f6c..54813ad 100755 --- a/OWNd/connection.py +++ b/OWNd/connection.py @@ -10,6 +10,7 @@ import secrets import socket import string +import time from collections.abc import Callable, Mapping from typing import Any from urllib.parse import urlparse @@ -72,6 +73,14 @@ # Longer pause when the failure was fatal (e.g. a genuinely wrong password): # retrying fast cannot help, and every attempt costs the gateway a session. RECONNECT_PAUSE_FATAL = 60 +# Routine event-session drops (e.g. the MH200/MH201 hourly session recycle) +# recover transparently and are logged at DEBUG. Only a *burst* of drops is +# worth a WARNING: DROP_BURST_COUNT drops within DROP_BURST_WINDOW seconds, +# repeated at most once every DROP_WARNING_INTERVAL seconds so a flapping +# link does not turn into a log flood. +DROP_BURST_COUNT = 3 +DROP_BURST_WINDOW = 600 # 10 minutes +DROP_WARNING_INTERVAL = 3600 # 1 hour def _first_scalar(value: Any, default: Any = None) -> Any: @@ -1014,6 +1023,8 @@ def __init__( gateway.profile.event_keepalive_interval if gateway is not None else None ) self._keepalive_task: asyncio.Task[None] | None = None + self._recent_drops: list[float] = [] + self._last_drop_warning: float | None = None async def connect(self) -> dict[str, Any] | None: await self._stop_keepalive() @@ -1124,17 +1135,47 @@ async def get_next(self) -> OWNMessage | str | None: ) await self._reconnect() return None + except asyncio.LimitOverrunError: + self._logger.warning( + "%s Received oversized or garbage frame, dropping connection and reconnecting...", + self._log_id, + ) + await self._reconnect() + return None except ( asyncio.IncompleteReadError, - asyncio.LimitOverrunError, ConnectionError, OSError, ): - # Covers EOF, RST (ConnectionResetError), aborted connections, - # over-long frames and other socket errors: reconnect in all cases. - self._logger.warning( - "%s Event connection lost, reconnecting...", self._log_id - ) + # Covers EOF, RST (ConnectionResetError), aborted connections and + # other socket errors: reconnect in all cases. A single drop is + # routine (MH200/MH201 recycle the session every hour) and is + # logged at DEBUG so healthy reconnects do not alarm downstream + # consumers; only a burst of drops is escalated to WARNING. + now = time.monotonic() + self._recent_drops = [ + t for t in self._recent_drops if now - t < DROP_BURST_WINDOW + ] + self._recent_drops.append(now) + if len(self._recent_drops) >= DROP_BURST_COUNT and ( + self._last_drop_warning is None + or now - self._last_drop_warning >= DROP_WARNING_INTERVAL + ): + # Report the span the drops actually covered, not the window + # they were measured in: three drops five seconds apart is a + # very different symptom from three spread over ten minutes. + self._logger.warning( + "%s Event connection dropped %d times in %.0fs; " + "network may be unstable. Reconnecting...", + self._log_id, + len(self._recent_drops), + now - self._recent_drops[0], + ) + self._last_drop_warning = now + else: + self._logger.debug( + "%s Event connection lost, reconnecting...", self._log_id + ) await self._reconnect() return None except Exception: # pylint: disable=broad-except diff --git a/tests/test_connection.py b/tests/test_connection.py index 8f05c3e..17535fe 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -1,11 +1,20 @@ """Regression tests for session negotiation and command responses.""" import asyncio +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest -from OWNd.connection import OWNCommandSession, OWNEventSession, OWNGateway, OWNSession +from OWNd.connection import ( + DROP_BURST_COUNT, + DROP_BURST_WINDOW, + DROP_WARNING_INTERVAL, + OWNCommandSession, + OWNEventSession, + OWNGateway, + OWNSession, +) from OWNd.message import OWNLightingEvent @@ -386,6 +395,174 @@ async def test_rejected_status_request_is_logged_at_debug() -> None: ) +@pytest.mark.asyncio +async def test_event_connection_loss_is_logged_at_debug() -> None: + """Routine event session drops are logged at DEBUG to avoid HA log spam.""" + session, _ = make_session(OWNEventSession) + assert isinstance(session, OWNEventSession) + logger = MagicMock() + session._logger = logger + session._stream_reader = AsyncMock() + session._stream_reader.readuntil = AsyncMock( + side_effect=asyncio.IncompleteReadError(b"", 0) + ) + session._reconnect = AsyncMock(return_value={"Success": True}) + + result = await session.get_next() + + assert result is None + session._reconnect.assert_awaited_once() + logger.warning.assert_not_called() + logger.debug.assert_any_call( + "%s Event connection lost, reconnecting...", session._log_id + ) + + +@pytest.mark.asyncio +async def test_event_connection_loss_oversized_frame_is_warning() -> None: + """An over-long frame is a protocol signal, not a session recycle.""" + session, _ = make_session(OWNEventSession) + logger = MagicMock() + session._logger = logger + session._stream_reader = AsyncMock() + session._stream_reader.readuntil = AsyncMock( + side_effect=asyncio.LimitOverrunError("overrun", 0) + ) + session._reconnect = AsyncMock(return_value={"Success": True}) + + result = await session.get_next() + + assert result is None + session._reconnect.assert_awaited_once() + logger.warning.assert_any_call( + "%s Received oversized or garbage frame, dropping connection and reconnecting...", + session._log_id, + ) + + +def _fake_clock(*times: float): + """Patch only the module's ``time`` so asyncio's own loop clock is untouched.""" + it = iter(times) + return patch("OWNd.connection.time", SimpleNamespace(monotonic=lambda: next(it))) + + +def _dropping_event_session() -> tuple[OWNEventSession, MagicMock]: + session, _ = make_session(OWNEventSession) + assert isinstance(session, OWNEventSession) + logger = MagicMock() + session._logger = logger + session._stream_reader = AsyncMock() + session._stream_reader.readuntil = AsyncMock( + side_effect=asyncio.IncompleteReadError(b"", 0) + ) + session._reconnect = AsyncMock(return_value={"Success": True}) + return session, logger + + +BURST_WARNING = "%s Event connection dropped %d times in %.0fs; network may be unstable. Reconnecting..." + + +@pytest.mark.asyncio +async def test_event_connection_drop_burst_is_warning_on_fresh_host() -> None: + """A burst warns even when monotonic() is small (host up < 1 h).""" + session, logger = _dropping_event_session() + # Fresh boot: monotonic() well below DROP_WARNING_INTERVAL. A 0.0 sentinel + # for "never warned" would silently suppress the warning here. + with _fake_clock(100.0, 100.5, 101.0, 101.5): + for _ in range(DROP_BURST_COUNT - 1): + await session.get_next() + logger.warning.assert_not_called() + logger.debug.assert_any_call( + "%s Event connection lost, reconnecting...", session._log_id + ) + logger.debug.reset_mock() + + await session.get_next() + # Drops at 100.0 / 100.5 / 101.0: the reported span is the real 1.0s, + # not the 600s window they were measured in. + logger.warning.assert_called_once_with( + BURST_WARNING, session._log_id, DROP_BURST_COUNT, 1.0 + ) + logger.warning.reset_mock() + + # Another drop right after the warning stays quiet (hourly rate limit). + await session.get_next() + logger.warning.assert_not_called() + logger.debug.assert_any_call( + "%s Event connection lost, reconnecting...", session._log_id + ) + + assert session._reconnect.await_count == DROP_BURST_COUNT + 1 + + +@pytest.mark.asyncio +async def test_event_connection_drop_burst_reports_real_span() -> None: + """A slow burst reports its own span, distinguishing it from a fast one.""" + session, logger = _dropping_event_session() + t0 = 5000.0 + spread = 200.0 + # Spread over most of DROP_BURST_WINDOW, but still inside it. + assert spread * (DROP_BURST_COUNT - 1) < DROP_BURST_WINDOW + with _fake_clock(*(t0 + i * spread for i in range(DROP_BURST_COUNT))): + for _ in range(DROP_BURST_COUNT): + await session.get_next() + + logger.warning.assert_called_once_with( + BURST_WARNING, + session._log_id, + DROP_BURST_COUNT, + spread * (DROP_BURST_COUNT - 1), + ) + + +@pytest.mark.asyncio +async def test_event_connection_drops_outside_window_stay_debug() -> None: + """Hourly session recycling never accumulates into a burst.""" + session, logger = _dropping_event_session() + with _fake_clock(*(float(i * 3460) for i in range(10))): + for _ in range(10): + await session.get_next() + + logger.warning.assert_not_called() + assert logger.debug.call_count == 10 + assert len(session._recent_drops) == 1 + + +@pytest.mark.asyncio +async def test_event_connection_drop_burst_warns_again_after_interval() -> None: + """A persistent flap is re-reported once per DROP_WARNING_INTERVAL.""" + session, logger = _dropping_event_session() + t0 = 5000.0 + # Burst one at t0, burst two well after the warning interval has elapsed. + times = [t0, t0 + 1, t0 + 2] + t1 = t0 + DROP_WARNING_INTERVAL + 5 + times += [t1, t1 + 1, t1 + 2] + with _fake_clock(*times): + for _ in times: + await session.get_next() + + assert logger.warning.call_count == 2 + assert session._last_drop_warning == t1 + 2 + + +@pytest.mark.asyncio +async def test_event_connection_drop_tracker_survives_reconnect() -> None: + """_reconnect() goes through connect(); that must not wipe the burst tracker.""" + session, logger = _dropping_event_session() + session._reconnect = AsyncMock(side_effect=session.connect) + + with ( + patch.object(OWNSession, "connect", AsyncMock(return_value=None)), + _fake_clock(100.0, 100.5, 101.0), + ): + for _ in range(DROP_BURST_COUNT): + await session.get_next() + + logger.warning.assert_called_once_with( + BURST_WARNING, session._log_id, DROP_BURST_COUNT, 1.0 + ) + + @pytest.mark.asyncio async def test_probe_gateway_uses_read_only_model_request() -> None: gateway = OWNGateway({"address": "192.0.2.1", "port": 20000})