Merge pull request #46 from 0xDEC0DE/issue/45

fix: RedisBroker: Remove non-idempotent jobs from running, too
NicolasLM · May 10, 2024 · 1545f18 · 1545f18
2 parents f1d2cb8 + 5e52315
commit 1545f18
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 9 deletions.
diff --git a/spinach/brokers/redis.py b/spinach/brokers/redis.py
@@ -226,15 +226,13 @@ def get_jobs_from_queue(self, queue: str, max_jobs: int) -> List[Job]:
         return jobs
 
     def remove_job_from_running(self, job: Job):
-        if job.max_retries > 0:
-            self._run_script(
-                self._remove_job_from_running,
-                self._to_namespaced(RUNNING_JOBS_KEY.format(self._id)),
-                self._to_namespaced(MAX_CONCURRENCY_KEY),
-                self._to_namespaced(CURRENT_CONCURRENCY_KEY),
-                job.serialize(),
-            )
-
+        self._run_script(
+            self._remove_job_from_running,
+            self._to_namespaced(RUNNING_JOBS_KEY.format(self._id)),
+            self._to_namespaced(MAX_CONCURRENCY_KEY),
+            self._to_namespaced(CURRENT_CONCURRENCY_KEY),
+            job.serialize(),
+        )
         self._something_happened.set()
 
     def _subscriber_func(self):

diff --git a/tests/test_redis_brokers.py b/tests/test_redis_brokers.py
@@ -86,6 +86,7 @@ def test_running_job(broker):
     )
     # Try to remove it, even if it doesn't exist in running
     broker.remove_job_from_running(job)
+    assert broker._r.hget(running_jobs_key, str(job.id)) is None
 
     # Idempotent job - get from queue
     job = Job('foo_task', 'foo_queue', datetime.now(timezone.utc), 10)