diff --git a/dispatcharr/db/__init__.py b/dispatcharr/db/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/dispatcharr/db/backends/__init__.py b/dispatcharr/db/backends/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/dispatcharr/db/backends/postgresql_psycopg3/__init__.py b/dispatcharr/db/backends/postgresql_psycopg3/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/dispatcharr/db/backends/postgresql_psycopg3/base.py b/dispatcharr/db/backends/postgresql_psycopg3/base.py new file mode 100644 index 00000000..f5a49ffd --- /dev/null +++ b/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 diff --git a/dispatcharr/db/backends/postgresql_psycopg3/pool.py b/dispatcharr/db/backends/postgresql_psycopg3/pool.py new file mode 100644 index 00000000..8d50a2ae --- /dev/null +++ b/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) diff --git a/dispatcharr/settings.py b/dispatcharr/settings.py index 51d5bc49..a22e5f9c 100644 --- a/dispatcharr/settings.py +++ b/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": { diff --git a/tests/test_geventpool_conn_lifetime.py b/tests/test_geventpool_conn_lifetime.py new file mode 100644 index 00000000..3f939034 --- /dev/null +++ b/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)