Worker reclaims this host's running jobs at start-up (restart-safe)
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 changed file +10 −2
modified
src/aiatlas/services/jobs.py
+10 −2
@@ -78,6 +78,10 @@ async def run_worker(*, concurrency: int | None = None, kinds: list[str] | None | |||||
| 78 | 78 | ||||
| 79 | worker = f"{socket.gethostname()}:{new_id('queue_job')[-6:]}" | 79 | worker = f"{socket.gethostname()}:{new_id('queue_job')[-6:]}" | ||
| 80 | stop = stop or asyncio.Event() | 80 | stop = stop or asyncio.Event() | ||
| 81 | + async with transaction() as conn: | ||||
| 82 | + reclaimed = await requeue_stale(conn, older_than_minutes=0, worker_prefix=socket.gethostname()) | ||||
| 83 | + if reclaimed: | ||||
| 84 | + log.info("reclaimed jobs left running by a previous worker on this host", extra={"n": reclaimed}) | ||||
| 81 | sem = asyncio.Semaphore(concurrency or settings.worker_concurrency) | 85 | sem = asyncio.Semaphore(concurrency or settings.worker_concurrency) | ||
| 82 | log.info("worker started", extra={"worker": worker, "kinds": kinds or "all"}) | 86 | log.info("worker started", extra={"worker": worker, "kinds": kinds or "all"}) | ||
| 83 | 87 | ||||
@@ -124,9 +128,13 @@ async def run_worker(*, concurrency: int | None = None, kinds: list[str] | None | |||||
| 124 | await asyncio.gather(*tasks, return_exceptions=True) | 128 | await asyncio.gather(*tasks, return_exceptions=True) | ||
| 125 | 129 | ||||
| 126 | 130 | ||||
| 127 | −async def requeue_stale(conn: AsyncConnection, *, older_than_minutes: int = 120) -> int: | 131 | +async def requeue_stale(conn: AsyncConnection, *, older_than_minutes: int = 120, worker_prefix: str | None = None) -> int: | ||
| 132 | + """Put back jobs left `running` by a dead worker. With `worker_prefix` (hostname), only this host's jobs are reclaimed — used at | ||||
| 133 | + worker start-up so a restart (deploy, reboot) never strands jobs for two hours.""" | ||||
| 128 | row = await fetch_one(conn, """with s as (update jobs set status = 'queued', locked_by = null, locked_at = null | 134 | row = await fetch_one(conn, """with s as (update jobs set status = 'queued', locked_by = null, locked_at = null | ||
| 129 | − where status = 'running' and locked_at < now() - make_interval(mins => :m) returning 1) select count(*) as n from s""", m=older_than_minutes) | 135 | + where status = 'running' and locked_at < now() - make_interval(mins => :m) | ||
| 136 | + and (cast(:p as text) = '' or locked_by like cast(:p as text) || ':%') returning 1) select count(*) as n from s""", | ||||
| 137 | + m=older_than_minutes, p=worker_prefix or "") | ||||
| 130 | return int(row["n"]) if row else 0 | 138 | return int(row["n"]) if row else 0 | ||
| 131 | 139 | ||||
| 132 | 140 | ||||
| 133 | 141 | ||||