fix(db): reap leaked idle connections from the gevent pool (#1418)

The custom psycopg3 gevent pool only evaluates connection age inside
get()/put(). A connection checked out by a greenlet that dies without
returning it — e.g. a live-stream / channel-switch teardown path that
never calls close_old_connections() — is never aged out. It stays counted
against MAX_CONNS but is no longer in the free queue, so once MAX_CONNS
connections leak per worker, get() blocks forever and every DB-backed
request on that worker hangs with a 60s/504 until the container is
restarted. DB-free endpoints (e.g. /api/core/version/) stay fast, which
is the tell.

Add a background reaper greenlet (one per worker, started lazily on first
get() so the gevent hub is up) that periodically closes connections which
are both expired (older than max_lifetime) and IDLE (no in-flight
transaction). The IDLE guard means an active stream or in-flight request
is never disrupted, and get() already refuses to hand out an expired
connection so a reaped connection can't be in active use. This frees the
leaked slot and lets the worker recover on its own — no restart.

Side effect: DATABASE_POOL_CONN_MAX_LIFETIME now actually reclaims leaked
connections (previously it only recycled healthy idle ones). Lower it to
shorten leak-recovery time. Reaper cadence is configurable via
DATABASE_POOL_REAP_INTERVAL (default 30s; 0 disables).

Also covers the M3U-refresh INTRANS leak family (#1338).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

dispatcharr/db/backends/postgresql_psycopg3/pool.py

@@ -4,8 +4,27 @@ django-db-geventpool keeps warm connections open indefinitely. psycopg3 accumula
 cache on long-lived handles. We close and replace after
 CONN_MAX_LIFETIME rather than recycling uWSGI workers, which would interrupt
 live stream backends.
+
+Leak reaper
+-----------
+The upstream pool only evaluates connection age inside ``get()``/``put()``. A
+connection checked out by a greenlet that dies without returning it (e.g. a live
+stream / channel-switch teardown path that never calls ``close_old_connections()``)
+is therefore never aged out: it stays counted against ``maxsize`` but is no longer
+in the free queue, so once ``maxsize`` connections leak the pool blocks forever in
+``get()`` and every DB-backed request on that worker hangs (Dispatcharr #1418).
+
+To close that gap we run a background greenlet that periodically closes connections
+which are both **expired** (older than ``max_lifetime``) **and IDLE** (no in-flight
+transaction). An IDLE+expired connection that is not being handed back is a leaked
+one; reaping it discards it from the (weak) connection set, which frees the slot.
+The IDLE guard guarantees we never close a connection that is mid-query, so an
+active stream or in-flight request is never disrupted. This also makes
+``DATABASE_POOL_CONN_MAX_LIFETIME`` actually effective against leaked connections
+(lower it to reclaim leaked slots faster).
 """
 import logging
+import os
 import time
 
 try:
@@ -13,15 +32,84 @@ try:
 except ImportError:
     from eventlet import queue
 
+try:
+    from psycopg.pq import TransactionStatus as _TxnStatus
+    _IDLE = _TxnStatus.IDLE
+except Exception:  # pragma: no cover - psycopg always present in prod
+    _IDLE = None
+
 from django_db_geventpool.backends.pool import DatabaseConnectionPool as BasePool
 
 logger = logging.getLogger("django.geventpool")
 
+# How often the leak reaper runs, in seconds. Reaping only closes connections
+# already older than max_lifetime, so a leaked slot is reclaimed within
+# ~(max_lifetime + this interval). Set 0 to disable the reaper.
+_REAP_INTERVAL = int(os.environ.get("DATABASE_POOL_REAP_INTERVAL", "30"))
+
 
 class DatabaseConnectionPool(BasePool):
     def __init__(self, maxsize: int = 100, reuse: int = 100, max_lifetime: float | None = None):
         super().__init__(maxsize, reuse)
         self.max_lifetime = max_lifetime
+        self._reaper_started = False
+
+    def _ensure_reaper(self) -> None:
+        """Start the background leak reaper once, lazily (so the gevent hub is up)."""
+        if self._reaper_started or not self.max_lifetime or not _REAP_INTERVAL or _IDLE is None:
+            return
+        try:
+            import gevent
+        except ImportError:
+            self._reaper_started = True  # not running under gevent; nothing to do
+            return
+        self._reaper_started = True
+        gevent.spawn(self._reaper_loop)
+        logger.info(
+            "DB pool leak reaper started (interval=%ss, max_lifetime=%ss)",
+            _REAP_INTERVAL,
+            int(self.max_lifetime),
+        )
+
+    def _reaper_loop(self) -> None:
+        import gevent
+
+        while True:
+            try:
+                gevent.sleep(_REAP_INTERVAL)
+                self._reap_leaked_connections()
+            except Exception:  # never let the reaper greenlet die
+                logger.debug("DB pool reaper iteration failed", exc_info=True)
+
+    def _reap_leaked_connections(self) -> None:
+        """Close expired connections that are sitting IDLE (leaked, never returned).
+
+        Never closes a connection with an active transaction, so an in-flight
+        request or stream is never disrupted. ``get()`` already refuses to hand
+        out an expired connection, so an expired IDLE connection cannot be in
+        active use by a live greenlet.
+        """
+        if not self.max_lifetime or _IDLE is None:
+            return
+        reaped = 0
+        for conn in list(self._conns):
+            try:
+                if not self._connection_expired(conn):
+                    continue
+                if conn.info.transaction_status != _IDLE:
+                    continue  # mid-transaction: in use, leave it alone
+                self._close_connection(conn)
+                reaped += 1
+            except Exception:
+                logger.debug("Error reaping DB connection", exc_info=True)
+        if reaped:
+            logger.info(
+                "DB pool reaper closed %d expired/leaked idle connection(s); "
+                "size now %d/%d",
+                reaped,
+                self.size,
+                self.maxsize,
+            )
 
     def _stamp_connection(self, conn) -> None:
         conn._dispatcharr_pool_created_at = time.monotonic()
@@ -43,6 +131,7 @@ class DatabaseConnectionPool(BasePool):
             self._conns.discard(conn)
 
     def get(self):
+        self._ensure_reaper()
         conn = None
         try:
             if self.size >= self.maxsize or self.pool.qsize():