From 2af41c561bfb5c296704bcfbf488f222fd3705df Mon Sep 17 00:00:00 2001 From: Chuck <33324927+ChuckBuilds@users.noreply.github.com> Date: Wed, 5 Aug 2026 19:38:11 -0400 Subject: [PATCH] fix(plugins): make update scheduling atomic so update() cannot run twice at once (#437) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(plugins): make update scheduling atomic so update() cannot run twice at once Closes #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()`. `update_all_plugins()` had the identical pattern. 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 that is never joined when `VegasModeCoordinator.play()` returns — a slow `update()` still in flight overlaps the next tick from the main loop. A plugin running `update()` twice concurrently is unsafe unless it happens to be reentrant; shared mutable state, a non-thread-safe HTTP session or cache all break. The async path was already covered by the `_pending_lock` dedup in `_enqueue_update`, so the live exposure was the synchronous kill-switch path and `update_all_plugins()`. Both now claim the plugin through `_reserve_for_update()`, which holds one lock across the eligibility check, the due-time check and the RUNNING transition — and nothing more. Holding it across `execute_update()` would serialize slow plugins behind each other and reintroduce the render stall the async worker exists to avoid. The due-time check moved inside the lock deliberately. Left outside, a thread that had already decided "due" could claim the plugin the instant the winner finished, running `update()` twice within one interval. Two supporting changes fall out of it: - `_enqueue_update()` no longer sets RUNNING (the reservation did), and hands the reservation back if the pending-dedup ever fires. Otherwise a reserved-but-unqueued plugin would sit in RUNNING with nothing left to release it, and `can_execute()` would refuse it forever. - `_finish()` now clears the pending entry *before* flipping the state back to ENABLED. The old order left a window where a scheduler saw ENABLED, reserved the plugin, then had its enqueue silently dropped by the dedup — harmless as a missed tick before, a stuck plugin once a reservation is involved. Regression suite added and enrolled in CI, along with test_async_plugin_updates.py which was not previously run there. The overlap tests delay `can_execute()` to hold every thread inside the check-then-act gap: the real window is a couple of bytecodes wide, so a plain hammering test passes against the unfixed scheduler and proves nothing. With that delay the suite reports `update() ran 8x concurrently` on both affected paths before the fix, and passes after. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Udr6MfaFLUPhX5Fgo67Jf5 * test: fix a race in the reservation suite's own wait loops `test_async_path_never_overlaps` failed in CI with "update() ran 0x concurrently" — the test's bug, not the scheduler's. It polled `plugin._active` to wait for the update to finish, but before the worker picks the item up nothing is active yet, so the loop fell straight through and asserted on a plugin that had never run. Both async waits now key on `update_calls >= 1` as well, so they wait for an update to have started *and* finished. The stranded-state test gets the same guard for a second reason: ENABLED is also the starting state, so without it that assertion passes vacuously on a plugin that was never scheduled. Verified over 12 consecutive local runs, 12 passed. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Udr6MfaFLUPhX5Fgo67Jf5 * fix(plugins): roll back the claim when dispatch fails _enqueue_update() reserved the plugin and added it to the pending set, then started the worker and queued the item. Thread.start() raises RuntimeError when the OS refuses a new thread — not hypothetical on a Pi under memory or thread pressure — and nothing is queued at that point to release the plugin. It stayed RUNNING with a stale pending entry, so can_execute() refused it for the rest of the process, and the exception escaped run_scheduled_updates() and skipped every remaining plugin in that tick. That is the same stranded-RUNNING failure the reservation was introduced to prevent, just reached through the dispatch rather than the dedup, so it is handled the same way: discard the pending entry, hand the reservation back, log the cause. Swallowed rather than raised so one plugin failing to queue cannot abort the others' turn. Both new tests fail against the un-rolled-back version — the second on the escaping RuntimeError itself — and pass with it. 74 tests across the reservation, async-update, plugin-system, health, Vegas-adapter and controller-toggle suites still pass. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WexvwNDtWLVymGVqKD7BGk --------- Co-authored-by: Claude Opus 5 (1M context) --- .github/workflows/test.yml | 4 +- src/plugin_system/plugin_manager.py | 138 +++++++--- test/test_plugin_update_reservation.py | 335 +++++++++++++++++++++++++ 3 files changed, 448 insertions(+), 29 deletions(-) create mode 100644 test/test_plugin_update_reservation.py 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"