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.
This commit is contained in:
JayChickenK
2026-08-19 18:33:22 +02:00
parent 9a45afe949
commit 43076352ae
4 changed files with 63 additions and 4 deletions
@@ -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)
+11 -4
View File
@@ -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)
+32
View File
@@ -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:
+18
View File
@@ -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()