peter
e49b77dbcd
The atomic claim added in trigger_job only covered the manual path; a scheduled fire sets running via _job without claiming, so a manual trigger racing it could still double-run, and the claim (set outside _job) could leak if _job threw before its own cleanup. Move _claim_running to the top of _job — the one point both the scheduled fire and the manual trigger pass through — and release it in _job's finally, so exactly one run wins whichever path it comes from and a failure can never wedge the job "running". trigger_job's _is_running check is now just a best-effort fast answer for the UI. Unit-test _claim_running (and is_messageable_user).
470 lines
18 KiB
Python
470 lines
18 KiB
Python
"""Background scheduler: free RSS detection, enrichment of new videos, and quota-aware
|
|
recent-first / deep backfill. Runs inside the API process (single worker)."""
|
|
import contextvars
|
|
import logging
|
|
import threading
|
|
from datetime import datetime, timezone
|
|
from typing import Callable
|
|
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
from sqlalchemy import select
|
|
|
|
from app import audit, progress, quota
|
|
from app.audit import AuditAction
|
|
from app.config import settings
|
|
from app.db import SessionLocal
|
|
from app.downloads.gc import run_download_gc
|
|
from app.models import SchedulerSetting
|
|
from app.plex.sync import sync as run_plex_sync
|
|
from app.plex.watch_sync import run_plex_watch_reconcile, run_plex_watch_sync
|
|
from app.notifications import create_notification
|
|
from app.state import is_sync_paused
|
|
from app.sync.autotag import run_autotag_all
|
|
from app.sync.explore import purge_ephemeral
|
|
from app.sync.maintenance import run_maintenance
|
|
from app.sync.playlists import sync_all_playlists
|
|
from app.sync.runner import (
|
|
run_deep_backfill,
|
|
run_enrich,
|
|
run_recent_backfill,
|
|
run_rss_poll,
|
|
run_shorts,
|
|
run_subscription_resync,
|
|
)
|
|
|
|
logger = logging.getLogger("siftlode.scheduler")
|
|
|
|
_scheduler: BackgroundScheduler | None = None
|
|
|
|
# Per-job run activity, kept in-memory for the admin Scheduler dashboard to read (the
|
|
# scheduler lives in the same single-worker process as the API, so a request can read it
|
|
# directly). Ephemeral by design — it reflects "what the scheduler is doing right now and
|
|
# how the last runs went" since this process started; durable history isn't the goal here.
|
|
_activity: dict[str, dict] = {}
|
|
_activity_lock = threading.Lock()
|
|
|
|
# Set (per-thread) by a manual "run now" trigger to the id of the admin who clicked, so that
|
|
# run — and only that run, not the recurring scheduled ones — posts a completion notification
|
|
# to their inbox. Unset (None) for scheduled runs.
|
|
_manual_actor: contextvars.ContextVar[int | None] = contextvars.ContextVar(
|
|
"manual_actor", default=None
|
|
)
|
|
|
|
# Each interval job's env/config default period (minutes). The admin can override any of
|
|
# these at runtime (stored in scheduler_settings, applied live); see load_intervals.
|
|
JOB_INTERVALS: dict[str, int] = {
|
|
"rss_poll": settings.rss_poll_minutes,
|
|
"enrich": settings.enrich_interval_minutes,
|
|
"backfill": settings.backfill_interval_minutes,
|
|
"autotag": settings.autotag_interval_minutes,
|
|
"shorts": settings.shorts_probe_interval_minutes,
|
|
"subscriptions": settings.subscriptions_resync_minutes,
|
|
"playlist_sync": settings.playlist_sync_minutes,
|
|
"maintenance": settings.maintenance_interval_minutes,
|
|
"demo_reset": settings.demo_reset_minutes,
|
|
"explore_cleanup": settings.explore_cleanup_minutes,
|
|
"download_gc": settings.download_gc_minutes,
|
|
"plex_sync": settings.plex_sync_interval_min,
|
|
"plex_watch_sync": settings.plex_watch_sync_interval_min,
|
|
"plex_watch_reconcile": settings.plex_watch_reconcile_interval_min,
|
|
"audit_gc": settings.audit_gc_minutes,
|
|
}
|
|
|
|
# Sane bounds for an admin-set interval (minutes).
|
|
MIN_INTERVAL = 1
|
|
MAX_INTERVAL = 1440 # one day
|
|
|
|
|
|
def load_intervals(db) -> dict[str, int]:
|
|
"""Effective per-job intervals: the env/config defaults overlaid with any admin overrides
|
|
saved in scheduler_settings."""
|
|
iv = dict(JOB_INTERVALS)
|
|
for row in db.execute(select(SchedulerSetting)).scalars():
|
|
if row.job_id in iv and row.interval_minutes:
|
|
iv[row.job_id] = row.interval_minutes
|
|
return iv
|
|
|
|
|
|
def apply_interval(db, job_id: str, minutes: int) -> int:
|
|
"""Persist a job's interval override and reschedule the live job immediately (if the
|
|
scheduler runs in this process). Returns the applied value."""
|
|
if job_id not in JOB_INTERVALS:
|
|
raise KeyError(job_id)
|
|
minutes = max(MIN_INTERVAL, min(MAX_INTERVAL, int(minutes)))
|
|
row = db.get(SchedulerSetting, job_id)
|
|
if row is None:
|
|
db.add(SchedulerSetting(job_id=job_id, interval_minutes=minutes))
|
|
else:
|
|
row.interval_minutes = minutes
|
|
db.commit()
|
|
if _scheduler is not None and _scheduler.get_job(job_id) is not None:
|
|
_scheduler.reschedule_job(job_id, trigger="interval", minutes=minutes)
|
|
logger.info("rescheduled job %s to every %s min", job_id, minutes)
|
|
return minutes
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def _record(name: str, **fields) -> None:
|
|
with _activity_lock:
|
|
_activity.setdefault(name, {}).update(fields)
|
|
|
|
|
|
def _notify_done(db, actor_id: int, job_id: str, status: str, summary: str | None) -> None:
|
|
"""Post a completion notice to the triggering admin's inbox (manual runs only). Best
|
|
effort — a notification failure must never mask the job's own outcome."""
|
|
try:
|
|
create_notification(
|
|
db,
|
|
user_id=actor_id,
|
|
type="scheduler",
|
|
title=f"{job_id}: {status}", # English fallback; UI renders translated from data
|
|
body=summary,
|
|
data={"job_id": job_id, "status": status, "summary": summary},
|
|
)
|
|
except Exception:
|
|
db.rollback()
|
|
logger.exception("failed to write completion notification for job %s", job_id)
|
|
|
|
|
|
def _run_changed(result) -> bool:
|
|
"""Did an automatic job run actually do something worth an audit row? (No-op polls don't.)
|
|
A "change" = a truthy NUMERIC count; status-string dicts like {"skipped": "disabled"} (Plex
|
|
off — the default) or {"skipped": "no demo account"} are no-ops, not changes, so they don't
|
|
flood the trail every interval."""
|
|
if result is None:
|
|
return False
|
|
if isinstance(result, dict):
|
|
return any(isinstance(v, (int, float)) and v for v in result.values())
|
|
if isinstance(result, (int, float)):
|
|
return result != 0
|
|
return bool(result)
|
|
|
|
|
|
def _audit_run(db, actor_id, job_id, status, summary, result) -> None:
|
|
"""Write ONE run-summary audit row per job run (never per-item). Manual runs (an admin
|
|
clicked "run now" → actor_id set) and errors are always logged; automatic runs only when
|
|
they changed something, so routine no-op polls don't spam the trail. Best-effort."""
|
|
if actor_id is None and status == "ok" and not _run_changed(result):
|
|
return
|
|
try:
|
|
audit.record(
|
|
db, actor_id, AuditAction.JOB_RUN, target_type="job", target_id=job_id,
|
|
summary=f"{job_id}: {status}" + (f" — {summary}" if summary else ""),
|
|
after={"status": status, "summary": summary},
|
|
)
|
|
db.commit()
|
|
except Exception:
|
|
db.rollback()
|
|
logger.exception("failed to write audit row for job %s", job_id)
|
|
|
|
|
|
def _job(name: str, fn) -> None:
|
|
# The one place both run paths converge (manual trigger AND scheduled fire), so claim the run
|
|
# HERE, atomically: whichever reaches this first wins, the other skips. This is the real
|
|
# single-run guard — trigger_job's pre-check is only a fast best-effort answer for the UI.
|
|
if not _claim_running(name):
|
|
logger.info("job %s already running — skipping this trigger", name)
|
|
return
|
|
db = SessionLocal()
|
|
actor_id = _manual_actor.get() # set only for a manual "run now" trigger
|
|
_record(name, last_started=_now_iso(), last_error=None, progress=None) # running already claimed
|
|
|
|
def sink(current, total, phase):
|
|
_record(name, progress={"current": current, "total": total, "phase": phase})
|
|
|
|
try:
|
|
if is_sync_paused(db):
|
|
logger.info("job %s skipped (sync paused)", name)
|
|
_record(name, status="skipped", last_finished=_now_iso(),
|
|
last_result="paused", progress=None)
|
|
if actor_id is not None:
|
|
_notify_done(db, actor_id, name, "skipped", "sync paused")
|
|
_audit_run(db, actor_id, name, "skipped", "sync paused", None)
|
|
return
|
|
with progress.bind(sink):
|
|
result = fn(db)
|
|
summary = _summarize(result)
|
|
logger.info("job %s -> %s", name, result)
|
|
_record(name, status="ok", last_finished=_now_iso(),
|
|
last_result=summary, progress=None)
|
|
if actor_id is not None:
|
|
_notify_done(db, actor_id, name, "ok", summary)
|
|
_audit_run(db, actor_id, name, "ok", summary, result)
|
|
except Exception as exc:
|
|
db.rollback()
|
|
logger.exception("job %s failed", name)
|
|
err = str(exc) or exc.__class__.__name__
|
|
_record(name, status="error", last_finished=_now_iso(),
|
|
last_error=err, progress=None)
|
|
if actor_id is not None:
|
|
_notify_done(db, actor_id, name, "error", err)
|
|
_audit_run(db, actor_id, name, "error", err, None)
|
|
finally:
|
|
# ALWAYS release the claim, even if the body raised before its own status update — so a
|
|
# failure can never wedge the job as permanently "running".
|
|
_record(name, running=False)
|
|
db.close()
|
|
|
|
|
|
def _summarize(result) -> str | None:
|
|
"""Compact one-line rendering of a job's return value for the dashboard."""
|
|
if result is None:
|
|
return None
|
|
if isinstance(result, dict):
|
|
return ", ".join(f"{k}={v}" for k, v in result.items()) or None
|
|
return str(result)
|
|
|
|
|
|
def scheduler_snapshot() -> dict:
|
|
"""What the scheduler is doing, for the admin dashboard. `running_here` is False in
|
|
environments where the scheduler is disabled (e.g. local dev shares the prod-ish DB but
|
|
runs no scheduler), so the UI can say so while still showing DB-derived queue/quota."""
|
|
running_here = _scheduler is not None
|
|
next_runs: dict[str, str | None] = {}
|
|
live_intervals: dict[str, int] = {}
|
|
if _scheduler is not None:
|
|
for job in _scheduler.get_jobs():
|
|
nr = getattr(job, "next_run_time", None)
|
|
next_runs[job.id] = nr.astimezone(timezone.utc).isoformat() if nr else None
|
|
iv = getattr(job.trigger, "interval", None)
|
|
if iv is not None:
|
|
live_intervals[job.id] = round(iv.total_seconds() / 60)
|
|
with _activity_lock:
|
|
acts = {k: dict(v) for k, v in _activity.items()}
|
|
jobs = []
|
|
for job_id, default_interval in JOB_INTERVALS.items():
|
|
a = acts.get(job_id, {})
|
|
jobs.append(
|
|
{
|
|
"id": job_id,
|
|
"interval_minutes": live_intervals.get(job_id, default_interval),
|
|
"next_run": next_runs.get(job_id),
|
|
"running": a.get("running", False),
|
|
"status": a.get("status"),
|
|
"last_started": a.get("last_started"),
|
|
"last_finished": a.get("last_finished"),
|
|
"last_result": a.get("last_result"),
|
|
"last_error": a.get("last_error"),
|
|
"progress": a.get("progress"),
|
|
}
|
|
)
|
|
return {"running_here": running_here, "enabled": settings.scheduler_enabled, "jobs": jobs}
|
|
|
|
|
|
def _rss_job() -> None:
|
|
_job("rss_poll", run_rss_poll)
|
|
|
|
|
|
def _enrich_job() -> None:
|
|
def work(db):
|
|
with quota.attribute(None, quota.QuotaAction.VIDEOS_ENRICH):
|
|
return run_enrich(db)
|
|
|
|
_job("enrich", work)
|
|
|
|
|
|
def _backfill_job() -> None:
|
|
# Recent-first for not-yet-synced channels, then deep backfill for the rest. All
|
|
# background spend is attributed to the system (no actor), split by action.
|
|
def work(db):
|
|
with quota.attribute(None, quota.QuotaAction.VIDEOS_BACKFILL_RECENT):
|
|
recent = run_recent_backfill(db, max_channels=25)
|
|
with quota.attribute(None, quota.QuotaAction.VIDEOS_BACKFILL_FULL):
|
|
deep = run_deep_backfill(db, max_channels=10)
|
|
return {"recent": recent, "deep": deep}
|
|
|
|
_job("backfill", work)
|
|
|
|
|
|
def _autotag_job() -> None:
|
|
_job("autotag", lambda db: run_autotag_all(db, only_missing=True))
|
|
|
|
|
|
def _shorts_job() -> None:
|
|
_job("shorts", run_shorts)
|
|
|
|
|
|
def _subscriptions_job() -> None:
|
|
def work(db):
|
|
with quota.attribute(None, quota.QuotaAction.SUBSCRIPTIONS_RESYNC):
|
|
return run_subscription_resync(db)
|
|
|
|
_job("subscriptions", work)
|
|
|
|
|
|
def _playlist_sync_job() -> None:
|
|
# Mirror each read-scope user's YouTube playlists; per-user quota attribution is
|
|
# handled inside sync_all_playlists.
|
|
_job("playlist_sync", sync_all_playlists)
|
|
|
|
|
|
def _maintenance_job() -> None:
|
|
def work(db):
|
|
with quota.attribute(None, quota.QuotaAction.MAINTENANCE_REVALIDATE):
|
|
return run_maintenance(db)
|
|
|
|
_job("maintenance", work)
|
|
|
|
|
|
def _demo_reset_job() -> None:
|
|
# Auto-clean the shared demo sandbox. No-op (and never creates one) if there's no demo
|
|
# account. Lazy import avoids a routes<->scheduler import cycle.
|
|
def work(db):
|
|
from app.models import User
|
|
demo = db.query(User).filter(User.is_demo.is_(True)).one_or_none()
|
|
if demo is None:
|
|
return {"skipped": "no demo account"}
|
|
from app.routes.admin import reset_demo_state
|
|
return {"playlists_seeded": reset_demo_state(db, demo)}
|
|
|
|
_job("demo_reset", work)
|
|
|
|
|
|
def _explore_cleanup_job() -> None:
|
|
# Reclaim un-kept ephemeral discovery content (explored channels + live-search results) after
|
|
# their grace periods. 0 quota (pure DB), so no quota attribution.
|
|
_job("explore_cleanup", purge_ephemeral)
|
|
|
|
|
|
def _download_gc_job() -> None:
|
|
# Retention GC for the download center: pre-expiry warnings, TTL deletion, LRU eviction.
|
|
# Pure disk/DB work (no YouTube quota).
|
|
_job("download_gc", run_download_gc)
|
|
|
|
|
|
def _plex_sync_job() -> None:
|
|
# Mirror the enabled Plex library sections into the local catalog. Pure metadata (no YouTube
|
|
# quota); a no-op when the Plex module is disabled.
|
|
_job("plex_sync", run_plex_sync)
|
|
|
|
|
|
def _plex_watch_sync_job() -> None:
|
|
# Incremental Plex→Siftlode watch-state pull (history + on-deck) + re-push of unsynced local
|
|
# states. Cheap; a no-op when Plex is disabled or no owner link has sync enabled.
|
|
_job("plex_watch_sync", run_plex_watch_sync)
|
|
|
|
|
|
def _plex_watch_reconcile_job() -> None:
|
|
# Full watch-state reconcile (whole-section rescan) — catches un-watches the incremental feed
|
|
# can't. Heavier, so it runs far less often (daily default).
|
|
_job("plex_watch_reconcile", run_plex_watch_reconcile)
|
|
|
|
|
|
def _audit_gc_job() -> None:
|
|
# Prune audit-log rows older than the admin-set retention (audit_retention_days; 0 = keep
|
|
# forever). Pure DB, no quota.
|
|
def work(db):
|
|
from app import sysconfig
|
|
return {"reaped": audit.gc(db, sysconfig.get_int(db, "audit_retention_days"))}
|
|
|
|
_job("audit_gc", work)
|
|
|
|
|
|
# job_id -> wrapper. The single source of truth for which jobs exist and how to run one,
|
|
# shared by start_scheduler (recurring registration) and trigger_job (manual "run now").
|
|
JOB_FUNCS: dict[str, Callable[[], None]] = {
|
|
"rss_poll": _rss_job,
|
|
"enrich": _enrich_job,
|
|
"backfill": _backfill_job,
|
|
"autotag": _autotag_job,
|
|
"shorts": _shorts_job,
|
|
"subscriptions": _subscriptions_job,
|
|
"playlist_sync": _playlist_sync_job,
|
|
"maintenance": _maintenance_job,
|
|
"demo_reset": _demo_reset_job,
|
|
"explore_cleanup": _explore_cleanup_job,
|
|
"download_gc": _download_gc_job,
|
|
"plex_sync": _plex_sync_job,
|
|
"plex_watch_sync": _plex_watch_sync_job,
|
|
"plex_watch_reconcile": _plex_watch_reconcile_job,
|
|
"audit_gc": _audit_gc_job,
|
|
}
|
|
|
|
|
|
def _is_running(job_id: str) -> bool:
|
|
with _activity_lock:
|
|
return bool(_activity.get(job_id, {}).get("running"))
|
|
|
|
|
|
def _claim_running(job_id: str) -> bool:
|
|
"""Atomically claim a job as running: True if it was idle (and marks it running now), False if
|
|
it was already running. Holding the lock across BOTH the check and the set is what makes the
|
|
single-run guard race-free — called once at the top of _job so a manual trigger and a scheduled
|
|
fire (or two rapid triggers) converge here and exactly one wins. Released in _job's finally."""
|
|
with _activity_lock:
|
|
if _activity.get(job_id, {}).get("running"):
|
|
return False
|
|
_activity.setdefault(job_id, {})["running"] = True
|
|
return True
|
|
|
|
|
|
def running_job_ids() -> set[str]:
|
|
"""Ids of jobs executing right now (any trigger). Lets non-admin surfaces — e.g. the
|
|
header's sync indicator — reflect real activity instead of just pending-work counts."""
|
|
with _activity_lock:
|
|
return {jid for jid, a in _activity.items() if a.get("running")}
|
|
|
|
|
|
def trigger_job(job_id: str, actor_id: int | None = None) -> str:
|
|
"""Run a job immediately in a background thread, independent of its interval schedule.
|
|
The wrapper handles its own DB session, activity tracking and pause-skip, so the live
|
|
dashboard reflects the run. Returns "started", "already_running", or raises KeyError for
|
|
an unknown job. Refusing a concurrent run keeps a manual trigger from overlapping a
|
|
scheduled run (APScheduler enforces max_instances=1 for the scheduled side).
|
|
|
|
`actor_id` (the admin who clicked "run now") is stashed in a contextvar inside the new
|
|
thread so the run posts a completion notification to that admin's inbox."""
|
|
fn = JOB_FUNCS.get(job_id)
|
|
if fn is None:
|
|
raise KeyError(job_id)
|
|
# Best-effort fast answer for the UI; the authoritative single-run guard is _job's atomic claim
|
|
# (so a manual trigger and a scheduled fire can't both run). A race here just means the loser's
|
|
# _job claims, sees it's taken, and skips — no double run.
|
|
if _is_running(job_id):
|
|
return "already_running"
|
|
|
|
def run() -> None:
|
|
if actor_id is not None:
|
|
_manual_actor.set(actor_id)
|
|
fn()
|
|
|
|
threading.Thread(target=run, name=f"trigger-{job_id}", daemon=True).start()
|
|
return "started"
|
|
|
|
|
|
def trigger_all(actor_id: int | None = None) -> list[str]:
|
|
"""Trigger every job that isn't already running; returns the ids actually started."""
|
|
started = []
|
|
for job_id in JOB_FUNCS:
|
|
if trigger_job(job_id, actor_id) == "started":
|
|
started.append(job_id)
|
|
return started
|
|
|
|
|
|
def start_scheduler() -> None:
|
|
global _scheduler
|
|
if not settings.scheduler_enabled or _scheduler is not None:
|
|
return
|
|
# Effective intervals = env defaults overlaid with any admin overrides from the DB.
|
|
db = SessionLocal()
|
|
try:
|
|
iv = load_intervals(db)
|
|
finally:
|
|
db.close()
|
|
scheduler = BackgroundScheduler(timezone="UTC")
|
|
for job_id, fn in JOB_FUNCS.items():
|
|
scheduler.add_job(fn, "interval", minutes=iv[job_id], id=job_id)
|
|
scheduler.start()
|
|
_scheduler = scheduler
|
|
logger.info("scheduler started")
|
|
|
|
|
|
def shutdown_scheduler() -> None:
|
|
global _scheduler
|
|
if _scheduler is not None:
|
|
_scheduler.shutdown(wait=False)
|
|
_scheduler = None
|