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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ
This commit is contained in:
ChuckBuilds
2026-07-13 15:12:47 -04:00
co-authored by Claude Sonnet 5
parent c32aaf02e5
commit cbc899126a
3 changed files with 307 additions and 91 deletions
+83 -19
View File
@@ -23,6 +23,8 @@ Entry point: :func:`main` — instantiates :class:`DisplayController` and calls
import time import time
import os import os
import json import json
import threading
import types
from contextlib import contextmanager from contextlib import contextmanager
from pathlib import Path from pathlib import Path
from typing import Dict, Any, List, Optional, Callable from typing import Dict, Any, List, Optional, Callable
@@ -1851,6 +1853,7 @@ class DisplayController:
plugin_id = getattr(manager_to_display, 'plugin_id', active_mode) plugin_id = getattr(manager_to_display, 'plugin_id', active_mode)
try: try:
logger.debug(f"Calling display() for {active_mode} with force_clear={self.force_change}") logger.debug(f"Calling display() for {active_mode} with force_clear={self.force_change}")
can_display = False
if hasattr(manager_to_display, 'display'): if hasattr(manager_to_display, 'display'):
# Opt #1: look up (or compute once) whether display() accepts display_mode # Opt #1: look up (or compute once) whether display() accepts display_mode
_cache_key = plugin_id _cache_key = plugin_id
@@ -1861,30 +1864,79 @@ class DisplayController:
) )
_accepts_display_mode = self._plugin_accepts_display_mode[_cache_key] _accepts_display_mode = self._plugin_accepts_display_mode[_cache_key]
with self._display_lock_or_skip(plugin_id) as can_display: pm = self.plugin_manager
if not can_display: display_lock = None
# update() in flight on the worker — hold can_display = True
# the last frame; not a plugin failure if pm and hasattr(pm, 'get_plugin_lock'):
result = True display_lock = pm.get_plugin_lock(plugin_id)
elif self.plugin_manager and hasattr(self.plugin_manager, 'plugin_executor'): can_display = display_lock.acquire(blocking=False)
# Use PluginExecutor for safe execution with timeout
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( result = self.plugin_manager.plugin_executor.execute_display(
manager_to_display, types.SimpleNamespace(display=_display_target),
plugin_id, plugin_id,
force_clear=self.force_change, force_clear=self.force_change,
display_mode=active_mode if _accepts_display_mode else None display_mode=active_mode if _accepts_display_mode else None
) )
# execute_display returns bool, convert to expected format except Exception: # pragma: no cover - defensive;
if result: # execute_display catches everything
result = True # Success # internally, but guarantee the lock is
else: # never leaked if something unexpected
result = False # Failed # slips through.
_release_display_lock()
raise
# execute_display returns bool, convert to expected format
if result:
result = True # Success
else: 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: if _accepts_display_mode:
result = manager_to_display.display(display_mode=active_mode, force_clear=self.force_change) result = manager_to_display.display(display_mode=active_mode, force_clear=self.force_change)
else: else:
result = manager_to_display.display(force_clear=self.force_change) 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)})") logger.debug(f"display() returned: {result} (type: {type(result)})")
# Check if display() returned a boolean (new behavior) # Check if display() returned a boolean (new behavior)
@@ -1893,11 +1945,15 @@ class DisplayController:
if not display_result: if not display_result:
logger.info(f"Plugin {plugin_id} display() returned False for mode {active_mode}") logger.info(f"Plugin {plugin_id} display() returned False for mode {active_mode}")
# Record success if display completed without exception # Record success only when display() actually ran this
if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker: # frame -- a skipped frame (lock busy) held the last
self.plugin_manager.health_tracker.record_success(plugin_id) # frame, not a real success, and must not clear
# force_change or the pending mode-switch clear will
self.force_change = False # 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 except Exception as exc: # pylint: disable=broad-except
logger.exception("Error displaying %s", self.current_display_mode) logger.exception("Error displaying %s", self.current_display_mode)
# Record failure # Record failure
@@ -2754,6 +2810,14 @@ class DisplayController:
def cleanup(self): def cleanup(self):
"""Clean up resources.""" """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 # Shutdown config service if it exists
if hasattr(self, 'config_service'): if hasattr(self, 'config_service'):
try: try:
+144 -65
View File
@@ -111,13 +111,31 @@ class PluginManager:
self._plugin_locks_guard = threading.Lock() self._plugin_locks_guard = threading.Lock()
self._update_worker: Optional[threading.Thread] = None self._update_worker: Optional[threading.Thread] = None
self._synchronous_updates = False 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 {} cfg = self.config_manager.get_config() or {}
self._synchronous_updates = bool( except (OSError, ValueError) as exc:
cfg.get('plugin_system', {}).get('synchronous_updates', False)) self.logger.warning(
except Exception: "Could not load config to check plugin_system.synchronous_updates "
self._synchronous_updates = False "(%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 # Ensure plugins directory exists with proper permissions
try: try:
@@ -804,84 +822,145 @@ class PluginManager:
self._update_worker.start() self._update_worker.start()
def _update_worker_loop(self) -> None: def _update_worker_loop(self) -> None:
"""Single worker: serializes all plugin updates (matching the old """Single worker: dispatches queued updates off the render thread
inline behavior no thundering herd of concurrent fetches), off the (matching the old inline behavior no thundering herd of
render thread.""" 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: while True:
item = self._update_queue.get() item = self._update_queue.get()
if item is None: # shutdown sentinel if item is None: # shutdown sentinel
return return
plugin_id, scheduled_time = item plugin_id, scheduled_time = item
try: lock = self.get_plugin_lock(plugin_id)
plugin_instance = self.plugins.get(plugin_id) lock.acquire()
if plugin_instance is None: # unloaded while queued plugin_instance = self.plugins.get(plugin_id)
self.state_manager.set_state(plugin_id, PluginState.ENABLED) if plugin_instance is None: # unloaded while queued; its
continue # lifecycle state was already cleared by unload_plugin —
lock = self.get_plugin_lock(plugin_id) # leave it alone rather than resurrecting it to ENABLED
with lock: lock.release()
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:
with self._pending_lock: with self._pending_lock:
self._pending_updates.discard(plugin_id) 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: def stop_update_worker(self, timeout: float = 5.0) -> None:
"""Signal the worker to exit (used by cleanup; thread is a daemon).""" """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(): if self._update_worker is not None and self._update_worker.is_alive():
self._update_queue.put(None) self._update_queue.put(None)
self._update_worker.join(timeout=timeout) 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, def _execute_update_now(self, plugin_id: str, plugin_instance: Any,
scheduled_time: float) -> None: scheduled_time: float,
"""The (unchanged) update execution block: executor + bookkeeping. 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 Caller is responsible for having set RUNNING state.
path, for holding the plugin's lock. plugin_executor's internal
thread.join(timeout) blocks only the calling thread; on timeout the On the synchronous path (``lock=None``) this is the original,
lingering daemon update-thread keeps running unkillable exactly the unchanged inline behavior. On the async worker path, PluginExecutor's
pre-async behavior, documented here so nobody "fixes" it into a internal thread.join(timeout) blocks only the calling thread -- on
thread leak hunt. 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: finish_guard = threading.Lock()
# Use PluginExecutor for safe execution finished = {'done': False}
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)
if success: def _finish(success: bool, exc: Optional[Exception] = None) -> None:
self.plugin_last_update[plugin_id] = scheduled_time with finish_guard:
self.state_manager.record_update(plugin_id) if finished['done']:
# Update state back to ENABLED return
self.state_manager.set_state(plugin_id, PluginState.ENABLED) finished['done'] = True
# Record success try:
if self.health_tracker: if success:
self.health_tracker.record_success(plugin_id) 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: else:
self._record_update_failure(plugin_id) _finish(True)
except Exception as exc: # pylint: disable=broad-except
self.logger.exception("Error updating plugin %s: %s", plugin_id, exc) try:
self._record_update_failure(plugin_id, exc=exc) 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: def update_all_plugins(self) -> None:
""" """
+80 -7
View File
@@ -4,8 +4,9 @@ The invariants that keep this change safe:
1. run_scheduled_updates returns immediately a slow update() can never 1. run_scheduled_updates returns immediately a slow update() can never
again freeze the render loop (the original defect: 30s scroll freezes). again freeze the render loop (the original defect: 30s scroll freezes).
2. A plugin's update() and display() are NEVER concurrent — the per-plugin 2. A plugin's update() and display() are NEVER concurrent — the per-plugin
lock makes the old implicit no-overlap guarantee explicit (and, unlike lock makes the old implicit no-overlap guarantee explicit, and it stays
before, holds it across the post-timeout window too). 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 3. Failure/timeout bookkeeping is unchanged (same executor, same
_record_update_failure path, same last-update stamping). _record_update_failure path, same last-update stamping).
4. The kill switch (plugin_system.synchronous_updates) restores the 4. The kill switch (plugin_system.synchronous_updates) restores the
@@ -138,6 +139,40 @@ class TestUpdateDisplayExclusion:
assert plugin.display_calls > 10 assert plugin.display_calls > 10
assert plugin.overlap_detected is False 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): def test_state_returns_to_enabled_after_update(self, pm):
"""RUNNING is set at enqueue (blocks re-entry via can_execute) and """RUNNING is set at enqueue (blocks re-entry via can_execute) and
must return to an executable state once the update finishes.""" must return to an executable state once the update finishes."""
@@ -175,11 +210,49 @@ class TestFailurePaths:
pm.get_plugin_lock(plugin_id).release() pm.get_plugin_lock(plugin_id).release()
def test_unloaded_while_queued_is_harmless(self, pm): def test_unloaded_while_queued_is_harmless(self, pm):
plugin_id = _install(pm, SlowPlugin(update_seconds=0.1)) """Exercise the public unload_plugin() lifecycle rather than
pm._enqueue_update(plugin_id, time.time()) deleting pm.plugins directly: queue the target's update behind a
del pm.plugins[plugin_id] deterministic blocker (occupying the single worker), unload the
time.sleep(0.3) # worker drains the item without crashing target while its item still sits queued, then release the blocker
assert pm._update_worker is None or pm._update_worker.is_alive() 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: class TestKillSwitch: