discovery: reclaim stale running discover jobs (dead workers) before claiming
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 changed file +5 −0
modified
src/companyatlas/services/discovery.py
+5 −0
@@ -695,8 +695,13 @@ async def _upsert_domain(conn: Any, company_id: str, domain: str, kind: str) -> | |||||
| 695 | # ------------------------------------------------------------------------------------------------------------ onboarding worker | 695 | # ------------------------------------------------------------------------------------------------------------ onboarding worker | ||
| 696 | 696 | ||||
| 697 | 697 | ||||
| 698 | +DISCOVER_CLAIM_TTL_MIN = 20 # a discovery never legitimately runs this long: older `running` claims belong to a dead worker | ||||
| 699 | + | ||||
| 700 | + | ||||
| 698 | async def claim_discover_jobs(limit: int, worker: str) -> list[dict[str, Any]]: | 701 | async def claim_discover_jobs(limit: int, worker: str) -> list[dict[str, Any]]: | ||
| 699 | async with transaction() as conn: | 702 | async with transaction() as conn: | ||
| 703 | + await execute(conn, """update queue_jobs set status = 'pending', locked_by = null, locked_at = null, run_at = now() | ||||
| 704 | + where kind = 'discover' and status = 'running' and locked_at < now() - make_interval(mins => :ttl)""", ttl=DISCOVER_CLAIM_TTL_MIN) | ||||
| 700 | rows = await fetch_all(conn, """ | 705 | rows = await fetch_all(conn, """ | ||
| 701 | with due as ( | 706 | with due as ( | ||
| 702 | select id from queue_jobs where kind = 'discover' and status = 'pending' and run_at <= now() | 707 | select id from queue_jobs where kind = 'discover' and status = 'pending' and run_at <= now() | ||
| 703 | 708 | ||||