From 9a45afe949b5a1054ab6a69338fb9ba7d26e6aab Mon Sep 17 00:00:00 2001 From: JayChickenK <19538983+JayChickenK@users.noreply.github.com> Date: Tue, 18 Aug 2026 11:19:29 +0200 Subject: [PATCH 1/4] observe: keep silent hrefs on the sub-poll cadence Issue #92: try_enter_observe_mode treated every subscribed href as push-covered once SUCCESS_FRACTION cleared, including ones that never notified. Those then skipped _run_subpolls for the rest of the session. Record subscribed-but-silent hrefs on fallback_hrefs (idle in observe mode) and keep just that set on the hot/warm cadence. --- custom_components/localthings/coordinator.py | 25 +++++---- custom_components/localthings/observe.py | 11 ++++ tests/localthings/test_coordinator.py | 56 ++++++++++++++++++++ tests/localthings/test_observe.py | 19 +++++++ 4 files changed, 102 insertions(+), 9 deletions(-) diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index a7da859..eabde62 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -975,16 +975,23 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): # ------------------------------------------------------------------ async def _run_subpolls(self, force: bool = False) -> None: - """Poll hot/warm hrefs in the gaps between summary polls. No-op in - observe-primary mode (those hrefs are already covered by push) - unless `force` is set -- set when this cycle's sweep found the - cache disagreeing with a still-live observe session (see - log_sweep_discrepancies): a bounded fallback for a channel gone - silent without a reconnect.""" + """Poll hot/warm hrefs in the gaps between summary polls. + + In observe-primary mode this is a no-op for hrefs the device is + actually pushing, unless `force` is set (sweep disagreed with the + cache -- see log_sweep_discrepancies). Hrefs that were subscribed + but stayed silent through the grace period (issue #92) stay on + the hot/warm cadence via `fallback_hrefs`. + """ if self._observe.mode == MODE_OBSERVE and not force: - return - hot = self._hot_hrefs - warm = self._warm_hrefs + silent = self._observe.fallback_hrefs + if not silent: + return + hot = [h for h in self._hot_hrefs if h in silent] + warm = [h for h in self._warm_hrefs if h in silent] + else: + hot = self._hot_hrefs + warm = self._warm_hrefs if not hot and not warm: return step = self._SUBPOLL_STEP_S diff --git a/custom_components/localthings/observe.py b/custom_components/localthings/observe.py index 1215766..0f547b5 100644 --- a/custom_components/localthings/observe.py +++ b/custom_components/localthings/observe.py @@ -97,6 +97,11 @@ class ObserveManager: # Wakes try_enter_observe_mode's grace wait early once enough hrefs # have notified. Guards only `_notified` mutations + the `wait_for`. self._notify_cond = threading.Condition() + # Idle while polling, except after downgrade_to_poll (every href + # that was subscribed). While in observe mode this is the set of + # hrefs that were subscribed but never notified during the grace + # period (issue #92) -- they stay on the hot/warm sub-poll cadence + # instead of waiting for the 30s /device/0 sweep. self.fallback_hrefs: set[str] = set() self._on_applied: Callable[[str, dict, str], None] | None = None self._refresh_task: ObserveRefreshTask | None = None @@ -270,6 +275,12 @@ class ObserveManager: #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) + with self._notify_cond: + notified = set(self._notified) + # Issue #92: subscribed-but-silent hrefs are counted as covered by + # push if we drop this, but they never emit a notify. Keep them on + # the poll cadence via fallback_hrefs (otherwise idle in observe). + self.fallback_hrefs = set(subscribed) - notified self._set_mode(MODE_OBSERVE) self.start_refresh_task(session) diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index ed1844e..16ec09e 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -1050,6 +1050,62 @@ async def test_sweep_mismatch_forces_subpolls_on_a_live_observe_session( mock_subpolls.assert_called_once_with(force=True) +async def test_observe_mode_subpolls_only_silent_hrefs( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """Issue #92: subscribed-but-silent hrefs keep the hot/warm cadence + instead of waiting for the 30s sweep.""" + 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._hot_hrefs + silent = coordinator._hot_hrefs[0] + coordinator._observe.mode = MODE_OBSERVE + coordinator._observe.fallback_hrefs = {silent} + + polled: list[list[str]] = [] + + def _capture(hrefs): + polled.append(list(hrefs)) + + with ( + patch.object(coordinator, "_poll_hrefs_blocking", side_effect=_capture), + patch( + "custom_components.localthings.coordinator.asyncio.sleep", + new_callable=AsyncMock, + ), + ): + await coordinator._run_subpolls() + + assert polled + for batch in polled: + assert set(batch) == {silent} + + +async def test_observe_mode_skips_subpolls_when_nothing_is_silent( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """The observe-mode no-op stays in place when every subscribed href + actually notified -- issue #92 only keeps the silent ones on poll.""" + 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] + coordinator._observe.mode = MODE_OBSERVE + coordinator._observe.fallback_hrefs = set() + + with ( + patch.object(coordinator, "_poll_hrefs_blocking") as mock_poll, + patch( + "custom_components.localthings.coordinator.asyncio.sleep", + new_callable=AsyncMock, + ) as mock_sleep, + ): + await coordinator._run_subpolls() + + mock_poll.assert_not_called() + mock_sleep.assert_not_called() + + async def test_write_marks_href_pending_before_post( hass: HomeAssistant, mock_entry, mock_coordinator_observe_session ) -> None: diff --git a/tests/localthings/test_observe.py b/tests/localthings/test_observe.py index 87af3e4..ec8e20f 100644 --- a/tests/localthings/test_observe.py +++ b/tests/localthings/test_observe.py @@ -190,6 +190,7 @@ def test_try_enter_observe_mode_succeeds_when_all_hrefs_notify(): assert entered is True assert mgr.mode == "observe" assert mgr.subscribed_hrefs == set(hrefs) + assert mgr.fallback_hrefs == set() finally: mgr.close() @@ -218,6 +219,23 @@ def test_try_enter_observe_mode_falls_back_when_subscribe_fails_for_all(): assert mgr.mode == "poll" +def test_enter_observe_mode_keeps_silent_hrefs_on_fallback(): + """Issue #92: subscribed hrefs that never notified stay on the poll + cadence via fallback_hrefs, rather than being treated as push-covered.""" + mgr = _manager() + session = _FakeSession() + subscribed = {"/a/vs/0", "/b/vs/0", "/c/vs/0"} + mgr.on_notification("/a/vs/0", cbor2.dumps({"x": 1})) + mgr.on_notification("/b/vs/0", cbor2.dumps({"x": 1})) + try: + mgr.enter_observe_mode(session, subscribed) + assert mgr.mode == "observe" + assert mgr.subscribed_hrefs == subscribed + assert mgr.fallback_hrefs == {"/c/vs/0"} + finally: + mgr.close() + + def test_try_enter_observe_mode_meets_success_fraction_with_partial_notifies(): mgr = _manager() session = _FakeSession() @@ -243,6 +261,7 @@ def test_try_enter_observe_mode_meets_success_fraction_with_partial_notifies(): assert entered is True assert mgr.mode == "observe" + assert mgr.fallback_hrefs == {hrefs[3]} finally: mgr.close() From 43076352aeae28eac92eb84176f2c29a1e682769 Mon Sep 17 00:00:00 2001 From: JayChickenK <19538983+JayChickenK@users.noreply.github.com> Date: Wed, 19 Aug 2026 18:33:22 +0200 Subject: [PATCH 2/4] observe: drop fallback hrefs on late notify; skip empty subpolls on_notification discards from fallback_hrefs so a late first push (after the 80% quorum snapshot) self-corrects instead of staying on the 3s GET cadence. Empty subpoll slots skip the session lock. --- custom_components/localthings/coordinator.py | 2 ++ custom_components/localthings/observe.py | 15 ++++++--- tests/localthings/test_coordinator.py | 32 ++++++++++++++++++++ tests/localthings/test_observe.py | 18 +++++++++++ 4 files changed, 63 insertions(+), 4 deletions(-) diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index eabde62..f4df4dd 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -998,6 +998,8 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): for i in range(1, 10): # slots 1..9 (T+3 s … T+27 s) await asyncio.sleep(step) hrefs = list(hot) + (list(warm) if i % 2 == 0 else []) + if not hrefs: + continue async with self._session_lock: try: await self.hass.async_add_executor_job(self._poll_hrefs_blocking, hrefs) diff --git a/custom_components/localthings/observe.py b/custom_components/localthings/observe.py index 0f547b5..ec1142c 100644 --- a/custom_components/localthings/observe.py +++ b/custom_components/localthings/observe.py @@ -95,13 +95,14 @@ class ObserveManager: self._notified: set[str] = set() self._last_notify_ts: float | None = None # Wakes try_enter_observe_mode's grace wait early once enough hrefs - # have notified. Guards only `_notified` mutations + the `wait_for`. + # have notified. Guards `_notified` mutations, the `wait_for`, and + # fallback_hrefs discards from on_notification. self._notify_cond = threading.Condition() # Idle while polling, except after downgrade_to_poll (every href # that was subscribed). While in observe mode this is the set of - # hrefs that were subscribed but never notified during the grace - # period (issue #92) -- they stay on the hot/warm sub-poll cadence - # instead of waiting for the 30s /device/0 sweep. + # subscribed hrefs that have not yet notified (issue #92) -- they + # stay on the hot/warm sub-poll cadence. on_notification discards + # an href once it pushes, so a late first notify self-corrects. self.fallback_hrefs: set[str] = set() self._on_applied: Callable[[str, dict, str], None] | None = None self._refresh_task: ObserveRefreshTask | None = None @@ -210,6 +211,10 @@ class ObserveManager: return with self._notify_cond: self._notified.add(href) + # Snapshot in enter_observe_mode is at the 80% quorum, not the + # full grace period -- a late first push (blockwise refetch) + # must drop the href so we don't keep GET-polling a live one. + self.fallback_hrefs.discard(href) self._last_notify_ts = time.monotonic() self._notify_cond.notify_all() self.log.debug("observe notify: %s", href) @@ -280,6 +285,8 @@ class ObserveManager: # Issue #92: subscribed-but-silent hrefs are counted as covered by # push if we drop this, but they never emit a notify. Keep them on # the poll cadence via fallback_hrefs (otherwise idle in observe). + # This is the 80% quorum snapshot; a later notify discards the + # href in on_notification. self.fallback_hrefs = set(subscribed) - notified self._set_mode(MODE_OBSERVE) self.start_refresh_task(session) diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index 16ec09e..8e3d51b 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -1106,6 +1106,38 @@ async def test_observe_mode_skips_subpolls_when_nothing_is_silent( mock_sleep.assert_not_called() +async def test_observe_mode_skips_empty_subpoll_slots( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """Once hot is empty, odd slots would otherwise take the session lock + and dispatch a no-op executor job. Skip those.""" + 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] + coordinator._observe.mode = MODE_OBSERVE + coordinator._hot_hrefs = [] + coordinator._warm_hrefs = ["/warm/vs/0"] + coordinator._observe.fallback_hrefs = {"/warm/vs/0"} + + polled: list[list[str]] = [] + + def _capture(hrefs): + polled.append(list(hrefs)) + + with ( + patch.object(coordinator, "_poll_hrefs_blocking", side_effect=_capture), + patch( + "custom_components.localthings.coordinator.asyncio.sleep", + new_callable=AsyncMock, + ), + ): + await coordinator._run_subpolls() + + assert polled + for batch in polled: + assert batch == ["/warm/vs/0"] + + async def test_write_marks_href_pending_before_post( hass: HomeAssistant, mock_entry, mock_coordinator_observe_session ) -> None: diff --git a/tests/localthings/test_observe.py b/tests/localthings/test_observe.py index ec8e20f..5fcb1fc 100644 --- a/tests/localthings/test_observe.py +++ b/tests/localthings/test_observe.py @@ -236,6 +236,24 @@ def test_enter_observe_mode_keeps_silent_hrefs_on_fallback(): mgr.close() +def test_on_notification_drops_href_from_fallback(): + """A late first notify (after the 80% snapshot) self-corrects + fallback_hrefs so a slow-but-pushing href is not polled for the rest + of the session -- multi-block resources being the reliable victim.""" + mgr = _manager() + session = _FakeSession() + subscribed = {"/a/vs/0", "/b/vs/0", "/c/vs/0"} + mgr.on_notification("/a/vs/0", cbor2.dumps({"x": 1})) + mgr.on_notification("/b/vs/0", cbor2.dumps({"x": 1})) + try: + mgr.enter_observe_mode(session, subscribed) + assert mgr.fallback_hrefs == {"/c/vs/0"} + mgr.on_notification("/c/vs/0", cbor2.dumps({"x": 1})) + assert mgr.fallback_hrefs == set() + finally: + mgr.close() + + def test_try_enter_observe_mode_meets_success_fraction_with_partial_notifies(): mgr = _manager() session = _FakeSession() From 35f144a2d12d415958b06a8931b030e00fceb1d6 Mon Sep 17 00:00:00 2001 From: JayChickenK <19538983+JayChickenK@users.noreply.github.com> Date: Wed, 19 Aug 2026 18:37:54 +0200 Subject: [PATCH 3/4] tests: cancel background subpolls before driving them directly Setup already starts `_run_subpolls` as a background task. Patching `_poll_hrefs_blocking` then captured its unfiltered hot/warm batches alongside the silent-only ones the test asked for. --- tests/localthings/test_coordinator.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index 8e3d51b..2a00caf 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -2,6 +2,8 @@ from __future__ import annotations +import asyncio +import contextlib import time from datetime import timedelta from unittest.mock import AsyncMock, patch @@ -1050,6 +1052,19 @@ async def test_sweep_mismatch_forces_subpolls_on_a_live_observe_session( mock_subpolls.assert_called_once_with(force=True) +async def _cancel_background_subpolls(coordinator: LocalThingsCoordinator) -> None: + """Setup starts `_run_subpolls` as a background task. Tests that drive + it directly have to cancel that one first, or a patched + `_poll_hrefs_blocking` also captures its batches.""" + task = coordinator._subpoll_task + if task is None: + return + task.cancel() + coordinator._subpoll_task = None + with contextlib.suppress(asyncio.CancelledError): + await task + + async def test_observe_mode_subpolls_only_silent_hrefs( hass: HomeAssistant, mock_entry, mock_coordinator_observe_session ) -> None: @@ -1058,6 +1073,7 @@ async def test_observe_mode_subpolls_only_silent_hrefs( 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] + await _cancel_background_subpolls(coordinator) assert coordinator._hot_hrefs silent = coordinator._hot_hrefs[0] coordinator._observe.mode = MODE_OBSERVE @@ -1090,6 +1106,7 @@ async def test_observe_mode_skips_subpolls_when_nothing_is_silent( 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] + await _cancel_background_subpolls(coordinator) coordinator._observe.mode = MODE_OBSERVE coordinator._observe.fallback_hrefs = set() @@ -1114,6 +1131,7 @@ async def test_observe_mode_skips_empty_subpoll_slots( 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] + await _cancel_background_subpolls(coordinator) coordinator._observe.mode = MODE_OBSERVE coordinator._hot_hrefs = [] coordinator._warm_hrefs = ["/warm/vs/0"] From 4edd5b7866a0ca550437fb966f90fb4674f48276 Mon Sep 17 00:00:00 2001 From: JayChickenK <19538983+JayChickenK@users.noreply.github.com> Date: Wed, 19 Aug 2026 22:22:11 +0200 Subject: [PATCH 4/4] observe: assign fallback_hrefs under the notify lock on_notification discards on the DTLS reader thread. Snapshotting _notified then assigning fallback_hrefs after releasing the lock let a notify in that window land on the set object being replaced, so a just-pushed href was classified silent until its next notify. --- custom_components/localthings/observe.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/custom_components/localthings/observe.py b/custom_components/localthings/observe.py index ec1142c..610f468 100644 --- a/custom_components/localthings/observe.py +++ b/custom_components/localthings/observe.py @@ -96,7 +96,8 @@ class ObserveManager: self._last_notify_ts: float | None = None # Wakes try_enter_observe_mode's grace wait early once enough hrefs # have notified. Guards `_notified` mutations, the `wait_for`, and - # fallback_hrefs discards from on_notification. + # fallback_hrefs (enter_observe_mode assignment, on_notification + # discard). self._notify_cond = threading.Condition() # Idle while polling, except after downgrade_to_poll (every href # that was subscribed). While in observe mode this is the set of @@ -280,14 +281,13 @@ class ObserveManager: #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) - with self._notify_cond: - notified = set(self._notified) # Issue #92: subscribed-but-silent hrefs are counted as covered by # push if we drop this, but they never emit a notify. Keep them on # the poll cadence via fallback_hrefs (otherwise idle in observe). - # This is the 80% quorum snapshot; a later notify discards the - # href in on_notification. - self.fallback_hrefs = set(subscribed) - notified + # Same lock as on_notification's discard so a notify in this window + # cannot land on a set object that is about to be replaced. + with self._notify_cond: + self.fallback_hrefs = set(subscribed) - self._notified self._set_mode(MODE_OBSERVE) self.start_refresh_task(session)