From 69f93be4dca396447a2abfcce12ff5d6b612e820 Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Thu, 6 Aug 2026 01:39:03 +0000 Subject: [PATCH] fix(coordinator): close the observe-mode race an Opus design review found The command-retry fix (issue #294) added a self._close_session() call to async_send_command that isn't synchronized against _attempt_observe_mode, which reads self._session and subscribes to it without holding _session_lock. A write's retry racing an in-flight subscribe attempt could tear down the session mid-subscribe -- or worse, land the close *after* the attempt's grace wait already succeeded, letting it commit observe mode against a session that's already gone: mode claims "Push" forever, with nothing left to notice the underlying socket is dead. Split ObserveManager.try_enter_observe_mode into four pieces (subscribe_hrefs / await_observe_notifies / enter_observe_mode / abandon_observe_attempt), keeping try_enter_observe_mode as a thin wrapper so its direct callers in test_observe.py are unaffected. _attempt_observe_mode now holds _session_lock only for the subscribe burst (each send is fire-and-forget, not a network round trip) and re-checks self._session is sess under the lock right before committing -- sess keeps the old session object alive, so identity can't be recycled onto a new one, which is what makes the check sufficient without a separate generation counter. The wait itself stays lock-free, so a command write is never blocked behind it. Two more bugs the same investigation turned up, fixed in the same pass since they're direct consequences of the design above: - async_send_command's own successful reconnect didn't downgrade observe mode the way the poll path's reconnect already does, leaving the same stale-commit problem reachable with zero concurrency at all -- just a write's retry succeeding while mode was observe. Replaced the poll path's local just_downgraded_from_observe with an instance flag both reconnect sites set, so either one triggers an immediate resubscribe. - _maybe_retry_observe_mode's 600s throttle gated solely on last_mode_change_ts, which _set_mode only stamps on an actual transition -- a device that never succeeds at observe mode leaves that timestamp stuck at construction time, so the throttle opens once and never closes again, re-attempting on every single poll cycle instead of every 600s. Now gates on the more recent of that timestamp and a new _last_observe_attempt_ts, stamped on every attempt regardless of outcome. --- custom_components/localthings/coordinator.py | 85 +++++++--- custom_components/localthings/observe.py | 80 +++++++--- tests/localthings/test_coordinator.py | 156 ++++++++++++++++++- 3 files changed, 279 insertions(+), 42 deletions(-) diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index d747406..c8d6d32 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -204,6 +204,14 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._consecutive_poll_timeouts = 0 self._unbound_hrefs: list[str] = [] self._reconnect_times: list[float] = [] + # See _maybe_retry_observe_mode: last_mode_change_ts alone doesn't + # move on a failed attempt, so this tracks attempts too. + self._last_observe_attempt_ts = 0.0 + # Set by both reconnect paths (poll and command) that hand back a + # session with zero OBSERVE registrations while mode was still + # observe; consumed once to trigger an immediate resubscribe + # instead of waiting out _RECOVERY_RETRY_S. + self._resubscribe_due = False # ------------------------------------------------------------------ # Session management (all blocking — must run in executor) @@ -731,29 +739,62 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): ir.async_delete_issue(self.hass, DOMAIN, issue_id) async def _attempt_observe_mode(self) -> None: - """Called once, right after first discovery. Blocking (sleeps for - the whole grace period) — must run in an executor.""" + """Called once, right after first discovery, after a reconnect + downgrades from observe, and periodically while polling. Two + phases: subscribing holds `_session_lock` (each send is fire-and- + forget, not a network round trip); the grace wait that follows + does not, so a concurrent command write isn't blocked for the + whole ~15s wait (issue #294) -- only the brief subscribe burst. + + The wait can outlast a reconnect elsewhere (the poll path's own + recovery, or a command retry), which would otherwise let a stale + success commit observe mode against a session that's already been + replaced -- claiming "Push" with nothing left to notice it's dead. + `self._session is sess` re-checked under the lock right before + committing closes that: `sess` keeps the old object alive, so + identity can't be recycled onto a new one. + """ hrefs = self._hot_hrefs + self._warm_hrefs if not hrefs: return - if self._session is None: - # _poll_once already connects on a real poll; only fires if the - # session was closed out from under us concurrently. - await self.hass.async_add_executor_job(self._connect_session) - sess = self._session - if sess is None: + self._last_observe_attempt_ts = time.monotonic() + async with self._session_lock: + if self._session is None: + # _poll_once already connects on a real poll; only fires if + # the session was closed out from under us concurrently. + await self.hass.async_add_executor_job(self._connect_session) + sess = self._session + if sess is None: + return + subscribed = await self.hass.async_add_executor_job( + self._observe.subscribe_hrefs, sess, hrefs + ) + if not subscribed: + self._observe.abandon_observe_attempt() return - await self.hass.async_add_executor_job( - self._observe.try_enter_observe_mode, - sess, - hrefs, - self._OBSERVE_GRACE_PERIOD_S, + reached = await self.hass.async_add_executor_job( + self._observe.await_observe_notifies, subscribed, self._OBSERVE_GRACE_PERIOD_S ) + async with self._session_lock: + if not reached or self._session is not sess: + self._observe.abandon_observe_attempt() + return + self._observe.enter_observe_mode(sess, subscribed) async def _maybe_retry_observe_mode(self) -> None: """While in poll-only mode, periodically re-attempt observe mode - so a device that gains internet access recovers push automatically.""" - if time.monotonic() - self._observe.last_mode_change_ts < _RECOVERY_RETRY_S: + so a device that gains internet access recovers push automatically. + + Gated on the more recent of the two timestamps, not just + `last_mode_change_ts`: `_set_mode` only stamps that on an actual + transition, so a device that never successfully enters observe + mode would otherwise leave this throttle open forever after the + first `_RECOVERY_RETRY_S` window -- re-attempting (and paying the + subscribe-burst lock) on every single poll cycle instead of every + `_RECOVERY_RETRY_S`. + """ + last_attempt = max(self._observe.last_mode_change_ts, self._last_observe_attempt_ts) + if time.monotonic() - last_attempt < _RECOVERY_RETRY_S: return await self._attempt_observe_mode() @@ -805,7 +846,6 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._subpoll_task.cancel() self._subpoll_task = None - just_downgraded_from_observe = False async with self._session_lock: try: resources = await self.hass.async_add_executor_job(self._poll_once) @@ -862,7 +902,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): "poll and resubscribing on the new session" ) self._observe.downgrade_to_poll() - just_downgraded_from_observe = True + self._resubscribe_due = True if not self._discovered: # One-time (issue #177): find sibling subdevices before the @@ -895,7 +935,8 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): for href, rep in resources.items(): self._observe.apply(href, rep, source=source) - if first_cycle or just_downgraded_from_observe: + if first_cycle or self._resubscribe_due: + self._resubscribe_due = False await self._attempt_observe_mode() elif self._observe.mode == MODE_POLL: await self._maybe_retry_observe_mode() @@ -1056,6 +1097,14 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._observe.mark_write_pending( write_href, settle_s=self._POST_TIMEOUT_S + self._POLL_TIMEOUT_S ) + # A reconnect hands back a session with zero OBSERVE + # registrations, same as the poll path's own reconnect -- + # keeping observe mode here would leave those subscriptions + # claimed on the closed session with no poll failure left + # to notice (issue #294). + if self._observe.mode == MODE_OBSERVE: + self._observe.downgrade_to_poll() + self._resubscribe_due = True await self.async_request_refresh() # ------------------------------------------------------------------ diff --git a/custom_components/localthings/observe.py b/custom_components/localthings/observe.py index 3d91d51..04226cf 100644 --- a/custom_components/localthings/observe.py +++ b/custom_components/localthings/observe.py @@ -171,17 +171,16 @@ class ObserveManager: self._last_notify_ts is not None and time.monotonic() - self._last_notify_ts < window_s ) - def try_enter_observe_mode( - self, - session, - hrefs: list[str], - grace_period_s: float = GRACE_PERIOD_S, - success_fraction: float = SUCCESS_FRACTION, - ) -> bool: - """Blocking — subscribes to every href then waits up to - `grace_period_s`, returning early once `success_fraction` of hrefs - have notified. Caller must run this in an executor, never on the - event loop.""" + def subscribe_hrefs(self, session, hrefs: list[str]) -> set[str]: + """Register OBSERVE on every href; returns the ones that took. + Blocking — run in an executor. + + Split from the grace wait below (issue #294) so the coordinator can + hold its session lock for just these sends -- each is a fire-and- + forget UDP datagram (DtlsCoapSession.subscribe doesn't wait for the + device's ack), unlike the wait, which can block for the whole grace + period and must not hold a lock a command write is also waiting on. + """ with self._notify_cond: self._notified.clear() subscribed: set[str] = set() @@ -192,30 +191,65 @@ class ObserveManager: subscribed.add(href) except Exception as e: self.log.warning("subscribe %s failed: %s", href, e) + return subscribed + + def await_observe_notifies( + self, + subscribed: set[str], + grace_period_s: float = GRACE_PERIOD_S, + success_fraction: float = SUCCESS_FRACTION, + ) -> bool: + """Blocking — waits up to `grace_period_s`, returning early once + `success_fraction` of `subscribed` have notified. Touches no + session; safe to run without holding a session lock.""" if not subscribed: - self._stop_refresh_task() - self._set_mode(MODE_POLL) - self.subscribed_hrefs = set() return False def _fraction_reached() -> bool: return len(set(self._notified) & subscribed) / len(subscribed) >= success_fraction with self._notify_cond: - reached = self._notify_cond.wait_for( - _fraction_reached, - timeout=grace_period_s, - ) + return self._notify_cond.wait_for(_fraction_reached, timeout=grace_period_s) - if reached: - self.subscribed_hrefs = subscribed - self._set_mode(MODE_OBSERVE) - self.start_refresh_task(session) - return True + def enter_observe_mode(self, session, subscribed: set[str]) -> None: + """Commit a successful attempt. Caller must have re-confirmed + `session` is still the live one under its session lock (issue + #294) -- committing against a session a reconnect already replaced + would claim observe mode with nothing left to notice it's dead.""" + self.subscribed_hrefs = set(subscribed) + self._set_mode(MODE_OBSERVE) + self.start_refresh_task(session) + def abandon_observe_attempt(self) -> None: + """Drop a failed or stale attempt: no subscriptions worth keeping.""" self._stop_refresh_task() self.subscribed_hrefs = set() self._set_mode(MODE_POLL) + + def try_enter_observe_mode( + self, + session, + hrefs: list[str], + grace_period_s: float = GRACE_PERIOD_S, + success_fraction: float = SUCCESS_FRACTION, + ) -> bool: + """Blocking — subscribes to every href then waits up to + `grace_period_s`, returning early once `success_fraction` of hrefs + have notified. Caller must run this in an executor, never on the + event loop. + + Single-threaded convenience wrapper around the phase split above + (subscribe_hrefs / await_observe_notifies / enter_observe_mode / + abandon_observe_attempt) for callers -- direct and most existing + tests -- that don't need the lock-scoping those phases exist for.""" + subscribed = self.subscribe_hrefs(session, hrefs) + if not subscribed: + self.abandon_observe_attempt() + return False + if self.await_observe_notifies(subscribed, grace_period_s, success_fraction): + self.enter_observe_mode(session, subscribed) + return True + self.abandon_observe_attempt() return False def _set_mode(self, mode: str) -> None: diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index 42cc435..131d3fa 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -2,6 +2,7 @@ from __future__ import annotations +import time from datetime import timedelta from unittest.mock import AsyncMock, patch @@ -21,6 +22,7 @@ from custom_components.localthings.const import ( SUMMARY_INTERVAL_S, ) from custom_components.localthings.coordinator import ( + _RECOVERY_RETRY_S, LocalThingsCoordinator, _local_source_port, ) @@ -30,7 +32,7 @@ from custom_components.localthings.registry.capabilities.common import ( remote_control_required_for_write, ) -from .conftest import ENTRY_DATA, MOCK_MODEL, MOCK_SERIAL +from .conftest import ENTRY_DATA, MOCK_MODEL, MOCK_SERIAL, FakeObserveSession from .conftest import _load_fridge_resources as _load_fridge @@ -742,6 +744,95 @@ async def test_reconnect_from_observe_mode_resubscribes_immediately( assert coordinator.observe_mode == MODE_OBSERVE +async def test_attempt_observe_mode_discards_stale_commit_after_session_swap( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """A reconnect (the poll path's own, or a command's retry) can swap + self._session while this attempt's grace wait is in flight -- it runs + without holding _session_lock precisely so a write isn't blocked behind + it (issue #294). Committing observe mode against the now-stale local + `sess` reference would claim "Push" on a session that's already gone, + with nothing left to notice -- the identity re-check under the lock + right before committing must catch this and abandon instead. + + Simulates the swap from inside await_observe_notifies itself rather + than via real concurrency: subscribe_hrefs (and its lock) has already + returned by the time that call runs, so this lands exactly in the + window the identity check exists to cover, deterministically.""" + 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] + + other = FakeObserveSession() + + def _swap_session_mid_wait(subscribed, grace_period_s, success_fraction=None): + coordinator._session = other + return True + + with patch.object( + coordinator._observe, "await_observe_notifies", side_effect=_swap_session_mid_wait + ): + await coordinator._attempt_observe_mode() + + assert coordinator.observe_mode == MODE_POLL + assert coordinator._observe.subscribed_hrefs == set() + assert coordinator._observe._refresh_thread is None + + +async def test_maybe_retry_observe_mode_uses_most_recent_attempt_not_just_mode_change( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """_set_mode only stamps last_mode_change_ts on an actual transition, + so a device that never successfully enters observe mode leaves that + timestamp stuck at construction time forever -- a failed attempt keeps + calling _set_mode(MODE_POLL) while already in MODE_POLL, a no-op. + Gating solely on that timestamp would make the 600s throttle open once + and then never close again, re-attempting (and paying the subscribe + burst) on every single poll cycle instead of every _RECOVERY_RETRY_S.""" + 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] + assert coordinator.observe_mode == MODE_POLL # never notified during setup + + # Simulate exactly the scenario above: mode_change_ts is old (as it + # would be forever, for a device that never gets push), but an attempt + # really did just run. + coordinator._observe.last_mode_change_ts = time.monotonic() - _RECOVERY_RETRY_S - 1 + coordinator._last_observe_attempt_ts = time.monotonic() + + with patch.object(fake, "subscribe") as mock_subscribe: + await coordinator._maybe_retry_observe_mode() + + mock_subscribe.assert_not_called() + + +async def test_attempt_observe_mode_releases_lock_before_the_grace_wait( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """The subscribe burst holds _session_lock (it touches the session); + the grace wait after it must not, or a command write could stall + behind up to _OBSERVE_GRACE_PERIOD_S of an unrelated observe-mode-entry + attempt (issue #294). By construction, subscribe_hrefs's own + `async with self._session_lock:` has already exited by the time + await_observe_notifies is even called -- checked here rather than + inferred from timing.""" + 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] + + locked_during_wait = {"value": None} + + def _check_lock(subscribed, grace_period_s, success_fraction=None): + locked_during_wait["value"] = coordinator._session_lock.locked() + return True + + with patch.object(coordinator._observe, "await_observe_notifies", side_effect=_check_lock): + await coordinator._attempt_observe_mode() + + assert locked_during_wait["value"] is False + + async def test_sweep_mismatch_never_downgrades_a_live_observe_session( hass: HomeAssistant, mock_entry, mock_coordinator_observe_session ) -> None: @@ -1110,6 +1201,69 @@ async def test_send_command_raises_after_reconnect_retry_also_fails( await coordinator.async_send_command(bound, 5) +async def test_send_command_reconnect_downgrades_observe_mode( + hass: HomeAssistant, + mock_entry, + mock_coordinator_observe_session, +) -> None: + """A command's own successful reconnect hands back a session with zero + OBSERVE registrations too, same as the poll path's reconnect -- must + downgrade the same way and flag a resubscribe, or observe mode stays + claimed against a session the write just replaced underneath it + (issue #294).""" + from custom_components.localthings.registry.discovery import BoundEntity + from custom_components.localthings.registry.entities import NumberDesc + + 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 + + fake.notify_on_subscribe = {"notified": True} + entered = await hass.async_add_executor_job( + coordinator._observe.try_enter_observe_mode, + fake, + hrefs, + 0.02, + 0.8, + ) + assert entered is True + assert coordinator.observe_mode == MODE_OBSERVE + + def _write_fn(payload, rep, href=None): + return (["test", "vs", "0"], {"value": payload}) + + desc = NumberDesc(key="test", field="value", write_fn=_write_fn) + bound = BoundEntity(href="/test/vs/0", capability=coordinator.bound[0].capability, desc=desc) + + calls = {"n": 0} + + def _post(*args, **kwargs): + calls["n"] += 1 + if calls["n"] == 1: + raise ConnectionError("socket closed") + return (0x44, b"") + + with ( + patch.object(fake, "subscribe") as mock_subscribe, + patch( + "custom_components.localthings.coordinator.asyncio.sleep", + new=AsyncMock(), + ), + ): + fake.post = _post + await coordinator.async_send_command(bound, 5) + + # _resubscribe_due is consumed by this same call's own trailing + # refresh (async_request_refresh is awaited, not fire-and-forget), + # so the visible effect is a resubscribe attempt, not a lingering + # flag value to assert on afterward. + assert mock_subscribe.called + + assert coordinator.observe_mode == MODE_POLL + + async def test_second_write_to_same_href_lands_during_first_writes_settle_window( hass: HomeAssistant, mock_entry,