From 7024b170fe0090324d1e3ee6cd12d9b73f3a54f9 Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Thu, 9 Jul 2026 20:36:56 -0500 Subject: [PATCH] fix(observe): re-sync to poll mode on reconnect, harden notified-set race A poll-failure-triggered reconnect while in observe mode left OBSERVE silently dead: the new session had zero subscriptions, mode stayed 'observe' so the hot/warm sub-poll fallback stayed disabled, and the ObserveRefreshTask kept retrying against the closed session instead of the live one. Downgrading to poll mode on reconnect hands recovery to the existing poll-mode retry path, which re-subscribes on the new session. Also snapshot ObserveManager._notified before intersecting it against the subscribed set, closing a rare set-mutated-during-iteration race between the DTLS reader thread (on_notification) and the executor thread computing the grace-period success fraction. --- custom_components/localthings/coordinator.py | 14 +++++ custom_components/localthings/observe.py | 2 +- tests/localthings/test_coordinator.py | 55 +++++++++++++++++++- 3 files changed, 69 insertions(+), 2 deletions(-) diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index fa8985e..f9f9be7 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -345,6 +345,20 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): _LOGGER.debug("Full error:", exc_info=e2) return flatten(self.bound, snapshot) raise UpdateFailed(f"poll failed after reconnect: {e2}") from e2 + else: + # The reconnect gave us a brand-new session with zero + # OBSERVE registrations. If we were in observe mode, + # that state is now stale — the refresh task is still + # pinned to the old (closed) session and nothing will + # ever re-subscribe on the new one. Tear it down now so + # the existing poll-mode retry path (_maybe_retry_observe_mode) + # re-subscribes on the live session on its next cycle. + if self._observe.mode == MODE_OBSERVE: + _LOGGER.debug( + "reconnect while in observe mode; downgrading to " + "poll so recovery re-subscribes on the new session" + ) + self._observe.downgrade_to_poll() source = 'sweep' if self._discovered else 'poll' if self._observe.mode == MODE_OBSERVE and self._observe.check_sweep_for_misses(resources): diff --git a/custom_components/localthings/observe.py b/custom_components/localthings/observe.py index dba6de0..bf02edc 100644 --- a/custom_components/localthings/observe.py +++ b/custom_components/localthings/observe.py @@ -111,7 +111,7 @@ class ObserveManager: time.sleep(grace_period_s) - fraction = len(self._notified & subscribed) / len(subscribed) + fraction = len(set(self._notified) & subscribed) / len(subscribed) if fraction >= success_fraction: self.subscribed_hrefs = subscribed self._set_mode(MODE_OBSERVE) diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index 5067a90..ee83309 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -4,7 +4,7 @@ from __future__ import annotations import threading import time from datetime import timedelta -from unittest.mock import patch +from unittest.mock import AsyncMock, patch import cbor2 import pytest @@ -224,6 +224,59 @@ async def test_enters_observe_mode_when_hot_warm_hrefs_notify( assert coordinator._observe.mode == MODE_OBSERVE +async def test_reconnect_while_observe_mode_downgrades_to_poll( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session, fridge_resources +) -> None: + """A poll-failure-triggered reconnect while in observe mode must tear + down the (now stale) observe state rather than leave it dangling on a + dead session — see _async_update_data's reconnect branch. + + After the reconnect the coordinator should be back in poll mode, which + hands recovery to the existing _maybe_retry_observe_mode path so it + re-subscribes on the live session on its next eligible cycle. + """ + fake = mock_coordinator_observe_session + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs + + # Get the coordinator into observe mode the same way + # test_enters_observe_mode_when_hot_warm_hrefs_notify does. + def _notify_during_grace_period(): + time.sleep(0.005) + for href in hrefs: + fake.on_notification(href, cbor2.dumps({'notified': True})) + + notifier = threading.Thread(target=_notify_during_grace_period, daemon=True) + notifier.start() + entered = await hass.async_add_executor_job( + coordinator._observe.try_enter_observe_mode, + fake, hrefs, 0.02, 0.8, + ) + notifier.join() + assert entered is True + assert coordinator.observe_mode == MODE_OBSERVE + + # Simulate the existing "poll failed, reconnecting" branch: _poll_once + # fails once (triggering the reconnect/backoff path), then succeeds. + with ( + patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once', + side_effect=[RuntimeError('connection lost'), fridge_resources], + ), + patch( + 'custom_components.localthings.coordinator.asyncio.sleep', + new=AsyncMock(), + ), + ): + await coordinator.async_request_refresh() + await hass.async_block_till_done() + + assert coordinator._observe.mode == MODE_POLL + + async def test_write_marks_href_pending_before_post( hass: HomeAssistant, mock_entry, mock_coordinator_observe_session ) -> None: