commit f3fd9e82643dc5e7de7682e7565b94f4691a9c79 equwal <truex@equwal.com> 2026-09-22 17:01:29 -0700 Give a server job one more attempt when its worker vanishes A burst worker can vanish in the middle of a job: a cloud takes a spot machine back, or the machine fails. rq puts such a job back on its queue when the job has retries left. Each job now has one retry, so it gets two attempts at most. run_job never raises, so a book that fails in the aligner is not tried again.
backend/queue.py | 6 ++++++ tests/test_paid_queue.py | 9 +++++++++ 2 files changed, 15 insertions(+)
diff --git a/backend/queue.py b/backend/queue.py index 2283d12..c3581fa 100644 --- a/backend/queue.py +++ b/backend/queue.py @@ -90,11 +90,17 @@ class RedisQueue(JobQueue): self._free = RQQueue(FREE_QUEUE, connection=self._conn) def enqueue(self, job_id: str, paid: bool = True) -> None: + from rq import Retry + (self._paid if paid else self._free).enqueue( "backend.runner.run_job", job_id, job_timeout=settings.job_timeout_seconds, result_ttl=86400, + # A worker can vanish during a job (a spot machine is taken back). + # rq then puts the job back on its queue once: two attempts at most. + # run_job never raises, so a failed book is not tried again. + retry=Retry(max=1), ) def depth(self) -> int: diff --git a/tests/test_paid_queue.py b/tests/test_paid_queue.py index e7ca08a..f57840f 100644 --- a/tests/test_paid_queue.py +++ b/tests/test_paid_queue.py @@ -81,6 +81,15 @@ def test_the_redis_queue_keeps_paid_and_free_jobs_apart(monkeypatch): assert q.depth() == 2 +def test_a_job_whose_worker_dies_gets_one_more_attempt(monkeypatch): + # A spot machine can vanish during a job. rq puts such a job back on its + # queue while the job has retries left: one retry, so two attempts at most. + conn = fakeredis.FakeRedis() + monkeypatch.setattr("redis.Redis.from_url", lambda url: conn) + jobqueue.RedisQueue("redis://unused").enqueue("job_1", paid=True) + assert Queue(jobqueue.PAID_QUEUE, connection=conn).jobs[0].retries_left == 1 + + def test_the_web_server_worker_takes_paid_jobs_first(monkeypatch): monkeypatch.delenv("SUBPLZ_WEB_WORKER_QUEUES", raising=False) assert worker.queue_names() == [jobqueue.PAID_QUEUE, jobqueue.FREE_QUEUE]