diff --git a/docs/design/connection-failures.md b/docs/design/connection-failures.md index bbf8f9b9..2d5b96ff 100644 --- a/docs/design/connection-failures.md +++ b/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). diff --git a/docs/guides/deployment.md b/docs/guides/deployment.md index 35807e0a..3e78bfea 100644 --- a/docs/guides/deployment.md +++ b/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 diff --git a/pgqueuer/core/applications.py b/pgqueuer/core/applications.py index fe94fa1f..cb4bef5b 100644 --- a/pgqueuer/core/applications.py +++ b/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: diff --git a/pgqueuer/core/qm.py b/pgqueuer/core/qm.py index 9c0a87e0..776f0fc5 100644 --- a/pgqueuer/core/qm.py +++ b/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 diff --git a/test/test_superviser.py b/test/test_superviser.py index 1ed161d4..76eba9d8 100644 --- a/test/test_superviser.py +++ b/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