From db40421faa228b59e47125f8e40b188f2d103d7b Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Mon, 15 Jun 2026 16:50:08 -0500 Subject: [PATCH] fix(plugins): ensure proper database connection management in plugin actions - Added calls to `close_old_connections()` in `run_action()` and `stop_plugin()` methods to prevent connection leaks after plugin execution. - Updated documentation to clarify connection handling for plugins, emphasizing the importance of cleanup in long-running tasks and event hooks. --- CHANGELOG.md | 1 + Plugins.md | 12 ++ apps/plugins/loader.py | 120 +++++++++--------- apps/plugins/tests/__init__.py | 0 .../tests/test_run_action_db_cleanup.py | 64 ++++++++++ 5 files changed, 140 insertions(+), 57 deletions(-) create mode 100644 apps/plugins/tests/__init__.py create mode 100644 apps/plugins/tests/test_run_action_db_cleanup.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 7e40e04a..01ab218f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Orphaned upstream threads kept writing Redis after teardown.** Partial cleanup could remove a channel from `stream_managers` while the `stream-{uuid}` OS thread kept running, so orphan metadata sweeps only deleted Redis keys and chunks/metadata were immediately recreated (bare `total_bytes` hash with TTL -1). Stream managers now stop upstream when ownership is lost, orphan cleanup stops local ffmpeg/stream threads even when registry entries are gone (`_live_stream_managers` tracks managers until the thread exits), and forced recovery stops processes before Redis deletion instead of dropping dict entries without calling `stop()`. - **Rapid channel switching could leave bare `buffer:index` keys in Redis.** `buffer.stop()` flushed a final partial chunk via `INCR` during teardown, creating a persistent index key (TTL -1) after chunk keys expired; under load overlapping disconnect and cleanup-thread stops could race on `_stop_local_stream_activity`, blocking on ffmpeg stderr join before `_clean_redis_keys` ever ran. Teardown no longer writes to Redis from `buffer.stop()`, disconnect uses coordinated stop only (no duplicate upstream stop), model `release_stream()` runs before Redis keys are scanned/deleted (so `profile_connections` counters are not left stuck), ownership is released after Redis cleanup but before the blocking local ffmpeg stop, and a `finally` block guarantees Redis cleanup if local teardown raises. - **Live proxy could leak geventpool DB checkouts outside HTTP requests.** With `django-db-geventpool`, `connection.close()` returns handles to the per-worker pool (`MAX_CONNS=8`); without it, greenlets and OS threads keep connections checked out until the pool blocks on `pool.get()`. Proxy paths that touch the ORM outside Django's request cycle now call `close_old_connections()` after work completes: `log_system_event()` (all system events), `_clean_redis_keys()`, channel init/start logging, client disconnect cleanup (TS and fMP4 generators), and profile lookup in `_establish_transcode_connection()` before spawning ffmpeg. +- **Plugins could leak geventpool DB checkouts after UI or Connect event runs.** Third-party plugins run inline on uWSGI greenlets (manual actions and Connect `"events"` hooks) with no guaranteed connection cleanup at the plugin boundary. `PluginManager.run_action()` and `stop_plugin()` now call `close_old_connections()` in a `finally` block so each action returns its pool slot whether the plugin succeeds or raises. - **Zombie ffmpeg could survive owner-lock expiry and orphan Redis sweeps.** When the 30s owner lock lapsed under single-worker load, disconnect handling treated the worker as non-owner so coordinated stop never ran, while ffmpeg kept writing and the orphan sweeper only deleted Redis keys (recreating them immediately). Disconnect now re-acquires ownership when local upstream is still active, stops locally when the last client leaves without a lock, orphan cleanup stops local ffmpeg before Redis deletion via `_has_local_upstream_activity`, re-init stops lingering upstream before starting a duplicate thread, and `is_channel_teardown_active` includes channels mid-`stop_channel` on this worker so rapid reconnect gets 503 during teardown. - **Stale `channel_stream` Redis keys after a channel stopped could skip connection accounting on retune.** On `dev`, `get_stream()` reused any existing `channel_stream` / `stream_profile` assignment without reserving a new slot. If those keys were left behind after a stop, the next tune-in could reach the provider without incrementing Redis counters. `get_stream()` now releases stale assignments when proxy metadata shows the channel is inactive, then reserves fresh slots. - **EPG auto-match reliability fixes.** diff --git a/Plugins.md b/Plugins.md index d528f1e4..97e95ed2 100644 --- a/Plugins.md +++ b/Plugins.md @@ -307,6 +307,18 @@ Plugins are server-side Python code running within the Django application. You c Prefer Celery tasks (`.delay()`) to keep `run` fast and non-blocking. +### Database connections + +Dispatcharr uses `django-db-geventpool` with a bounded per-uWSGI-worker pool (`MAX_CONNS=8`). Each greenlet or OS thread that runs ORM code checks out a connection until Django closes it. + +`PluginManager.run_action()` and `stop_plugin()` always call `close_old_connections()` in a `finally` block after your plugin returns (success or error). That returns the current greenlet's checkout to the pool. **You do not need to call `close_old_connections()` yourself for normal inline ORM inside `run()` or `stop()`.** + +Still follow these rules: + +- **Heavy or long work:** dispatch a Celery task (`.delay()`) and return quickly from `run()`. Celery workers close connections after each task; blocking the uWSGI gevent hub with `time.sleep`, sync HTTP, or large CPU work can freeze the whole worker regardless of DB cleanup. +- **Background threads or greenlets you spawn:** each thread/greenlet that uses the ORM must call `close_old_connections()` (or `connection.close()`) in its own `finally` block when done. The wrapper only covers the thread/greenlet that called `run_action()`. +- **Connect event hooks:** actions with an `"events"` list run synchronously inside whatever caller triggered the event (for example a live-proxy system event). Keep event handlers short or defer to Celery. + ### Important: Don’t Ask Users for URL/User/Password Dispatcharr plugins run **inside** the Dispatcharr backend process. That means they already have direct access to the app’s models, tasks, and internal utilities. Plugins **should not** ask users for “Dispatcharr URL”, “Admin Username”, or “Admin Password” just to call the API. That is unnecessary and unsafe because: diff --git a/apps/plugins/loader.py b/apps/plugins/loader.py index a7ecb255..3c275f69 100644 --- a/apps/plugins/loader.py +++ b/apps/plugins/loader.py @@ -10,7 +10,7 @@ import types from dataclasses import dataclass, field from typing import Any, Dict, List, Optional -from django.db import transaction +from django.db import close_old_connections, transaction from .models import PluginConfig @@ -492,79 +492,85 @@ class PluginManager: return cfg.settings def run_action(self, key: str, action_id: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: - lp = self.get_plugin(key) - if not lp or not lp.instance: - # Attempt a lightweight re-discovery in case the registry was rebuilt - self.discover_plugins(sync_db=False, force_reload=False, use_cache=False) + try: lp = self.get_plugin(key) if not lp or not lp.instance: - raise ValueError(f"Plugin '{key}' not found") + # Attempt a lightweight re-discovery in case the registry was rebuilt + self.discover_plugins(sync_db=False, force_reload=False, use_cache=False) + lp = self.get_plugin(key) + if not lp or not lp.instance: + raise ValueError(f"Plugin '{key}' not found") - cfg = PluginConfig.objects.get(key=key) - if not cfg.enabled: - raise PermissionError(f"Plugin '{key}' is disabled") - params = params or {} + cfg = PluginConfig.objects.get(key=key) + if not cfg.enabled: + raise PermissionError(f"Plugin '{key}' is disabled") + params = params or {} - context = self._build_context(lp, cfg) + context = self._build_context(lp, cfg) - # Run either via Celery if plugin provides a delayed method, or inline - run_method = getattr(lp.instance, "run", None) - if not callable(run_method): - raise ValueError(f"Plugin '{key}' has no runnable 'run' method") + run_method = getattr(lp.instance, "run", None) + if not callable(run_method): + raise ValueError(f"Plugin '{key}' has no runnable 'run' method") - try: - result = run_method(action_id, params, context) - except Exception: - logger.exception(f"Plugin '{key}' action '{action_id}' failed") - raise + try: + result = run_method(action_id, params, context) + except Exception: + logger.exception(f"Plugin '{key}' action '{action_id}' failed") + raise - # Normalize return - if isinstance(result, dict): - return result - return {"status": "ok", "result": result} + if isinstance(result, dict): + return result + return {"status": "ok", "result": result} + finally: + # Return geventpool checkouts for this greenlet/thread after every action, + # including Connect event hooks and manual UI runs. + close_old_connections() def stop_plugin(self, key: str, reason: Optional[str] = None) -> bool: - lp = self.get_plugin(key) - if not lp or not lp.instance: - return False try: - cfg = PluginConfig.objects.get(key=key) - except PluginConfig.DoesNotExist: - return False - if not cfg.enabled: - return False - - context = self._build_context(lp, cfg) - if reason: - context["reason"] = reason - - stop_method = getattr(lp.instance, "stop", None) - if callable(stop_method): + lp = self.get_plugin(key) + if not lp or not lp.instance: + return False try: - stop_method(context) - return True - except TypeError: + cfg = PluginConfig.objects.get(key=key) + except PluginConfig.DoesNotExist: + return False + if not cfg.enabled: + return False + + context = self._build_context(lp, cfg) + if reason: + context["reason"] = reason + + stop_method = getattr(lp.instance, "stop", None) + if callable(stop_method): try: - stop_method() + stop_method(context) return True + except TypeError: + try: + stop_method() + return True + except Exception: + logger.exception("Plugin '%s' stop() failed", key) + return False except Exception: logger.exception("Plugin '%s' stop() failed", key) return False - except Exception: - logger.exception("Plugin '%s' stop() failed", key) - return False - run_method = getattr(lp.instance, "run", None) - if callable(run_method): - actions = {a.get("id") for a in (lp.actions or []) if isinstance(a, dict)} - if "stop" in actions: - try: - run_method("stop", {}, context) - return True - except Exception: - logger.exception("Plugin '%s' stop action failed", key) - return False - return False + run_method = getattr(lp.instance, "run", None) + if callable(run_method): + actions = {a.get("id") for a in (lp.actions or []) if isinstance(a, dict)} + if "stop" in actions: + try: + run_method("stop", {}, context) + return True + except Exception: + logger.exception("Plugin '%s' stop action failed", key) + return False + return False + finally: + close_old_connections() def stop_all_plugins(self, reason: Optional[str] = None) -> int: stopped = 0 diff --git a/apps/plugins/tests/__init__.py b/apps/plugins/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/apps/plugins/tests/test_run_action_db_cleanup.py b/apps/plugins/tests/test_run_action_db_cleanup.py new file mode 100644 index 00000000..d9a45c88 --- /dev/null +++ b/apps/plugins/tests/test_run_action_db_cleanup.py @@ -0,0 +1,64 @@ +"""PluginManager must release geventpool checkouts after every run/stop.""" + +from contextlib import contextmanager +from unittest.mock import MagicMock, patch + +from django.test import SimpleTestCase + +from apps.plugins.loader import LoadedPlugin, PluginManager + + +class PluginRunActionDbCleanupTests(SimpleTestCase): + @contextmanager + def _manager_with_plugin(self, run_impl): + instance = MagicMock() + instance.run = run_impl + lp = LoadedPlugin( + key="test_plugin", + name="Test Plugin", + instance=instance, + actions=[{"id": "do_work"}], + ) + pm = PluginManager() + cfg = MagicMock(enabled=True, settings={}) + with patch.object(pm, "get_plugin", return_value=lp), patch( + "apps.plugins.loader.PluginConfig.objects.get", return_value=cfg + ): + yield pm + + @patch("apps.plugins.loader.close_old_connections") + def test_run_action_closes_connections_on_success(self, mock_close): + with self._manager_with_plugin(lambda *_a, **_k: {"status": "ok"}) as pm: + result = pm.run_action("test_plugin", "do_work") + + self.assertEqual(result, {"status": "ok"}) + mock_close.assert_called_once() + + @patch("apps.plugins.loader.close_old_connections") + def test_run_action_closes_connections_on_plugin_error(self, mock_close): + def _boom(*_a, **_k): + raise RuntimeError("plugin failed") + + with self._manager_with_plugin(_boom) as pm: + with self.assertRaises(RuntimeError): + pm.run_action("test_plugin", "do_work") + + mock_close.assert_called_once() + + @patch("apps.plugins.loader.close_old_connections") + def test_stop_plugin_closes_connections(self, mock_close): + instance = MagicMock() + instance.stop = MagicMock() + lp = LoadedPlugin( + key="test_plugin", + name="Test Plugin", + instance=instance, + ) + pm = PluginManager() + cfg = MagicMock(enabled=True, settings={}) + with patch.object(pm, "get_plugin", return_value=lp), patch( + "apps.plugins.loader.PluginConfig.objects.get", return_value=cfg + ): + self.assertTrue(pm.stop_plugin("test_plugin", reason="shutdown")) + + mock_close.assert_called_once()