From 0f8336f6626fa9d6d15c0683abbb82296d127b7f Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Sun, 20 Sep 2026 21:57:13 +0200 Subject: [PATCH 1/4] fix(connection): log routine event connection loss at DEBUG --- OWNd/connection.py | 4 +++- tests/test_connection.py | 23 +++++++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/OWNd/connection.py b/OWNd/connection.py index 2304f6c..8d1891e 100755 --- a/OWNd/connection.py +++ b/OWNd/connection.py @@ -1132,7 +1132,9 @@ async def get_next(self) -> OWNMessage | str | None: ): # Covers EOF, RST (ConnectionResetError), aborted connections, # over-long frames and other socket errors: reconnect in all cases. - self._logger.warning( + # Routine drops (e.g. MH200/MH201 hourly session recycling) are logged + # at DEBUG so healthy reconnects do not alarm downstream consumers. + self._logger.debug( "%s Event connection lost, reconnecting...", self._log_id ) await self._reconnect() diff --git a/tests/test_connection.py b/tests/test_connection.py index 8f05c3e..93f6c67 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -386,6 +386,29 @@ 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_probe_gateway_uses_read_only_model_request() -> None: gateway = OWNGateway({"address": "192.0.2.1", "port": 20000}) From 8a2d5f7b278f5db4cb3fb90fd6af57e626f2bd64 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Mon, 21 Sep 2026 20:32:00 +0200 Subject: [PATCH 2/4] Address review comments on reconnect logging --- OWNd/connection.py | 37 ++++++++++++++++++----- tests/test_connection.py | 65 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 94 insertions(+), 8 deletions(-) diff --git a/OWNd/connection.py b/OWNd/connection.py index 8d1891e..2e4102b 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 @@ -1014,6 +1015,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 = 0.0 async def connect(self) -> dict[str, Any] | None: await self._stop_keepalive() @@ -1124,19 +1127,37 @@ 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. - # Routine drops (e.g. MH200/MH201 hourly session recycling) are logged - # at DEBUG so healthy reconnects do not alarm downstream consumers. - self._logger.debug( - "%s Event connection lost, reconnecting...", self._log_id - ) + now = time.monotonic() + self._recent_drops = [t for t in self._recent_drops if now - t < 600] + self._recent_drops.append(now) + + if len(self._recent_drops) >= 3 and (now - self._last_drop_warning) >= 3600: + self._logger.warning( + "%s Event connection dropped %d times in 10 minutes; network may be unstable. Reconnecting...", + self._log_id, + len(self._recent_drops), + ) + self._last_drop_warning = now + else: + # Covers EOF, RST (ConnectionResetError), aborted connections, + # and other socket errors: reconnect in all cases. + # Routine drops (e.g. MH200/MH201 hourly session recycling) are logged + # at DEBUG so healthy reconnects do not alarm downstream consumers. + 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 93f6c67..4c8cb0e 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -409,6 +409,71 @@ async def test_event_connection_loss_is_logged_at_debug() -> None: ) +@pytest.mark.asyncio +async def test_event_connection_loss_oversized_frame_is_warning() -> None: + 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, + ) + + +@pytest.mark.asyncio +async def test_event_connection_drops_clustering() -> None: + session, _ = make_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}) + + # Drop 1 (DEBUG) + 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() + + # Drop 2 (DEBUG) + 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() + + # Drop 3 within 10 minutes (WARNING) + await session.get_next() + logger.warning.assert_any_call( + "%s Event connection dropped %d times in 10 minutes; network may be unstable. Reconnecting...", + session._log_id, + 3, + ) + logger.warning.reset_mock() + + # Drop 4 right away (DEBUG because we already warned in the last hour) + await session.get_next() + logger.warning.assert_not_called() + logger.debug.assert_any_call( + "%s Event connection lost, reconnecting...", session._log_id + ) + + @pytest.mark.asyncio async def test_probe_gateway_uses_read_only_model_request() -> None: gateway = OWNGateway({"address": "192.0.2.1", "port": 20000}) From 5f245683eeaa837e4d38088374e3275aff26a5ee Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Mon, 21 Sep 2026 21:29:29 +0200 Subject: [PATCH 3/4] fix(connection): burst warning never fired on hosts with < 1 h uptime `_last_drop_warning` used `0.0` as its "never warned" sentinel and was compared against `time.monotonic()`, which counts seconds since boot. So `now - 0.0 >= 3600` really meant "has this host been up for an hour": on a freshly booted machine (every CI runner, and a Home Assistant box right after a reboot) the "dropped N times in 10 minutes" warning was silently suppressed. This is why test_event_connection_drops_clustering passed locally but failed on all four CI matrix entries. - Use `None` as the sentinel. - Lift 3 / 600 / 3600 into DROP_BURST_COUNT, DROP_BURST_WINDOW and DROP_WARNING_INTERVAL next to the other reconnect constants, and format the window into the log message instead of hard-coding it. - Move the "covers EOF, RST, ..." comment back to the except clause it describes. - Replace the wall-clock-dependent burst test with clock-patched tests covering: burst on a fresh host, drops spread over > 10 min staying at DEBUG, a second warning after DROP_WARNING_INTERVAL, and the tracker surviving a real `_reconnect()` (which goes through `connect()`). The clock patch replaces only `OWNd.connection.time`, so asyncio's own loop clock is untouched. Co-Authored-By: Claude Opus 5 --- OWNd/connection.py | 33 ++++++++--- tests/test_connection.py | 125 ++++++++++++++++++++++++++++++--------- 2 files changed, 120 insertions(+), 38 deletions(-) diff --git a/OWNd/connection.py b/OWNd/connection.py index 2e4102b..17f9cb2 100755 --- a/OWNd/connection.py +++ b/OWNd/connection.py @@ -73,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: @@ -1016,7 +1024,7 @@ def __init__( ) self._keepalive_task: asyncio.Task[None] | None = None self._recent_drops: list[float] = [] - self._last_drop_warning: float = 0.0 + self._last_drop_warning: float | None = None async def connect(self) -> dict[str, Any] | None: await self._stop_keepalive() @@ -1139,22 +1147,29 @@ async def get_next(self) -> OWNMessage | str | None: ConnectionError, OSError, ): + # 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 < 600] + 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) >= 3 and (now - self._last_drop_warning) >= 3600: + if len(self._recent_drops) >= DROP_BURST_COUNT and ( + self._last_drop_warning is None + or now - self._last_drop_warning >= DROP_WARNING_INTERVAL + ): self._logger.warning( - "%s Event connection dropped %d times in 10 minutes; network may be unstable. Reconnecting...", + "%s Event connection dropped %d times in %ss; " + "network may be unstable. Reconnecting...", self._log_id, len(self._recent_drops), + DROP_BURST_WINDOW, ) self._last_drop_warning = now else: - # Covers EOF, RST (ConnectionResetError), aborted connections, - # and other socket errors: reconnect in all cases. - # Routine drops (e.g. MH200/MH201 hourly session recycling) are logged - # at DEBUG so healthy reconnects do not alarm downstream consumers. self._logger.debug( "%s Event connection lost, reconnecting...", self._log_id ) diff --git a/tests/test_connection.py b/tests/test_connection.py index 4c8cb0e..019da73 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 @@ -411,6 +420,7 @@ async def test_event_connection_loss_is_logged_at_debug() -> None: @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 @@ -430,9 +440,15 @@ async def test_event_connection_loss_oversized_frame_is_warning() -> None: ) -@pytest.mark.asyncio -async def test_event_connection_drops_clustering() -> None: +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() @@ -440,37 +456,88 @@ async def test_event_connection_drops_clustering() -> None: side_effect=asyncio.IncompleteReadError(b"", 0) ) session._reconnect = AsyncMock(return_value={"Success": True}) + return session, logger - # Drop 1 (DEBUG) - 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() - # Drop 2 (DEBUG) - 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() +BURST_WARNING = "%s Event connection dropped %d times in %ss; network may be unstable. Reconnecting..." - # Drop 3 within 10 minutes (WARNING) - await session.get_next() - logger.warning.assert_any_call( - "%s Event connection dropped %d times in 10 minutes; network may be unstable. Reconnecting...", - session._log_id, - 3, - ) - logger.warning.reset_mock() - # Drop 4 right away (DEBUG because we already warned in the last hour) - await session.get_next() +@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() + logger.warning.assert_called_once_with( + BURST_WARNING, session._log_id, DROP_BURST_COUNT, DROP_BURST_WINDOW + ) + 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_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() - logger.debug.assert_any_call( - "%s Event connection lost, reconnecting...", session._log_id + 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, DROP_BURST_WINDOW ) From d0d277d37f0ca57d7f7930e755b8fa7cf79d3066 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Tue, 22 Sep 2026 13:38:26 +0200 Subject: [PATCH 4/4] fix(connection): report the real drop span in the burst warning Per review: the warning always read "dropped 3 times in 600s" because it formatted the measurement window rather than the observed span, so three drops five seconds apart looked like ten minutes of trouble. Log `now - self._recent_drops[0]` instead, which is the span the drops in the current window actually covered. Adds test_event_connection_drop_burst_reports_real_span (a burst spread over 400s reports 400s), and the two existing burst assertions now pin the real 1.0s span instead of the window constant. Co-Authored-By: Claude Opus 5 --- OWNd/connection.py | 7 +++++-- tests/test_connection.py | 28 +++++++++++++++++++++++++--- 2 files changed, 30 insertions(+), 5 deletions(-) diff --git a/OWNd/connection.py b/OWNd/connection.py index 17f9cb2..54813ad 100755 --- a/OWNd/connection.py +++ b/OWNd/connection.py @@ -1161,12 +1161,15 @@ async def get_next(self) -> OWNMessage | str | None: 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 %ss; " + "%s Event connection dropped %d times in %.0fs; " "network may be unstable. Reconnecting...", self._log_id, len(self._recent_drops), - DROP_BURST_WINDOW, + now - self._recent_drops[0], ) self._last_drop_warning = now else: diff --git a/tests/test_connection.py b/tests/test_connection.py index 019da73..17535fe 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -459,7 +459,7 @@ def _dropping_event_session() -> tuple[OWNEventSession, MagicMock]: return session, logger -BURST_WARNING = "%s Event connection dropped %d times in %ss; network may be unstable. Reconnecting..." +BURST_WARNING = "%s Event connection dropped %d times in %.0fs; network may be unstable. Reconnecting..." @pytest.mark.asyncio @@ -478,8 +478,10 @@ async def test_event_connection_drop_burst_is_warning_on_fresh_host() -> None: 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, DROP_BURST_WINDOW + BURST_WARNING, session._log_id, DROP_BURST_COUNT, 1.0 ) logger.warning.reset_mock() @@ -493,6 +495,26 @@ async def test_event_connection_drop_burst_is_warning_on_fresh_host() -> None: 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.""" @@ -537,7 +559,7 @@ async def test_event_connection_drop_tracker_survives_reconnect() -> None: await session.get_next() logger.warning.assert_called_once_with( - BURST_WARNING, session._log_id, DROP_BURST_COUNT, DROP_BURST_WINDOW + BURST_WARNING, session._log_id, DROP_BURST_COUNT, 1.0 )