diff --git a/src/plugin_system/plugin_state.py b/src/plugin_system/plugin_state.py index 8b52823e..f9b9d7d0 100644 --- a/src/plugin_system/plugin_state.py +++ b/src/plugin_system/plugin_state.py @@ -6,22 +6,38 @@ with state transitions and queries. """ 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 @@ class PluginStateManager: 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 @@ class PluginStateManager: 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 @@ class PluginStateManager: """ 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 @@ class PluginStateManager: 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 @@ class PluginStateManager: 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: diff --git a/test/test_plugin_state_history_retention.py b/test/test_plugin_state_history_retention.py new file mode 100644 index 00000000..a7cf1edf --- /dev/null +++ b/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]}"