feat(scheduler): admin dashboard endpoint + per-job activity tracking

Record each scheduler job's run activity in-process (running / last result /
error / timestamps) and expose a snapshot. New admin-only GET
/api/admin/scheduler returns the per-job activity (with APScheduler next-run
times), the DB-derived work still queued (channels/videos pending, deep ETA,
live-refresh backlog) and the shared quota picture.
This commit is contained in:
npeter83 2026-06-16 14:38:51 +02:00
parent 30728dfe7c
commit 02d913f133
3 changed files with 177 additions and 2 deletions

View file

@ -26,7 +26,19 @@ from starlette.middleware.sessions import SessionMiddleware
from app import auth from app import auth
from app.config import settings from app.config import settings
from app.routes import admin, channels, feed, health, me, playlists, quota, sync, tags, version from app.routes import (
admin,
channels,
feed,
health,
me,
playlists,
quota,
scheduler as scheduler_routes,
sync,
tags,
version,
)
from app.scheduler import shutdown_scheduler, start_scheduler from app.scheduler import shutdown_scheduler, start_scheduler
@ -68,6 +80,7 @@ app.include_router(me.router)
app.include_router(channels.router) app.include_router(channels.router)
app.include_router(playlists.router) app.include_router(playlists.router)
app.include_router(admin.router) app.include_router(admin.router)
app.include_router(scheduler_routes.router)
app.include_router(quota.router) app.include_router(quota.router)
app.include_router(version.router) app.include_router(version.router)

View file

@ -0,0 +1,86 @@
"""Admin Scheduler dashboard: a live view of what the background scheduler is doing —
per-job run activity (from the in-process scheduler), the work still queued (DB-derived),
and the shared quota picture."""
from fastapi import APIRouter, Depends
from sqlalchemy import and_, func, or_, select
from sqlalchemy.orm import Session
from app import quota, state
from app.config import settings
from app.db import get_db
from app.models import Channel, Subscription, User, Video
from app.routes.admin import admin_user
from app.scheduler import scheduler_snapshot
from app.sync.runner import estimate_deep_backfill
router = APIRouter(prefix="/api/admin/scheduler", tags=["admin"])
@router.get("")
def get_scheduler(
_: User = Depends(admin_user), db: Session = Depends(get_db)
) -> dict:
snap = scheduler_snapshot()
# Work still queued (shared catalog, so these are instance-wide, not per-user).
deep_requested = (
select(Subscription.id)
.where(
Subscription.channel_id == Channel.id,
Subscription.deep_requested.is_(True),
)
.exists()
)
deep_channels = (
db.execute(
select(Channel).where(Channel.backfill_done.is_(False), deep_requested)
)
.scalars()
.all()
)
deep_eta = estimate_deep_backfill(db, deep_channels)
queue = {
"channels_recent_pending": db.scalar(
select(func.count()).select_from(Channel).where(Channel.recent_synced_at.is_(None))
)
or 0,
"channels_deep_pending": deep_eta["channels_pending"],
"deep_eta_seconds": deep_eta["eta_seconds"],
"videos_pending_enrich": db.scalar(
select(func.count()).select_from(Video).where(Video.enriched_at.is_(None))
)
or 0,
"videos_pending_shorts": db.scalar(
select(func.count())
.select_from(Video)
.where(Video.shorts_probed.is_(False), Video.enriched_at.is_not(None))
)
or 0,
"videos_live_refresh": db.scalar(
select(func.count())
.select_from(Video)
.where(
or_(
Video.live_status.in_(("live", "upcoming")),
and_(Video.live_status == "was_live", Video.duration_seconds.is_(None)),
)
)
)
or 0,
}
used = quota.units_used_today(db)
quota_info = {
"used_today": used,
"remaining_today": quota.remaining_today(db),
"daily_budget": settings.quota_daily_budget,
"backfill_reserve": settings.backfill_quota_reserve,
}
return {
**snap,
"paused": state.is_sync_paused(db),
"queue": queue,
"quota": quota_info,
}

View file

@ -1,6 +1,8 @@
"""Background scheduler: free RSS detection, enrichment of new videos, and quota-aware """Background scheduler: free RSS detection, enrichment of new videos, and quota-aware
recent-first / deep backfill. Runs inside the API process (single worker).""" recent-first / deep backfill. Runs inside the API process (single worker)."""
import logging import logging
import threading
from datetime import datetime, timezone
from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.schedulers.background import BackgroundScheduler
@ -23,22 +25,96 @@ logger = logging.getLogger("subfeed.scheduler")
_scheduler: BackgroundScheduler | None = None _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()
# Each interval job's configured period, surfaced so the dashboard can show "every N min".
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,
}
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 _job(name: str, fn) -> None: def _job(name: str, fn) -> None:
db = SessionLocal() db = SessionLocal()
_record(name, running=True, last_started=_now_iso(), last_error=None)
try: try:
if is_sync_paused(db): if is_sync_paused(db):
logger.info("job %s skipped (sync paused)", name) logger.info("job %s skipped (sync paused)", name)
_record(name, running=False, status="skipped", last_finished=_now_iso(),
last_result="paused")
return return
result = fn(db) result = fn(db)
logger.info("job %s -> %s", name, result) logger.info("job %s -> %s", name, result)
except Exception: _record(name, running=False, status="ok", last_finished=_now_iso(),
last_result=_summarize(result))
except Exception as exc:
db.rollback() db.rollback()
logger.exception("job %s failed", name) logger.exception("job %s failed", name)
_record(name, running=False, status="error", last_finished=_now_iso(),
last_error=str(exc) or exc.__class__.__name__)
finally: finally:
db.close() 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] = {}
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
with _activity_lock:
acts = {k: dict(v) for k, v in _activity.items()}
jobs = []
for job_id, interval in JOB_INTERVALS.items():
a = acts.get(job_id, {})
jobs.append(
{
"id": job_id,
"interval_minutes": 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"),
}
)
return {"running_here": running_here, "enabled": settings.scheduler_enabled, "jobs": jobs}
def _rss_job() -> None: def _rss_job() -> None:
_job("rss_poll", lambda db: run_rss_poll(db)) _job("rss_poll", lambda db: run_rss_poll(db))