Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 66 additions & 21 deletions src/plugin_system/plugin_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,14 +6,24 @@
"""

import threading
from collections import deque
from enum import Enum
from typing import Optional, Dict, Any
from typing import Optional, Dict, Any, Deque, List
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


class PluginState(Enum):
"""Plugin state enumeration."""
UNLOADED = "unloaded" # Plugin not loaded
Expand All @@ -37,11 +47,34 @@ 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, list] = {}
self._state_history: Dict[str, Deque[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] = {}
self._error_info: Dict[str, Dict[str, Any]] = {}
self._last_update: Dict[str, datetime] = {}
self._last_display: Dict[str, datetime] = {}

def _record_transition(
self,
plugin_id: str,
transition: Dict[str, Any]
) -> None:
"""Append a transition to the plugin's bounded history.

Callers must already hold ``_lock``. The deque discards its oldest
entry once it is full, so the history cannot grow without bound; the
lifetime total is tracked separately for get_state_info().
"""
history = self._state_history.get(plugin_id)
if history is None:
history = deque(maxlen=MAX_STATE_HISTORY_PER_PLUGIN)
self._state_history[plugin_id] = history
history.append(transition)
self._state_transition_counts[plugin_id] = (
self._state_transition_counts.get(plugin_id, 0) + 1
)

def set_state(
self,
plugin_id: str,
Expand All @@ -60,16 +93,13 @@ def set_state(
old_state = self._states.get(plugin_id, PluginState.UNLOADED)
self._states[plugin_id] = state

if plugin_id not in self._state_history:
self._state_history[plugin_id] = []

transition = {
'timestamp': datetime.now(),
'from': old_state.value,
'to': state.value,
'error': str(error) if error else None
}
self._state_history[plugin_id].append(transition)
self._record_transition(plugin_id, transition)

# Store error info if transitioning to ERROR state
if state == PluginState.ERROR and error:
Expand Down Expand Up @@ -126,17 +156,27 @@ def can_execute(self, plugin_id: str) -> bool:
state = self.get_state(plugin_id)
return state == PluginState.ENABLED

def get_state_history(self, plugin_id: str) -> list:
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.

Args:
plugin_id: Plugin identifier

Returns:
List of state transitions
List of recent state transitions, oldest first. Both the list and
the transition dicts are copies, so callers cannot mutate the
manager's own history. The values inside a transition are all
immutable, so a shallow copy per entry is enough.
"""
return self._state_history.get(plugin_id, [])
with self._lock:
return [
dict(transition)
for transition in self._state_history.get(plugin_id, ())
]

def set_error_info(self, plugin_id: str, error_info: Dict[str, Any]) -> None:
"""
Expand Down Expand Up @@ -179,9 +219,7 @@ def set_state_with_error(
old_state = self._states.get(plugin_id, PluginState.UNLOADED)
self._states[plugin_id] = state

if plugin_id not in self._state_history:
self._state_history[plugin_id] = []
self._state_history[plugin_id].append({
self._record_transition(plugin_id, {
'timestamp': datetime.now(),
'from': old_state.value,
'to': state.value,
Expand Down Expand Up @@ -252,15 +290,22 @@ def get_state_info(self, plugin_id: str) -> Dict[str, Any]:
'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': len(self.get_state_history(plugin_id))
'state_history_count': self._state_transition_counts.get(plugin_id, 0)
}
return info

def clear_state(self, plugin_id: str) -> None:
"""Clear all state information for a plugin."""
self._states.pop(plugin_id, None)
self._state_history.pop(plugin_id, None)
self._error_info.pop(plugin_id, None)
self._last_update.pop(plugin_id, None)
self._last_display.pop(plugin_id, None)
"""Clear all state information for a plugin.

Held under ``_lock`` so the five dicts are dropped as one unit: every
other mutator takes the lock, and without it a concurrent set_state()
could interleave and leave a plugin with history but no state.
"""
with self._lock:
self._states.pop(plugin_id, None)
self._state_history.pop(plugin_id, None)
self._state_transition_counts.pop(plugin_id, None)
self._error_info.pop(plugin_id, None)
self._last_update.pop(plugin_id, None)
self._last_display.pop(plugin_id, None)

166 changes: 166 additions & 0 deletions test/test_plugin_state_history_cap.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,166 @@
"""Plugin state history must not grow without bound.

`PluginStateManager` recorded every state transition in a per-plugin list and
never trimmed it. The only code that removed entries was `clear_state()`, called
solely from `PluginManager.unload_plugin()`, so a plugin that stays loaded --
i.e. normal operation -- never released a single entry.

The list is written on the hot scheduling path. Every update cycle appends
twice: `_reserve_for_update()` sets RUNNING and `_finish()` sets ENABLED back
again. At the default 60-second update interval that is 2,880 entries per
plugin per day, and nothing ever reads the entries -- `get_state_info()` only
takes their `len()`. It is pure dead weight.

Measured against the unpatched class, ten plugins on a 60s interval retain
864,010 transitions after thirty simulated days, for 231 MB of heap. On a 1 GB
Pi that is fatal on its own, and the failure is not a clean OOM: once
MemAvailable falls far enough, fork() starts returning ENOMEM, so sshd accepts
connections and closes them before its banner while the kernel still answers
pings. The board looks like a hardware fault and needs a power cycle.

These tests pin the cap, the retention order, and the one piece of behaviour the
cap must not change: `state_history_count` is surfaced through the web API, so
it has to keep reporting the lifetime total rather than plateauing at the cap.
"""

import os
import sys

import pytest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from src.plugin_system.plugin_state import ( # noqa: E402
MAX_STATE_HISTORY_PER_PLUGIN,
PluginState,
PluginStateManager,
)


def _cycle_updates(manager, plugin_id, cycles):
"""Drive the real scheduling path: RUNNING on reserve, ENABLED on finish."""
for _ in range(cycles):
manager.set_state(plugin_id, PluginState.RUNNING)
manager.set_state(plugin_id, PluginState.ENABLED)


def test_state_history_is_capped():
"""A day of updates must not retain a day of transitions."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)

# One simulated day at the default 60s update interval.
_cycle_updates(manager, "clock", 1440)

history = manager.get_state_history("clock")
assert len(history) <= MAX_STATE_HISTORY_PER_PLUGIN, (
f"history grew to {len(history)} entries; it is never trimmed"
)


def test_state_history_keeps_the_most_recent_transitions():
"""Trimming drops the oldest entries, not the newest."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
_cycle_updates(manager, "clock", MAX_STATE_HISTORY_PER_PLUGIN)

history = manager.get_state_history("clock")

# The scheduling cycle ends on ENABLED, so the newest entry is the
# RUNNING -> ENABLED half of the last cycle.
assert history[-1]["from"] == PluginState.RUNNING.value
assert history[-1]["to"] == PluginState.ENABLED.value

# And the very first ENABLED transition has aged out.
assert history[0]["from"] != PluginState.UNLOADED.value


def test_state_history_count_reports_lifetime_total():
"""The count exposed through the API must not plateau at the cap.

`get_state_info()['state_history_count']` is surfaced by the web UI. Capping
the retained list must not turn it into "entries we happen to still hold".
"""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
total = 1

cycles = MAX_STATE_HISTORY_PER_PLUGIN * 2
_cycle_updates(manager, "clock", cycles)
total += cycles * 2

info = manager.get_state_info("clock")
assert info["state_history_count"] == total
assert len(manager.get_state_history("clock")) <= MAX_STATE_HISTORY_PER_PLUGIN


def test_error_transitions_are_capped_too():
"""set_state_with_error() appends to the same list and needs the same cap."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)

for _ in range(MAX_STATE_HISTORY_PER_PLUGIN * 2):
manager.set_state_with_error(
"clock",
PluginState.ENABLED,
{"reason": "update timeout"},
error=RuntimeError("boom"),
)

assert len(manager.get_state_history("clock")) <= MAX_STATE_HISTORY_PER_PLUGIN


def test_history_is_isolated_per_plugin():
"""The cap is per plugin, not shared across the manager."""
manager = PluginStateManager()
for plugin_id in ("clock", "weather"):
manager.set_state(plugin_id, PluginState.ENABLED)
_cycle_updates(manager, plugin_id, 50)

assert len(manager.get_state_history("clock")) == 101
assert len(manager.get_state_history("weather")) == 101


def test_get_state_history_returns_a_copy():
"""Callers must not be able to mutate the manager's internal history."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)

history = manager.get_state_history("clock")
history.clear()

assert len(manager.get_state_history("clock")) == 1


def test_get_state_history_entries_are_copies():
"""Copying the outer list is not enough -- the entries are handed out too.

A caller holding a returned transition must not be able to rewrite the
manager's record of what happened.
"""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)

entry = manager.get_state_history("clock")[0]
entry["to"] = "tampered"
entry["error"] = "injected"

stored = manager.get_state_history("clock")[0]
assert stored["to"] == PluginState.ENABLED.value
assert stored["error"] is None


def test_clear_state_drops_history():
"""Unloading a plugin still releases everything it accumulated."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
_cycle_updates(manager, "clock", 10)

manager.clear_state("clock")

assert manager.get_state_history("clock") == []
assert manager.get_state_info("clock")["state_history_count"] == 0


if __name__ == "__main__":
sys.exit(pytest.main([__file__, "-v"]))
Loading