feat(scheduler): Hintergrund-Jobs im Serverprozess
Das Backend hatte bisher keinerlei Hintergrund-Mechanik. Neues Modul mit einer Job-Registry, angebunden ueber den FastAPI-Lifespan. Der Server laeuft als genau ein Prozess, ein In-Process-Loop laeuft damit garantiert nur einmal -- steht als Vorbehalt im Modulkopf, falls spaeter Worker dazukommen. Jeder Job laeuft im Thread mit eigener Session, weil SQLAlchemy hier synchron ist und Datei- oder Netz-I/O sonst den Event-Loop blockiert. Fehler werden pro Job gefangen. Zwei Jobs: der Anhang-Sweep alle 24 h und die iCal-Aktualisierung alle 15 min. Letztere treibt nur den vorhandenen _refresh_if_needed an, der das Intervall je Abo schon respektiert -- neu ist allein, dass Abos auch aktuell bleiben, wenn niemand die App offen hat. Jedes Abo einzeln gekapselt, weil der Helfer ohne Cache weiterwirft. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
b885e731ee
commit
4b9c09df74
+3
-1
@@ -16,6 +16,7 @@ STATIC_CACHE = f"public, max-age={STATIC_MAX_AGE_SECONDS}, must-revalidate"
|
|||||||
|
|
||||||
sys.path.insert(0, str(Path(__file__).parent))
|
sys.path.insert(0, str(Path(__file__).parent))
|
||||||
|
|
||||||
|
import scheduler
|
||||||
from database import Base, engine
|
from database import Base, engine
|
||||||
from routers import admin_router, attachments_router, auth_router, birthdays_router, caldav_router, dav_router, google_router, groups_router, homeassistant_router, ical_router, local_router, oidc_router, profile_router, settings_router, users_router
|
from routers import admin_router, attachments_router, auth_router, birthdays_router, caldav_router, dav_router, google_router, groups_router, homeassistant_router, ical_router, local_router, oidc_router, profile_router, settings_router, users_router
|
||||||
|
|
||||||
@@ -384,7 +385,8 @@ def _migrate():
|
|||||||
|
|
||||||
_migrate()
|
_migrate()
|
||||||
|
|
||||||
app = FastAPI(title="Calendarr", docs_url=None, redoc_url=None)
|
app = FastAPI(title="Calendarr", docs_url=None, redoc_url=None,
|
||||||
|
lifespan=scheduler.lifespan)
|
||||||
|
|
||||||
|
|
||||||
@app.middleware("http")
|
@app.middleware("http")
|
||||||
|
|||||||
@@ -0,0 +1,157 @@
|
|||||||
|
"""A small in-process scheduler for periodic maintenance jobs.
|
||||||
|
|
||||||
|
The server runs as a single uvicorn process (``main.py`` calls ``uvicorn.run``
|
||||||
|
with no ``workers=``), so an in-process loop runs exactly once — no locking or
|
||||||
|
leader election needed. If the deployment ever grows to multiple workers this
|
||||||
|
has to be revisited, because every worker would run every job.
|
||||||
|
|
||||||
|
Jobs are plain synchronous callables taking a ``Session``. They run in a worker
|
||||||
|
thread with their own session: SQLAlchemy is synchronous here, and a job doing
|
||||||
|
file I/O or an HTTP fetch on the event loop would stall every request.
|
||||||
|
|
||||||
|
Failures are caught per job, so a broken iCal feed cannot stop the attachment
|
||||||
|
sweep and neither can stop the loop.
|
||||||
|
|
||||||
|
Environment:
|
||||||
|
SCHEDULER_ENABLED "0" disables the loop entirely (default on)
|
||||||
|
ATTACHMENT_SWEEP_HOURS how often orphaned attachments are cleaned (24)
|
||||||
|
ICAL_REFRESH_MINUTES how often subscriptions are checked (15)
|
||||||
|
"""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import contextlib
|
||||||
|
import logging
|
||||||
|
import os
|
||||||
|
import time
|
||||||
|
from contextlib import asynccontextmanager
|
||||||
|
from typing import Callable
|
||||||
|
|
||||||
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
|
from database import SessionLocal
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
ENABLED = os.environ.get("SCHEDULER_ENABLED", "1") != "0"
|
||||||
|
# How often the loop wakes to see whether anything is due. Jobs have their own,
|
||||||
|
# much longer intervals; this only bounds how late a job can start.
|
||||||
|
TICK_SECONDS = 30
|
||||||
|
# Delay before the first run so a restart is not slowed by maintenance work.
|
||||||
|
FIRST_RUN_DELAY_SECONDS = 20
|
||||||
|
|
||||||
|
|
||||||
|
class _Job:
|
||||||
|
def __init__(self, name: str, interval: float, fn: Callable[[Session], None]):
|
||||||
|
self.name = name
|
||||||
|
self.interval = interval
|
||||||
|
self.fn = fn
|
||||||
|
self.next_run = time.monotonic() + FIRST_RUN_DELAY_SECONDS
|
||||||
|
|
||||||
|
|
||||||
|
_jobs: list[_Job] = []
|
||||||
|
|
||||||
|
|
||||||
|
def register(name: str, interval_seconds: float, fn: Callable[[Session], None]) -> None:
|
||||||
|
"""Add a job. Registering the same name twice replaces the first."""
|
||||||
|
_jobs[:] = [j for j in _jobs if j.name != name]
|
||||||
|
_jobs.append(_Job(name, interval_seconds, fn))
|
||||||
|
|
||||||
|
|
||||||
|
def _run_job(job: _Job) -> None:
|
||||||
|
"""Run one job in a worker thread, with its own session."""
|
||||||
|
db = SessionLocal()
|
||||||
|
try:
|
||||||
|
job.fn(db)
|
||||||
|
except Exception:
|
||||||
|
# Never let one job's failure escape: the loop must survive it, and so
|
||||||
|
# must the other jobs.
|
||||||
|
logger.exception("Scheduled job %s failed", job.name)
|
||||||
|
with contextlib.suppress(Exception):
|
||||||
|
db.rollback()
|
||||||
|
finally:
|
||||||
|
db.close()
|
||||||
|
|
||||||
|
|
||||||
|
async def _runner() -> None:
|
||||||
|
logger.info(
|
||||||
|
"Scheduler started with %d job(s): %s",
|
||||||
|
len(_jobs),
|
||||||
|
", ".join(j.name for j in _jobs),
|
||||||
|
)
|
||||||
|
while True:
|
||||||
|
await asyncio.sleep(TICK_SECONDS)
|
||||||
|
now = time.monotonic()
|
||||||
|
for job in list(_jobs):
|
||||||
|
if now < job.next_run:
|
||||||
|
continue
|
||||||
|
job.next_run = now + job.interval
|
||||||
|
await asyncio.to_thread(_run_job, job)
|
||||||
|
|
||||||
|
|
||||||
|
# ── The jobs ──────────────────────────────────────────────
|
||||||
|
|
||||||
|
def sweep_attachments(db: Session) -> None:
|
||||||
|
"""Remove attachment rows whose event is gone, and files with no row.
|
||||||
|
|
||||||
|
The explicit purge at each delete path is the normal mechanism; this is the
|
||||||
|
safety net for what slips past it — an external CalDAV client deleting an
|
||||||
|
event, a request that died between writing the file and committing the row,
|
||||||
|
or a delete path added later that forgets the helper.
|
||||||
|
"""
|
||||||
|
import attachments_store
|
||||||
|
|
||||||
|
attachments_store.sweep(db)
|
||||||
|
|
||||||
|
|
||||||
|
def refresh_ical_subscriptions(db: Session) -> None:
|
||||||
|
"""Re-fetch iCal subscriptions that are due.
|
||||||
|
|
||||||
|
``_refresh_if_needed`` already honours each subscription's own
|
||||||
|
``refresh_minutes`` and swallows network errors — except when a
|
||||||
|
subscription has no cached data at all, where it re-raises. So every
|
||||||
|
subscription is wrapped separately: one unreachable URL must not stop the
|
||||||
|
rest.
|
||||||
|
|
||||||
|
The lazy refresh in the merged read stays as it is. This only adds the case
|
||||||
|
nobody was covering: keeping subscriptions current while no one has the app
|
||||||
|
open.
|
||||||
|
"""
|
||||||
|
import models
|
||||||
|
from routers.ical_router import _refresh_if_needed
|
||||||
|
|
||||||
|
subs = (
|
||||||
|
db.query(models.ICalSubscription)
|
||||||
|
.filter(models.ICalSubscription.enabled == True) # noqa: E712
|
||||||
|
.all()
|
||||||
|
)
|
||||||
|
for sub in subs:
|
||||||
|
try:
|
||||||
|
_refresh_if_needed(sub, db)
|
||||||
|
except Exception:
|
||||||
|
logger.warning("Scheduled refresh of iCal subscription %s failed", sub.id)
|
||||||
|
with contextlib.suppress(Exception):
|
||||||
|
db.rollback()
|
||||||
|
|
||||||
|
|
||||||
|
def install_default_jobs() -> None:
|
||||||
|
sweep_hours = float(os.environ.get("ATTACHMENT_SWEEP_HOURS", "24"))
|
||||||
|
ical_minutes = float(os.environ.get("ICAL_REFRESH_MINUTES", "15"))
|
||||||
|
register("attachments.sweep", sweep_hours * 3600, sweep_attachments)
|
||||||
|
register("ical.refresh", ical_minutes * 60, refresh_ical_subscriptions)
|
||||||
|
|
||||||
|
|
||||||
|
@asynccontextmanager
|
||||||
|
async def lifespan(app):
|
||||||
|
task = None
|
||||||
|
if ENABLED:
|
||||||
|
install_default_jobs()
|
||||||
|
task = asyncio.create_task(_runner())
|
||||||
|
else:
|
||||||
|
logger.info("Scheduler disabled via SCHEDULER_ENABLED=0")
|
||||||
|
try:
|
||||||
|
yield
|
||||||
|
finally:
|
||||||
|
if task is not None:
|
||||||
|
task.cancel()
|
||||||
|
with contextlib.suppress(asyncio.CancelledError):
|
||||||
|
await task
|
||||||
@@ -0,0 +1,120 @@
|
|||||||
|
"""The in-process job scheduler and its two jobs."""
|
||||||
|
|
||||||
|
import time
|
||||||
|
|
||||||
|
import models
|
||||||
|
import scheduler
|
||||||
|
from conftest import auth, register_admin
|
||||||
|
from database import SessionLocal
|
||||||
|
|
||||||
|
|
||||||
|
def _run(fn):
|
||||||
|
db = SessionLocal()
|
||||||
|
try:
|
||||||
|
fn(db)
|
||||||
|
finally:
|
||||||
|
db.close()
|
||||||
|
|
||||||
|
|
||||||
|
def test_default_jobs_are_registered_with_sane_intervals():
|
||||||
|
scheduler.install_default_jobs()
|
||||||
|
names = {j.name: j for j in scheduler._jobs}
|
||||||
|
assert set(names) == {"attachments.sweep", "ical.refresh"}
|
||||||
|
assert names["attachments.sweep"].interval == 24 * 3600
|
||||||
|
assert names["ical.refresh"].interval == 15 * 60
|
||||||
|
# Nothing runs immediately: a restart must not be slowed by maintenance.
|
||||||
|
assert all(j.next_run > time.monotonic() for j in scheduler._jobs)
|
||||||
|
|
||||||
|
|
||||||
|
def test_registering_the_same_name_twice_replaces_it():
|
||||||
|
scheduler.register("demo", 60, lambda db: None)
|
||||||
|
scheduler.register("demo", 120, lambda db: None)
|
||||||
|
matching = [j for j in scheduler._jobs if j.name == "demo"]
|
||||||
|
assert len(matching) == 1 and matching[0].interval == 120
|
||||||
|
scheduler._jobs[:] = [j for j in scheduler._jobs if j.name != "demo"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_failing_job_is_contained(caplog):
|
||||||
|
"""One broken job must not take down the loop or the other jobs."""
|
||||||
|
def boom(db):
|
||||||
|
raise RuntimeError("kaputt")
|
||||||
|
|
||||||
|
job = scheduler._Job("explodes", 60, boom)
|
||||||
|
scheduler._run_job(job) # must not raise
|
||||||
|
assert "explodes" in caplog.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_sweep_job_cleans_an_orphaned_row(client):
|
||||||
|
"""End-to-end through the job function, not just the store helper."""
|
||||||
|
token = register_admin(client)
|
||||||
|
r = client.post("/api/local/calendars", headers=auth(token),
|
||||||
|
json={"name": "Privat", "color": "#4285f4"})
|
||||||
|
cal_id = r.json()["id"]
|
||||||
|
ev = client.post("/api/local/events", headers=auth(token), json={
|
||||||
|
"calendar_id": cal_id, "title": "Termin",
|
||||||
|
"start": "2026-06-10T10:00:00+00:00", "end": "2026-06-10T11:00:00+00:00",
|
||||||
|
}).json()
|
||||||
|
client.post(f"/api/local/events/{ev['id']}/attachments", headers=auth(token),
|
||||||
|
files={"file": ("x.pdf", b"%PDF-1.4\ntrailer\n", "application/pdf")})
|
||||||
|
|
||||||
|
db = SessionLocal()
|
||||||
|
ev_row = db.query(models.LocalEvent).filter(models.LocalEvent.uid == ev["id"]).first()
|
||||||
|
db.query(models.LocalEvent).filter(models.LocalEvent.id == ev_row.id).delete()
|
||||||
|
db.commit()
|
||||||
|
db.close()
|
||||||
|
|
||||||
|
_run(scheduler.sweep_attachments)
|
||||||
|
|
||||||
|
db = SessionLocal()
|
||||||
|
assert db.query(models.EventAttachment).count() == 0
|
||||||
|
db.close()
|
||||||
|
|
||||||
|
|
||||||
|
def test_ical_job_survives_an_unreachable_subscription(client):
|
||||||
|
"""A subscription with no cached data makes _refresh_if_needed re-raise —
|
||||||
|
the job has to wrap each one separately or a single bad URL stops the rest."""
|
||||||
|
token = register_admin(client)
|
||||||
|
db = SessionLocal()
|
||||||
|
user = db.query(models.User).first()
|
||||||
|
db.add(models.ICalSubscription(
|
||||||
|
user_id=user.id, name="Kaputt",
|
||||||
|
url="http://127.0.0.1:9/nicht-erreichbar.ics",
|
||||||
|
color="#ea4335", enabled=True, refresh_minutes=60,
|
||||||
|
))
|
||||||
|
db.commit()
|
||||||
|
db.close()
|
||||||
|
|
||||||
|
_run(scheduler.refresh_ical_subscriptions) # must not raise
|
||||||
|
|
||||||
|
|
||||||
|
def test_the_loop_runs_due_jobs_and_stops_with_the_app(monkeypatch):
|
||||||
|
"""Drive the real lifespan with a fast tick and a probe job."""
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
import main
|
||||||
|
|
||||||
|
monkeypatch.setattr(scheduler, "TICK_SECONDS", 0.02)
|
||||||
|
monkeypatch.setattr(scheduler, "FIRST_RUN_DELAY_SECONDS", 0)
|
||||||
|
ran = []
|
||||||
|
scheduler.register("probe", 3600, lambda db: ran.append(1))
|
||||||
|
try:
|
||||||
|
with TestClient(main.app): # entering runs the lifespan
|
||||||
|
deadline = time.time() + 3
|
||||||
|
while not ran and time.time() < deadline:
|
||||||
|
time.sleep(0.02)
|
||||||
|
assert ran, "the scheduler loop never ran the registered job"
|
||||||
|
finally:
|
||||||
|
scheduler._jobs[:] = [j for j in scheduler._jobs if j.name != "probe"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_plain_test_client_starts_no_background_work():
|
||||||
|
"""conftest builds TestClient without a context manager, so the lifespan —
|
||||||
|
and with it the loop — never starts. Keeps the rest of the suite free of
|
||||||
|
background jobs touching the same database."""
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
import main
|
||||||
|
|
||||||
|
before = len(scheduler._jobs)
|
||||||
|
TestClient(main.app)
|
||||||
|
assert len(scheduler._jobs) == before
|
||||||
Reference in New Issue
Block a user