Merge pull request #49 from knifecake/fix/issue-48-fork-safety

Fix supervisor fork crash loop with Django psycopg pools

Author: knifecake

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 ## Unreleased
 
+**Fixed:**
+
+- 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).
+
 ## v0.1.8 - 2026-03-08
 
 **Fixed:**

steady_queue/processes/base.py

@@ -1,10 +1,13 @@
+import logging
 import os
 import secrets
 import socket
 from typing import Any
 
 from django.db import connections
 
+logger = logging.getLogger("steady_queue")
+
 
 class Base:
     name: str
@@ -51,44 +54,77 @@ 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 to prevent
-        pool-related errors in steady_queue workers.
+        This method removes pool configuration from database settings and from
+        already-instantiated Django connection wrappers.
         """
-        import logging
-
         from django.conf import settings
 
-        logger = logging.getLogger("steady_queue")
-
-        # Disable pooling in database configuration
         if hasattr(settings, "DATABASES"):
             for alias, db_config in settings.DATABASES.items():
-                if db_config.get("ENGINE") == "django.db.backends.postgresql":
-                    # Remove pool configuration if it exists
-                    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"]
-
-        # Also disable on any existing connections
+                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.
+
+        Django stores psycopg pools in DatabaseWrapper._connection_pools, so we
+        must clear those references to avoid inheriting stale pools across fork.
+        """
+        closed_pool_maps: set[int] = set()
+
         for alias in connections:
             connection = connections[alias]
-            if hasattr(connection, "pool") and connection.pool is not None:
+            if (
+                connection.settings_dict.get("ENGINE")
+                != "django.db.backends.postgresql"
+            ):
+                continue
+
+            pool_map = getattr(connection.__class__, "_connection_pools", None)
+            if not isinstance(pool_map, dict) or id(pool_map) in closed_pool_maps:
+                continue
+
+            for pool_alias, pool in list(pool_map.items()):
                 try:
-                    connection.pool.close()
-                    connection.pool = None
+                    pool.close()
                     logger.debug(
-                        "%(name)s removed existing pool for '%(alias)s'",
-                        {"name": self.name, "alias": alias},
+                        "%(name)s closed psycopg pool for '%(alias)s'",
+                        {"name": self.name, "alias": pool_alias},
                     )
                 except Exception as e:
                     logger.debug(
-                        "%(name)s failed to close existing pool for '%(alias)s': %(e)s",
-                        {"name": self.name, "alias": alias, "e": e},
+                        "%(name)s failed to close pool for '%(alias)s': %(e)s",
+                        {"name": self.name, "alias": pool_alias, "e": e},
                     )
+                finally:
+                    pool_map.pop(pool_alias, None)
+
+            closed_pool_maps.add(id(pool_map))
 
     def reset_database_connections(self):
         """
@@ -98,5 +134,5 @@ def reset_database_connections(self):
         issues with shared connections between parent and child processes.
         """
         self.disable_connection_pooling()
-
         connections.close_all()
+        self.close_postgresql_connection_pools()

steady_queue/processes/supervisor.py

@@ -43,6 +43,8 @@ def start(self) -> None:
         logger.info("starting supervisor with PID %(pid)d", {"pid": self.pid})
         try:
             self.boot()
+            # Fork only after resetting DB state (connections + psycopg pools).
+            self.reset_database_connections()
             self.start_processes()
             self.launch_maintenance_task()
         except SystemExit:

tests/test_fork_safety.py

@@ -0,0 +1,125 @@
+import warnings
+from unittest.mock import patch
+
+from django.conf import settings
+from django.test import SimpleTestCase
+
+from steady_queue.configuration import Configuration
+from steady_queue.processes.base import Base
+from steady_queue.processes.supervisor import Supervisor
+
+
+class SupervisorForkSafetyTest(SimpleTestCase):
+    def build_supervisor(self) -> Supervisor:
+        options = Configuration.Options(
+            workers=[],
+            dispatchers=[],
+            recurring_tasks=[],
+            skip_recurring=True,
+        )
+        return Supervisor(Configuration(options))
+
+    def test_supervisor_start_resets_connections_before_forking(self):
+        supervisor = self.build_supervisor()
+        calls = []
+
+        supervisor.boot = lambda: calls.append("boot")
+        supervisor.reset_database_connections = lambda: calls.append("reset")
+        supervisor.start_processes = lambda: calls.append("start_processes")
+        supervisor.launch_maintenance_task = lambda: calls.append("launch_maintenance")
+        supervisor.supervise = lambda: calls.append("supervise")
+
+        supervisor.start()
+
+        self.assertEqual(
+            calls,
+            [
+                "boot",
+                "reset",
+                "start_processes",
+                "launch_maintenance",
+                "supervise",
+            ],
+        )
+
+
+class ResetDatabaseConnectionsTest(SimpleTestCase):
+    def test_reset_connections_disables_and_clears_psycopg_pool_cache(self):
+        class FakePool:
+            def __init__(self):
+                self.closed = False
+
+            def close(self):
+                self.closed = True
+
+        class FakeConnection:
+            _connection_pools = {}
+
+            def __init__(self, engine: str):
+                self.settings_dict = {
+                    "ENGINE": engine,
+                    "OPTIONS": {"pool": {"min_size": 1, "max_size": 4}},
+                }
+
+        class FakeConnections(dict):
+            close_all_called = False
+
+            def __iter__(self):
+                return iter(self.keys())
+
+            def close_all(self):
+                self.close_all_called = True
+
+        pool_default = FakePool()
+        pool_queue = FakePool()
+        FakeConnection._connection_pools = {
+            "default": pool_default,
+            "queue": pool_queue,
+        }
+
+        fake_connections = FakeConnections(
+            {
+                "default": FakeConnection("django.db.backends.postgresql"),
+                "queue": FakeConnection("django.db.backends.postgresql"),
+                "sqlite": FakeConnection("django.db.backends.sqlite3"),
+            }
+        )
+
+        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"])
+
+        self.assertNotIn("pool", fake_connections["default"].settings_dict["OPTIONS"])
+        self.assertNotIn("pool", fake_connections["queue"].settings_dict["OPTIONS"])
+        self.assertIn("pool", fake_connections["sqlite"].settings_dict["OPTIONS"])
+
+        self.assertTrue(fake_connections.close_all_called)
+        self.assertTrue(pool_default.closed)
+        self.assertTrue(pool_queue.closed)
+        self.assertEqual(FakeConnection._connection_pools, {})