refactor: extract ocf/ — fold sensors.index_links into StateCache.index_device_tree

ocf/poll_scheduler.py

@@ -0,0 +1,311 @@
+"""Tiered adaptive polling against a DtlsCoapSession.
+
+Tiers are descriptor-declared (hot/warm/cold + sweep). Per tick:
+each tier whose deadline has passed polls all its paths sequentially
+on the shared session, writing into the StateCache. The sweep tier
+issues one Block2 GET of /device/0 and uses index_links to fan its
+result into many href reps.
+
+Adaptive cadence: when descriptor.is_active(cache.links) returns True
+and tier.active_interval_s is set, that tier uses the tighter cadence.
+If the previous health window saw `active_throttle_threshold` timeouts,
+the throttle drops back to idle cadence even when active=True — the
+RT-OCF stack wedges under load and stacking poll attempts only makes
+it worse.
+
+Per-tier timeouts: tier.timeout_s overrides the scheduler default. Hot
+tiers want a tight ceiling (e.g. 2s) so one wedged path can't eat
+several poll cycles.
+
+Cooldown on timeout: when a path times out, it's deferred for ~3 of
+its tier's intervals (clamped 5–60s) via the same _defer_until mechanism
+used for write_in_progress. Breaks the cluster-cascade where the next
+tier tick fires immediately after an 8s wedge and reattempts the same
+stalled path.
+
+Post-write defer: bridge calls write_in_progress(href) before POSTing
+a write; the scheduler skips that href for settle_s to avoid Samsung's
+fetchback-revert bug.
+"""
+from __future__ import annotations
+
+import threading
+import time
+from dataclasses import dataclass
+from typing import Callable, Optional, TYPE_CHECKING
+
+import cbor2
+
+from protocol.dtls_session import DtlsCoapSession, fmt_code
+
+if TYPE_CHECKING:
+    from .state_cache import StateCache
+
+
+@dataclass(frozen=True)
+class PollTier:
+    name: str
+    interval_s: float
+    paths: tuple[tuple[str, ...], ...]
+    active_interval_s: Optional[float] = None
+    is_sweep: bool = False
+    # Per-tier CoAP request timeout. Falls back to PollScheduler.timeout_s
+    # when None. Hot tiers want a tight ceiling (e.g. 2s) so one wedged
+    # path can't eat several poll cycles; sweep tiers tolerate longer.
+    timeout_s: Optional[float] = None
+
+
+class PollScheduler:
+
+    def __init__(self,
+                 session: DtlsCoapSession,
+                 cache: 'StateCache',
+                 tiers: list[PollTier],
+                 is_active_fn: Optional[Callable[[dict[str, dict]], bool]] = None,
+                 logger=None,
+                 timeout_s: float = 8.0,
+                 active_throttle_timeout_threshold: int = 3):
+        self.session = session
+        self.cache = cache
+        self.tiers = tiers
+        self.is_active_fn = is_active_fn
+        self.log = logger
+        self.timeout_s = timeout_s
+        self.active_throttle_threshold = active_throttle_timeout_threshold
+
+        now = time.monotonic()
+        self._next_due: dict[str, float] = {t.name: now for t in tiers}
+        self._defer_until: dict[str, float] = {}
+        self._defer_lock = threading.Lock()
+        self._poll_count = 0
+        self._poll_error_count = 0
+        self._last_active: Optional[bool] = None
+        self._last_throttled: bool = False
+        # Real-liveness signal for KeepaliveTask: updated on every 2.05
+        # we receive from the wire. Initialized to "now" so the first
+        # keepalive tick after start doesn't fire a false unreachable.
+        self._last_success_ts: float = now
+
+        # Per-window tail-latency tracking. Bridge consumes-and-resets
+        # these via take_window_stats() once per HEALTH_INTERVAL_S.
+        # _window_max_rtt_ms tracks SUCCESSFUL polls only — timeouts go
+        # into _window_timeout_count so the dashboard sees the real tail
+        # instead of an 8000ms wall.
+        self._stats_lock = threading.Lock()
+        self._window_max_rtt_ms = 0.0
+        self._window_slow_count = 0
+        self._window_timeout_count = 0
+        # Snapshot of the previous window's timeout count, used by the
+        # active-window throttle to back off when polls are wedging.
+        self._last_window_timeouts = 0
+        self.slow_threshold_ms = 1000.0
+
+    def write_in_progress(self, href: str, settle_s: float = 4.0) -> None:
+        with self._defer_lock:
+            self._defer_until[href] = time.monotonic() + settle_s
+
+    def run_forever(self, stop: threading.Event) -> None:
+        while not stop.is_set():
+            self._run_due_tiers()
+            sleep_for = max(0.05, min(1.0, self._earliest_deadline() - time.monotonic()))
+            if stop.wait(sleep_for):
+                return
+
+    @property
+    def poll_count(self) -> int:
+        return self._poll_count
+
+    @property
+    def poll_error_count(self) -> int:
+        return self._poll_error_count
+
+    @property
+    def last_success_ts(self) -> float:
+        """Monotonic timestamp of the most recent 2.05 response from any
+        tier. KeepaliveTask uses this as its half-open-detection signal:
+        if no 2.05 has landed in `liveness_window_s`, the session is
+        wedged regardless of whether ping sends succeed."""
+        return self._last_success_ts
+
+    def take_window_stats(self) -> tuple[float, int, int]:
+        """Return (max RTT ms over successful polls, slow-poll count,
+        timeout count) seen since the last call, and reset all three.
+        Slow threshold is `self.slow_threshold_ms`. The timeout count
+        is snapshotted into `_last_window_timeouts` for the throttle."""
+        with self._stats_lock:
+            out = (self._window_max_rtt_ms,
+                   self._window_slow_count,
+                   self._window_timeout_count)
+            self._last_window_timeouts = self._window_timeout_count
+            self._window_max_rtt_ms = 0.0
+            self._window_slow_count = 0
+            self._window_timeout_count = 0
+        return out
+
+    def _record_rtt(self, rtt_ms: float, *, timed_out: bool = False) -> None:
+        with self._stats_lock:
+            if timed_out:
+                self._window_timeout_count += 1
+                return
+            if rtt_ms > self._window_max_rtt_ms:
+                self._window_max_rtt_ms = rtt_ms
+            if rtt_ms >= self.slow_threshold_ms:
+                self._window_slow_count += 1
+
+    def _earliest_deadline(self) -> float:
+        return min(self._next_due.values())
+
+    def _run_due_tiers(self) -> None:
+        now = time.monotonic()
+        active = False
+        if self.is_active_fn is not None:
+            try:
+                active = bool(self.is_active_fn(self.cache.snapshot()))
+            except Exception as e:
+                if self.log: self.log.warning("is_active: %s", e)
+        if active != self._last_active:
+            if self.log and self._last_active is not None:
+                self.log.info("active=%s", active)
+            self._last_active = active
+        # Active-window throttle: if the previous health window saw a
+        # cluster of timeouts (RT-OCF wedging under load), drop back to
+        # idle cadence even when is_active=True. Lets the device breathe
+        # instead of stacking poll attempts on a stalled responder.
+        with self._stats_lock:
+            recent_to = self._last_window_timeouts
+        throttled = (active
+                     and recent_to >= self.active_throttle_threshold)
+        if throttled != self._last_throttled:
+            if self.log:
+                if throttled:
+                    self.log.warning(
+                        "active-throttle ON (%d timeouts last window) — "
+                        "using idle cadence", recent_to)
+                else:
+                    self.log.info("active-throttle OFF")
+            self._last_throttled = throttled
+        effective_active = active and not throttled
+        for tier in self.tiers:
+            if self._next_due[tier.name] > now:
+                continue
+            interval = (tier.active_interval_s
+                        if (effective_active and tier.active_interval_s is not None)
+                        else tier.interval_s)
+            self._next_due[tier.name] = now + interval
+            try:
+                if tier.is_sweep:
+                    self._do_sweep(tier)
+                else:
+                    self._do_tier(tier)
+            except Exception as e:
+                self._poll_error_count += 1
+                if self.log: self.log.warning("tier %s: %s", tier.name, e)
+
+    def _tier_timeout(self, tier: PollTier) -> float:
+        return tier.timeout_s if tier.timeout_s is not None else self.timeout_s
+
+    def _cooldown_for(self, tier: PollTier) -> float:
+        # On timeout, defer the wedged href for ~3 cycles (clamped to a
+        # 5–60s band) so repeated tier ticks don't stack attempts on a
+        # stalled responder. Breaks the cluster-cascade we see in the
+        # Poll Max RTT chart during heavy device use.
+        return max(5.0, min(60.0, tier.interval_s * 3.0))
+
+    def _set_cooldown(self, href: str, cooldown_s: float) -> None:
+        with self._defer_lock:
+            self._defer_until[href] = time.monotonic() + cooldown_s
+
+    def _do_tier(self, tier: PollTier) -> None:
+        timeout = self._tier_timeout(tier)
+        cooldown = self._cooldown_for(tier)
+        for path in tier.paths:
+            href = '/' + '/'.join(path)
+            with self._defer_lock:
+                if self._defer_until.get(href, 0) > time.monotonic():
+                    continue
+            self._poll_count += 1
+            t0 = time.monotonic()
+            try:
+                code, body = self.session.get(list(path), timeout=timeout)
+            except TimeoutError:
+                self._poll_error_count += 1
+                self._record_rtt(0.0, timed_out=True)
+                self._set_cooldown(href, cooldown)
+                if self.log:
+                    self.log.warning("poll %s timeout (cooldown %.0fs)",
+                                     href, cooldown)
+                return
+            except ConnectionError as e:
+                self._poll_error_count += 1
+                self._record_rtt((time.monotonic() - t0) * 1000.0)
+                if self.log: self.log.debug("poll %s: %s", href, e)
+                return
+            except Exception as e:
+                self._poll_error_count += 1
+                self._record_rtt((time.monotonic() - t0) * 1000.0)
+                if self.log: self.log.warning("poll %s: %s", href, e)
+                return
+            self._record_rtt((time.monotonic() - t0) * 1000.0)
+            if code != 0x45 or not body:
+                self._poll_error_count += 1
+                if self.log: self.log.warning("poll %s -> %s", href, fmt_code(code))
+                continue
+            self._last_success_ts = time.monotonic()
+            try:
+                rep = cbor2.loads(body)
+            except Exception as e:
+                self._poll_error_count += 1
+                if self.log: self.log.warning("poll %s cbor: %s", href, e)
+                continue
+            if isinstance(rep, dict):
+                self.cache.apply_rep(href, rep, source='poll')
+
+    def _do_sweep(self, tier: PollTier) -> None:
+        timeout = self._tier_timeout(tier)
+        cooldown = self._cooldown_for(tier)
+        path = list(tier.paths[0])
+        href = '/' + '/'.join(path)
+        t0 = time.monotonic()
+        self._poll_count += 1
+        try:
+            code, body = self.session.get(path, timeout=timeout)
+        except TimeoutError:
+            self._poll_error_count += 1
+            self._record_rtt(0.0, timed_out=True)
+            self._set_cooldown(href, cooldown)
+            if self.log:
+                self.log.warning("sweep %s timeout (cooldown %.0fs)",
+                                 path, cooldown)
+            return
+        except ConnectionError as e:
+            self._poll_error_count += 1
+            self._record_rtt((time.monotonic() - t0) * 1000.0)
+            if self.log: self.log.debug("sweep %s: %s", path, e)
+            return
+        except Exception as e:
+            self._poll_error_count += 1
+            self._record_rtt((time.monotonic() - t0) * 1000.0)
+            if self.log: self.log.warning("sweep %s: %s", path, e)
+            return
+        self._record_rtt((time.monotonic() - t0) * 1000.0)
+        if code != 0x45 or not body:
+            self._poll_error_count += 1
+            if self.log: self.log.warning("sweep -> %s", fmt_code(code))
+            return
+        self._last_success_ts = time.monotonic()
+        try:
+            tree = cbor2.loads(body)
+        except Exception as e:
+            self._poll_error_count += 1
+            if self.log: self.log.warning("sweep cbor: %s", e)
+            return
+        indexed = self.cache.index_device_tree(tree)
+        for href, rep in indexed.items():
+            with self._defer_lock:
+                if self._defer_until.get(href, 0) > time.monotonic():
+                    continue
+            self.cache.apply_rep(href, rep, source='sweep')
+        if self.log:
+            elapsed_ms = (time.monotonic() - t0) * 1000.0
+            self.log.info("sweep complete (%d links, %.0fms)",
+                          len(indexed), elapsed_ms)