From cbc899126ae2d6c1e15c6995bb515f18f9199149 Mon Sep 17 00:00:00 2001 From: ChuckBuilds Date: Mon, 13 Jul 2026 15:12:47 -0400 Subject: [PATCH] fix(plugins): fix skipped-frame health/force_change tracking, config validation, and lock-lifetime gaps in async updates Addresses PR #407 review findings: - display_controller: only clear force_change / record health success when display() actually ran this frame, not when the frame was skipped because the plugin's lock was busy (a skip must preserve a pending mode-switch force_clear). - plugin_manager: replace bool() coercion of synchronous_updates with explicit isinstance validation of plugin_system/synchronous_updates, failing safe to synchronous mode (with a logged reason) on malformed config instead of silently defaulting to async. - plugin_manager: _update_worker_loop now acquires the plugin lock before looking up its instance and re-checks under the lock, so an unloaded plugin's lifecycle state is never resurrected to ENABLED. - plugin_manager + display_controller: move lock ownership (and, for updates, RUNNING/pending lifecycle bookkeeping) into the actual update()/ display() call itself rather than the timeout-wrapped caller, so the lock stays held for the real operation's duration even after PluginExecutor's own join(timeout) elapses and a lingering daemon thread keeps running in the background. - DisplayController.cleanup() now stops the update worker before tearing down display/cache resources; stop_update_worker() logs when the join times out instead of failing silently. - test_async_plugin_updates: rewrite test_unloaded_while_queued_is_harmless to exercise the public unload_plugin() lifecycle (via a deterministic blocker) instead of deleting pm.plugins directly, and add a regression test proving the plugin lock stays held through PluginExecutor's own timeout while the real update() call is still running. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ --- src/display_controller.py | 102 +++++++++++--- src/plugin_system/plugin_manager.py | 209 +++++++++++++++++++--------- test/test_async_plugin_updates.py | 87 +++++++++++- 3 files changed, 307 insertions(+), 91 deletions(-) diff --git a/src/display_controller.py b/src/display_controller.py index 79680291..b59bba7d 100644 --- a/src/display_controller.py +++ b/src/display_controller.py @@ -23,6 +23,8 @@ Entry point: :func:`main` — instantiates :class:`DisplayController` and calls import time import os import json +import threading +import types from contextlib import contextmanager from pathlib import Path from typing import Dict, Any, List, Optional, Callable @@ -1851,6 +1853,7 @@ class DisplayController: plugin_id = getattr(manager_to_display, 'plugin_id', active_mode) try: logger.debug(f"Calling display() for {active_mode} with force_clear={self.force_change}") + can_display = False if hasattr(manager_to_display, 'display'): # Opt #1: look up (or compute once) whether display() accepts display_mode _cache_key = plugin_id @@ -1861,30 +1864,79 @@ class DisplayController: ) _accepts_display_mode = self._plugin_accepts_display_mode[_cache_key] - with self._display_lock_or_skip(plugin_id) as can_display: - if not can_display: - # update() in flight on the worker — hold - # the last frame; not a plugin failure - result = True - elif self.plugin_manager and hasattr(self.plugin_manager, 'plugin_executor'): - # Use PluginExecutor for safe execution with timeout + pm = self.plugin_manager + display_lock = None + can_display = True + if pm and hasattr(pm, 'get_plugin_lock'): + display_lock = pm.get_plugin_lock(plugin_id) + can_display = display_lock.acquire(blocking=False) + + if not can_display: + # update() in flight on the worker — hold + # the last frame; not a plugin failure + result = True + elif pm and hasattr(pm, 'plugin_executor'): + # PluginExecutor's own thread.join(timeout) can + # return before the real display() call + # finishes (a lingering daemon thread keeps + # running it) -- so the lock is released from + # inside the wrapped call itself, whichever + # thread actually finishes it, rather than + # here when this dispatch merely returns. + release_guard = threading.Lock() + released = {'done': False} + + def _release_display_lock(): + with release_guard: + if released['done']: + return + released['done'] = True + if display_lock is not None: + display_lock.release() + + if _accepts_display_mode: + def _display_target(display_mode=None, force_clear=False): + try: + return manager_to_display.display( + display_mode=display_mode, force_clear=force_clear) + finally: + _release_display_lock() + else: + def _display_target(force_clear=False): + try: + return manager_to_display.display(force_clear=force_clear) + finally: + _release_display_lock() + + try: result = self.plugin_manager.plugin_executor.execute_display( - manager_to_display, + types.SimpleNamespace(display=_display_target), plugin_id, force_clear=self.force_change, display_mode=active_mode if _accepts_display_mode else None ) - # execute_display returns bool, convert to expected format - if result: - result = True # Success - else: - result = False # Failed + except Exception: # pragma: no cover - defensive; + # execute_display catches everything + # internally, but guarantee the lock is + # never leaked if something unexpected + # slips through. + _release_display_lock() + raise + # execute_display returns bool, convert to expected format + if result: + result = True # Success else: - # Fallback to direct call if executor not available + result = False # Failed + else: + # Fallback to direct call if executor not available + try: if _accepts_display_mode: result = manager_to_display.display(display_mode=active_mode, force_clear=self.force_change) else: result = manager_to_display.display(force_clear=self.force_change) + finally: + if display_lock is not None: + display_lock.release() logger.debug(f"display() returned: {result} (type: {type(result)})") # Check if display() returned a boolean (new behavior) @@ -1893,11 +1945,15 @@ class DisplayController: if not display_result: logger.info(f"Plugin {plugin_id} display() returned False for mode {active_mode}") - # Record success if display completed without exception - if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker: - self.plugin_manager.health_tracker.record_success(plugin_id) - - self.force_change = False + # Record success only when display() actually ran this + # frame -- a skipped frame (lock busy) held the last + # frame, not a real success, and must not clear + # force_change or the pending mode-switch clear will + # be lost when display() finally does run. + if can_display: + if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker: + self.plugin_manager.health_tracker.record_success(plugin_id) + self.force_change = False except Exception as exc: # pylint: disable=broad-except logger.exception("Error displaying %s", self.current_display_mode) # Record failure @@ -2754,6 +2810,14 @@ class DisplayController: def cleanup(self): """Clean up resources.""" + # Stop the async update worker first so no in-flight update() call + # is still touching display/cache-backed resources while they're + # torn down below. + if self.plugin_manager and hasattr(self.plugin_manager, 'stop_update_worker'): + try: + self.plugin_manager.stop_update_worker() + except Exception as e: + logger.warning("Error stopping plugin update worker: %s", e) # Shutdown config service if it exists if hasattr(self, 'config_service'): try: diff --git a/src/plugin_system/plugin_manager.py b/src/plugin_system/plugin_manager.py index f6b68bfb..74b4cc46 100644 --- a/src/plugin_system/plugin_manager.py +++ b/src/plugin_system/plugin_manager.py @@ -111,13 +111,31 @@ class PluginManager: self._plugin_locks_guard = threading.Lock() self._update_worker: Optional[threading.Thread] = None self._synchronous_updates = False - try: - if self.config_manager is not None: + if self.config_manager is not None: + try: cfg = self.config_manager.get_config() or {} - self._synchronous_updates = bool( - cfg.get('plugin_system', {}).get('synchronous_updates', False)) - except Exception: - self._synchronous_updates = False + except (OSError, ValueError) as exc: + self.logger.warning( + "Could not load config to check plugin_system.synchronous_updates " + "(%s: %s); defaulting to synchronous updates", type(exc).__name__, exc) + self._synchronous_updates = True + else: + plugin_system_cfg = cfg.get('plugin_system', {}) + if not isinstance(plugin_system_cfg, dict): + self.logger.warning( + "config plugin_system must be a mapping, got %s; " + "defaulting to synchronous updates", + type(plugin_system_cfg).__name__) + self._synchronous_updates = True + else: + sync_value = plugin_system_cfg.get('synchronous_updates', False) + if not isinstance(sync_value, bool): + self.logger.warning( + "config plugin_system.synchronous_updates must be a boolean, " + "got %r; defaulting to synchronous updates", sync_value) + self._synchronous_updates = True + else: + self._synchronous_updates = sync_value # Ensure plugins directory exists with proper permissions try: @@ -804,84 +822,145 @@ class PluginManager: self._update_worker.start() def _update_worker_loop(self) -> None: - """Single worker: serializes all plugin updates (matching the old - inline behavior — no thundering herd of concurrent fetches), off the - render thread.""" + """Single worker: dispatches queued updates off the render thread + (matching the old inline behavior — no thundering herd of + concurrent fetches). + + The plugin's lock is acquired here, before its instance is looked + up, and the instance is re-fetched under the lock — a concurrent + unload_plugin() can't leave this loop about to run update() on an + instance that's already been torn down. The lock — and RUNNING/ + pending lifecycle state — is released by the update itself once the + real update() call genuinely finishes (see _execute_update_now), + which can be after this dispatch returns if PluginExecutor's own + timeout elapses first. + """ while True: item = self._update_queue.get() if item is None: # shutdown sentinel return plugin_id, scheduled_time = item - try: - plugin_instance = self.plugins.get(plugin_id) - if plugin_instance is None: # unloaded while queued - self.state_manager.set_state(plugin_id, PluginState.ENABLED) - continue - lock = self.get_plugin_lock(plugin_id) - with lock: - self._execute_update_now(plugin_id, plugin_instance, - scheduled_time) - except Exception: # pylint: disable=broad-except - self.logger.exception("update worker: unexpected error for %s", - plugin_id) - finally: + lock = self.get_plugin_lock(plugin_id) + lock.acquire() + plugin_instance = self.plugins.get(plugin_id) + if plugin_instance is None: # unloaded while queued; its + # lifecycle state was already cleared by unload_plugin — + # leave it alone rather than resurrecting it to ENABLED + lock.release() with self._pending_lock: self._pending_updates.discard(plugin_id) + continue + try: + self._execute_update_now(plugin_id, plugin_instance, + scheduled_time, lock=lock) + except Exception: # pylint: disable=broad-except + # _execute_update_now guarantees the lock/pending bookkeeping + # is released via its own _finish() before returning or + # raising; this is a last-resort log only. + self.logger.exception("update worker: unexpected error for %s", + plugin_id) def stop_update_worker(self, timeout: float = 5.0) -> None: """Signal the worker to exit (used by cleanup; thread is a daemon).""" if self._update_worker is not None and self._update_worker.is_alive(): self._update_queue.put(None) self._update_worker.join(timeout=timeout) + if self._update_worker.is_alive(): + self.logger.warning( + "Update worker did not stop within %.1fs; it is a daemon " + "thread and will be abandoned on shutdown", timeout) def _execute_update_now(self, plugin_id: str, plugin_instance: Any, - scheduled_time: float) -> None: - """The (unchanged) update execution block: executor + bookkeeping. + scheduled_time: float, + lock: Optional[threading.Lock] = None) -> None: + """Execute a plugin's update() via PluginExecutor, then bookkeep. - Caller is responsible for having set RUNNING state and, on the async - path, for holding the plugin's lock. plugin_executor's internal - thread.join(timeout) blocks only the calling thread; on timeout the - lingering daemon update-thread keeps running unkillable — exactly the - pre-async behavior, documented here so nobody "fixes" it into a - thread leak hunt. + Caller is responsible for having set RUNNING state. + + On the synchronous path (``lock=None``) this is the original, + unchanged inline behavior. On the async worker path, PluginExecutor's + internal thread.join(timeout) blocks only the calling thread -- on + timeout the lingering daemon update-thread keeps running the real + plugin.update() call unkillable in the background. So that the + plugin's lock (and its RUNNING/pending lifecycle state) stays held + for that real duration rather than just this bounded wait, ownership + of both is carried by the wrapped update callable itself, released + from whichever thread actually finishes it -- see _finish() below. """ - try: - # Use PluginExecutor for safe execution - success = False - if self.resource_monitor: - # If resource monitor exists, wrap the call - def monitored_update(): - self.resource_monitor.monitor_call(plugin_id, plugin_instance.update) - # SimpleNamespace stores `update` as an *instance* - # attribute, so attribute lookup returns the plain - # function object as-is. A dynamically-built class - # (`type(..., {'update': monitored_update})`) instead - # stores it as a *class* attribute, which the - # descriptor protocol turns into a bound method on - # access -- silently prepending the instance as an - # implicit first argument to a function that takes - # none, raising "monitored_update() takes 0 - # positional arguments but 1 was given" on every call. - success = self.plugin_executor.execute_update( - types.SimpleNamespace(update=monitored_update), - plugin_id - ) - else: - success = self.plugin_executor.execute_update(plugin_instance, plugin_id) + finish_guard = threading.Lock() + finished = {'done': False} - if success: - self.plugin_last_update[plugin_id] = scheduled_time - self.state_manager.record_update(plugin_id) - # Update state back to ENABLED - self.state_manager.set_state(plugin_id, PluginState.ENABLED) - # Record success - if self.health_tracker: - self.health_tracker.record_success(plugin_id) + def _finish(success: bool, exc: Optional[Exception] = None) -> None: + with finish_guard: + if finished['done']: + return + finished['done'] = True + try: + if success: + self.plugin_last_update[plugin_id] = scheduled_time + self.state_manager.record_update(plugin_id) + self.state_manager.set_state(plugin_id, PluginState.ENABLED) + if self.health_tracker: + self.health_tracker.record_success(plugin_id) + else: + self._record_update_failure(plugin_id, exc=exc) + 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. + try: + if self.resource_monitor: + def monitored_update(): + self.resource_monitor.monitor_call(plugin_id, plugin_instance.update) + # SimpleNamespace stores `update` as an *instance* + # attribute, so attribute lookup returns the plain + # function object as-is. A dynamically-built class + # (`type(..., {'update': monitored_update})`) instead + # stores it as a *class* attribute, which the + # descriptor protocol turns into a bound method on + # access -- silently prepending the instance as an + # implicit first argument to a function that takes + # none, raising "monitored_update() takes 0 + # positional arguments but 1 was given" on every call. + success = self.plugin_executor.execute_update( + types.SimpleNamespace(update=monitored_update), + plugin_id + ) + else: + success = self.plugin_executor.execute_update(plugin_instance, plugin_id) + _finish(success) + except Exception as exc: # pylint: disable=broad-except + self.logger.exception("Error updating plugin %s: %s", plugin_id, exc) + _finish(False, exc=exc) + return + + # Async worker path: the real update() call -- through the resource + # monitor, if configured -- owns finishing the lock/lifecycle + # bookkeeping, from whichever thread actually runs it to completion. + def _target_update() -> None: + try: + if self.resource_monitor: + self.resource_monitor.monitor_call(plugin_id, plugin_instance.update) + else: + plugin_instance.update() + except Exception as exc: + _finish(False, exc=exc) + raise else: - self._record_update_failure(plugin_id) - except Exception as exc: # pylint: disable=broad-except - self.logger.exception("Error updating plugin %s: %s", plugin_id, exc) - self._record_update_failure(plugin_id, exc=exc) + _finish(True) + + try: + self.plugin_executor.execute_update( + types.SimpleNamespace(update=_target_update), plugin_id) + except Exception as exc: # pragma: no cover - defensive; execute_update + # catches everything internally, but guarantee _finish still + # runs (releasing the lock) if something unexpected slips through. + self.logger.exception("Unexpected error dispatching update for %s: %s", plugin_id, exc) + _finish(False, exc=exc) def update_all_plugins(self) -> None: """ diff --git a/test/test_async_plugin_updates.py b/test/test_async_plugin_updates.py index 0d3ebfc5..a27fea11 100644 --- a/test/test_async_plugin_updates.py +++ b/test/test_async_plugin_updates.py @@ -4,8 +4,9 @@ The invariants that keep this change safe: 1. run_scheduled_updates returns immediately — a slow update() can never again freeze the render loop (the original defect: 30s scroll freezes). 2. A plugin's update() and display() are NEVER concurrent — the per-plugin - lock makes the old implicit no-overlap guarantee explicit (and, unlike - before, holds it across the post-timeout window too). + lock makes the old implicit no-overlap guarantee explicit, and it stays + held through PluginExecutor's own timeout: a lingering, still-running + update() keeps the lock even after PluginExecutor gives up waiting on it. 3. Failure/timeout bookkeeping is unchanged (same executor, same _record_update_failure path, same last-update stamping). 4. The kill switch (plugin_system.synchronous_updates) restores the @@ -138,6 +139,40 @@ class TestUpdateDisplayExclusion: assert plugin.display_calls > 10 assert plugin.overlap_detected is False + def test_lock_held_through_timeout_until_real_update_finishes(self, pm): + """PluginExecutor's join(timeout) can return before the real + update() call does -- the lingering daemon thread keeps running it. + The plugin's lock must stay held for that whole real duration, not + just PluginExecutor's bounded wait, or display() could run + concurrently with a still-executing update().""" + plugin = SlowPlugin(update_seconds=0.3) + plugin_id = _install(pm, plugin) + pm.plugin_executor.default_timeout = 0.05 # times out well before + # update_seconds elapses + + pm.run_scheduled_updates() + + deadline = time.monotonic() + 2 + while not plugin.in_update and time.monotonic() < deadline: + time.sleep(0.01) + assert plugin.in_update is True + + # PluginExecutor's own timeout has now elapsed, but the real + # update() (0.3s) is still running in its lingering daemon thread. + time.sleep(0.15) + assert plugin.in_update is True, "test setup: update should still be running" + lock = pm.get_plugin_lock(plugin_id) + assert lock.acquire(blocking=False) is False, \ + "lock must stay held through PluginExecutor's timeout while the real update() runs" + + # Once the real update() genuinely finishes, the lock is released. + deadline = time.monotonic() + 2 + while plugin.in_update and time.monotonic() < deadline: + time.sleep(0.02) + time.sleep(0.1) + assert lock.acquire(blocking=False) is True + lock.release() + def test_state_returns_to_enabled_after_update(self, pm): """RUNNING is set at enqueue (blocks re-entry via can_execute) and must return to an executable state once the update finishes.""" @@ -175,11 +210,49 @@ class TestFailurePaths: pm.get_plugin_lock(plugin_id).release() def test_unloaded_while_queued_is_harmless(self, pm): - plugin_id = _install(pm, SlowPlugin(update_seconds=0.1)) - pm._enqueue_update(plugin_id, time.time()) - del pm.plugins[plugin_id] - time.sleep(0.3) # worker drains the item without crashing - assert pm._update_worker is None or pm._update_worker.is_alive() + """Exercise the public unload_plugin() lifecycle rather than + deleting pm.plugins directly: queue the target's update behind a + deterministic blocker (occupying the single worker), unload the + target while its item still sits queued, then release the blocker + and confirm the target's update never ran and its state stayed + unloaded rather than being resurrected to ENABLED.""" + blocker_event = threading.Event() + + class BlockerPlugin(SlowPlugin): + def update(self): + self.update_calls += 1 + blocker_event.wait(timeout=5) + return True + + blocker_id = _install(pm, BlockerPlugin(), plugin_id="blocker-plugin") + target = SlowPlugin(update_seconds=0.05) + target_id = _install(pm, target, plugin_id="slow-plugin") + + # Dispatch the blocker first so it occupies the single worker + # thread, then enqueue the target behind it -- deterministically + # queued, not yet started. + pm._enqueue_update(blocker_id, time.time()) + deadline = time.monotonic() + 2 + while pm.plugins[blocker_id].update_calls == 0 and time.monotonic() < deadline: + time.sleep(0.01) + assert pm.plugins[blocker_id].update_calls == 1 + + pm._enqueue_update(target_id, time.time()) + assert target_id in pm._pending_updates + + assert pm.unload_plugin(target_id) is True + assert target_id not in pm.plugins + + blocker_event.set() # let the blocker finish; worker moves on to + # the target's queued item + + deadline = time.monotonic() + 3 + while target_id in pm._pending_updates and time.monotonic() < deadline: + time.sleep(0.02) + + assert target.update_calls == 0, "update() must not run for an unloaded plugin" + assert target_id not in pm._pending_updates + assert pm.state_manager.get_state(target_id) == PluginState.UNLOADED class TestKillSwitch: