feat: add typed qBittorrent scheduled polling
This commit is contained in:
Developer
@@ -0,0 +1,296 @@
|
||||
"""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"]
|
||||
Reference in New Issue
Block a user