disable connection pooling in workers :/

Author: knifecake

Makefile

@@ -11,3 +11,7 @@ steady_queue:
 .PHONY: lint
 lint:
 	uv run pre-commit run --all-files
+
+.PHONY: force-kill
+force-kill:
+	ps | grep steady_queue | cut -f 1  -d ' '  | xargs kill -9 && rm -f tmp/pids/steady_queue_supervisor.pid

steady_queue/__init__.py

@@ -1,7 +1,7 @@
 from datetime import timedelta
 from typing import Optional
 
-VERSION = (0, 1, "0b3")
+VERSION = (0, 1, "0b4")
 
 __version__ = ".".join(map(str, VERSION))
 

steady_queue/processes/base.py

@@ -46,5 +46,61 @@ 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 to prevent
+        pool-related errors in steady_queue workers.
+        """
+        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(
+                            "PID %d disabling connection pooling for database '%s'",
+                            os.getpid(),
+                            alias,
+                        )
+                        del options["pool"]
+
+        # Also disable on any existing connections
+        for alias in connections:
+            connection = connections[alias]
+            if hasattr(connection, "pool") and connection.pool is not None:
+                try:
+                    connection.pool.close()
+                    connection.pool = None
+                    logger.debug(
+                        "PID %d removed existing pool for '%s'", os.getpid(), alias
+                    )
+                except Exception as e:
+                    logger.debug(
+                        "PID %d failed to close existing pool for '%s': %s",
+                        os.getpid(),
+                        alias,
+                        e,
+                    )
+
     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.
+        """
+        # First disable connection pooling for steady_queue processes
+        self.disable_connection_pooling()
+
+        # Close all existing connections
         connections.close_all()

tests/dummy/tasks.py

@@ -1,13 +1,22 @@
+import time
+
 from django_tasks import task
 
-from steady_queue import recurring
+from steady_queue.recurring_task import recurring
 
 
 @task()
 def dummy_task():
     print("dummy task")
 
 
+@task()
+def long_running_task():
+    print("long running task")
+    time.sleep(10)
+    print("long running task finished")
+
+
 @recurring(schedule="*/1 * * * *", key="dummy_recurring_task")
 @task()
 def dummy_recurring_task():

tests/settings.py

@@ -39,6 +39,10 @@
     ),
 }
 
+if DATABASES["default"]["ENGINE"] == "django.db.backends.postgresql":
+    print("Enabling connection pooling")
+    DATABASES["default"]["OPTIONS"] = {"pool": {"min_size": 2, "max_size": 4}}
+
 
 # Tasks
 TASKS = {