diff --git a/custom_components/localthings/observe.py b/custom_components/localthings/observe.py index 7d51109..ca13ce0 100644 --- a/custom_components/localthings/observe.py +++ b/custom_components/localthings/observe.py @@ -73,6 +73,9 @@ class ObserveManager: self.subscribed_hrefs: set[str] = set() 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`. + self._notify_cond = threading.Condition() self.fallback_hrefs: set[str] = set() self._refresh_task: ObserveRefreshTask | None = None self._refresh_stop: threading.Event | None = None @@ -147,8 +150,10 @@ class ObserveManager: return if not isinstance(rep, dict): return - self._notified.add(href) - self._last_notify_ts = time.monotonic() + with self._notify_cond: + self._notified.add(href) + self._last_notify_ts = time.monotonic() + self._notify_cond.notify_all() self.log.debug("observe notify: %s", href) self.apply(href, rep, source='observe') @@ -172,10 +177,12 @@ class ObserveManager: grace_period_s: float = GRACE_PERIOD_S, success_fraction: float = SUCCESS_FRACTION, ) -> bool: - """Blocking — subscribes to every href then sleeps for the whole - grace period. Caller must run this in an executor, never on the + """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.""" - self._notified.clear() + with self._notify_cond: + self._notified.clear() subscribed: set[str] = set() for href in hrefs: segs = [s for s in href.strip('/').split('/') if s] @@ -190,10 +197,18 @@ class ObserveManager: self.subscribed_hrefs = set() return False - time.sleep(grace_period_s) + def _fraction_reached() -> bool: + return ( + len(set(self._notified) & subscribed) / len(subscribed) + >= success_fraction + ) - fraction = len(set(self._notified) & subscribed) / len(subscribed) - if fraction >= success_fraction: + with self._notify_cond: + reached = 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) diff --git a/tests/localthings/test_observe.py b/tests/localthings/test_observe.py index 4ea057d..1a87724 100644 --- a/tests/localthings/test_observe.py +++ b/tests/localthings/test_observe.py @@ -452,3 +452,71 @@ def test_downgrade_to_poll_stops_refresh_task(): assert not thread.is_alive() assert mgr._refresh_thread is None + + +def test_try_enter_observe_mode_exits_early_when_fraction_reached(): + """The grace wait returns as soon as enough hrefs notify, not after the + whole grace ceiling. Production data (fridge TP2X_REF_20K, 2026-07-30): + all subscribed hrefs notify within ~0.34s of subscribe, yet the old fixed + time.sleep(15) waited the remaining ~14.6s anyway — every successful + first-refresh / observe-retry fetch paid that dead time.""" + mgr = _manager() + session = _FakeSession() + hrefs = ['/a/vs/0', '/b/vs/0'] + + def _notify_after_subscribe(): + # Let try_enter clear _notified + subscribe first (matches the existing + # test pattern's 0.005 lead), then notify well before the 2s ceiling. + time.sleep(0.01) + for href in hrefs: + mgr.on_notification(href, cbor2.dumps({'x': 1})) + + notifier = threading.Thread(target=_notify_after_subscribe, daemon=True) + notifier.start() + + try: + t0 = time.monotonic() + entered = mgr.try_enter_observe_mode(session, hrefs, grace_period_s=2.0) + elapsed = time.monotonic() - t0 + notifier.join() + + assert entered is True + assert mgr.mode == 'observe' + # Early exit: returns near the notify (~0.01s), not the 2.0s ceiling. + # The old fixed sleep would make this >= 2.0. + assert elapsed < 0.5 + finally: + mgr.close() + + +def test_try_enter_observe_mode_waits_full_ceiling_when_fraction_not_reached(): + """Early-exit must NOT fire below the success fraction. A partial set of + notifies that doesn't meet the threshold waits the full grace ceiling, + then falls back to poll — identical to the old fixed sleep. Guards + against an over-eager predicate that enters observe on too few notifies.""" + mgr = _manager() + session = _FakeSession() + hrefs = ['/a/vs/0', '/b/vs/0', '/c/vs/0', '/d/vs/0'] # 4 hrefs + + def _notify_one(): + time.sleep(0.01) + mgr.on_notification(hrefs[0], cbor2.dumps({'x': 1})) # 1/4 = 0.25 + + notifier = threading.Thread(target=_notify_one, daemon=True) + notifier.start() + + try: + t0 = time.monotonic() + entered = mgr.try_enter_observe_mode( + session, hrefs, grace_period_s=0.2, success_fraction=0.8, + ) + elapsed = time.monotonic() - t0 + notifier.join() + + assert entered is False + assert mgr.mode == 'poll' + assert mgr.subscribed_hrefs == set() + # Waited the full 0.2s ceiling (1/4 = 0.25 < 0.8); did not early-exit. + assert elapsed >= 0.18 + finally: + mgr.close()