fix(qm): clear unwritten job ends per run and test the restart write

docs/design/connection-failures.md

@@ -19,7 +19,7 @@ supervisor                                                         reruns the fa
       │  ├─ main loop                ─► dequeue + probes           raises, run ends
       │  ├─ job task (per job)                                     waited for
       │  │  ├─ heartbeats, buffered  ─► update_heartbeat           logged, kept
-      │  │  └─ results, buffered     ─► log_jobs                   at exit: 5 tries, then dropped
+      │  │  └─ results, buffered     ─► log_jobs                   at exit: 5 tries, then kept
       │  ├─ health check, 10 s       ─► notify                     ends run (opt-in)
       │  └─ log aggregation          ─► aggregate_logs             swallowed
       └─ SchedulerManager.run
@@ -74,14 +74,18 @@ running without NOTIFY.
 
 ## Known gaps
 
-Jobs running at the drop run twice. The queue manager waits for them to finish,
-but their results sit in `JobStatusLogBuffer`, whose last flush runs on the dead
-connection. It tries five times and then drops them. The rows stay `picked` and
-are picked again after `heartbeat_timeout`. In the test, all 10 in-flight jobs
-ran twice on both asyncpg and psycopg. ADR-0004 allows this; the cost is that a
-drop reruns every job in flight. With psycopg it also happened once to jobs that
-had finished before the kill, probably because their flush was still waiting
-for the connection lock that `notifies()` holds.
+Jobs running at the drop still finish. Their results sit in `JobStatusLogBuffer`,
+whose last flush runs on the dead connection. It tries five times.
+`QueueManager.run` then keeps whatever remains on `unwritten_job_logs`.
+`--restart-on-failure` writes those ends on the next cycle's connection before
+dequeuing. A caller that handles the error itself can pass the same list to
+`log_jobs`. A write after another worker owns the job is ignored. A new process
+does not have the list, so those rows stay `picked` and run again after
+`heartbeat_timeout`. ADR-0004 allows that second run. In the 2026-10-14 run,
+before the ends were kept, all 10 in-flight jobs ran twice on both asyncpg and
+psycopg. With psycopg it also happened once to jobs that had finished before the
+kill, probably because their flush was still waiting for the connection lock
+that `notifies()` holds.
 
 The asyncpg pool hides the failure. `AsyncpgPoolDriver` keeps its LISTEN
 connection out of the pool, and that connection dies without an error. Queries
@@ -105,10 +109,11 @@ process die and relies on a supervisor, and PgQueuer works the same way.
 
 PgQueuer keeps its current model: the user passes a connection, and recovery
 means the supervisor reruns the factory. PgQueuer does not reconnect in place
-and does not take a driver factory. Open follow-ups:
+and does not take a driver factory. A failed shutdown flush keeps the job ends
+in the process, on `QueueManager.unwritten_job_logs`. An in-process restart
+writes them; a new process does not have them, and ADR-0004 allows the second
+run. Open follow-ups:
 
-1. Keep the results of in-flight jobs across a restart, or accept the second
-   run that ADR-0004 allows.
-2. Turn on `restart_on_failure` and `shutdown_on_listener_failure` by default.
+1. Turn on `restart_on_failure` and `shutdown_on_listener_failure` by default.
    This changes behaviour, so it waits for v1.7.
-3. Deprecate the pool drivers (#951).
+2. Deprecate the pool drivers (#951).

docs/guides/deployment.md

@@ -185,8 +185,13 @@ next query, which can be up to `--dequeue-timeout` (30 s) away. With an asyncpg
 LISTEN connection is never noticed: the worker keeps running, but only picks up jobs by polling
 every `--dequeue-timeout`.
 
-Jobs running when the connection drops still finish, but their results can't be saved, so they
-run again after `--heartbeat-timeout`. Keep jobs idempotent.
+Jobs running when the connection drops still finish. If their status write fails, the
+ends stay on `QueueManager.unwritten_job_logs` until the next `run`. `pgq run
+--restart-on-failure` writes them on the new connection before it takes more work. A
+caller that handles the error itself can pass the same list to `Queries.log_jobs`. A
+write after another worker owns the job does nothing. A new process does not have the
+list: those jobs stay `picked` and run again after `--heartbeat-timeout`. Keep jobs
+idempotent.
 
 ## Graceful shutdown
 

pgqueuer/core/applications.py

@@ -88,7 +88,7 @@ def __post_init__(self) -> None:
 
     @property
     def unwritten_job_logs(self) -> list[JobLogEntry]:
-        """The queue manager's unwritten job logs."""
+        """Job ends the queue manager could not write on its last run."""
         return self.qm.unwritten_job_logs
 
     async def write_job_ends(self, logs: list[JobLogEntry]) -> None:

pgqueuer/core/qm.py

@@ -409,6 +409,8 @@ async def run(
         ``pgqueuer_log`` into ``pgqueuer_statistics`` on a timer instead of only
         on stats reads; pass ``timedelta(0)`` to disable (pure on-demand).
         """
+        # Drop the previous run's ends before anything here can raise.
+        self.unwritten_job_logs = []
         await self.verify_structure()
 
         # 0 has always meant unlimited; keep unlimited as None all the way to

test/test_superviser.py

@@ -7,6 +7,7 @@
 from contextlib import asynccontextmanager
 from datetime import timedelta
 from functools import partial
+from typing import Any
 from unittest.mock import ANY, MagicMock, patch
 
 import pytest
@@ -345,6 +346,63 @@ async def write_job_ends(self, logs: list[JobLogEntry]) -> None:
     assert held.entries == [fresh]
 
 
+async def test_restart_writes_job_end_the_first_run_could_not() -> None:
+    """An in-process restart writes the job end the failed flush kept."""
+    shared = InMemoryQueries(driver=InMemoryDriver())
+    real_log = shared.log_jobs
+    real_dequeue = shared.dequeue
+    cycle = 0
+    polls = 0
+    ran = 0
+
+    async def log_jobs(job_status: list[JobLogEntry]) -> None:
+        if cycle == 1:
+            raise RuntimeError("connection lost")
+        await real_log(job_status)
+
+    async def dequeue(*args: Any, **kwargs: Any) -> list[Job]:
+        nonlocal polls
+        if cycle == 1:
+            polls += 1
+            if polls > 1:
+                raise RuntimeError("poll failed")
+        return await real_dequeue(*args, **kwargs)
+
+    shared.log_jobs = log_jobs
+    shared.dequeue = dequeue
+
+    @asynccontextmanager
+    async def factory() -> AsyncIterator[QueueManager]:
+        nonlocal cycle
+        cycle += 1
+        qm = QueueManager(shared)
+
+        @qm.entrypoint("work")
+        async def work(job: Job) -> None:
+            nonlocal ran
+            ran += 1
+
+        yield qm
+
+    await shared.enqueue(["work"], [None], [0])
+    await supervisor.runit(
+        factory=factory,
+        dequeue_timeout=timedelta(seconds=0.05),
+        batch_size=10,
+        restart_delay=timedelta(0),
+        restart_on_failure=True,
+        shutdown=asyncio.Event(),
+        mode=QueueExecutionMode.drain,
+        max_concurrent_tasks=None,
+        shutdown_on_listener_failure=False,
+    )
+
+    assert cycle == 2
+    assert ran == 1
+    assert sum(row.count for row in await shared.queue_size()) == 0
+    assert any(log.status == "successful" for log in await shared.queue_log())
+
+
 async def test_runit_drain_exits_once_with_restart_on_failure() -> None:
     """Successful drain must not loop just because restart_on_failure is enabled."""
     calls = 0