137 lines
5.1 KiB
Python
137 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
from backend.jobs.registry import JobRegistry
|
|
from backend.jobs.repository import SQLiteJobRunRepository
|
|
|
|
|
|
JobAction = Callable[[], Any]
|
|
|
|
|
|
class InProcessJobRunner:
|
|
def __init__(self, registry: JobRegistry, repository: SQLiteJobRunRepository) -> None:
|
|
self.registry = registry
|
|
self.repository = repository
|
|
self._locks: dict[str, threading.Lock] = {}
|
|
self._locks_guard = threading.Lock()
|
|
self._scheduler_guard = threading.Lock()
|
|
self._scheduler_stop = threading.Event()
|
|
self._scheduler_thread: threading.Thread | None = None
|
|
|
|
def submit(
|
|
self, job_id: str, idempotency_key: str, action: JobAction,
|
|
metadata: dict[str, Any] | None = None,
|
|
) -> bool:
|
|
definition = self.registry.get(job_id)
|
|
if self.repository.completed(job_id, idempotency_key):
|
|
return False
|
|
lock = self._lock(definition.lock_key)
|
|
if not lock.acquire(blocking=False):
|
|
return False
|
|
thread = threading.Thread(
|
|
target=self._execute_locked,
|
|
args=(job_id, idempotency_key, action, metadata, lock),
|
|
name=f"job-{job_id}-{idempotency_key}"[:80],
|
|
daemon=True,
|
|
)
|
|
thread.start()
|
|
return True
|
|
|
|
def run_inline(
|
|
self, job_id: str, idempotency_key: str, action: JobAction,
|
|
metadata: dict[str, Any] | None = None,
|
|
) -> bool:
|
|
definition = self.registry.get(job_id)
|
|
if self.repository.completed(job_id, idempotency_key):
|
|
return False
|
|
lock = self._lock(definition.lock_key)
|
|
if not lock.acquire(blocking=False):
|
|
return False
|
|
self._execute_locked(job_id, idempotency_key, action, metadata, lock)
|
|
return True
|
|
|
|
def start_scheduler(
|
|
self, callback: Callable[[], None], interval_seconds: float,
|
|
initial_delay_seconds: float = 0,
|
|
) -> threading.Thread:
|
|
with self._scheduler_guard:
|
|
current = self._scheduler_thread
|
|
if current is not None and current.is_alive():
|
|
return current
|
|
self._scheduler_stop.clear()
|
|
|
|
def schedule_loop() -> None:
|
|
if self._scheduler_stop.wait(initial_delay_seconds):
|
|
return
|
|
while not self._scheduler_stop.is_set():
|
|
try:
|
|
callback()
|
|
except Exception:
|
|
# Submitted jobs persist failures; the scheduler must stay alive.
|
|
pass
|
|
self._scheduler_stop.wait(interval_seconds)
|
|
|
|
thread = threading.Thread(
|
|
target=schedule_loop,
|
|
name="background-job-scheduler",
|
|
daemon=True,
|
|
)
|
|
self._scheduler_thread = thread
|
|
thread.start()
|
|
return thread
|
|
|
|
def stop_scheduler(self, timeout_seconds: float = 5) -> bool:
|
|
with self._scheduler_guard:
|
|
thread = self._scheduler_thread
|
|
self._scheduler_stop.set()
|
|
if thread is not None and thread is not threading.current_thread():
|
|
thread.join(max(0, timeout_seconds))
|
|
return thread is None or not thread.is_alive()
|
|
|
|
def wait_for_idle(self, timeout_seconds: float = 5) -> bool:
|
|
deadline = time.monotonic() + max(0, timeout_seconds)
|
|
while time.monotonic() <= deadline:
|
|
with self._locks_guard:
|
|
busy = any(lock.locked() for lock in self._locks.values())
|
|
if not busy:
|
|
return True
|
|
time.sleep(0.01)
|
|
return False
|
|
|
|
def _execute_locked(
|
|
self, job_id: str, idempotency_key: str, action: JobAction,
|
|
metadata: dict[str, Any] | None, lock: threading.Lock,
|
|
) -> None:
|
|
definition = self.registry.get(job_id)
|
|
try:
|
|
for attempt in range(1, definition.max_attempts + 1):
|
|
run_id = self.repository.start(
|
|
job_id, idempotency_key, definition.output_version, metadata
|
|
)
|
|
started = time.perf_counter()
|
|
try:
|
|
result = action()
|
|
if isinstance(result, dict) and result.get("status") == "failed":
|
|
raise RuntimeError(str(result.get("error") or "Job reported failure"))
|
|
elapsed_ms = round((time.perf_counter() - started) * 1000)
|
|
self.repository.finish(run_id, "success", elapsed_ms)
|
|
return
|
|
except Exception as exc:
|
|
elapsed_ms = round((time.perf_counter() - started) * 1000)
|
|
self.repository.finish(
|
|
run_id, "failed", elapsed_ms,
|
|
type(exc).__name__, str(exc),
|
|
)
|
|
if attempt >= definition.max_attempts:
|
|
return
|
|
finally:
|
|
lock.release()
|
|
|
|
def _lock(self, lock_key: str) -> threading.Lock:
|
|
with self._locks_guard:
|
|
return self._locks.setdefault(lock_key, threading.Lock())
|