fix(plugins): retain state history by age, with the count as a ceiling

src/plugin_system/plugin_state.py

@@ -6,22 +6,38 @@
 """
 
 import threading
+import time
 from collections import deque
 from enum import Enum
-from typing import Optional, Dict, Any, Deque, List
+from typing import Optional, Dict, Any, Deque, List, Tuple
 from datetime import datetime
 import logging
 
 from src.logging_config import get_logger
 
 
-# Transitions retained per plugin. The history is diagnostic only -- nothing
-# reads the entries themselves, just their count -- but it is appended to on the
-# hot scheduling path: every update cycle records RUNNING on reserve and ENABLED
-# on finish. Unbounded, that is 2,880 entries per plugin per day at the default
-# 60s interval, which on a 1 GB Pi exhausts memory in weeks. Keep the recent
-# tail for debugging and let the rest age out.
-MAX_STATE_HISTORY_PER_PLUGIN = 200
+# The history is diagnostic only -- nothing reads the entries themselves, just
+# their count -- but it is appended to on the hot scheduling path: every update
+# cycle records RUNNING on reserve and ENABLED on finish. Unbounded, that is
+# 2,880 entries per plugin per day at the default 60s interval, which on a 1 GB
+# Pi exhausts memory in weeks.
+#
+# Two limits, because a single entry count answers the wrong question. What a
+# reader wants is "the last couple of hours", and how many transitions that is
+# depends entirely on the plugin's update interval -- which on a real board
+# spans 2s to 3600s. A flat 200 entries is 4.2 days for the slowest plugin and
+# 3.3 minutes for the fastest, so the plugin churning hardest, the one worth
+# looking at, keeps the least history.
+#
+# So: trim by AGE first, which makes the retained window comparable across
+# plugins whatever their cadence...
+STATE_HISTORY_MAX_AGE_SECONDS = 2 * 60 * 60
+
+# ...and cap by COUNT second, purely as a memory ceiling for the fast pollers
+# whose age window would otherwise run to thousands of entries. At ~230 bytes
+# an entry this is ~0.5 MB per plugin worst case, and only plugins updating
+# faster than roughly every 4s can reach it.
+MAX_STATE_HISTORY_PER_PLUGIN = 2000
 
 
 class PluginState(Enum):
@@ -47,7 +63,10 @@ def __init__(self, logger: Optional[logging.Logger] = None) -> None:
         self.logger = logger or get_logger(__name__)
         self._lock = threading.RLock()
         self._states: Dict[str, PluginState] = {}
-        self._state_history: Dict[str, Deque[Dict[str, Any]]] = {}
+        # (monotonic timestamp, transition). The clock is monotonic so a DST
+        # shift or an NTP step cannot make entries look old and flush the
+        # history; the human-readable timestamp lives inside the transition.
+        self._state_history: Dict[str, Deque[Tuple[float, Dict[str, Any]]]] = {}
         # Lifetime transition totals, kept separately so the count reported by
         # get_state_info() stays truthful once the history above starts rolling.
         self._state_transition_counts: Dict[str, int] = {}
@@ -70,7 +89,13 @@ def _record_transition(
         if history is None:
             history = deque(maxlen=MAX_STATE_HISTORY_PER_PLUGIN)
             self._state_history[plugin_id] = history
-        history.append(transition)
+        now = time.monotonic()
+        history.append((now, transition))
+        # Age out first; the deque's maxlen is the backstop for plugins that
+        # produce more than the ceiling within the window.
+        cutoff = now - STATE_HISTORY_MAX_AGE_SECONDS
+        while history and history[0][0] < cutoff:
+            history.popleft()
         self._state_transition_counts[plugin_id] = (
             self._state_transition_counts.get(plugin_id, 0) + 1
         )
@@ -160,8 +185,10 @@ def get_state_history(self, plugin_id: str) -> List[Dict[str, Any]]:
         """
         Get state transition history for a plugin.
 
-        Only the most recent MAX_STATE_HISTORY_PER_PLUGIN transitions are
-        retained; older ones age out.
+        Retention is by age first -- transitions older than
+        STATE_HISTORY_MAX_AGE_SECONDS are dropped -- and by count second, at
+        MAX_STATE_HISTORY_PER_PLUGIN, which only binds for plugins updating
+        fast enough to exceed it inside that window.
 
         Args:
             plugin_id: Plugin identifier
@@ -175,7 +202,7 @@ def get_state_history(self, plugin_id: str) -> List[Dict[str, Any]]:
         with self._lock:
             return [
                 dict(transition)
-                for transition in self._state_history.get(plugin_id, ())
+                for _stamp, transition in self._state_history.get(plugin_id, ())
             ]
 
     def set_error_info(self, plugin_id: str, error_info: Dict[str, Any]) -> None:
@@ -279,19 +306,26 @@ def get_state_info(self, plugin_id: str) -> Dict[str, Any]:
         Returns:
             Dictionary with state information
         """
-        state = self.get_state(plugin_id)
-        info = {
-            'state': state.value,
-            'is_loaded': self.is_loaded(plugin_id),
-            'is_enabled': self.is_enabled(plugin_id),
-            'is_running': self.is_running(plugin_id),
-            'is_error': self.is_error(plugin_id),
-            'can_execute': self.can_execute(plugin_id),
-            'last_update': self.get_last_update(plugin_id),
-            'last_display': self.get_last_display(plugin_id),
-            'error_info': self.get_error_info(plugin_id),
-            'state_history_count': self._state_transition_counts.get(plugin_id, 0)
-        }
+        # One snapshot, one critical section. Each field was read under its own
+        # lock, so an unload running concurrently could be observed half-done:
+        # 'state' read before clear_state() removed it and
+        # 'state_history_count' read after, giving a caller a plugin that is
+        # ENABLED with zero transitions. _lock is an RLock, so the helpers
+        # below can still take it.
+        with self._lock:
+            state = self.get_state(plugin_id)
+            info = {
+                'state': state.value,
+                'is_loaded': self.is_loaded(plugin_id),
+                'is_enabled': self.is_enabled(plugin_id),
+                'is_running': self.is_running(plugin_id),
+                'is_error': self.is_error(plugin_id),
+                'can_execute': self.can_execute(plugin_id),
+                'last_update': self.get_last_update(plugin_id),
+                'last_display': self.get_last_display(plugin_id),
+                'error_info': self.get_error_info(plugin_id),
+                'state_history_count': self._state_transition_counts.get(plugin_id, 0)
+            }
         return info
 
     def clear_state(self, plugin_id: str) -> None:

test/test_plugin_state_history_retention.py

@@ -0,0 +1,209 @@
+"""Retention is bounded by age first and by count second.
+
+The cap added in the parent change is a flat entry count, and an entry count
+answers the wrong question. What a reader wants from this history is "the last
+couple of hours"; how many transitions that is depends entirely on the
+plugin's update interval, which on a real board spans 2s to 3600s. A flat 200
+entries is 4.2 days of history for the slowest plugin and 3.3 minutes for the
+fastest -- so the plugin churning hardest, the one actually worth looking at,
+keeps the least.
+
+Trimming by age makes the retained window comparable whatever the cadence, and
+the count then serves only as a memory ceiling for pollers fast enough to
+produce thousands of transitions inside that window.
+"""
+
+import time
+import pytest
+
+from src.plugin_system.plugin_state import (
+    PluginState,
+    PluginStateManager,
+    MAX_STATE_HISTORY_PER_PLUGIN,
+    STATE_HISTORY_MAX_AGE_SECONDS,
+)
+
+
+class FakeClock:
+    """A monotonic clock the test drives, so no test has to sleep."""
+
+    def __init__(self):
+        self.t = 1000.0
+
+    def __call__(self):
+        return self.t
+
+    def advance(self, seconds):
+        self.t += seconds
+
+
+@pytest.fixture
+def clock(monkeypatch):
+    c = FakeClock()
+    monkeypatch.setattr("src.plugin_system.plugin_state.time.monotonic", c)
+    return c
+
+
+def _cycle(manager, plugin_id, clock, interval, cycles):
+    """One update cycle: RUNNING on reserve, ENABLED on finish."""
+    for _ in range(cycles):
+        manager.set_state(plugin_id, PluginState.RUNNING)
+        manager.set_state(plugin_id, PluginState.ENABLED)
+        clock.advance(interval)
+
+
+def test_transitions_older_than_the_window_are_dropped(clock):
+    m = PluginStateManager()
+    _cycle(m, "clock", clock, interval=60, cycles=10)
+    assert len(m.get_state_history("clock")) == 20
+
+    # Nothing happens for longer than the window, then one more cycle.
+    clock.advance(STATE_HISTORY_MAX_AGE_SECONDS + 1)
+    _cycle(m, "clock", clock, interval=60, cycles=1)
+
+    assert len(m.get_state_history("clock")) == 2, (
+        "only the transitions inside the window should survive")
+
+
+def test_every_plugin_keeps_the_same_WINDOW_not_the_same_COUNT(clock):
+    """The point of the age policy, stated as the property that distinguishes it.
+
+    Run both plugins for three times the retention window. Under a flat count
+    cap the slow one would still be holding transitions from hours before the
+    window, because it never produces enough entries to evict them. Under the
+    age policy each plugin retains its own last two hours and no more --
+    different entry counts, same span of time.
+    """
+    window = STATE_HISTORY_MAX_AGE_SECONDS
+    m = PluginStateManager()
+
+    _cycle(m, "slow", clock, interval=60, cycles=(3 * window) // 60)
+    slow = len(m.get_state_history("slow"))
+
+    # Assert the property directly rather than a derived count. The guarantee
+    # is about the SPAN of retained history, not its age against the current
+    # clock: trimming happens on append, so a plugin that has gone quiet keeps
+    # its last window until it writes again. That is intentional -- it is
+    # bounded either way, and a lazy trim costs nothing on the hot path.
+    stamps = [stamp for stamp, _ in m._state_history["slow"]]
+    assert stamps[-1] - stamps[0] <= window, (
+        f"retained history spans {stamps[-1] - stamps[0]:.0f}s, "
+        f"window is {window}s")
+    assert slow < 2 * ((3 * window) // 60), (
+        f"slow plugin kept {slow} entries -- three windows' worth was retained")
+
+    clock.t = 1000.0
+    _cycle(m, "fast", clock, interval=2, cycles=(3 * window) // 2)
+    fast = len(m.get_state_history("fast"))
+
+    # Different counts, and the fast poller keeps more of them -- under a flat
+    # count cap these would be equal and the fast one would cover minutes.
+    assert fast > slow, f"fast={fast} slow={slow}"
+
+
+def test_the_count_ceiling_still_bounds_a_fast_poller(clock):
+    """Age alone would let a 2s plugin hold 7,200 entries."""
+    m = PluginStateManager()
+    _cycle(m, "flights", clock, interval=2, cycles=STATE_HISTORY_MAX_AGE_SECONDS)
+    assert len(m.get_state_history("flights")) <= MAX_STATE_HISTORY_PER_PLUGIN
+
+
+def test_a_burst_inside_the_window_is_capped_not_kept(clock):
+    """Transitions with no time between them still cannot grow without bound."""
+    m = PluginStateManager()
+    for _ in range(MAX_STATE_HISTORY_PER_PLUGIN * 3):
+        m.set_state("flapping", PluginState.RUNNING)   # clock never advances
+    assert len(m.get_state_history("flapping")) <= MAX_STATE_HISTORY_PER_PLUGIN
+
+
+def test_ageing_out_does_not_disturb_the_lifetime_count(clock):
+    m = PluginStateManager()
+    _cycle(m, "clock", clock, interval=60, cycles=10)
+    clock.advance(STATE_HISTORY_MAX_AGE_SECONDS + 1)
+    _cycle(m, "clock", clock, interval=60, cycles=1)
+
+    assert len(m.get_state_history("clock")) == 2
+    assert m.get_state_info("clock")["state_history_count"] == 22, (
+        "the lifetime total must survive trimming, it is the flap signal")
+
+
+def test_the_surviving_entries_are_the_recent_ones(clock):
+    m = PluginStateManager()
+    _cycle(m, "clock", clock, interval=60, cycles=5)
+    clock.advance(STATE_HISTORY_MAX_AGE_SECONDS + 1)
+    m.set_state("clock", PluginState.ERROR)
+
+    history = m.get_state_history("clock")
+    assert [h["to"] for h in history] == ["error"]
+
+
+def test_a_monotonic_clock_is_used_not_the_wall_clock(clock):
+    """A DST shift or NTP step must not flush the history.
+
+    The trim reads time.monotonic(); the human-readable datetime inside each
+    transition is for display only.
+    """
+    m = PluginStateManager()
+    _cycle(m, "clock", clock, interval=60, cycles=3)
+    before = len(m.get_state_history("clock"))
+
+    import datetime as real_datetime
+
+    class ShiftedDatetime(real_datetime.datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return real_datetime.datetime(1999, 1, 1)   # clock jumps backwards
+
+    import src.plugin_system.plugin_state as ps
+    original = ps.datetime
+    ps.datetime = ShiftedDatetime
+    try:
+        m.set_state("clock", PluginState.ENABLED)
+    finally:
+        ps.datetime = original
+
+    assert len(m.get_state_history("clock")) == before + 1, (
+        "a wall-clock jump must not trim anything")
+
+
+def test_get_state_info_is_a_consistent_snapshot():
+    """An unload running concurrently must not be observed half-done.
+
+    Each field used to be read under its own lock, so clear_state() could
+    interleave: 'state' read before the removal, 'state_history_count' after,
+    handing a caller a plugin that is ENABLED with zero transitions. The whole
+    payload is now built in one critical section.
+    """
+    import threading
+
+    m = PluginStateManager()
+    for _ in range(50):
+        m.set_state("clock", PluginState.RUNNING)
+        m.set_state("clock", PluginState.ENABLED)
+
+    inconsistent = []
+    stop = threading.Event()
+
+    def reader():
+        while not stop.is_set():
+            info = m.get_state_info("clock")
+            # Either fully present or fully cleared -- never a live state with
+            # a wiped count.
+            if info["state"] != PluginState.UNLOADED.value and \
+                    info["state_history_count"] == 0:
+                inconsistent.append(info)
+                return
+
+    def clearer():
+        for _ in range(200):
+            for _ in range(20):
+                m.set_state("clock", PluginState.ENABLED)
+            m.clear_state("clock")
+
+    t = threading.Thread(target=reader, daemon=True)
+    t.start()
+    clearer()
+    stop.set()
+    t.join(timeout=5)
+
+    assert not inconsistent, f"observed a torn snapshot: {inconsistent[:1]}"