File size: 5,173 Bytes
eea689d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | """The in-process worker that drains the Intake follow-up queue.
Intake enqueues report induction (#92) and dictionary compilation (#91) for each submitted
source-and-dictionary pair (`jobs.py`). This module claims those jobs from the durable queue and runs
them one at a time, in a background thread started from the API lifespan. It matches the app's
local-first, single-process posture: no external queue or worker process. The claim goes through
`storage.claim_next_job`, a conditional update, so even if a second process ran this loop a job would
still run once.
An executor runs one job to completion or raises; the worker records done or failed around it, so a
single job's failure never stops the loop and always leaves a recorded reason. The executors register
themselves here: report induction with #107, dictionary compilation with #108.
Tests drive `process_next_job` directly and leave the background thread off, so the fast suite never
starts a thread or spends the Anthropic key.
"""
from __future__ import annotations
import logging
import os
import threading
from pathlib import Path
from typing import Callable, Optional
from sqlalchemy import Connection
from endopath import storage
logger = logging.getLogger(__name__)
# kind -> executor. Populated by the executor tickets. An executor gets the connection and the claimed
# job row (dict), reads what it needs from the row (dictionary_id, source_kind, counts), and writes its
# artifact; it returns nothing and raises on failure.
Executor = Callable[[Connection, dict], None]
EXECUTORS: dict[str, Executor] = {}
def register_executor(kind: str, executor: Executor) -> None:
"""Register the executor for a job kind. Called at import by the executor modules."""
EXECUTORS[kind] = executor
def process_next_job(conn: Connection) -> Optional[dict]:
"""Claim and run the next enqueued job, returning the claimed job row with its final status, or None
when the queue is drained. An executor that raises marks the job failed with the error text; a kind
with no registered executor fails the same way rather than silently completing."""
job = storage.claim_next_job(conn)
if job is None:
return None
executor = EXECUTORS.get(job["kind"])
try:
if executor is None:
raise RuntimeError(f"no executor registered for job kind {job['kind']!r}")
executor(conn, job)
except Exception as exc: # noqa: BLE001 - a failed job records its reason, it does not crash the loop
logger.exception("job %s (%s) failed", job["id"], job["kind"])
storage.finish_job(conn, job_id=job["id"], status="failed", error=str(exc))
job["status"] = "failed"
job["error"] = str(exc)
return job
storage.finish_job(conn, job_id=job["id"], status="done")
job["status"] = "done"
return job
def drain(conn: Connection) -> int:
"""Run every currently enqueued job, returning how many ran. For a synchronous drain in a test or a
CLI, where the background thread is not wanted."""
count = 0
while process_next_job(conn) is not None:
count += 1
return count
# --- The background thread -----------------------------------------------------
POLL_SECONDS = 2.0
WORKER_DISABLED_ENV = "ENDOPATH_WORKER_DISABLED"
_thread: Optional[threading.Thread] = None
_stop = threading.Event()
def _loop(db_path: Path | str) -> None:
conn = storage.connect(db_path)
try:
while not _stop.is_set():
try:
ran = process_next_job(conn)
except Exception: # noqa: BLE001 - a storage hiccup must not kill the loop
logger.exception("worker loop error")
ran = None
if ran is None:
# Queue drained (or errored): wait before polling again, and wake immediately on stop.
_stop.wait(POLL_SECONDS)
finally:
conn.close()
def start(db_path: Path | str) -> bool:
"""Start the background worker thread, returning whether one was started. Disabled under pytest and
when ENDOPATH_WORKER_DISABLED is set, so tests drive process_next_job directly and never start a
thread. A no-op if a worker thread is already running."""
global _thread
if os.environ.get(WORKER_DISABLED_ENV) or "PYTEST_CURRENT_TEST" in os.environ:
return False
if _thread is not None and _thread.is_alive():
return False
# Importing the executors module registers report induction (#107) and dictionary compilation
# (#108) into EXECUTORS. Imported here, at the point the worker starts, so an import cycle never
# forms at module load (executors imports worker) and the heavy extraction stack loads only when a
# deployment actually runs jobs.
from endopath import executors # noqa: F401
_stop.clear()
_thread = threading.Thread(target=_loop, args=(db_path,), name="endopath-worker", daemon=True)
_thread.start()
return True
def stop() -> None:
"""Signal the worker to stop and wait briefly for it. For a clean shutdown and for tests."""
_stop.set()
if _thread is not None:
_thread.join(timeout=5.0)
|