v0.0.14: Replace broken signal-based auto-run with DB polling monitor#61
v0.0.14: Replace broken signal-based auto-run with DB polling monitor#61cmc0619 wants to merge 1 commit into
Conversation
PROBLEM: Dispatcharr's VOD refresh now uses bulk_create/bulk_update which bypass Django post_save signals. Additionally, signals only registered in the web process but VOD refresh runs in the Celery worker process. SOLUTION: Replace the signal-based approach with a daemon thread that polls M3UAccount.updated_at every N seconds to detect completed VOD refreshes. Changes: - New: _monitor_loop/_monitor_check DB polling daemon thread - New: _is_vod_refresh_complete() heuristic on last_message - New: Start/Stop/Status auto-monitor action buttons - New: monitor_interval setting (default 60s) - Kept: AUTO_RUN_CACHE_KEY distributed lock for concurrent run prevention - Kept: auto_run_after_vod_refresh setting controls generation trigger - Removed: _ensure_signals_registered() and all post_save signal handlers - Removed: _schedule_auto_run_after_vod_refresh() debounce timer - Added: M3UAccount import from apps.m3u.models - Cost: ~2 lightweight DB queries per poll cycle
WalkthroughThe PR refactors the auto-run mechanism from a signal-based trigger system to a polling-based auto-monitor running in a daemon thread. The new system periodically checks M3U accounts for completed VOD refreshes and uses distributed cache locking to prevent concurrent generations. Version bumped to 0.0.14. Changes
Sequence DiagramsequenceDiagram
participant Plugin as Plugin Core
participant Monitor as Auto-Monitor Thread
participant M3U as M3U Accounts
participant Cache as Distributed Cache Lock
participant VOD as VOD Refresh System
Monitor->>Monitor: _monitor_loop (periodic)
Monitor->>M3U: Check all accounts
M3U-->>Monitor: Account list
loop For each M3U account
Monitor->>VOD: _is_vod_refresh_complete(account)
VOD-->>Monitor: Refresh status
alt Refresh complete & auto_run enabled
Monitor->>Cache: Acquire distributed lock
alt Lock acquired
Cache-->>Monitor: Lock granted
Monitor->>Plugin: _auto_run_generation(account)
Plugin->>Plugin: Trigger generation
Monitor->>Cache: Release lock
Cache-->>Monitor: Released
else Lock held by other
Cache-->>Monitor: Lock unavailable
Monitor->>Monitor: Skip this iteration
end
else Skip
Monitor->>Monitor: Continue to next account
end
end
Possibly related PRs
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In `@plugin.py`:
- Around line 2087-2128: The monitor thread never closes Django DB connections
causing leaks: in _monitor_check (which queries M3UAccount and may call
_auto_run_generation → _run_job_sync) ensure you close the DB connection at the
end of each poll cycle (or immediately after calling _auto_run_generation) by
invoking django.db.connection.close() (or close_old_connections()) so worker
threads don't accumulate stale connections; add the close call in the finally
block or after the trigger handling in _monitor_check to guarantee it always
runs.
- Around line 2230-2246: The stop/start race happens because _monitor_stop_event
is shared and cleared by _start_auto_monitor while an old thread (running
_auto_run_generation) is still alive, letting the old thread continue; fix by
making the stop event tied to the thread lifecycle: modify _start_auto_monitor
to create a fresh threading.Event for each new thread (e.g., new_event) and
pass/store that event alongside the thread (replace the global shared
_monitor_stop_event with a per-thread event reference), and modify
_stop_auto_monitor to set and join the specific thread's event/thread (use
_monitor_lock to atomically read the current _monitor_thread and its associated
event, set that event, join it and only clear or replace the global reference
after the join confirms the old thread is dead); ensure _auto_run_generation
reads the per-thread event reference rather than a global so a new monitor start
cannot clear the old thread's stop signal.
🧹 Nitpick comments (4)
plugin.py (4)
2066-2084: Fragile heuristic — consider documenting the Dispatcharr message source.The substring matches (
"streams processed","refreshed successfully", etc.) are brittle and could false-positive on non-VOD operations or silently break if Dispatcharr changes its message format. Consider adding a comment pointing to where these messages originate in Dispatcharr, so future maintainers can update them if needed.
2146-2153: Usegetattrforplugin_config.enabledto be defensive.
plugin_config.enabledis accessed directly without checking if the field exists. Per coding guidelines, usegetattr(obj, 'field', default)instead of direct attribute access on Django model fields.Proposed fix
- if not plugin_config or not plugin_config.enabled: + if not plugin_config or not getattr(plugin_config, "enabled", True):As per coding guidelines, "Always check if Django model fields exist using
getattr(obj, 'field', default)instead of direct attribute access".
2176-2176: Redundant exception object inlogging.exceptioncall.
LOGGER.exception()already captures the traceback. The%s, eformat arg duplicates the error message in the log line.Proposed fix
- LOGGER.exception("Auto-monitor: STRM generation failed: %s", e) + LOGGER.exception("Auto-monitor: STRM generation failed")
2160-2173: Consider dispatching generation via Celery instead of running synchronously in the monitor thread.
_auto_run_generation()calls_run_job_sync()directly in the daemon thread, blocking all polling for the entire duration of the generation job (potentially minutes to hours). This also compounds the DB connection leak issue since_run_job_syncspawnsThreadPoolExecutorworkers that each open their own connections.Dispatching via Celery (
celery_run_job.delay(args)) would:
- Keep the monitor thread responsive for subsequent poll cycles
- Leverage Celery's
task_postrunconnection cleanup (per coding guidelines)- Maintain the existing cache lock pattern for concurrency control
Proposed refactor
def _auto_run_generation() -> None: from django.core.cache import cache LOGGER.info("Auto-monitor: triggering STRM generation") lock_acquired = cache.add(AUTO_RUN_CACHE_KEY, "locked", timeout=3600) if not lock_acquired: LOGGER.info("Auto-monitor: generation already in progress (lock held), skipping") return try: from apps.plugins.models import PluginConfig plugin_config = PluginConfig.objects.filter(key="vod2strm").first() - if not plugin_config or not plugin_config.enabled: + if not plugin_config or not getattr(plugin_config, "enabled", True): LOGGER.info("Auto-monitor: plugin disabled, skipping generation") return current_settings = plugin_config.settings or {} if not current_settings.get("auto_run_after_vod_refresh", False): LOGGER.info("Auto-monitor: 'Auto-run after VOD Refresh' setting is disabled, skipping") return - _run_job_sync( - mode="all", - output_root=current_settings.get("output_root") or DEFAULT_ROOT, - base_url=current_settings.get("base_url") or DEFAULT_BASE_URL, - write_nfos=bool(current_settings.get("write_nfos", True)), - cleanup_mode=current_settings.get("cleanup_mode", CLEANUP_OFF), - concurrency=int(current_settings.get("concurrency") or 4), - debug_logging=bool(current_settings.get("debug_logging", False)), - dry_run=False, - adaptive_throttle=bool(current_settings.get("adaptive_throttle", True)), - clean_regex=(current_settings.get("name_clean_regex") or "").strip() or None, - use_direct_urls=bool(current_settings.get("use_direct_urls", False)), - multi_provider_mode=bool(current_settings.get("multi_provider_mode", False)), - ) - LOGGER.info("Auto-monitor: STRM generation completed successfully") - except Exception as e: - LOGGER.exception("Auto-monitor: STRM generation failed") + args = { + "mode": "all", + "output_root": current_settings.get("output_root") or DEFAULT_ROOT, + "base_url": current_settings.get("base_url") or DEFAULT_BASE_URL, + "write_nfos": bool(current_settings.get("write_nfos", True)), + "cleanup_mode": current_settings.get("cleanup_mode", CLEANUP_OFF), + "concurrency": int(current_settings.get("concurrency") or 4), + "debug_logging": bool(current_settings.get("debug_logging", False)), + "dry_run": False, + "adaptive_throttle": bool(current_settings.get("adaptive_throttle", True)), + "clean_regex": (current_settings.get("name_clean_regex") or "").strip() or None, + "use_direct_urls": bool(current_settings.get("use_direct_urls", False)), + "multi_provider_mode": bool(current_settings.get("multi_provider_mode", False)), + } + if celery_run_job is not None: + celery_run_job.delay(args) + LOGGER.info("Auto-monitor: STRM generation task dispatched to Celery") + else: + celery_app.send_task("vod2strm.plugin.run_job", args=[args]) + LOGGER.info("Auto-monitor: STRM generation task sent by name") + except Exception: + LOGGER.exception("Auto-monitor: failed to dispatch STRM generation") finally: - cache.delete(AUTO_RUN_CACHE_KEY) - LOGGER.debug("Auto-monitor: generation lock released") + # NOTE: When using Celery, move lock release to the task's finally block + # or use cache timeout as the safety net. The lock is released here only + # if dispatch fails; successful task should release it upon completion. + passNote: If you switch to Celery dispatch, the lock lifecycle needs rethinking — the Celery task should release the lock, not
_auto_run_generation. The existing 3600s cache timeout serves as a safety net.
| def _monitor_check() -> None: | ||
| """ | ||
| Single poll cycle: check all VOD-enabled M3U accounts for completed refreshes. | ||
| If a refresh is detected, trigger full STRM generation. | ||
| """ | ||
| global _monitor_last_seen | ||
|
|
||
| try: | ||
| accounts = M3UAccount.objects.filter( | ||
| is_active=True, | ||
| enable_vod=True, | ||
| ).only("id", "name", "updated_at", "last_message") | ||
|
|
||
| triggered = False | ||
| for acct in accounts: | ||
| updated = acct.updated_at | ||
| if updated is None: | ||
| continue | ||
|
|
||
| last_seen = _monitor_last_seen.get(acct.id) | ||
|
|
||
| # First time seeing this account — record timestamp, don't trigger | ||
| if last_seen is None: | ||
| _monitor_last_seen[acct.id] = updated | ||
| continue | ||
|
|
||
| # Account was updated since we last checked | ||
| if updated > last_seen: | ||
| _monitor_last_seen[acct.id] = updated | ||
|
|
||
| if _is_vod_refresh_complete(acct): | ||
| LOGGER.info( | ||
| "Auto-monitor: VOD refresh detected for account '%s' (id=%s, updated=%s, msg='%s')", | ||
| acct.name, acct.id, updated, (acct.last_message or "")[:80] | ||
| ) | ||
| triggered = True | ||
|
|
||
| if triggered: | ||
| _auto_run_generation() | ||
|
|
||
| except Exception as e: | ||
| LOGGER.warning("Auto-monitor check failed: %s", e) |
There was a problem hiding this comment.
Django DB connections are never closed in the daemon thread — connection leak risk.
The monitor thread repeatedly queries the database (M3UAccount.objects.filter(...)) and, on trigger, runs _auto_run_generation() → _run_job_sync() which spawns ThreadPoolExecutor workers that also touch the DB. Django does not automatically close connections in non-request threads, so these connections accumulate or go stale.
Call connection.close() at the end of each poll cycle (or at least after _auto_run_generation()) to return the connection to the pool.
Proposed fix in `_monitor_loop`
def _monitor_loop(interval: int) -> None:
LOGGER.info("Auto-monitor thread started (poll interval: %ds)", interval)
while not _monitor_stop_event.is_set():
- _monitor_check()
+ try:
+ _monitor_check()
+ finally:
+ # Django doesn't auto-close DB connections in non-request threads.
+ # Close after each cycle to prevent stale / leaked connections.
+ from django.db import connection as _conn
+ _conn.close()
_monitor_stop_event.wait(timeout=interval)
LOGGER.info("Auto-monitor thread stopped")As per coding guidelines, "Close Django database connections in Celery task_postrun signal handlers to prevent connection leaks".
🧰 Tools
🪛 Ruff (0.14.14)
[warning] 2127-2127: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
In `@plugin.py` around lines 2087 - 2128, The monitor thread never closes Django
DB connections causing leaks: in _monitor_check (which queries M3UAccount and
may call _auto_run_generation → _run_job_sync) ensure you close the DB
connection at the end of each poll cycle (or immediately after calling
_auto_run_generation) by invoking django.db.connection.close() (or
close_old_connections()) so worker threads don't accumulate stale connections;
add the close call in the finally block or after the trigger handling in
_monitor_check to guarantee it always runs.
| def _stop_auto_monitor() -> dict: | ||
| """Stop the auto-monitor daemon thread.""" | ||
| global _monitor_thread | ||
|
|
||
| Module-level code executes during phase 1 (DB not ready), but Python's | ||
| module caching prevents it from running again in phase 2. This function | ||
| provides lazy initialization that runs when Plugin.run() is first called, | ||
| ensuring signals register after the database is fully ready. | ||
| """ | ||
| global _SIGNALS_REGISTERED | ||
| with _monitor_lock: | ||
| if _monitor_thread is None or not _monitor_thread.is_alive(): | ||
| _monitor_thread = None | ||
| return {"status": "ok", "message": "Auto-monitor is not running."} | ||
|
|
||
| # Use lock to prevent race conditions during concurrent initialization | ||
| with _LOG_LOCK: | ||
| if _SIGNALS_REGISTERED: | ||
| return # Already registered | ||
| _monitor_stop_event.set() | ||
| _monitor_thread.join(timeout=10) | ||
| alive = _monitor_thread.is_alive() | ||
| _monitor_thread = None | ||
|
|
||
| try: | ||
| from django.db.models.signals import post_save | ||
| from django.dispatch import receiver | ||
| if alive: | ||
| return {"status": "ok", "message": "Auto-monitor stop requested (thread still winding down)."} | ||
| return {"status": "ok", "message": "Auto-monitor stopped."} |
There was a problem hiding this comment.
Race condition: rapid stop → start can leave two monitor threads running.
If _stop_auto_monitor is called while the thread is blocked in _auto_run_generation(), the join(timeout=10) returns without the thread having exited. _monitor_thread is set to None. A subsequent _start_auto_monitor calls _monitor_stop_event.clear(), which also clears the stop signal for the old thread. When the old thread finishes its generation and re-enters the loop, it sees _monitor_stop_event.is_set() → False and continues polling — now you have two monitor threads.
Since both threads call _monitor_check() → _auto_run_generation(), the cache lock prevents concurrent generation, but you still get double polling and double DB queries.
Fix: use a fresh Event per thread lifecycle, or keep a reference to the old thread and ensure it has exited before clearing the event.
Proposed fix — don't clear the event until the old thread is dead
def _start_auto_monitor(settings: dict) -> dict:
global _monitor_thread
with _monitor_lock:
- # Check if already running
- if _monitor_thread is not None and _monitor_thread.is_alive():
- return {"status": "ok", "message": "Auto-monitor is already running."}
+ # If a previous thread is still winding down, wait for it
+ if _monitor_thread is not None:
+ if _monitor_thread.is_alive():
+ _monitor_stop_event.set()
+ _monitor_thread.join(timeout=30)
+ if _monitor_thread.is_alive():
+ return {
+ "status": "error",
+ "message": "Previous monitor thread is still running (stuck in generation?). Try again later.",
+ }
+ _monitor_thread = None
if not settings.get("auto_run_after_vod_refresh", False):🤖 Prompt for AI Agents
In `@plugin.py` around lines 2230 - 2246, The stop/start race happens because
_monitor_stop_event is shared and cleared by _start_auto_monitor while an old
thread (running _auto_run_generation) is still alive, letting the old thread
continue; fix by making the stop event tied to the thread lifecycle: modify
_start_auto_monitor to create a fresh threading.Event for each new thread (e.g.,
new_event) and pass/store that event alongside the thread (replace the global
shared _monitor_stop_event with a per-thread event reference), and modify
_stop_auto_monitor to set and join the specific thread's event/thread (use
_monitor_lock to atomically read the current _monitor_thread and its associated
event, set that event, join it and only clear or replace the global reference
after the join confirms the old thread is dead); ensure _auto_run_generation
reads the per-thread event reference rather than a global so a new monitor start
cannot clear the old thread's stop signal.
…ith DB polling monitor
1 participant
Problem
The auto-run feature stopped working after Dispatcharr refactored its VOD refresh pipeline to use
bulk_create/bulk_update, which bypass Django'spost_savesignals entirely. Additionally, the signal handlers only registered in the web process (viaPlugin.run()→_ensure_signals_registered()), but VOD refresh runs in the Celery worker process where plugin code isn't loaded.Solution
Replace the signal-based approach with a lightweight DB polling daemon thread.
How it works
M3UAccount.updated_atevery N seconds (configurable, default 60s)updated_atadvances ANDlast_messageindicates a completed VOD refresh → triggers_run_job_sync(mode="all")with fresh settings from DBAUTO_RUN_CACHE_KEY) prevents concurrent runsCost
~2 lightweight DB queries per poll cycle. No process boundary issues, no monkey-patching, no upstream code changes.
Changes
Added
_monitor_loop/_monitor_check— daemon thread + poll cycle_is_vod_refresh_complete()— heuristic checkinglast_messagefor completion patterns_auto_run_generation()— fresh-settings loader + distributed lock +_run_job_synccall_start_auto_monitor/_stop_auto_monitor/_monitor_status— action handlersmonitor_intervalsetting (default 60s, minimum 10s)M3UAccountimport fromapps.m3u.modelsRemoved
_ensure_signals_registered()and all@receiver(post_save, ...)signal handlers_schedule_auto_run_after_vod_refresh()debounce timer_SIGNALS_REGISTEREDglobal flagKept (unchanged)
AUTO_RUN_CACHE_KEYdistributed lock patternauto_run_after_vod_refreshsetting (now controls whether monitor triggers generation)Summary by CodeRabbit