From 54a2a68dabdffdec249698e2d3b03112483b7c66 Mon Sep 17 00:00:00 2001 From: Elias Hernandis Date: Sun, 6 Sep 2026 19:53:01 +0200 Subject: [PATCH 1/2] Strengthen regression coverage for claiming and pool logging --- CHANGELOG.md | 5 +++ tests/test_executions.py | 54 ++++++++++++++++++++++++++++++++- tests/test_process_lifecycle.py | 34 +++++++++++++++------ 3 files changed, 82 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 136e53f..1c3a4a8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,11 @@ ## Unreleased +**Added:** + +- Deterministic regression coverage for job arrivals during claiming (#25) and + propagation of thread-pool logging failures in the missing-process test (#26). + **Fixed:** - Avoid duplicate recurring-task enqueues when multiple schedulers race on the diff --git a/tests/test_executions.py b/tests/test_executions.py index 44b66df..d57eb1a 100644 --- a/tests/test_executions.py +++ b/tests/test_executions.py @@ -1,6 +1,9 @@ +from concurrent.futures import ThreadPoolExecutor from datetime import timedelta +from unittest.mock import patch -from django.test import TestCase +from django.db import connections +from django.test import TestCase, TransactionTestCase, skipUnlessDBFeature from django.utils import timezone import steady_queue @@ -14,6 +17,7 @@ ScheduledExecution, Semaphore, ) +from steady_queue.models.claimed_execution import ClaimedExecutionQuerySet from tests.dummy.tasks import dummy_task, limited_task, limited_task_with_lambda_key @@ -106,6 +110,54 @@ def test_claim_with_zero_limit_returns_empty(self): self.assertEqual(ReadyExecution.objects.count(), 1) +class ConcurrentClaimTestCase(TestHelperMixin, TransactionTestCase): + databases = {"default", "queue"} + + @skipUnlessDBFeature("has_select_for_update_skip_locked") + def test_higher_priority_arrival_does_not_change_claimed_candidates(self): + """#25: keep the selected jobs stable across insertion and retrieval.""" + process = self.create_test_process() + original = self.create_job_in_queue("default", priority=0) + bulk_create = ClaimedExecutionQuerySet.bulk_create + + def enqueue_higher_priority_job(): + try: + return self.create_job_in_queue("default", priority=10).pk + finally: + connections.close_all() + + def insert_claims_then_enqueue(queryset, *args, **kwargs): + result = bulk_create(queryset, *args, **kwargs) + # Commit an arrival on another connection after claims are inserted, + # but before claiming() retrieves them. No sleeps or timing races. + with ThreadPoolExecutor(max_workers=1) as executor: + arrival = executor.submit(enqueue_higher_priority_job) + arrivals.append(arrival.result(timeout=10)) + return result + + arrivals = [] + with patch.object( + ClaimedExecutionQuerySet, "bulk_create", insert_claims_then_enqueue + ): + claimed = ReadyExecution.objects.claim(["*"], 1, process.pk) + + self.assertEqual([execution.job_id for execution in claimed], [original.pk]) + self.assertFalse( + ReadyExecution.objects.filter( + job_id__in=ClaimedExecution.objects.values("job_id") + ).exists() + ) + self.assertEqual( + list(ReadyExecution.objects.values_list("job_id", flat=True)), arrivals + ) + + # Draining the remaining queue must not try to claim the original again. + next_claimed = ReadyExecution.objects.claim(["*"], 1, process.pk) + self.assertEqual([execution.job_id for execution in next_claimed], arrivals) + self.assertEqual(ReadyExecution.objects.claim(["*"], 1, process.pk), []) + self.assertEqual(ClaimedExecution.objects.count(), 2) + + class ClaimedExecutionTestCase(TestHelperMixin, TestCase): """Tests for ClaimedExecution behavior.""" diff --git a/tests/test_process_lifecycle.py b/tests/test_process_lifecycle.py index 13613b1..900fdb8 100644 --- a/tests/test_process_lifecycle.py +++ b/tests/test_process_lifecycle.py @@ -1,5 +1,5 @@ from datetime import timedelta -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch from django.test import TestCase from django.utils import timezone @@ -272,11 +272,11 @@ class PoolPostNilProcessTest(TestCase): def test_pool_post_does_not_access_execution_process(self): """Pool.post should not crash when execution.process is None.""" - import time - from steady_queue.processes.pool import Pool - pool = Pool(size=1, on_idle=lambda: None) + on_idle = MagicMock() + pool = Pool(size=1, on_idle=on_idle, worker_name="test-worker") + self.addCleanup(pool.executor.shutdown, wait=True) execution = MagicMock() execution.process = None @@ -284,11 +284,25 @@ def test_pool_post_does_not_access_execution_process(self): execution.job.class_name = "test_task" execution.pk = 1 - # Should not raise AttributeError - pool.post(execution) - # Give the thread pool time to execute - time.sleep(0.2) - pool.shutdown() + submit = pool.executor.submit + futures = [] + + def capture_future(*args, **kwargs): + future = submit(*args, **kwargs) + futures.append(future) + return future + + with self.assertLogs("steady_queue", level="INFO") as logs: + with patch.object(pool.executor, "submit", capture_future): + pool.post(execution) + # Exceptions after perform() live in the Future. Checking only that + # perform() ran allowed the original #26 logging crash to pass. + futures[0].result(timeout=5) - # Verify perform was called successfully execution.perform.assert_called_once() + self.assertEqual( + logs.output, + ["INFO:steady_queue:test-worker completed job 1 test_task"], + ) + self.assertEqual(pool.idle_threads, 1) + on_idle.assert_called_once_with() From c28a373862eeeb8ceb5f8d186549b791de44b787 Mon Sep 17 00:00:00 2001 From: Elias Hernandis Date: Sun, 6 Sep 2026 20:03:19 +0200 Subject: [PATCH 2/2] Check queue database features for concurrent claim regression --- tests/test_executions.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/tests/test_executions.py b/tests/test_executions.py index d57eb1a..e295776 100644 --- a/tests/test_executions.py +++ b/tests/test_executions.py @@ -1,9 +1,10 @@ from concurrent.futures import ThreadPoolExecutor from datetime import timedelta +from unittest import skipUnless from unittest.mock import patch from django.db import connections -from django.test import TestCase, TransactionTestCase, skipUnlessDBFeature +from django.test import TestCase, TransactionTestCase from django.utils import timezone import steady_queue @@ -113,7 +114,10 @@ def test_claim_with_zero_limit_returns_empty(self): class ConcurrentClaimTestCase(TestHelperMixin, TransactionTestCase): databases = {"default", "queue"} - @skipUnlessDBFeature("has_select_for_update_skip_locked") + @skipUnless( + connections["queue"].features.has_select_for_update_skip_locked, + "The queue database must support SELECT FOR UPDATE SKIP LOCKED", + ) def test_higher_priority_arrival_does_not_change_claimed_candidates(self): """#25: keep the selected jobs stable across insertion and retrieval.""" process = self.create_test_process()