Merge pull request #1355 from Dispatcharr/custom-psycopg3-pool

Custom-psycopg3-pool

CHANGELOG.md

@@ -32,6 +32,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Performance
 
+- **EPG programme parse now streams through PostgreSQL staging instead of holding the full catalogue in Python.** `parse_programs_for_source` writes parsed rows into a session-scoped temp table in batches (`_EPG_PARSE_BATCH_SIZE=2500`), then atomically swaps them into `ProgramData` with batched `DELETE ... RETURNING` + `INSERT` (`_EPG_SWAP_BATCH_SIZE=5000`) so Postgres never materializes the entire guide in one statement. Peak Celery memory during large XMLTV refreshes drops sharply compared with building a monolithic in-memory list before bulk insert. The byte-offset programme index build is deferred until after the swap completes so index construction no longer competes with parse for memory and I/O. `refresh_epg_data` closes its DB connection in `finally`, and `build_programme_index_task` is registered as a memory-intensive Celery task. SQLite/dev installs keep an in-memory fallback path.
+- **Pooled PostgreSQL connections now rotate on a bounded lifetime.** psycopg3 client-side cache grows on handles kept open indefinitely by `django-db-geventpool`; recycling uWSGI workers would interrupt live streams. A thin custom backend (`dispatcharr.db.backends.postgresql_psycopg3`) closes and replaces pool connections after `DATABASE_POOL_CONN_MAX_LIFETIME` seconds (default 600, env-overridable; set `0` to disable) on checkout and return while preserving warm-pool reuse within the window. (Fixes #1343)
 - **Schedules Direct refresh only fetches guide data for mapped channels.** Schedule MD5 checks and schedule downloads now target mapped lineup stations instead of the entire lineup, and schedule MD5 cache for unmapped stations is pruned each refresh. Unmapped lineup entries no longer trigger wasted schedule API calls when their MD5 changes.
 - **EPG auto-match memory and throughput improvements.**
   - Single-channel matching streams active EPG rows and keeps only the best match plus the top 20 candidates in memory; ML validates at most 21 names per channel instead of embedding the full catalog.

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)