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

@@ -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.**

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:

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

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()