diff --git a/backend/main.py b/backend/main.py index df49359..e1bf01d 100644 --- a/backend/main.py +++ b/backend/main.py @@ -16,6 +16,7 @@ STATIC_CACHE = f"public, max-age={STATIC_MAX_AGE_SECONDS}, must-revalidate" sys.path.insert(0, str(Path(__file__).parent)) +import scheduler 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 @@ -384,7 +385,8 @@ def _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") diff --git a/backend/scheduler.py b/backend/scheduler.py new file mode 100644 index 0000000..27f557a --- /dev/null +++ b/backend/scheduler.py @@ -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 diff --git a/backend/tests/test_scheduler.py b/backend/tests/test_scheduler.py new file mode 100644 index 0000000..010edbd --- /dev/null +++ b/backend/tests/test_scheduler.py @@ -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