"""Single-worker scheduler for typed backend actions.""" from __future__ import annotations import logging import threading import time from dataclasses import dataclass from typing import Any from media_library_viewer_api.observability import record_scheduled_action from media_library_viewer_api.widgets.sources import ServiceRecord, build_service_record from .scheduler_actions import ( # type: ignore[reportMissingImports] QBITTORRENT_SPEED_ACTION, get_scheduled_action, ) from .scheduler_store import SchedulerRunStore # type: ignore[reportMissingImports] from .settings_store import SettingsStore, get_settings_store logger = logging.getLogger(__name__) DEFAULT_INTERVAL_SECONDS = 15 MIN_INTERVAL_SECONDS = 5 MAX_INTERVAL_SECONDS = 300 DEFAULT_RETENTION_SECONDS = 1_800 MAX_RETENTION_SECONDS = 86_400 DEFAULT_MAX_ROWS = 1_200 MAX_MAX_ROWS = 1_200 SCHEDULER_LOOP_SECONDS = 1.0 BACKOFF_CAP_SECONDS = 300 class SchedulerBusyError(RuntimeError): """Raised when a manual action overlaps an existing service run.""" def _safe_int(value: Any, default: int = 0) -> int: try: return int(value) except (TypeError, ValueError): return default def _bounded_config(config: dict[str, Any]) -> tuple[int, int, int]: interval = max( MIN_INTERVAL_SECONDS, min(_safe_int(config.get("poll_interval_seconds"), DEFAULT_INTERVAL_SECONDS), MAX_INTERVAL_SECONDS), ) retention = max( 60, min(_safe_int(config.get("sample_retention_seconds"), DEFAULT_RETENTION_SECONDS), MAX_RETENTION_SECONDS), ) max_rows = max(60, min(_safe_int(config.get("sample_max_rows"), DEFAULT_MAX_ROWS), MAX_MAX_ROWS)) return interval, retention, max_rows @dataclass class _ServiceState: signature: tuple[Any, ...] next_run_at: float | None = None running: bool = False last_attempt_at: int | None = None last_success_at: int | None = None last_error: str = "" consecutive_failures: int = 0 backoff_until: int | None = None class Scheduler: """Run registered service actions from one lifespan-managed worker.""" action_key = QBITTORRENT_SPEED_ACTION def __init__(self) -> None: self._thread: threading.Thread | None = None self._stop_event = threading.Event() self._state_lock = threading.RLock() self._service_locks: dict[str, threading.Lock] = {} self._states: dict[str, _ServiceState] = {} self._run_store = SchedulerRunStore() def start(self) -> None: with self._state_lock: if self._thread and self._thread.is_alive(): return self._stop_event.clear() self._thread = threading.Thread(target=self._run, name="scheduled-actions", daemon=True) self._thread.start() logger.info("Scheduled action worker started") def stop(self, timeout: float = 5.0) -> None: with self._state_lock: thread = self._thread if not thread: return self._stop_event.set() thread.join(timeout=timeout) if thread.is_alive(): logger.warning("Scheduled action worker did not stop within %.1fs", timeout) else: logger.info("Scheduled action worker stopped") def status(self, service_id: str, store: SettingsStore | None = None) -> dict[str, Any]: store = store or get_settings_store() service = store.get_service(service_id) if not service or service.get("service_type") != "qbittorrent": raise ValueError("qBittorrent service not found") config = service.get("config") or {} interval, retention, max_rows = _bounded_config(config) with self._state_lock: state = self._states.get(service_id) worker_running = bool(self._thread and self._thread.is_alive()) if state is None: state = _ServiceState(signature=()) last_success = state.last_success_at is_stale = last_success is None or time.time() - last_success > max(2 * interval, 60) return { "service_id": service_id, "action_key": self.action_key, "worker_running": worker_running, "enabled": bool(service.get("enabled", True)) and bool(config.get("polling_enabled", True)), "running": state.running, "poll_interval_seconds": interval, "sample_retention_seconds": retention, "sample_max_rows": max_rows, "next_run_at": _safe_int(state.next_run_at) if state.next_run_at is not None else None, "last_attempt_at": state.last_attempt_at, "last_success_at": last_success, "last_error": state.last_error, "consecutive_failures": state.consecutive_failures, "backoff_until": state.backoff_until, "is_stale": is_stale, } def run_now(self, service_id: str, store: SettingsStore | None = None) -> dict[str, Any]: store = store or get_settings_store() service_row = store.get_service(service_id) if not service_row or service_row.get("service_type") != "qbittorrent": raise ValueError("qBittorrent service not found") if not service_row.get("enabled", True): raise ValueError("Service is disabled") config = service_row.get("config") or {} if not config.get("polling_enabled", True): raise ValueError("Polling is disabled") service = build_service_record(store, service_row) run = self._execute(service, "manual") return {"run": run, "status": self.status(service_id, store)} def _run(self) -> None: while not self._stop_event.is_set(): try: self._reconcile_and_run() except Exception: logger.exception("Scheduled action cycle failed") self._stop_event.wait(SCHEDULER_LOOP_SECONDS) def _reconcile_and_run(self) -> None: store = get_settings_store() services = sorted(store.list_services("qbittorrent"), key=lambda row: str(row.get("id") or "")) active_ids = {str(row.get("id") or "") for row in services} with self._state_lock: for service_id in set(self._states) - active_ids: self._states.pop(service_id, None) self._service_locks.pop(service_id, None) now = time.time() for index, service_row in enumerate(services): service_id = str(service_row.get("id") or "") if not service_id: continue config = service_row.get("config") or {} interval, _, _ = _bounded_config(config) enabled = bool(service_row.get("enabled", True)) and bool(config.get("polling_enabled", True)) signature = (enabled, interval, config.get("sample_retention_seconds"), config.get("sample_max_rows")) with self._state_lock: state = self._states.get(service_id) if state is None: state = _ServiceState(signature=signature, next_run_at=now + min(index * 0.5, 5.0)) self._states[service_id] = state elif state.signature != signature: state.signature = signature state.next_run_at = now if enabled else None if not enabled: state.next_run_at = None continue due = state.next_run_at is not None and now >= state.next_run_at if due: service = build_service_record(store, service_row) try: self._execute(service, "schedule") except SchedulerBusyError: logger.debug("Scheduled action already running service_id=%s", service_id) with self._state_lock: state = self._states.get(service_id) if state: delay = interval if state.backoff_until: delay = max(delay, state.backoff_until - _safe_int(time.time())) state.next_run_at = time.time() + max(1, delay) def _execute(self, service: ServiceRecord, trigger: str) -> dict[str, Any]: lock = self._service_lock(service.id) if not lock.acquire(blocking=False): raise SchedulerBusyError(f"Action already running for service {service.id}") try: return self._execute_locked(service, trigger) finally: lock.release() def _execute_locked(self, service: ServiceRecord, trigger: str) -> dict[str, Any]: config = service.config interval, _, _ = _bounded_config(config) state = self._state_for(service.id, config) attempt = state.consecutive_failures start = time.perf_counter() run = self._run_store.start_run(service.id, self.action_key, trigger, attempt=attempt) with self._state_lock: state.running = True state.last_attempt_at = _safe_int(time.time()) action = get_scheduled_action(self.action_key) try: if action is None: raise RuntimeError(f"Scheduled action is not registered: {self.action_key}") action.run(service) except Exception as exc: duration_ms = _safe_int((time.perf_counter() - start) * 1000) error = str(exc)[:1000] finished = self._run_store.finish_run(run["id"], "failure", duration_ms=duration_ms, error=error) self._run_store.prune(service.id, self.action_key) with self._state_lock: state.running = False state.last_error = error state.consecutive_failures += 1 delay = min(BACKOFF_CAP_SECONDS, max(interval, 2**state.consecutive_failures)) state.backoff_until = _safe_int(time.time()) + delay record_scheduled_action( service.id, self.action_key, "failure", duration_seconds=duration_ms / 1000.0, consecutive_failures=state.consecutive_failures, ) logger.warning("Scheduled action failed service_id=%s action=%s: %s", service.id, self.action_key, error) return finished if finished is not None else run duration_ms = _safe_int((time.perf_counter() - start) * 1000) finished = self._run_store.finish_run(run["id"], "success", duration_ms=duration_ms) self._run_store.prune(service.id, self.action_key) with self._state_lock: state.running = False state.last_success_at = _safe_int(time.time()) state.last_error = "" state.consecutive_failures = 0 state.backoff_until = None if trigger == "manual": state.next_run_at = max(state.next_run_at or 0, time.time() + interval) record_scheduled_action( service.id, self.action_key, "success", duration_seconds=duration_ms / 1000.0, success=True, consecutive_failures=0, ) return finished if finished is not None else run def _state_for(self, service_id: str, config: dict[str, Any]) -> _ServiceState: with self._state_lock: state = self._states.get(service_id) if state is None: interval, _, _ = _bounded_config(config) state = _ServiceState(signature=(), next_run_at=time.time() + interval) self._states[service_id] = state return state def _service_lock(self, service_id: str) -> threading.Lock: with self._state_lock: return self._service_locks.setdefault(service_id, threading.Lock()) _SCHEDULER = Scheduler() def get_scheduler() -> Scheduler: return _SCHEDULER def reset_scheduler() -> None: """Reset the singleton for tests.""" global _SCHEDULER _SCHEDULER.stop() _SCHEDULER = Scheduler() __all__ = ["Scheduler", "SchedulerBusyError", "get_scheduler", "reset_scheduler"]