diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 0000c47f..2c29fd44 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -80,4 +80,6 @@ jobs: test/test_version_consistency.py \ test/test_plugin_compatibility_gate.py \ test/test_install_preserves_existing.py \ - test/test_core_owned_config_keys.py + test/test_core_owned_config_keys.py \ + test/test_async_plugin_updates.py \ + test/test_plugin_update_reservation.py diff --git a/src/plugin_system/plugin_manager.py b/src/plugin_system/plugin_manager.py index 50d6419e..8bfc13ab 100644 --- a/src/plugin_system/plugin_manager.py +++ b/src/plugin_system/plugin_manager.py @@ -113,6 +113,15 @@ class PluginManager: self._update_queue: "queue.Queue[Optional[Tuple[str, float]]]" = queue.Queue() self._pending_updates: set = set() self._pending_lock = threading.Lock() + # Serializes the "is this plugin eligible?" -> "claim it (RUNNING)" + # transition. Two schedulers run concurrently in practice — the render + # loop's _tick_plugin_updates() and Vegas mode's vegas-plugin-tick + # daemon thread, which is never joined — so without this both can + # observe ENABLED and both call update() on the same plugin. Held only + # across the check and the state transition, never across update() + # itself: that would serialize slow plugins behind each other and + # reintroduce the stall the async worker exists to avoid. + self._reservation_lock = threading.Lock() self._plugin_locks: Dict[str, threading.Lock] = {} self._plugin_locks_guard = threading.Lock() self._update_worker: Optional[threading.Thread] = None @@ -808,25 +817,69 @@ class PluginManager: if self.health_tracker and self.health_tracker.should_skip_plugin(plugin_id): continue - # Check if plugin can execute - if not self.state_manager.can_execute(plugin_id): - continue - interval = self._get_plugin_update_interval(plugin_id, plugin_instance) if interval is None: continue - with self._plugin_last_update_lock: - last_update = self.plugin_last_update.get(plugin_id, 0.0) + # Eligibility check, due check and the RUNNING transition happen + # together, so a concurrent scheduler cannot claim the same plugin. + if not self._reserve_for_update(plugin_id, current_time, interval): + continue - if last_update == 0.0 or (current_time - last_update) >= interval: - if self._synchronous_updates: - # Kill-switch path: the original inline execution - # (blocks the caller until update() completes/times out) - self.state_manager.set_state(plugin_id, PluginState.RUNNING) - self._execute_update_now(plugin_id, plugin_instance, current_time) - else: - self._enqueue_update(plugin_id, current_time) + if self._synchronous_updates: + # Kill-switch path: the original inline execution + # (blocks the caller until update() completes/times out) + self._execute_update_now(plugin_id, plugin_instance, current_time) + else: + self._enqueue_update(plugin_id, current_time) + + def _reserve_for_update( + self, + plugin_id: str, + current_time: Optional[float] = None, + interval: Optional[float] = None, + ) -> bool: + """Atomically claim a plugin for update, returning True if we won it. + + can_execute() and the RUNNING transition have to happen under one lock. + As two separate calls, two scheduler threads can both see ENABLED and + both go on to run the same plugin's update() concurrently — unsafe for + any plugin that isn't reentrant (shared mutable state, a non-thread-safe + HTTP session or cache). + + The due-time check is inside the lock too. Leaving it outside would let + a second thread that had already decided "due" claim the plugin the + instant the first finished, running update() twice in one interval. + + Args: + plugin_id: Plugin to claim. + current_time: Now, for the due check. Omit to skip that check. + interval: Seconds between updates. Omit to skip the due check. + + Returns: + True if this caller reserved the plugin and must dispatch it, + False if it is ineligible, not yet due, or already claimed. + """ + with self._reservation_lock: + if not self.state_manager.can_execute(plugin_id): + return False + + if current_time is not None and interval is not None: + with self._plugin_last_update_lock: + last_update = self.plugin_last_update.get(plugin_id, 0.0) + if last_update != 0.0 and (current_time - last_update) < interval: + return False + + self.state_manager.set_state(plugin_id, PluginState.RUNNING) + return True + + def _release_reservation(self, plugin_id: str) -> None: + """Hand a claimed plugin back when it never got dispatched. + + Without this a plugin reserved but not queued would sit in RUNNING + forever, and can_execute() would refuse it on every later tick. + """ + self.state_manager.set_state(plugin_id, PluginState.ENABLED) def get_plugin_lock(self, plugin_id: str) -> threading.Lock: """Per-plugin lock keeping update() and display() mutually exclusive. @@ -843,16 +896,40 @@ class PluginManager: return lock def _enqueue_update(self, plugin_id: str, scheduled_time: float) -> None: - """Queue a due update for the background worker (dedup while pending).""" + """Queue an already-reserved update for the background worker. + + The caller has reserved the plugin (RUNNING), which is what blocks + re-entry and shows the truthful state in the web UI while the item + waits its turn. The pending set stays as a second line of defence; if + it ever fires the reservation has to be handed back, or the plugin + would sit in RUNNING with nothing queued to release it. + """ with self._pending_lock: if plugin_id in self._pending_updates: + self.logger.warning( + "Plugin %s reserved for update but already queued; " + "releasing the reservation", plugin_id) + self._release_reservation(plugin_id) return self._pending_updates.add(plugin_id) - # RUNNING is set at enqueue time so can_execute() blocks re-entry and - # the web UI shows the truthful state while the item waits its turn. - self.state_manager.set_state(plugin_id, PluginState.RUNNING) - self._ensure_update_worker() - self._update_queue.put((plugin_id, scheduled_time)) + try: + self._ensure_update_worker() + self._update_queue.put((plugin_id, scheduled_time)) + except Exception as exc: # pylint: disable=broad-except + # Thread.start() raises RuntimeError when the OS refuses a new + # thread — a real condition on a Pi under memory pressure. Nothing + # is queued to release the plugin at that point, so the claim has to + # be undone here, or it sits in RUNNING with nothing to clear it and + # can_execute() refuses it for the rest of the process. Swallowed + # rather than raised so the remaining plugins in this tick still get + # their turn. + self.logger.error( + "Could not queue update for plugin %s (%s: %s); releasing the " + "reservation so the next tick can retry", + plugin_id, type(exc).__name__, exc, exc_info=True) + with self._pending_lock: + self._pending_updates.discard(plugin_id) + self._release_reservation(plugin_id) def _ensure_update_worker(self) -> None: if self._update_worker is not None and self._update_worker.is_alive(): @@ -937,6 +1014,14 @@ class PluginManager: return finished['done'] = True try: + # Drop the queue reservation *before* the state goes back to + # ENABLED. The other order leaves a window where a scheduler + # sees ENABLED, reserves the plugin, then finds it still in + # _pending_updates -- the enqueue is dropped and the plugin + # would sit in RUNNING with nothing left to release it. + if lock is not None: + with self._pending_lock: + self._pending_updates.discard(plugin_id) if success: with self._plugin_last_update_lock: self.plugin_last_update[plugin_id] = scheduled_time @@ -949,8 +1034,6 @@ class PluginManager: finally: if lock is not None: lock.release() - with self._pending_lock: - self._pending_updates.discard(plugin_id) if lock is None: # Synchronous / no-lock path: unchanged behavior. @@ -1041,13 +1124,12 @@ class PluginManager: if not hasattr(plugin_instance, "update"): continue - # Check if plugin can execute - if not self.state_manager.can_execute(plugin_id): + # Eligibility check and the RUNNING transition together, so a + # concurrent scheduler cannot claim the same plugin (see + # _reserve_for_update). + if not self._reserve_for_update(plugin_id): continue - - # Update state to RUNNING - self.state_manager.set_state(plugin_id, PluginState.RUNNING) - + try: success = self.plugin_executor.execute_update(plugin_instance, plugin_id) if success: diff --git a/test/test_plugin_update_reservation.py b/test/test_plugin_update_reservation.py new file mode 100644 index 00000000..90ea62ae --- /dev/null +++ b/test/test_plugin_update_reservation.py @@ -0,0 +1,335 @@ +"""Plugin update scheduling must be atomic (issue #401). + +`run_scheduled_updates()` decided whether to update a plugin with a +check-then-act sequence: `can_execute()` and `set_state(RUNNING)` were separate +calls with nothing between them, so two scheduler threads could both observe +ENABLED and both go on to call the same plugin's `update()`. + +Two schedulers really do run at once. The render loop calls +`_tick_plugin_updates()`, and Vegas mode fires its own `vegas-plugin-tick` +daemon thread every few seconds; that thread is never joined when +`VegasModeCoordinator.play()` returns, so a slow `update()` still in flight can +overlap the next tick from the main loop. + +A plugin whose `update()` runs twice at once is unsafe unless it happens to be +reentrant — shared mutable state, a non-thread-safe HTTP session or cache all +break. These tests pin the reservation that closes it, and the state +bookkeeping that has to survive it: a plugin reserved but never dispatched must +not be stranded in RUNNING, because `can_execute()` would then refuse it +forever. +""" + +import os +import sys +import threading +import time + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from src.plugin_system.plugin_manager import PluginManager # noqa: E402 +from src.plugin_system.plugin_state import PluginState # noqa: E402 + + +class OverlapDetectingPlugin: + """Records the high-water mark of concurrent update() calls.""" + + def __init__(self, update_seconds=0.25): + self.enabled = True + self.update_seconds = update_seconds + self.update_calls = 0 + self.max_concurrent = 0 + self._active = 0 + self._guard = threading.Lock() + + def update(self): + with self._guard: + self._active += 1 + self.update_calls += 1 + self.max_concurrent = max(self.max_concurrent, self._active) + try: + time.sleep(self.update_seconds) + finally: + with self._guard: + self._active -= 1 + return True + + def display(self, force_clear=False): + return True + + +@pytest.fixture +def pm(tmp_path): + manager = PluginManager(plugins_dir=str(tmp_path), config_manager=None, + display_manager=None, cache_manager=None) + yield manager + manager.stop_update_worker() + + +def _install(pm, plugin, plugin_id="racy-plugin", interval=0.01): + pm.plugins[plugin_id] = plugin + pm._update_interval_cache[plugin_id] = interval + pm.state_manager.set_state(plugin_id, PluginState.ENABLED) + return plugin_id + + +def _widen_check_then_act_window(pm, delay=0.02): + """Hold every scheduler thread inside the eligibility check at once. + + The real gap between can_execute() and the RUNNING transition is a couple + of bytecodes wide, so a plain thread race under the GIL almost never lands + in it — an unfixed scheduler looks correct in a test that just hammers it. + Delaying the check reproduces the interleaving that Vegas's tick thread and + the render loop actually produce when a slow update() overlaps the next + tick, and it is what makes these tests fail without the reservation. + + Once the check and the transition are under one lock, the delay only + serializes the schedulers: the losers observe RUNNING and back off. + """ + real_can_execute = pm.state_manager.can_execute + + def slow_can_execute(plugin_id): + result = real_can_execute(plugin_id) + time.sleep(delay) + return result + + pm.state_manager.can_execute = slow_can_execute + + +def _hammer(target, threads=8, rounds=1): + """Run `target` on N threads released simultaneously by a barrier.""" + errors = [] + barrier = threading.Barrier(threads) + + def runner(): + try: + barrier.wait(timeout=5) + for _ in range(rounds): + target() + except Exception as exc: # noqa: BLE001 - surfaced by the assertion + errors.append(exc) + + workers = [threading.Thread(target=runner) for _ in range(threads)] + for worker in workers: + worker.start() + for worker in workers: + worker.join(timeout=30) + assert not errors, f"worker raised: {errors[0]!r}" + + +class TestReservationAtomicity: + """The check-and-claim itself, independent of any dispatch path.""" + + def test_only_one_caller_wins_the_reservation(self, pm): + plugin_id = _install(pm, OverlapDetectingPlugin()) + _widen_check_then_act_window(pm) + wins = [] + lock = threading.Lock() + + def claim(): + if pm._reserve_for_update(plugin_id): + with lock: + wins.append(threading.current_thread().name) + + _hammer(claim, threads=16) + assert len(wins) == 1, ( + f"{len(wins)} threads reserved the same plugin concurrently; " + "the eligibility check and the RUNNING transition are not atomic") + assert pm.state_manager.get_state(plugin_id) == PluginState.RUNNING + + def test_reservation_refused_while_running(self, pm): + plugin_id = _install(pm, OverlapDetectingPlugin()) + assert pm._reserve_for_update(plugin_id) is True + assert pm._reserve_for_update(plugin_id) is False + + def test_reservation_can_be_handed_back(self, pm): + plugin_id = _install(pm, OverlapDetectingPlugin()) + assert pm._reserve_for_update(plugin_id) is True + pm._release_reservation(plugin_id) + assert pm.state_manager.get_state(plugin_id) == PluginState.ENABLED + assert pm._reserve_for_update(plugin_id) is True, \ + "a released reservation must be claimable again" + + def test_due_check_is_inside_the_reservation(self, pm): + """Two threads that both decided 'due' must not both get a turn. + + With the due check outside the lock, the loser of the race could claim + the plugin the moment the winner finished, running update() twice + inside one interval. + """ + plugin_id = _install(pm, OverlapDetectingPlugin(), interval=60.0) + now = time.time() + assert pm._reserve_for_update(plugin_id, now, 60.0) is True + pm.plugin_last_update[plugin_id] = now + pm._release_reservation(plugin_id) + assert pm._reserve_for_update(plugin_id, now, 60.0) is False, \ + "plugin updated just now must not be due again" + + +class TestNoConcurrentUpdate: + """The end-to-end invariant the issue is actually about.""" + + def test_synchronous_path_never_overlaps(self, pm): + """The kill-switch path ran update() inline with no dedup at all.""" + pm._synchronous_updates = True + plugin = OverlapDetectingPlugin(update_seconds=0.25) + _install(pm, plugin) + _widen_check_then_act_window(pm) + + _hammer(pm.run_scheduled_updates, threads=8) + + assert plugin.max_concurrent == 1, ( + f"update() ran {plugin.max_concurrent}x concurrently on the " + "synchronous path") + + def test_update_all_plugins_never_overlaps(self, pm): + plugin = OverlapDetectingPlugin(update_seconds=0.25) + _install(pm, plugin) + _widen_check_then_act_window(pm) + + _hammer(pm.update_all_plugins, threads=8) + + assert plugin.max_concurrent == 1, ( + f"update() ran {plugin.max_concurrent}x concurrently via " + "update_all_plugins()") + + def test_async_path_never_overlaps(self, pm): + plugin = OverlapDetectingPlugin(update_seconds=0.2) + plugin_id = _install(pm, plugin) + + _hammer(pm.run_scheduled_updates, threads=8, rounds=3) + + # Wait for an update to have both started and finished. Polling only + # `_active` races the worker: before it picks the item up nothing is + # active yet, so the loop would fall straight through and assert on a + # plugin that never ran. + deadline = time.monotonic() + 15 + while time.monotonic() < deadline: + if plugin.update_calls >= 1 and plugin._active == 0: + break + time.sleep(0.05) + + assert plugin.update_calls >= 1, "no update ran on the async path" + assert plugin.max_concurrent == 1, ( + f"update() ran {plugin.max_concurrent}x concurrently on the " + "async path") + assert plugin_id in pm.plugins + + +class TestNoStrandedState: + """A reservation that is never dispatched must not wedge the plugin.""" + + def test_plugin_returns_to_enabled_after_async_updates(self, pm): + plugin = OverlapDetectingPlugin(update_seconds=0.1) + plugin_id = _install(pm, plugin) + + _hammer(pm.run_scheduled_updates, threads=6, rounds=2) + + deadline = time.monotonic() + 15 + while time.monotonic() < deadline: + if (plugin.update_calls >= 1 + and pm.state_manager.get_state(plugin_id) == PluginState.ENABLED + and not pm._pending_updates): + break + time.sleep(0.05) + + # ENABLED is also the starting state, so without this the assertion + # below would pass on a plugin that never got scheduled at all. + assert plugin.update_calls >= 1, "no update ran; the state assertion would be vacuous" + assert pm.state_manager.get_state(plugin_id) == PluginState.ENABLED, \ + "plugin stranded in RUNNING; can_execute() would refuse it forever" + assert not pm._pending_updates, "pending set not drained" + + def test_pending_cleared_before_state_reset(self, pm): + """Invariant: never-RUNNING implies never-pending. + + _finish() used to clear the pending entry *after* flipping the state + back to ENABLED. In that window a scheduler could reserve the plugin + and then have its enqueue dropped by the pending-dedup. + """ + plugin = OverlapDetectingPlugin(update_seconds=0.05) + plugin_id = _install(pm, plugin) + + violations = [] + stop = threading.Event() + + def watcher(): + while not stop.is_set(): + state = pm.state_manager.get_state(plugin_id) + if state != PluginState.RUNNING and plugin_id in pm._pending_updates: + violations.append(state) + time.sleep(0.001) + + thread = threading.Thread(target=watcher, daemon=True) + thread.start() + try: + for _ in range(15): + pm.run_scheduled_updates() + time.sleep(0.05) + finally: + stop.set() + thread.join(timeout=5) + + assert not violations, ( + f"{len(violations)} sample(s) saw a non-RUNNING plugin still in " + "_pending_updates") + + +class TestDispatchFailure: + """A reservation must survive the dispatch itself failing. + + _enqueue_update() claims the plugin, adds it to the pending set, and only + then starts the worker and queues the item. threading.Thread.start() raises + RuntimeError when the OS refuses a new thread — not hypothetical on a Pi + under memory or thread pressure. Nothing is queued to release the plugin at + that point, so without an explicit rollback it stays RUNNING with a stale + pending entry and can_execute() refuses it for the rest of the process. + """ + + def test_reservation_released_when_the_worker_cannot_start(self, pm): + plugin_id = _install(pm, OverlapDetectingPlugin()) + + def refuse_to_start(): + raise RuntimeError("can't start new thread") + + pm._ensure_update_worker = refuse_to_start + + assert pm._reserve_for_update(plugin_id) is True + pm._enqueue_update(plugin_id, time.time()) + + assert pm.state_manager.get_state(plugin_id) == PluginState.ENABLED, \ + "plugin left in RUNNING after a failed dispatch; can_execute() " \ + "would refuse it forever" + assert plugin_id not in pm._pending_updates, \ + "stale pending entry would make the next enqueue hit the dedup" + assert pm._reserve_for_update(plugin_id) is True, \ + "plugin must be claimable again on the next tick" + + def test_dispatch_failure_does_not_abort_the_rest_of_the_tick(self, pm): + """One plugin failing to queue must not skip the others in that tick.""" + first = OverlapDetectingPlugin() + second = OverlapDetectingPlugin() + _install(pm, first, plugin_id="plugin-a") + _install(pm, second, plugin_id="plugin-b") + + calls = [] + real_ensure = pm._ensure_update_worker + + def fail_first_only(): + calls.append(1) + if len(calls) == 1: + raise RuntimeError("can't start new thread") + return real_ensure() + + pm._ensure_update_worker = fail_first_only + + pm.run_scheduled_updates() # must not propagate the RuntimeError + + assert len(calls) == 2, ( + "run_scheduled_updates() stopped after the failing plugin; the " + "exception escaped the enqueue") + for plugin_id in ("plugin-a", "plugin-b"): + assert pm.state_manager.get_state(plugin_id) in ( + PluginState.ENABLED, PluginState.RUNNING), \ + f"{plugin_id} left in an unexpected state"