feat(observe): manage ObserveRefreshTask lifecycle

custom_components/localthings/observe.py

@@ -17,9 +17,12 @@ import time
 import cbor2
 
 from smartthings_local.ocf.state_cache import StateCache
+from smartthings_local.ocf.observe_refresh import ObserveRefreshTask
 
 _LOGGER = logging.getLogger(__name__)
 
+REFRESH_INTERVAL_S = 6 * 3600.0
+
 MODE_OBSERVE = 'observe'
 MODE_POLL = 'poll'
 
@@ -45,6 +48,9 @@ class ObserveManager:
         self.subscribed_hrefs: set[str] = set()
         self._notified: set[str] = set()
         self.fallback_hrefs: set[str] = set()
+        self._refresh_task: ObserveRefreshTask | None = None
+        self._refresh_stop: threading.Event | None = None
+        self._refresh_thread: threading.Thread | None = None
 
     def mark_write_pending(self, href: str, settle_s: float = DEFAULT_SETTLE_S) -> None:
         with self._settle_lock:
@@ -98,6 +104,7 @@ class ObserveManager:
             except Exception as e:
                 self.log.warning("subscribe %s failed: %s", href, e)
         if not subscribed:
+            self._stop_refresh_task()
             self._set_mode(MODE_POLL)
             self.subscribed_hrefs = set()
             return False
@@ -108,8 +115,10 @@ class ObserveManager:
         if fraction >= success_fraction:
             self.subscribed_hrefs = subscribed
             self._set_mode(MODE_OBSERVE)
+            self.start_refresh_task(session)
             return True
 
+        self._stop_refresh_task()
         self.subscribed_hrefs = set()
         self._set_mode(MODE_POLL)
         return False
@@ -144,4 +153,29 @@ class ObserveManager:
     def downgrade_to_poll(self) -> None:
         self.fallback_hrefs = set(self.subscribed_hrefs)
         self.subscribed_hrefs = set()
+        self._stop_refresh_task()
         self._set_mode(MODE_POLL)
+
+    def start_refresh_task(self, session) -> None:
+        self._stop_refresh_task()
+        paths = [tuple(h.strip('/').split('/')) for h in self.subscribed_hrefs]
+        self._refresh_task = ObserveRefreshTask(
+            session, paths, interval_s=REFRESH_INTERVAL_S, logger=self.log,
+        )
+        self._refresh_stop = threading.Event()
+        self._refresh_thread = threading.Thread(
+            target=self._refresh_task.run_forever,
+            args=(self._refresh_stop,),
+            daemon=True, name='localthings-observe-refresh',
+        )
+        self._refresh_thread.start()
+
+    def _stop_refresh_task(self) -> None:
+        if self._refresh_stop is not None:
+            self._refresh_stop.set()
+        self._refresh_thread = None
+        self._refresh_task = None
+        self._refresh_stop = None
+
+    def close(self) -> None:
+        self._stop_refresh_task()

tests/localthings/test_observe.py

@@ -89,12 +89,15 @@ def test_try_enter_observe_mode_succeeds_when_all_hrefs_notify():
     notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
     notifier.start()
 
-    entered = mgr.try_enter_observe_mode(session, hrefs, grace_period_s=0.02)
-    notifier.join()
+    try:
+        entered = mgr.try_enter_observe_mode(session, hrefs, grace_period_s=0.02)
+        notifier.join()
 
-    assert entered is True
-    assert mgr.mode == 'observe'
-    assert mgr.subscribed_hrefs == set(hrefs)
+        assert entered is True
+        assert mgr.mode == 'observe'
+        assert mgr.subscribed_hrefs == set(hrefs)
+    finally:
+        mgr.close()
 
 
 def test_try_enter_observe_mode_falls_back_when_no_notifies_arrive():
@@ -135,13 +138,16 @@ def test_try_enter_observe_mode_meets_success_fraction_with_partial_notifies():
     notifier = threading.Thread(target=_notify_partial, daemon=True)
     notifier.start()
 
-    entered = mgr.try_enter_observe_mode(
-        session, hrefs, grace_period_s=0.02, success_fraction=0.7,
-    )
-    notifier.join()
+    try:
+        entered = mgr.try_enter_observe_mode(
+            session, hrefs, grace_period_s=0.02, success_fraction=0.7,
+        )
+        notifier.join()
 
-    assert entered is True
-    assert mgr.mode == 'observe'
+        assert entered is True
+        assert mgr.mode == 'observe'
+    finally:
+        mgr.close()
 
 
 def test_on_notification_ignores_malformed_cbor():
@@ -158,37 +164,40 @@ def test_try_enter_observe_mode_clears_stale_notifications_on_retry():
     session = _FakeSession()
     hrefs = ['/a/vs/0', '/b/vs/0']
 
-    # First call: all hrefs notify -> succeeds and enters observe mode
-    def _notify_all_first():
-        time.sleep(0.005)
-        for href in hrefs:
-            mgr.on_notification(href, cbor2.dumps({'x': 1}))
+    try:
+        # First call: all hrefs notify -> succeeds and enters observe mode
+        def _notify_all_first():
+            time.sleep(0.005)
+            for href in hrefs:
+                mgr.on_notification(href, cbor2.dumps({'x': 1}))
 
-    notifier1 = threading.Thread(target=_notify_all_first, daemon=True)
-    notifier1.start()
-    entered = mgr.try_enter_observe_mode(session, hrefs, grace_period_s=0.02)
-    notifier1.join()
-    assert entered is True
-    assert mgr.mode == 'observe'
+        notifier1 = threading.Thread(target=_notify_all_first, daemon=True)
+        notifier1.start()
+        entered = mgr.try_enter_observe_mode(session, hrefs, grace_period_s=0.02)
+        notifier1.join()
+        assert entered is True
+        assert mgr.mode == 'observe'
 
-    # Second call: retry with same hrefs, but only one notifies.
-    # Without the fix (missing self._notified.clear()), the old notifications
-    # would leak in, making this appear successful (2/2 instead of 1/2).
-    # With the fix, it should fail because only 1/2 < 0.8.
-    session.subscribed = []  # reset for clean test
+        # Second call: retry with same hrefs, but only one notifies.
+        # Without the fix (missing self._notified.clear()), the old notifications
+        # would leak in, making this appear successful (2/2 instead of 1/2).
+        # With the fix, it should fail because only 1/2 < 0.8.
+        session.subscribed = []  # reset for clean test
 
-    def _notify_partial_second():
-        time.sleep(0.005)
-        mgr.on_notification(hrefs[0], cbor2.dumps({'x': 2}))  # only one notifies
+        def _notify_partial_second():
+            time.sleep(0.005)
+            mgr.on_notification(hrefs[0], cbor2.dumps({'x': 2}))  # only one notifies
 
-    notifier2 = threading.Thread(target=_notify_partial_second, daemon=True)
-    notifier2.start()
-    entered = mgr.try_enter_observe_mode(session, hrefs, grace_period_s=0.02)
-    notifier2.join()
+        notifier2 = threading.Thread(target=_notify_partial_second, daemon=True)
+        notifier2.start()
+        entered = mgr.try_enter_observe_mode(session, hrefs, grace_period_s=0.02)
+        notifier2.join()
 
-    assert entered is False
-    assert mgr.mode == 'poll'
-    assert mgr.subscribed_hrefs == set()
+        assert entered is False
+        assert mgr.mode == 'poll'
+        assert mgr.subscribed_hrefs == set()
+    finally:
+        mgr.close()
 
 
 def test_check_sweep_for_misses_false_when_not_in_observe_mode():
@@ -208,14 +217,17 @@ def test_check_sweep_for_misses_detects_mismatch():
     notifier = threading.Thread(target=_notify_during_grace_period, daemon=True)
     notifier.start()
 
-    mgr.try_enter_observe_mode(session, [href], grace_period_s=0.02)
-    notifier.join()
-    assert mgr.mode == 'observe'
+    try:
+        mgr.try_enter_observe_mode(session, [href], grace_period_s=0.02)
+        notifier.join()
+        assert mgr.mode == 'observe'
 
-    # Sweep sees a value the cache never got via notify -> observe missed it.
-    missed = mgr.check_sweep_for_misses({href: {'a': 2}})
+        # Sweep sees a value the cache never got via notify -> observe missed it.
+        missed = mgr.check_sweep_for_misses({href: {'a': 2}})
 
-    assert missed is True
+        assert missed is True
+    finally:
+        mgr.close()
 
 
 def test_check_sweep_for_misses_ignores_href_during_settle_window():
@@ -251,3 +263,45 @@ def test_downgrade_to_poll_moves_subscribed_to_fallback():
     assert mgr.mode == 'poll'
     assert mgr.subscribed_hrefs == set()
     assert mgr.fallback_hrefs == {href}
+
+
+def test_start_refresh_task_spawns_a_daemon_thread():
+    mgr = _manager()
+    mgr.subscribed_hrefs = {'/oven/vs/0'}
+    session = _FakeSession()
+
+    mgr.start_refresh_task(session)
+    try:
+        assert mgr._refresh_thread is not None
+        assert mgr._refresh_thread.is_alive()
+        assert mgr._refresh_thread.daemon is True
+    finally:
+        mgr.close()
+
+
+def test_close_stops_the_refresh_thread():
+    mgr = _manager()
+    mgr.subscribed_hrefs = {'/oven/vs/0'}
+    session = _FakeSession()
+    mgr.start_refresh_task(session)
+    thread = mgr._refresh_thread
+
+    mgr.close()
+    thread.join(timeout=2.0)
+
+    assert not thread.is_alive()
+    assert mgr._refresh_thread is None
+
+
+def test_downgrade_to_poll_stops_refresh_task():
+    mgr = _manager()
+    mgr.subscribed_hrefs = {'/oven/vs/0'}
+    session = _FakeSession()
+    mgr.start_refresh_task(session)
+    thread = mgr._refresh_thread
+
+    mgr.downgrade_to_poll()
+    thread.join(timeout=2.0)
+
+    assert not thread.is_alive()
+    assert mgr._refresh_thread is None