From 46504f161a18315b0fd8c07aca528580ddc5a569 Mon Sep 17 00:00:00 2001 From: JeyBee Date: Sat, 24 Oct 2026 22:14:54 +0200 Subject: [PATCH 1/2] fix(qm): retry a failed retry_job write until shutdown A transient failure while persisting RetryRequested was recorded as a terminal failure, so the handler never ran again. Keep the job picked and heartbeating and retry the write. Shutdown leaves the row picked for stale recovery. Fixes #1030 --- RELEASE.md | 4 ++ docs/guides/hold-failed-jobs.md | 1 + docs/guides/retry.md | 5 +++ pgqueuer/core/qm.py | 36 ++++++++++++++-- test/test_retry_requeue.py | 75 ++++++++++++++++++++++++--------- 5 files changed, 97 insertions(+), 24 deletions(-) diff --git a/RELEASE.md b/RELEASE.md index e8c82caf..ceff97d4 100644 --- a/RELEASE.md +++ b/RELEASE.md @@ -111,6 +111,10 @@ `pgq schedules`. They printed the driver's wall clock with no zone: UTC on asyncpg, the session `TimeZone` on psycopg, so the same row read hours apart depending on the driver. `--json` output is unchanged. +- A failed `retry_job` write is retried while the worker keeps the job + `picked` and heartbeating. Shutdown before the write lands leaves the row + `picked` for stale recovery, and `attempts` stays where it was. One dropped + statement no longer records the job as `exception` or `failed`. ## v1.6.0 diff --git a/docs/guides/hold-failed-jobs.md b/docs/guides/hold-failed-jobs.md index 27f865f1..817d1126 100644 --- a/docs/guides/hold-failed-jobs.md +++ b/docs/guides/hold-failed-jobs.md @@ -148,5 +148,6 @@ await queries.requeue_jobs([JobId(42), JobId(43)]) |----------|--------------------------------|---------------------| | Handler raises exception | Job deleted, logged as `exception` | Job kept with `status='failed'` | | Handler raises `RetryRequested` | Re-queued with delay (both modes) | Re-queued with delay (both modes) | +| Retry write fails until shutdown | Job stays `picked` for stale recovery | Job stays `picked` for stale recovery | | Handler completes | Job deleted, logged as `successful` | Job deleted, logged as `successful` | | `DatabaseRetryEntrypointExecutor` exhausted | Job deleted | Job held with `status='failed'` | diff --git a/docs/guides/retry.md b/docs/guides/retry.md index f72f017f..37489c82 100644 --- a/docs/guides/retry.md +++ b/docs/guides/retry.md @@ -16,6 +16,10 @@ are preserved across retries. 3. The job becomes eligible for dequeue again after the delay expires. 4. On the next execution, `job.attempts` reflects how many previous attempts occurred. +If that write fails, the worker keeps the job `picked` and heartbeating and retries +the write until it succeeds. Shutting the manager down first leaves the row +`picked`. Stale recovery runs the handler again, and `attempts` is unchanged. + Because the job row is **updated** (not deleted), the same `job.id` is stable across all retry attempts. This makes it straightforward to trace the full retry graph by querying the log table. @@ -184,6 +188,7 @@ ORDER BY created; | Scenario | What happens | |----------|-------------| | Handler raises `RetryRequested` | Job updated to `queued`, attempts incremented, log entry written | +| Retry write fails | Retried while the job stays `picked` and heartbeating. Shutdown leaves it `picked` for stale recovery | | Handler raises any other exception | Job deleted, logged as `exception` (or held if `on_failure="hold"`: see [Holding Failed Jobs](hold-failed-jobs.md)) | | Handler completes normally | Job deleted, logged as `successful` (existing behavior) | | `DatabaseRetryEntrypointExecutor` + exception + attempts < max | Converted to `RetryRequested` with backoff | diff --git a/pgqueuer/core/qm.py b/pgqueuer/core/qm.py index 776f0fc5..4eaaea15 100644 --- a/pgqueuer/core/qm.py +++ b/pgqueuer/core/qm.py @@ -537,6 +537,37 @@ async def log_failure( ) await jbuff.add((job, status, tbr)) + async def write_retry( + self, + job: models.Job, + delay: timedelta, + traceback_record: models.TracebackRecord, + ) -> None: + """Persist a requested retry until the write succeeds. + + Failures keep the job picked and heartbeating. Shutdown leaves the row + picked so stale recovery runs the handler again, with ``attempts`` unchanged. + """ + pause = timedelta(milliseconds=50) + cap = timedelta(seconds=1) + while True: + try: + await self.queries.retry_job(job, delay, traceback_record) + return + except Exception: + logconfig.logger.exception( + "Retry write failed for entrypoint/job-id: %s/%s", + job.entrypoint, + job.id, + ) + if self.shutdown.is_set(): + return + with suppress(TimeoutError): + async with asyncio.timeout(pause.total_seconds()): + await self.shutdown.wait() + return + pause = min(pause * 2, cap) + async def _dispatch( self, job: models.Job, @@ -594,10 +625,7 @@ async def _dispatch( ) # Release before re-queueing, or our own dequeue skips the re-picked row. self.job_context.pop(job.id, None) - try: - await self.queries.retry_job(job, retry_exc.delay, tbr) - except Exception as e: # noqa: BLE001 - recorded as a job failure - await self.log_failure(job, executor, e, jbuff) + await self.write_retry(job, retry_exc.delay, tbr) except asyncio.CancelledError: logconfig.logger.debug( "Job canceled mid-flight for entrypoint/id: %s/%s", diff --git a/test/test_retry_requeue.py b/test/test_retry_requeue.py index b24386b6..d745f4cc 100644 --- a/test/test_retry_requeue.py +++ b/test/test_retry_requeue.py @@ -22,7 +22,6 @@ from pgqueuer.domain.models import Context, Job, TracebackRecord from pgqueuer.domain.types import ( JobId, - OnFailure, QueueEntrypoint, QueueExecutionMode, QueueManagerId, @@ -208,43 +207,79 @@ async def handler(job: Job) -> None: assert await pq.qm.queries.queue_size() == [] -@pytest.mark.parametrize( - ("on_failure", "status", "queue_status"), - (("delete", "exception", None), ("hold", "failed", "failed")), -) -async def test_failing_retry_job_records_the_job_as_failed( +async def test_retry_write_is_retried_until_it_lands(monkeypatch: pytest.MonkeyPatch) -> None: + """One failed retry write is tried again, and the handler runs with attempts 1 (#1030).""" + pq = PgQueuer.in_memory() + seen: list[int] = [] + writes = 0 + original = pq.qm.queries.retry_job + + @pq.entrypoint("retry_ep") + async def handler(job: Job) -> None: + seen.append(job.attempts) + if job.attempts == 0: + raise RetryRequested(delay=timedelta(0)) + + async def flaky_retry_job( + job: Job, + delay: timedelta, + traceback_record: TracebackRecord | None, + ) -> None: + nonlocal writes + writes += 1 + if writes == 1: + raise ConnectionError("retry write failed") + await original(job, delay, traceback_record) + + monkeypatch.setattr(pq.qm.queries, "retry_job", flaky_retry_job) + await pq.qm.queries.enqueue("retry_ep", b"data", priority=0) + + async with asyncio.timeout(5): + await pq.qm.run( + batch_size=10, + mode=QueueExecutionMode.drain, + max_concurrent_tasks=100, + dequeue_timeout=timedelta(seconds=1), + ) + + assert seen == [0, 1] + assert writes == 2 + assert await pq.qm.queries.queue_size() == [] + assert [log.status for log in await pq.qm.queries.queue_log()][-1] == "successful" + + +async def test_retry_write_stops_on_shutdown_and_leaves_the_job_picked( monkeypatch: pytest.MonkeyPatch, - on_failure: OnFailure, - status: str, - queue_status: str | None, ) -> None: - """A retry_job error goes down the failure path instead of leaving the row picked (#883).""" + """Shutdown during a failing retry write leaves the job picked (#1030).""" pq = PgQueuer.in_memory() - @pq.entrypoint("retry_ep", on_failure=on_failure) + @pq.entrypoint("retry_ep") async def handler(job: Job) -> None: raise RetryRequested - async def failing_retry_job(*_: object) -> None: + async def failing_retry_job(*_args: object) -> None: + pq.qm.shutdown.set() raise ConnectionError("retry write failed") monkeypatch.setattr(pq.qm.queries, "retry_job", failing_retry_job) - await pq.qm.queries.enqueue("retry_ep", b"data", priority=0) + [job_id] = await pq.qm.queries.enqueue("retry_ep", b"data", priority=0) async with asyncio.timeout(5): await pq.qm.run( batch_size=10, - mode=QueueExecutionMode.drain, + mode=QueueExecutionMode.continuous, max_concurrent_tasks=100, dequeue_timeout=timedelta(seconds=1), ) - (log,) = [log for log in await pq.qm.queries.queue_log() if log.status == status] - assert log.traceback is not None - assert log.traceback.exception_type == "ConnectionError" - assert [row.status for row in await pq.qm.queries.queue_size()] == ( - [queue_status] if queue_status else [] - ) + job = await pq.qm.queries.queue_job_by_id(job_id) + assert job is not None + assert job.status == "picked" + assert job.attempts == 0 + logs = await pq.qm.queries.queue_log() + assert [log.status for log in logs] == ["queued", "picked"] + assert all(log.traceback is None for log in logs) async def raise_from_task_group(*errors: Exception) -> None: From 9d3381125ae16238dfb02ffe8a74311d7ded3ae0 Mon Sep 17 00:00:00 2001 From: JeyBee Date: Mon, 26 Oct 2026 23:00:40 +0100 Subject: [PATCH 2/2] fix(qm): satisfy ruff TRY300 and mypy in the retry write --- pgqueuer/core/qm.py | 3 ++- test/test_retry_requeue.py | 1 + 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/pgqueuer/core/qm.py b/pgqueuer/core/qm.py index 4eaaea15..137e907a 100644 --- a/pgqueuer/core/qm.py +++ b/pgqueuer/core/qm.py @@ -553,13 +553,14 @@ async def write_retry( while True: try: await self.queries.retry_job(job, delay, traceback_record) - return except Exception: logconfig.logger.exception( "Retry write failed for entrypoint/job-id: %s/%s", job.entrypoint, job.id, ) + else: + return if self.shutdown.is_set(): return with suppress(TimeoutError): diff --git a/test/test_retry_requeue.py b/test/test_retry_requeue.py index d745f4cc..69e82ea8 100644 --- a/test/test_retry_requeue.py +++ b/test/test_retry_requeue.py @@ -273,6 +273,7 @@ async def failing_retry_job(*_args: object) -> None: dequeue_timeout=timedelta(seconds=1), ) + assert isinstance(pq.qm.queries, InMemoryQueries) job = await pq.qm.queries.queue_job_by_id(job_id) assert job is not None assert job.status == "picked"