1"""Durable job queue on PostgreSQL (SELECT … FOR UPDATE SKIP LOCKED). Survives reboots; retries with2exponential backoff + jitter; dead-letter after max_attempts; checkpoints keyed by scope."""3from __future__ import annotations45import json6import random7import socket8from datetime import UTC, datetime, timedelta9from typing import Any1011from findata_core.config import get_settings12from findata_core.db import connection, transaction13from findata_core.redact import redact1415WORKER_ID = f"{socket.gethostname()}:{random.randint(1000, 9999)}"161718def enqueue(kind: str, dataset_id: str | None = None, params: dict[str, Any] | None = None, priority: int = 50,19 dedupe_key: str | None = None, max_attempts: int = 5, created_by: str = "system",20 run_at: datetime | None = None) -> str | None:21 """Insert a job; returns id, or None if a job with the same dedupe_key is already queued/running."""22 with transaction() as conn:23 if dedupe_key:24 r = conn.execute("SELECT id FROM ingest_jobs WHERE dedupe_key=%s", (dedupe_key,)).fetchone()25 if r:26 st = conn.execute("SELECT status FROM ingest_jobs WHERE id=%s", (r["id"],)).fetchone()["status"]27 if st in ("queued", "running"):28 return None29 conn.execute("UPDATE ingest_jobs SET dedupe_key=NULL WHERE id=%s", (r["id"],))30 r = conn.execute(31 "INSERT INTO ingest_jobs(kind, dataset_id, params, priority, dedupe_key, max_attempts, created_by, next_run_at) "32 "VALUES (%s,%s,%s,%s,%s,%s,%s,%s) RETURNING id",33 (kind, dataset_id, json.dumps(params or {}), priority, dedupe_key, max_attempts, created_by,34 run_at or datetime.now(UTC))).fetchone()35 return str(r["id"])363738def claim_next(kinds: list[str] | None = None) -> dict[str, Any] | None:39 with transaction() as conn:40 sql = ("SELECT * FROM ingest_jobs WHERE status='queued' AND next_run_at <= now() "41 + ("AND kind = ANY(%s) " if kinds else "") +42 "ORDER BY priority, next_run_at LIMIT 1 FOR UPDATE SKIP LOCKED")43 job = conn.execute(sql, (kinds,) if kinds else None).fetchone()44 if not job:45 return None46 conn.execute("UPDATE ingest_jobs SET status='running', attempts=attempts+1, locked_by=%s, locked_at=now(), "47 "started_at=COALESCE(started_at, now()) WHERE id=%s", (WORKER_ID, job["id"]))48 run = conn.execute("INSERT INTO ingest_runs(job_id, worker, code_version) VALUES (%s,%s,%s) RETURNING id",49 (job["id"], WORKER_ID, _code_version())).fetchone()50 job = dict(job)51 job["attempts"] += 152 job["run_id"] = str(run["id"])53 return job545556def finish(job: dict[str, Any], ok: bool, result: dict[str, Any] | None = None, error: str | None = None) -> None:57 secrets = get_settings().secret_values()58 with transaction() as conn:59 if ok:60 conn.execute("UPDATE ingest_jobs SET status='succeeded', finished_at=now(), locked_by=NULL, result=%s, last_error=NULL WHERE id=%s",61 (json.dumps(result or {}, default=str), job["id"]))62 conn.execute("UPDATE ingest_runs SET status='succeeded', finished_at=now(), stats=%s WHERE id=%s",63 (json.dumps(result or {}, default=str), job["run_id"]))64 return65 msg = redact(error or "unknown error", secrets)[:4000]66 dead = job["attempts"] >= job["max_attempts"]67 if dead:68 conn.execute("UPDATE ingest_jobs SET status='dead', finished_at=now(), locked_by=NULL, last_error=%s WHERE id=%s", (msg, job["id"]))69 else:70 delay = min(3600, 30 * 2 ** job["attempts"]) + random.uniform(0, 30)71 conn.execute("UPDATE ingest_jobs SET status='queued', locked_by=NULL, last_error=%s, next_run_at=%s WHERE id=%s",72 (msg, datetime.now(UTC) + timedelta(seconds=delay), job["id"]))73 conn.execute("UPDATE ingest_runs SET status=%s, finished_at=now(), stats=%s WHERE id=%s",74 ("dead" if dead else "failed", json.dumps({"error": msg}), job["run_id"]))757677def progress(job_id: str, data: dict[str, Any]) -> None:78 with transaction() as conn:79 conn.execute("UPDATE ingest_jobs SET progress=%s WHERE id=%s", (json.dumps(data, default=str), job_id))808182def set_checkpoint(scope: str, key: str, value: dict[str, Any]) -> None:83 with transaction() as conn:84 conn.execute("INSERT INTO checkpoints(scope, key, value) VALUES (%s,%s,%s) ON CONFLICT (scope, key) DO UPDATE SET value=EXCLUDED.value, updated_at=now()",85 (scope, key, json.dumps(value, default=str)))868788def get_checkpoint(scope: str, key: str) -> dict[str, Any] | None:89 with connection() as conn:90 r = conn.execute("SELECT value FROM checkpoints WHERE scope=%s AND key=%s", (scope, key)).fetchone()91 return r["value"] if r else None929394def requeue_stale(max_age_minutes: int = 240) -> int:95 """Jobs left 'running' by a crashed worker are re-queued (idempotent tasks + checkpoints make this safe)."""96 with transaction() as conn:97 r = conn.execute("UPDATE ingest_jobs SET status='queued', locked_by=NULL, next_run_at=now() WHERE status='running' "98 "AND locked_at < now() - make_interval(mins => %s) RETURNING id", (max_age_minutes,)).fetchall()99 return len(r)100101102def _code_version() -> str:103 import subprocess104 try:105 return subprocess.run(["git", "rev-parse", "--short", "HEAD"], capture_output=True, text=True, timeout=5,106 cwd=str(get_settings().root)).stdout.strip() or "unknown"107 except Exception:108 return "unknown"109