From 7039f663b81d3668d004d7a982ba79fbfe7fba6b Mon Sep 17 00:00:00 2001 From: Elias Hernandis Date: Thu, 16 Apr 2026 21:10:31 +0200 Subject: [PATCH] Keep DB pooling enabled after fork reset and validate pool sizing --- CHANGELOG.md | 4 +++ steady_queue/configuration.py | 43 +++++++++++++++++++++++++++++ steady_queue/processes/base.py | 44 ++---------------------------- tests/settings.py | 2 +- tests/test_configuration.py | 49 ++++++++++++++++++++++++++++++++++ tests/test_fork_safety.py | 41 +++++----------------------- 6 files changed, 106 insertions(+), 77 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7fb2f9b..ca7d177 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/steady_queue/configuration.py b/steady_queue/configuration.py index 1367621..ea963a2 100644 --- a/steady_queue/configuration.py +++ b/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 diff --git a/steady_queue/processes/base.py b/steady_queue/processes/base.py index ddc4da7..a0e4711 100644 --- a/steady_queue/processes/base.py +++ b/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() diff --git a/tests/settings.py b/tests/settings.py index 16bf129..337b8e0 100644 --- a/tests/settings.py +++ b/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"} diff --git a/tests/test_configuration.py b/tests/test_configuration.py index 5b2910b..1b17fb9 100644 --- a/tests/test_configuration.py +++ b/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( diff --git a/tests/test_fork_safety.py b/tests/test_fork_safety.py index 78cc340..c181506 100644 --- a/tests/test_fork_safety.py +++ b/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)