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..e295776 100644 --- a/tests/test_executions.py +++ b/tests/test_executions.py @@ -1,6 +1,10 @@ +from concurrent.futures import ThreadPoolExecutor from datetime import timedelta +from unittest import skipUnless +from unittest.mock import patch -from django.test import TestCase +from django.db import connections +from django.test import TestCase, TransactionTestCase from django.utils import timezone import steady_queue @@ -14,6 +18,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 +111,57 @@ def test_claim_with_zero_limit_returns_empty(self): self.assertEqual(ReadyExecution.objects.count(), 1) +class ConcurrentClaimTestCase(TestHelperMixin, TransactionTestCase): + databases = {"default", "queue"} + + @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() + 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()