diff --git a/CHANGELOG.md b/CHANGELOG.md index 1c3a4a8..d2fba4b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ **Fixed:** +- Reset database connections and PostgreSQL pools immediately before each + supervised fork, including replacements after crashed-worker job recovery + (#48). This addresses a remaining cleanup gap; the reported macOS segfault + has not been reproduced. - Avoid duplicate recurring-task enqueues when multiple schedulers race on the same `run_at`. We now create recurring execution records atomically and skip already-recorded runs, matching Solid Queue's behavior. diff --git a/steady_queue/processes/supervisor.py b/steady_queue/processes/supervisor.py index 5132048..d423a77 100644 --- a/steady_queue/processes/supervisor.py +++ b/steady_queue/processes/supervisor.py @@ -86,6 +86,9 @@ def start_process(self, process: Configuration.Process) -> None: instance.supervisor = self.process instance.mode = "fork" + # Replacement recovery queries the database after the startup reset. + # Clear that connection and pool state before the child inherits it. + self.reset_database_connections() if (pid := os.fork()) == 0: # child instance.start() diff --git a/tests/test_fork_safety.py b/tests/test_fork_safety.py index c181506..b0a1c60 100644 --- a/tests/test_fork_safety.py +++ b/tests/test_fork_safety.py @@ -1,10 +1,20 @@ +from unittest import skipUnless from unittest.mock import patch -from django.test import SimpleTestCase +from django.db import connections +from django.test import SimpleTestCase, TransactionTestCase from steady_queue.configuration import Configuration +from steady_queue.models import ( + ClaimedExecution, + FailedExecution, + Job, + Process, + ReadyExecution, +) from steady_queue.processes.base import Base from steady_queue.processes.supervisor import Supervisor +from tests.dummy.tasks import dummy_task class SupervisorForkSafetyTest(SimpleTestCase): @@ -41,6 +51,81 @@ def test_supervisor_start_resets_connections_before_forking(self): ) +@skipUnless( + connections["queue"].vendor == "postgresql", + "Requires PostgreSQL connection pooling on the queue database", +) +class SupervisorPostgreSQLForkTest(TransactionTestCase): + databases = {"default", "queue"} + + def setUp(self): + options = Configuration.Options( + workers=[], dispatchers=[], recurring_tasks=[], skip_recurring=True + ) + self.supervisor = Supervisor(Configuration(options)) + self.supervisor.register() + self.worker_config = Configuration.Process( + kind="worker", attributes=Configuration.Worker() + ) + self.connection = connections["queue"] + self.pool_options = self.connection.settings_dict["OPTIONS"]["pool"].copy() + self.addCleanup(self.supervisor.reset_database_connections) + + def assert_reset_at_fork(self): + # Intercept only the OS fork; exercise real Django connections, psycopg + # pools, and replacement recovery up to the boundary children inherit. + self.assertIsNone(self.connection.connection) + self.assertEqual(self.connection.__class__._connection_pools, {}) + self.assertEqual( + self.connection.settings_dict["OPTIONS"]["pool"], self.pool_options + ) + return 12346 + + def test_start_process_resets_open_connection_and_pool_before_fork(self): + self.assertIsNotNone(self.connection.connection) + self.assertIn("queue", self.connection.__class__._connection_pools) + + with patch( + "steady_queue.processes.supervisor.os.fork", + side_effect=self.assert_reset_at_fork, + ) as fork: + self.supervisor.start_process(self.worker_config) + + fork.assert_called_once_with() + self.assertIn(12346, self.supervisor.forks) + # Pooling remains usable by the parent after the reset. + self.assertTrue(Process.objects.filter(pk=self.supervisor.process.pk).exists()) + self.assertIn("queue", self.connection.__class__._connection_pools) + + def test_replacement_resets_connection_reopened_by_job_recovery(self): + old_worker = self.worker_config.instantiate() + self.addCleanup(old_worker.pool.shutdown) + registered_worker = Process.register( + kind="worker", + name=old_worker.name, + pid=12345, + hostname="test-host", + supervisor=self.supervisor.process, + ) + job = Job.objects.enqueue(dummy_task, [], {}) + ReadyExecution.objects.claim(["*"], 1, registered_worker.pk) + self.supervisor.forks[12345] = old_worker + self.supervisor.configured_processes[12345] = self.worker_config + self.supervisor.reset_database_connections() + + with patch( + "steady_queue.processes.supervisor.os.fork", + side_effect=self.assert_reset_at_fork, + ) as fork: + self.supervisor.replace_fork(12345, 11) + + fork.assert_called_once_with() + self.assertNotIn(12345, self.supervisor.forks) + self.assertIn(12346, self.supervisor.forks) + self.assertFalse(ClaimedExecution.objects.filter(job=job).exists()) + self.assertTrue(FailedExecution.objects.filter(job=job).exists()) + + class ResetDatabaseConnectionsTest(SimpleTestCase): def test_reset_connections_clears_psycopg_pool_cache_without_disabling_pooling( self,