Merge pull request #50 from knifecake/fix/issue-48-keep-db-pooling

Keep DB pooling enabled and add pool-size validation

Author: knifecake

CHANGELOG.md

@@ -7,6 +7,10 @@
 - Fixed supervisor child-process crash loops (`exit code 11`) seen with
   Django/PostgreSQL pooling by resetting DB state before forking and clearing
   Django's class-level psycopg pool cache in the forking path (#48).
+- Keep database pooling enabled while resetting fork-inherited connection state
+  so child processes can establish fresh pools after forking.
+- Validate worker thread sizing against PostgreSQL `OPTIONS.pool.max_size`
+  (when explicitly configured), mirroring Solid Queue's pool sizing check.
 
 ## v0.1.8 - 2026-03-08
 

steady_queue/configuration.py

@@ -3,9 +3,11 @@
 from typing import Optional
 
 from crontab import CronTab
+from django.conf import settings
 from django.core.exceptions import ValidationError
 from django.utils.module_loading import import_string
 
+from steady_queue.db_router import steady_queue_database_alias
 from steady_queue.processes.base import Base
 
 
@@ -167,6 +169,7 @@ def schedulers(self) -> list["Configuration.Process"]:
     def is_valid(self):
         self.errors = []
         self.errors.extend(self.validate_configured_processes())
+        self.errors.extend(self.validate_database_pool_size())
         self.errors.extend(self.validate_recurring_tasks())
 
         return len(self.errors) == 0
@@ -177,6 +180,39 @@ def validate_configured_processes(self) -> list[ValidationError]:
 
         return []
 
+    def validate_database_pool_size(self) -> list[ValidationError]:
+        # Match Solid Queue behavior by validating worker thread count against
+        # the queue DB connection pool size when a max_size is explicitly set.
+        if len(self.options.workers) == 0:
+            return []
+
+        db_alias = steady_queue_database_alias()
+        db_config = settings.DATABASES.get(db_alias, {})
+
+        if db_config.get("ENGINE") != "django.db.backends.postgresql":
+            return []
+
+        pool_options = db_config.get("OPTIONS", {}).get("pool")
+        if not isinstance(pool_options, dict):
+            return []
+
+        pool_max_size = pool_options.get("max_size")
+        if not isinstance(pool_max_size, int):
+            return []
+
+        if pool_max_size < self.estimated_number_of_threads:
+            return [
+                ValidationError(
+                    "Steady Queue is configured to use "
+                    f"{self.estimated_number_of_threads} threads but the "
+                    f"database connection pool max_size for '{db_alias}' is "
+                    f"{pool_max_size}. Increase "
+                    f"DATABASES['{db_alias}']['OPTIONS']['pool']['max_size']."
+                )
+            ]
+
+        return []
+
     def validate_recurring_tasks(self) -> list[ValidationError]:
         if self.skip_recurring:
             return []
@@ -195,6 +231,13 @@ def validate_recurring_tasks(self) -> list[ValidationError]:
 
         return errors
 
+    @property
+    def estimated_number_of_threads(self) -> int:
+        # At most `threads` in each worker + 2 additional threads (worker loop
+        # and heartbeat), mirroring Solid Queue's sizing heuristic.
+        max_worker_threads = max((w.threads for w in self.options.workers), default=1)
+        return max_worker_threads + 2
+
     @property
     def skip_recurring(self) -> bool:
         return self.options.skip_recurring or self.options.only_work

steady_queue/processes/base.py

@@ -49,45 +49,6 @@ def is_stopped(self) -> bool:
     def generate_name(self) -> str:
         return "-".join((self.kind, secrets.token_hex(10)))
 
-    def disable_connection_pooling(self):
-        """
-        Disable connection pooling for steady_queue processes.
-
-        Connection pooling with psycopg doesn't work with forked processes.
-        This method removes pool configuration from database settings and from
-        already-instantiated Django connection wrappers.
-        """
-        from django.conf import settings
-
-        if hasattr(settings, "DATABASES"):
-            for alias, db_config in settings.DATABASES.items():
-                if db_config.get("ENGINE") != "django.db.backends.postgresql":
-                    continue
-
-                options = db_config.setdefault("OPTIONS", {})
-                if "pool" in options:
-                    logger.info(
-                        "%(name)s disabling connection pooling for database '%(alias)s'",
-                        {"name": self.name, "alias": alias},
-                    )
-                    del options["pool"]
-
-        for alias in connections:
-            connection = connections[alias]
-            if (
-                connection.settings_dict.get("ENGINE")
-                != "django.db.backends.postgresql"
-            ):
-                continue
-
-            options = connection.settings_dict.setdefault("OPTIONS", {})
-            if "pool" in options:
-                logger.debug(
-                    "%(name)s removing pool option from instantiated connection '%(alias)s'",
-                    {"name": self.name, "alias": alias},
-                )
-                del options["pool"]
-
     def close_postgresql_connection_pools(self):
         """
         Close and clear Django's class-level psycopg pool cache.
@@ -130,9 +91,8 @@ def reset_database_connections(self):
         """
         Reset database connections for forked processes.
 
-        This disables connection pooling and resets connection state to prevent
-        issues with shared connections between parent and child processes.
+        This closes all current connections and clears Django's class-level
+        psycopg pool cache so child processes don't inherit parent pool state.
         """
-        self.disable_connection_pooling()
         connections.close_all()
         self.close_postgresql_connection_pools()

tests/settings.py

@@ -47,7 +47,7 @@
 DATABASE_ROUTERS = ["steady_queue.db_router.SteadyQueueRouter"]
 
 if DATABASES["queue"]["ENGINE"] == "django.db.backends.postgresql":
-    DATABASES["queue"]["OPTIONS"] = {"pool": {"min_size": 2, "max_size": 4}}
+    DATABASES["queue"]["OPTIONS"] = {"pool": {"min_size": 2, "max_size": 8}}
     DATABASES["queue"]["TEST"] = {"NAME": "test_queue"}
     DATABASES["default"]["TEST"] = {"NAME": "test_default"}
 

tests/test_configuration.py

@@ -1,3 +1,4 @@
+import warnings
 from datetime import timedelta
 
 from django.core.exceptions import ValidationError
@@ -153,6 +154,54 @@ def test_configuration_with_no_processes_fails_validation(self):
         self.assertGreater(len(config.errors), 0)
         self.assertIn("No processes configured", str(config.errors[0]))
 
+    def test_small_postgres_pool_fails_validation(self):
+        """Configured worker threads must fit in postgres pool max_size."""
+        options = Configuration.Options(workers=[Configuration.Worker(threads=3)])
+        config = Configuration(options)
+
+        with warnings.catch_warnings():
+            warnings.filterwarnings(
+                "ignore",
+                message="Overriding setting DATABASES can lead to unexpected behavior.",
+                category=UserWarning,
+            )
+
+            with self.settings(
+                DATABASES={
+                    "queue": {
+                        "ENGINE": "django.db.backends.postgresql",
+                        "OPTIONS": {"pool": {"min_size": 1, "max_size": 4}},
+                    }
+                }
+            ):
+                self.assertFalse(config.is_valid)
+
+        self.assertTrue(
+            any("pool max_size" in error.message for error in config.errors)
+        )
+
+    def test_sufficient_postgres_pool_passes_validation(self):
+        """Validation passes when postgres pool max_size is large enough."""
+        options = Configuration.Options(workers=[Configuration.Worker(threads=3)])
+        config = Configuration(options)
+
+        with warnings.catch_warnings():
+            warnings.filterwarnings(
+                "ignore",
+                message="Overriding setting DATABASES can lead to unexpected behavior.",
+                category=UserWarning,
+            )
+
+            with self.settings(
+                DATABASES={
+                    "queue": {
+                        "ENGINE": "django.db.backends.postgresql",
+                        "OPTIONS": {"pool": {"min_size": 1, "max_size": 5}},
+                    }
+                }
+            ):
+                self.assertTrue(config.is_valid)
+
     def test_invalid_recurring_task_schedule_fails_validation(self):
         """Invalid cron schedule in recurring task fails validation."""
         invalid_task = Configuration.RecurringTask(

tests/test_fork_safety.py

@@ -1,7 +1,5 @@
-import warnings
 from unittest.mock import patch
 
-from django.conf import settings
 from django.test import SimpleTestCase
 
 from steady_queue.configuration import Configuration
@@ -44,7 +42,9 @@ def test_supervisor_start_resets_connections_before_forking(self):
 
 
 class ResetDatabaseConnectionsTest(SimpleTestCase):
-    def test_reset_connections_disables_and_clears_psycopg_pool_cache(self):
+    def test_reset_connections_clears_psycopg_pool_cache_without_disabling_pooling(
+        self,
+    ):
         class FakePool:
             def __init__(self):
                 self.closed = False
@@ -85,38 +85,11 @@ def close_all(self):
             }
         )
 
-        with warnings.catch_warnings():
-            warnings.filterwarnings(
-                "ignore",
-                message="Overriding setting DATABASES can lead to unexpected behavior.",
-                category=UserWarning,
-            )
-
-            with self.settings(
-                DATABASES={
-                    "default": {
-                        "ENGINE": "django.db.backends.postgresql",
-                        "OPTIONS": {"pool": {"min_size": 1, "max_size": 4}},
-                    },
-                    "queue": {
-                        "ENGINE": "django.db.backends.postgresql",
-                        "OPTIONS": {"pool": {"min_size": 1, "max_size": 4}},
-                    },
-                    "sqlite": {
-                        "ENGINE": "django.db.backends.sqlite3",
-                        "OPTIONS": {"pool": {"min_size": 1, "max_size": 4}},
-                    },
-                }
-            ):
-                with patch("steady_queue.processes.base.connections", fake_connections):
-                    Base().reset_database_connections()
-
-                self.assertNotIn("pool", settings.DATABASES["default"]["OPTIONS"])
-                self.assertNotIn("pool", settings.DATABASES["queue"]["OPTIONS"])
-                self.assertIn("pool", settings.DATABASES["sqlite"]["OPTIONS"])
+        with patch("steady_queue.processes.base.connections", fake_connections):
+            Base().reset_database_connections()
 
-        self.assertNotIn("pool", fake_connections["default"].settings_dict["OPTIONS"])
-        self.assertNotIn("pool", fake_connections["queue"].settings_dict["OPTIONS"])
+        self.assertIn("pool", fake_connections["default"].settings_dict["OPTIONS"])
+        self.assertIn("pool", fake_connections["queue"].settings_dict["OPTIONS"])
         self.assertIn("pool", fake_connections["sqlite"].settings_dict["OPTIONS"])
 
         self.assertTrue(fake_connections.close_all_called)