fix(qm): retry a failed retry_job write until shutdown

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
 

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'` |

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 |

pgqueuer/core/qm.py

@@ -537,6 +537,38 @@ 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)
+            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):
+                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 +626,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",

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,80 @@ 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 []
-    )
+    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"
+    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: