fix(vegas): lock plugin_last_update snapshot/diff against concurrent mutation (#398)

* fix(vegas): restore live plugin-update refresh dropped by sync refactor

Investigating a user report that Vegas scroll mode doesn't update scores
or game status. Root cause: PR #299 (Mar 28) added a mechanism so a live
score change reached the ticker within a few seconds instead of waiting
for a full scroll cycle -- _tick_plugin_updates_for_vegas() diffed
plugin_last_update timestamps to detect which plugins got fresh data and
called coordinator.mark_plugin_updated() for each, and should_recompose()
checked has_pending_updates_for_visible_segments() to trigger an immediate
hot-swap.

PR #330 (May 14, multi-display wireless sync) refactored both call sites
while adding sync support and silently deleted this entire mechanism --
not just gated it behind the new sync-mode deferral it legitimately
needed, but removed it outright. The result: VegasModeCoordinator.
mark_plugin_updated() and StreamManager.has_pending_updates_for_visible_
segments() have been fully implemented but never called from anywhere
since. Vegas mode's only remaining freshness sources are a 5s content
cache TTL (fine) and full recompose at cycle boundaries, which depending
on min/max_cycle_duration can be minutes away -- so live scores/status
can sit stale far longer than a user would expect from a "live" ticker.

Fix:
- Restored _tick_plugin_updates_for_vegas() in display_controller.py,
  wired as the Vegas coordinator's update callback in place of the plain
  _tick_plugin_updates(). Diffs plugin_last_update before/after the tick
  and calls vegas_coordinator.mark_plugin_updated(plugin_id) for each
  plugin that actually got new data (rather than returning the list, since
  the callback interface no longer consumes a return value).
- Restored the has_pending_updates_for_visible_segments() check in
  render_pipeline.should_recompose(), positioned after (not instead of)
  the sync-mode early return PR #330 added, so standalone installations
  regain immediate refresh while synced leader/follower pairs correctly
  keep deferring hot-swaps to cycle boundaries as PR #330 intended.

Test plan:
- Added test_display_controller_vegas_tick.py and
  test_vegas_render_pipeline_recompose.py -- neither area had any prior
  test coverage, which is very likely why this regression went unnoticed
  for ~2.5 months.
- Verified both new test files fail against the pre-fix code (swapped in
  the current main versions of both files) with exactly the expected
  errors -- AttributeError for the deleted method, and the recompose
  assertion returning False instead of True -- then pass against the fix.
- Confirmed the sync-mode deferral this restoration must not break still
  holds: test_sync_active_defers_pending_updates_to_cycle_boundary.
- Full related suite (test_vegas_plugin_adapter, test_vegas_config,
  test_display_controller_plugin_toggle, test_display_controller_
  optimizations, test_plugin_system): 108 passed, 1 pre-existing failure
  unrelated to this change (test_circuit_breaker, stale mock signature).
- Full CI plugin-safety suite (test_harness, test_visual_rendering,
  test_plugin_matrix): 52 passed, 2 pre-existing skips.

* fix(vegas): lock plugin_last_update snapshot/diff against concurrent mutation

_tick_plugin_updates_for_vegas() snapshotted and later re-iterated
plugin_manager.plugin_last_update from the Vegas background update-tick
thread while the main render loop (or other callers) could mutate the
same dict concurrently — a real race (unprotected dict iteration/mutation
across threads), not just a style nit.

Move the snapshot/update/diff into a new locked
PluginManager.run_scheduled_updates_with_changes() so all reads and
mutations of plugin_last_update happen under one lock, and update
DisplayController to use it. The lock is only held around the dict
accesses, not the update pass itself, so slow plugin update() calls don't
serialize against other callers.

Also add a regression test covering that the Vegas coordinator is wired
to the Vegas-aware tick callback rather than the plain one.

Skipped as not worth the change:
- Narrowing the broad `except Exception` around
  vc.mark_plugin_updated(plugin_id) to specific types: it's a deliberate
  per-plugin isolation boundary (matches the same pattern used elsewhere
  in this file for plugin/coordinator calls) and there's no documented,
  stable set of exceptions that call can raise to narrow to.
- Adding an inactive-DisplaySyncManager test to
  test_vegas_render_pipeline_recompose.py: verified
  VegasModeCoordinator.set_sync_manager() already normalizes a
  SyncRole.STANDALONE manager to None before handing it to the render
  pipeline (src/vegas_mode/coordinator.py:152-156), so should_recompose()'s
  `is not None` check is correct in practice; the suggested case is
  already covered by that normalization.

---------

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-07-12 10:52:18 -04:00
committed by GitHub
co-authored by Claude
parent 6052a60d22
commit 1c7a0cef66
3 changed files with 87 additions and 38 deletions
+8 -8
View File
@@ -831,7 +831,7 @@ class DisplayController:
if hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker: if hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
self.plugin_manager.health_tracker.record_failure(plugin_id, exc) self.plugin_manager.health_tracker.record_failure(plugin_id, exc)
def _tick_plugin_updates_for_vegas(self): def _tick_plugin_updates_for_vegas(self) -> None:
"""Run scheduled plugin updates and tell Vegas mode which plugins """Run scheduled plugin updates and tell Vegas mode which plugins
actually got fresh data, so it can hot-swap them into the scroll actually got fresh data, so it can hot-swap them into the scroll
without waiting for a full cycle to complete. without waiting for a full cycle to complete.
@@ -843,22 +843,22 @@ class DisplayController:
minutes away). Restores wiring that PR #299 added and PR #330's minutes away). Restores wiring that PR #299 added and PR #330's
sync-mode refactor inadvertently dropped: coordinator.mark_plugin_updated() sync-mode refactor inadvertently dropped: coordinator.mark_plugin_updated()
has been unreachable dead code since. has been unreachable dead code since.
Delegates the before/after plugin_last_update snapshot to
PluginManager.run_scheduled_updates_with_changes() so the snapshot,
update pass, and diff are lock-protected against this callback's own
background update-tick thread racing the main render loop.
""" """
if not self.plugin_manager or not hasattr(self.plugin_manager, "plugin_last_update"): if not self.plugin_manager or not hasattr(self.plugin_manager, "run_scheduled_updates_with_changes"):
self._tick_plugin_updates() self._tick_plugin_updates()
return return
old_times = dict(self.plugin_manager.plugin_last_update) updated = self.plugin_manager.run_scheduled_updates_with_changes()
self._tick_plugin_updates()
vc = getattr(self, "vegas_coordinator", None) vc = getattr(self, "vegas_coordinator", None)
if vc is None: if vc is None:
return return
updated = [
plugin_id for plugin_id, new_time in self.plugin_manager.plugin_last_update.items()
if new_time > old_times.get(plugin_id, 0.0)
]
if updated: if updated:
logger.info("Vegas update tick: %d plugin(s) updated: %s", len(updated), updated) logger.info("Vegas update tick: %d plugin(s) updated: %s", len(updated), updated)
for plugin_id in updated: for plugin_id in updated:
+37
View File
@@ -76,6 +76,12 @@ class PluginManager:
# concurrent mutation (background reconciliation) and reads (requests). # concurrent mutation (background reconciliation) and reads (requests).
self._discovery_lock = threading.RLock() self._discovery_lock = threading.RLock()
# Lock protecting plugin_last_update from concurrent mutation/iteration.
# It's written from run_scheduled_updates()/update_all_plugins() (main
# loop) and read/diffed by run_scheduled_updates_with_changes(), which
# Vegas mode calls from its own background update-tick thread.
self._plugin_last_update_lock = threading.RLock()
# Active plugins # Active plugins
self.plugins: Dict[str, Any] = {} self.plugins: Dict[str, Any] = {}
self.plugin_manifests: Dict[str, Dict[str, Any]] = {} self.plugin_manifests: Dict[str, Dict[str, Any]] = {}
@@ -317,6 +323,7 @@ class PluginManager:
# Store plugin instance # Store plugin instance
self.plugins[plugin_id] = plugin_instance self.plugins[plugin_id] = plugin_instance
with self._plugin_last_update_lock:
self.plugin_last_update[plugin_id] = 0.0 self.plugin_last_update[plugin_id] = 0.0
# Invalidate cached interval so next tick re-derives it for this plugin # Invalidate cached interval so next tick re-derives it for this plugin
self._update_interval_cache.pop(plugin_id, None) self._update_interval_cache.pop(plugin_id, None)
@@ -429,6 +436,7 @@ class PluginManager:
# Remove from active plugins # Remove from active plugins
del self.plugins[plugin_id] del self.plugins[plugin_id]
with self._plugin_last_update_lock:
self.plugin_last_update.pop(plugin_id, None) self.plugin_last_update.pop(plugin_id, None)
self._update_interval_cache.pop(plugin_id, None) self._update_interval_cache.pop(plugin_id, None)
@@ -698,6 +706,7 @@ class PluginManager:
'recoverable': True, 'recoverable': True,
} }
self.logger.warning("Plugin %s update() failed; will retry after interval", plugin_id) self.logger.warning("Plugin %s update() failed; will retry after interval", plugin_id)
with self._plugin_last_update_lock:
self.plugin_last_update[plugin_id] = failure_time self.plugin_last_update[plugin_id] = failure_time
self.state_manager.set_state_with_error(plugin_id, PluginState.ENABLED, error_info, error=err) self.state_manager.set_state_with_error(plugin_id, PluginState.ENABLED, error_info, error=err)
if self.health_tracker: if self.health_tracker:
@@ -731,6 +740,7 @@ class PluginManager:
if interval is None: if interval is None:
continue continue
with self._plugin_last_update_lock:
last_update = self.plugin_last_update.get(plugin_id, 0.0) last_update = self.plugin_last_update.get(plugin_id, 0.0)
if last_update == 0.0 or (current_time - last_update) >= interval: if last_update == 0.0 or (current_time - last_update) >= interval:
@@ -762,6 +772,7 @@ class PluginManager:
success = self.plugin_executor.execute_update(plugin_instance, plugin_id) success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
if success: if success:
with self._plugin_last_update_lock:
self.plugin_last_update[plugin_id] = current_time self.plugin_last_update[plugin_id] = current_time
self.state_manager.record_update(plugin_id) self.state_manager.record_update(plugin_id)
# Update state back to ENABLED # Update state back to ENABLED
@@ -775,6 +786,31 @@ class PluginManager:
self.logger.exception("Error updating plugin %s: %s", plugin_id, exc) self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
self._record_update_failure(plugin_id, exc=exc) self._record_update_failure(plugin_id, exc=exc)
def run_scheduled_updates_with_changes(self, current_time: Optional[float] = None) -> List[str]:
"""
Like run_scheduled_updates(), but also returns the plugin_ids whose
plugin_last_update timestamp actually advanced during this call.
The before/after snapshots and the update pass itself are each
individually lock-protected against concurrent plugin_last_update
mutation (Vegas mode calls this from its own background
update-tick thread, racing the main render loop's plugin updates),
so callers get an atomic "who got fresh data" answer without
reaching into plugin_last_update themselves. The lock is not held
across the update pass so slow/blocking plugin update() calls don't
serialize against other plugin_last_update readers.
"""
with self._plugin_last_update_lock:
old_times = dict(self.plugin_last_update)
self.run_scheduled_updates(current_time)
with self._plugin_last_update_lock:
return [
plugin_id for plugin_id, new_time in self.plugin_last_update.items()
if new_time > old_times.get(plugin_id, 0.0)
]
def update_all_plugins(self) -> None: def update_all_plugins(self) -> None:
""" """
Update all enabled plugins. Update all enabled plugins.
@@ -797,6 +833,7 @@ class PluginManager:
try: try:
success = self.plugin_executor.execute_update(plugin_instance, plugin_id) success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
if success: if success:
with self._plugin_last_update_lock:
self.plugin_last_update[plugin_id] = time.time() self.plugin_last_update[plugin_id] = time.time()
self.state_manager.record_update(plugin_id) self.state_manager.record_update(plugin_id)
self.state_manager.set_state(plugin_id, PluginState.ENABLED) self.state_manager.set_state(plugin_id, PluginState.ENABLED)
+36 -24
View File
@@ -11,66 +11,78 @@ orphaning VegasModeCoordinator.mark_plugin_updated() -- it has had zero
callers since. callers since.
""" """
from typing import Dict, List, Optional
from unittest.mock import MagicMock from unittest.mock import MagicMock
from src.display_controller import DisplayController from src.display_controller import DisplayController
def _make_controller(plugin_last_update, vegas_coordinator=None): def _make_controller(updated: Optional[List[str]] = None, vegas_coordinator: Optional[MagicMock] = None) -> DisplayController:
dc = object.__new__(DisplayController) dc = object.__new__(DisplayController)
dc.plugin_manager = MagicMock() dc.plugin_manager = MagicMock()
dc.plugin_manager.plugin_last_update = dict(plugin_last_update) dc.plugin_manager.run_scheduled_updates_with_changes.return_value = list(updated or [])
dc.vegas_coordinator = vegas_coordinator dc.vegas_coordinator = vegas_coordinator
return dc return dc
class TestTickPluginUpdatesForVegas: class TestTickPluginUpdatesForVegas:
def test_marks_only_plugins_whose_timestamp_advanced(self): def test_marks_only_plugins_whose_timestamp_advanced(self):
dc = _make_controller({"stock-news": 100.0, "odds-ticker": 100.0})
vc = MagicMock() vc = MagicMock()
dc.vegas_coordinator = vc dc = _make_controller(updated=["stock-news"], vegas_coordinator=vc)
# Simulate run_scheduled_updates() advancing only stock-news.
def fake_tick():
dc.plugin_manager.plugin_last_update["stock-news"] = 200.0
dc._tick_plugin_updates = fake_tick
dc._tick_plugin_updates_for_vegas() dc._tick_plugin_updates_for_vegas()
vc.mark_plugin_updated.assert_called_once_with("stock-news") vc.mark_plugin_updated.assert_called_once_with("stock-news")
def test_no_advance_marks_nothing(self): def test_no_advance_marks_nothing(self):
dc = _make_controller({"stock-news": 100.0})
vc = MagicMock() vc = MagicMock()
dc.vegas_coordinator = vc dc = _make_controller(updated=[], vegas_coordinator=vc)
dc._tick_plugin_updates = lambda: None
dc._tick_plugin_updates_for_vegas() dc._tick_plugin_updates_for_vegas()
vc.mark_plugin_updated.assert_not_called() vc.mark_plugin_updated.assert_not_called()
def test_no_vegas_coordinator_does_not_raise(self): def test_no_vegas_coordinator_does_not_raise(self):
dc = _make_controller({"stock-news": 100.0}, vegas_coordinator=None) dc = _make_controller(updated=["stock-news"], vegas_coordinator=None)
def fake_tick():
dc.plugin_manager.plugin_last_update["stock-news"] = 200.0
dc._tick_plugin_updates = fake_tick
dc._tick_plugin_updates_for_vegas() # must not raise dc._tick_plugin_updates_for_vegas() # must not raise
def test_mark_plugin_updated_exception_does_not_propagate(self): def test_mark_plugin_updated_exception_does_not_propagate(self):
"""One plugin's mark_plugin_updated failing must not stop the tick """One plugin's mark_plugin_updated failing must not stop the tick
or crash the update loop it runs in.""" or crash the update loop it runs in."""
dc = _make_controller({"a": 1.0, "b": 1.0})
vc = MagicMock() vc = MagicMock()
vc.mark_plugin_updated.side_effect = [RuntimeError("boom"), None] vc.mark_plugin_updated.side_effect = [RuntimeError("boom"), None]
dc.vegas_coordinator = vc dc = _make_controller(updated=["a", "b"], vegas_coordinator=vc)
def fake_tick():
dc.plugin_manager.plugin_last_update["a"] = 2.0
dc.plugin_manager.plugin_last_update["b"] = 2.0
dc._tick_plugin_updates = fake_tick
dc._tick_plugin_updates_for_vegas() # must not raise dc._tick_plugin_updates_for_vegas() # must not raise
assert vc.mark_plugin_updated.call_count == 2 assert vc.mark_plugin_updated.call_count == 2
class TestVegasCoordinatorCallbackWiring:
def test_initialize_wires_vegas_aware_tick_as_update_callback(self):
"""The Vegas coordinator must be given the Vegas-aware
_tick_plugin_updates_for_vegas as its update callback, not the plain
_tick_plugin_updates() -- that's the exact wiring PR #330 dropped."""
dc = object.__new__(DisplayController)
dc.config = {"display": {"vegas_scroll": {"enabled": True}}, "sync": {}}
dc.display_manager = MagicMock()
dc.plugin_manager = MagicMock()
dc.sync_manager = MagicMock()
dc._check_live_priority = MagicMock()
dc._check_vegas_interrupt = MagicMock(return_value=False)
fake_coordinator = MagicMock()
import src.display_controller as dc_module
original_imported = dc_module._vegas_mode_imported
original_class = dc_module.VegasModeCoordinator
try:
dc_module._vegas_mode_imported = True
dc_module.VegasModeCoordinator = MagicMock(return_value=fake_coordinator)
dc._initialize_vegas_mode()
finally:
dc_module._vegas_mode_imported = original_imported
dc_module.VegasModeCoordinator = original_class
fake_coordinator.set_update_callback.assert_called_once_with(dc._tick_plugin_updates_for_vegas)