feat(settings): add DATABASE_POOL_CONN_MAX_LIFETIME for connection management and update database engine path (Fixes #1343)

- Introduced DATABASE_POOL_CONN_MAX_LIFETIME to manage pooled connection lifecycle, improving performance and resource management.
- Updated database engine path to use 'dispatcharr.db.backends.postgresql_psycopg3' for consistency.
- Enhanced logging configuration for geventpool connection lifecycle tracking.

dispatcharr/db/backends/postgresql_psycopg3/base.py

@@ -0,0 +1,44 @@
+try:
+    import psycopg
+except ImportError as e:
+    from django.core.exceptions import ImproperlyConfigured
+
+    raise ImproperlyConfigured("Error loading psycopg3 module: %s" % e) from e
+
+from django.db.backends.postgresql.base import DatabaseWrapper as OriginalDatabaseWrapper
+from django_db_geventpool.backends.base import DatabaseWrapperMixin
+
+from .pool import DatabaseConnectionPool
+
+
+class PostgresConnectionPool(DatabaseConnectionPool):
+    DBERROR = psycopg.DatabaseError
+
+    def __init__(self, *args, **kwargs):
+        self.connect = kwargs.pop("connect", psycopg.connect)
+        self.connection = None
+        maxsize = kwargs.pop("MAX_CONNS", 4)
+        reuse = kwargs.pop("REUSE_CONNS", maxsize)
+        max_lifetime = kwargs.pop("CONN_MAX_LIFETIME", None)
+        self.args = args
+        self.kwargs = kwargs
+        self.kwargs["client_encoding"] = "UTF8"
+        super().__init__(maxsize, reuse, max_lifetime=max_lifetime)
+
+    def create_connection(self):
+        return self.connect(*self.args, **self.kwargs)
+
+    def check_usable(self, connection):
+        connection.cursor().execute("SELECT 1")
+
+
+class DatabaseWrapper(DatabaseWrapperMixin, OriginalDatabaseWrapper):
+    pool_class = PostgresConnectionPool
+    INTRANS = psycopg.pq.TransactionStatus.INTRANS
+
+    def get_connection_params(self) -> dict:
+        conn_params = super().get_connection_params()
+        for attr in ("MAX_CONNS", "REUSE_CONNS", "CONN_MAX_LIFETIME"):
+            if attr in self.settings_dict["OPTIONS"]:
+                conn_params[attr] = self.settings_dict["OPTIONS"][attr]
+        return conn_params

dispatcharr/db/backends/postgresql_psycopg3/pool.py

@@ -0,0 +1,92 @@
+"""geventpool with bounded connection lifetime.
+
+django-db-geventpool keeps warm connections open indefinitely. psycopg3 accumulates
+cache on long-lived handles. We close and replace after
+CONN_MAX_LIFETIME rather than recycling uWSGI workers, which would interrupt
+live stream backends.
+"""
+import logging
+import time
+
+try:
+    from gevent import queue
+except ImportError:
+    from eventlet import queue
+
+from django_db_geventpool.backends.pool import DatabaseConnectionPool as BasePool
+
+logger = logging.getLogger("django.geventpool")
+
+
+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
+
+    def _stamp_connection(self, conn) -> None:
+        conn._dispatcharr_pool_created_at = time.monotonic()
+
+    def _connection_expired(self, conn) -> bool:
+        if not self.max_lifetime:
+            return False
+        created_at = getattr(conn, "_dispatcharr_pool_created_at", None)
+        if created_at is None:
+            return False
+        return (time.monotonic() - created_at) >= self.max_lifetime
+
+    def _close_connection(self, conn) -> None:
+        try:
+            conn.close()
+        except Exception:
+            logger.debug("Error closing pool connection", exc_info=True)
+        finally:
+            self._conns.discard(conn)
+
+    def get(self):
+        conn = None
+        try:
+            if self.size >= self.maxsize or self.pool.qsize():
+                conn = self.pool.get()
+            else:
+                conn = self.pool.get_nowait()
+
+            if conn is not None and self._connection_expired(conn):
+                logger.debug(
+                    "DB connection expired after %ss, replacing",
+                    int(self.max_lifetime),
+                )
+                self._close_connection(conn)
+                conn = None
+            elif conn is not None:
+                try:
+                    self.check_usable(conn)
+                    logger.trace("DB connection reused")
+                except self.DBERROR:
+                    logger.debug("DB connection was closed, creating a new one")
+                    self._close_connection(conn)
+                    conn = None
+        except queue.Empty:
+            conn = None
+            logger.trace("DB connection queue empty, creating a new one")
+
+        if conn is None:
+            conn = self.create_connection()
+            self._stamp_connection(conn)
+            self._conns.add(conn)
+
+        return conn
+
+    def put(self, item):
+        if self._connection_expired(item):
+            logger.debug(
+                "DB connection expired after %ss on return, closing",
+                int(self.max_lifetime),
+            )
+            self._close_connection(item)
+            return
+
+        try:
+            self.pool.put_nowait(item)
+            logger.trace("DB connection returned to the pool")
+        except queue.Full:
+            self._close_connection(item)

dispatcharr/settings.py

@@ -114,6 +114,12 @@ XC_PROFILE_REFRESH_DELAY = float(os.environ.get('XC_PROFILE_REFRESH_DELAY', '2.5
 # Database optimization settings
 DATABASE_STATEMENT_TIMEOUT = 300  # Seconds before timing out long-running queries
 DATABASE_CONN_MAX_AGE = 0  # geventpool intercepts close(); pool handles reuse
+# Pooled connections are closed and replaced after this many seconds (per worker).
+# psycopg3 cache grows on long-lived handles; rotating bounds RAM without recycling
+# uWSGI workers (which would interrupt live streams). Reuse within the window keeps
+# the warm-pool performance win over opening a new TCP session every request.
+# Override via DATABASE_POOL_CONN_MAX_LIFETIME; set 0 to disable. Default 600 (10 min).
+DATABASE_POOL_CONN_MAX_LIFETIME = int(os.environ.get("DATABASE_POOL_CONN_MAX_LIFETIME", "600"))
 
 # Disable atomic requests for performance-sensitive views
 ATOMIC_REQUESTS = False
@@ -223,7 +229,7 @@ if os.getenv("DB_ENGINE", None) == "sqlite":
 else:
     DATABASES = {
         "default": {
-            "ENGINE": "django_db_geventpool.backends.postgresql_psycopg3",
+            "ENGINE": "dispatcharr.db.backends.postgresql_psycopg3",
             "NAME": os.environ.get("POSTGRES_DB", "dispatcharr"),
             "USER": os.environ.get("POSTGRES_USER", "dispatch"),
             "PASSWORD": os.environ.get("POSTGRES_PASSWORD", "secret"),
@@ -234,6 +240,7 @@ else:
                 "MAX_CONNS": 8,   # Per-worker pool size; 4 workers × 8 = 32 total < pg max_connections=100
                 "REUSE_CONNS": 3, # Connections to keep warm between requests
                 "pool": False,    # Disable Django's native psycopg3 pool; geventpool manages connections
+                "CONN_MAX_LIFETIME": DATABASE_POOL_CONN_MAX_LIFETIME or None,
             },
         }
     }
@@ -557,6 +564,12 @@ LOGGING = {
             "level": LOG_LEVEL,  # Use configured log level for scheduler logs
             "propagate": False,
         },
+        # geventpool connection lifecycle
+        "django.geventpool": {
+            "handlers": ["console"],
+            "level": LOG_LEVEL,
+            "propagate": False,
+        },
         # Add any other loggers you need to capture TRACE logs from
     },
     "root": {

tests/test_geventpool_conn_lifetime.py

@@ -0,0 +1,65 @@
+"""Tests for dispatcharr geventpool connection max lifetime."""
+from unittest import TestCase
+from unittest.mock import MagicMock, patch
+
+from dispatcharr.db.backends.postgresql_psycopg3.pool import DatabaseConnectionPool
+
+
+class _TestPool(DatabaseConnectionPool):
+    DBERROR = Exception
+
+    def create_connection(self):
+        return MagicMock(name="connection", closed=False)
+
+    def check_usable(self, connection):
+        return None
+
+
+class GeventPoolConnLifetimeTests(TestCase):
+    def test_fresh_connection_is_stamped_on_create(self):
+        pool = _TestPool(maxsize=2, reuse=2, max_lifetime=3600)
+        conn = pool.get()
+        self.assertIsNotNone(getattr(conn, "_dispatcharr_pool_created_at", None))
+
+    def test_expired_connection_on_get_is_replaced(self):
+        pool = _TestPool(maxsize=2, reuse=2, max_lifetime=60)
+        conn = pool.create_connection()
+        pool._stamp_connection(conn)
+        conn._dispatcharr_pool_created_at = 0
+        conn.close = MagicMock()
+        pool._conns.add(conn)
+        pool.pool.put_nowait(conn)
+
+        with patch("dispatcharr.db.backends.postgresql_psycopg3.pool.time.monotonic", return_value=120):
+            new_conn = pool.get()
+
+        conn.close.assert_called_once()
+        self.assertIsNot(new_conn, conn)
+
+    def test_expired_connection_on_put_is_closed_not_pooled(self):
+        pool = _TestPool(maxsize=2, reuse=2, max_lifetime=60)
+        conn = pool.get()
+        conn.close = MagicMock()
+        conn._dispatcharr_pool_created_at = 0
+
+        with patch("dispatcharr.db.backends.postgresql_psycopg3.pool.time.monotonic", return_value=120):
+            pool.put(conn)
+
+        conn.close.assert_called_once()
+        self.assertEqual(pool.pool.qsize(), 0)
+
+    def test_non_expired_connection_is_returned_to_pool(self):
+        pool = _TestPool(maxsize=2, reuse=2, max_lifetime=3600)
+        conn = pool.get()
+        pool.put(conn)
+        self.assertEqual(pool.pool.qsize(), 1)
+
+    def test_max_lifetime_disabled_when_none(self):
+        pool = _TestPool(maxsize=2, reuse=2, max_lifetime=None)
+        conn = pool.get()
+        conn._dispatcharr_pool_created_at = 0
+
+        with patch("dispatcharr.db.backends.postgresql_psycopg3.pool.time.monotonic", return_value=999999):
+            pool.put(conn)
+
+        self.assertEqual(pool.pool.qsize(), 1)