From 4248b450ea5f0bcfd3b4db93c28a50cea48681ce Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 15 Oct 2026 17:34:01 -0400 Subject: [PATCH] feat(523): the backup job - an hourly tick that runs at the chosen hour, Back up now, and the last 30 runs services/backup_job.py reads its config when it fires, so startup reads nothing (#4181) and a changed hour needs no reschedule. A run is held to a 60-minute deadline (a hung NFS mount would otherwise hold the lock and skip every later backup), never runs twice at once, and keeps its outcome - file, size, duration, or the reason in words - in the bounded backup_runs setting, which no backup carries. summarise() gives the page its verdict: overdue when on and no success for 26h. Admin API: GET/PUT /api/admin/backups, POST /backups/check (tests unsaved form values, 45s deadline), POST /backups/run (202, background). Co-Authored-By: Claude Opus 5.5 --- src/scribe/app.py | 7 + src/scribe/routes/admin.py | 114 +++++++++++ src/scribe/services/backup.py | 6 +- src/scribe/services/backup_destination.py | 26 +++ src/scribe/services/backup_job.py | 188 ++++++++++++++++++ tests/test_services_backup_job.py | 223 ++++++++++++++++++++++ 6 files changed, 563 insertions(+), 1 deletion(-) create mode 100644 src/scribe/services/backup_job.py create mode 100644 tests/test_services_backup_job.py diff --git a/src/scribe/app.py b/src/scribe/app.py index 17f542ae..64bd549d 100644 --- a/src/scribe/app.py +++ b/src/scribe/app.py @@ -304,6 +304,11 @@ def create_app() -> Quart: asyncio.get_running_loop(), await get_maintenance_hour() ) + # Scheduled backups (milestone 523). An hourly tick that reads its + # config when it fires, so starting it reads nothing here (#4181). + from scribe.services.backup_job import start_backup_scheduler + start_backup_scheduler(asyncio.get_running_loop()) + # Diagnostic instrumentation — heartbeat, signal handlers, asyncio # exception hook. Cheap (~1 log line/min), high diagnostic value when # the app crashes mysteriously. See services/diagnostics.py. @@ -330,6 +335,8 @@ def create_app() -> Quart: stop_db_maintenance_scheduler, ) stop_db_maintenance_scheduler() + from scribe.services.backup_job import stop_backup_scheduler + stop_backup_scheduler() from scribe.services.diagnostics import stop_diagnostics stop_diagnostics() diff --git a/src/scribe/routes/admin.py b/src/scribe/routes/admin.py index 9edf548c..611754cb 100644 --- a/src/scribe/routes/admin.py +++ b/src/scribe/routes/admin.py @@ -315,6 +315,120 @@ async def run_db_maintenance_now(): return jsonify(summary) +# ── Scheduled backups (milestone 523) ──────────────────────────────────────── +# One dated, uncompressed tar per run to a folder or an S3 bucket — see +# services/backup_file.py for why it is shaped that way, backup_destination.py +# for where it goes, backup_job.py for when. + +# How long "Test" waits on the destination. A folder on a hung NFS mount has +# no timeout of its own (rule 156), and the request must come back either way. +_BACKUP_CHECK_TIMEOUT = 45.0 + + +async def _backups_payload() -> dict: + from datetime import datetime, timezone + + from scribe.services import backup_destination as dest + from scribe.services import backup_job as job + + config = await dest.load_config() + secret, stored = await dest.load_secret() + runs = await job.list_runs() + return { + "config": config.to_dict(), + # Never the secret itself: whether one is kept, and whether this + # instance can still open it (it can't after a restore onto another + # SECRET_KEY, and then it has to be entered again). + "secret": {"set": stored, "readable": secret is not None}, + "problems": config.problems(has_secret=secret is not None), + "running": job.is_running(), + "runs": runs, + **job.summarise(runs, enabled=config.enabled, now=datetime.now(timezone.utc)), + } + + +@admin_bp.route("/backups", methods=["GET"]) +@admin_required +async def get_backups(): + """The backup destination, its schedule, and the recent runs.""" + return jsonify(await _backups_payload()) + + +@admin_bp.route("/backups", methods=["PUT"]) +@admin_required +async def update_backups(): + """Change the destination or schedule. `s3_secret` replaces the stored + secret when it is sent and isn't the mask; `clear_s3_secret` removes it.""" + from scribe.services import backup_destination as dest + + data = await request.get_json() or {} + try: + config = dest.apply_changes(await dest.load_config(), data) + except ValueError as exc: + return jsonify({"error": str(exc)}), 400 + await dest.save_config(config) + secret = data.get("s3_secret") + if data.get("clear_s3_secret"): + await dest.save_secret("") + elif isinstance(secret, str) and secret and secret != SECRET_MASK: + await dest.save_secret(secret) + uid = get_current_user_id() + await log_audit( + "backup_config", user_id=uid, username=g.user.username, + ip_address=request.remote_addr, + details={**config.to_dict(), "secret_changed": bool( + data.get("clear_s3_secret") or (isinstance(secret, str) and secret and secret != SECRET_MASK) + )}, + ) + return jsonify(await _backups_payload()) + + +@admin_bp.route("/backups/check", methods=["POST"]) +@admin_required +async def check_backups_destination(): + """Ask the destination whether it would take a backup, without sending + one. The body may carry unsaved form values, so a destination can be + tested before it is saved; the stored secret is used unless one is sent.""" + from scribe.services import backup_destination as dest + + data = await request.get_json() or {} + try: + config = dest.apply_changes(await dest.load_config(), data) + except ValueError as exc: + return jsonify({"ok": False, "message": str(exc)}) + secret = data.get("s3_secret") + if not (isinstance(secret, str) and secret and secret != SECRET_MASK): + secret, _ = await dest.load_secret() + try: + message = await asyncio.wait_for( + dest.check_destination(config, secret), _BACKUP_CHECK_TIMEOUT, + ) + except dest.DestinationError as exc: + return jsonify({"ok": False, "message": str(exc)}) + except (TimeoutError, asyncio.TimeoutError): + return jsonify({"ok": False, "message": ( + f"The destination didn't answer within {int(_BACKUP_CHECK_TIMEOUT)} seconds." + )}) + return jsonify({"ok": True, "message": message}) + + +@admin_bp.route("/backups/run", methods=["POST"]) +@admin_required +async def run_backup_now(): + """Start a backup now. It runs in the background; the page follows it + through GET /backups (`running`, then the newest run).""" + from scribe.services import backup_job as job + + try: + job.start_backup_now() + except job.BackupBusy: + return jsonify({"error": "A backup is already running."}), 409 + uid = get_current_user_id() + await log_audit("backup_run", user_id=uid, username=g.user.username, + ip_address=request.remote_addr, details={}) + return jsonify({"status": "started"}), 202 + + @admin_bp.route("/invitations", methods=["POST"]) @admin_required async def create_invite(): diff --git a/src/scribe/services/backup.py b/src/scribe/services/backup.py index a57ab811..9adda6a5 100644 --- a/src/scribe/services/backup.py +++ b/src/scribe/services/backup.py @@ -776,7 +776,11 @@ def _rule_version_rows(rows) -> list[dict]: # under SECRET_KEY, but a backup is exactly the thing copied off-site — it must # not carry the key that writes to the place it is copied to. A restore leaves # the destination configured and asks for the secret again. -_SETTINGS_NOT_EXPORTED = frozenset({"backup_s3_secret"}) +# +# `backup_runs` is the scheduled backup's own run history: a record of what +# THIS instance sent where. Restored onto another instance it would claim +# backups that instance never made. +_SETTINGS_NOT_EXPORTED = frozenset({"backup_s3_secret", "backup_runs"}) def _setting_rows(rows) -> list[dict]: diff --git a/src/scribe/services/backup_destination.py b/src/scribe/services/backup_destination.py index f339a981..84849a7b 100644 --- a/src/scribe/services/backup_destination.py +++ b/src/scribe/services/backup_destination.py @@ -108,6 +108,32 @@ class BackupConfig: return out +def apply_changes(current: BackupConfig, data: dict) -> BackupConfig: + """`current` with the fields an admin sent changed, checked strictly. + + `from_dict` forgives a stored value it can't read; a value a person just + typed is refused instead, with the reason, so a typo in the hour does not + silently become 02:00. Keys that aren't config fields (the secret, say) + are ignored here. + """ + names = {f.name for f in fields(BackupConfig)} + changes = {k: v for k, v in (data or {}).items() if k in names} + if "hour" in changes: + try: + hour = int(changes["hour"]) + except (TypeError, ValueError): + raise ValueError("The hour must be a whole number from 0 to 23.") from None + if not 0 <= hour <= 23: + raise ValueError("The hour must be from 0 to 23 (UTC).") + changes["hour"] = hour + if "destination" in changes and changes["destination"] not in DESTINATION_KINDS: + raise ValueError("The destination must be a folder or an S3 bucket.") + for name in ("enabled", "s3_verify_tls"): + if name in changes and not isinstance(changes[name], bool): + raise ValueError(f"{name} must be true or false.") + return BackupConfig.from_dict({**current.to_dict(), **changes}) + + async def load_config() -> BackupConfig: raw = await get_admin_setting(CONFIG_SETTING, "") try: diff --git a/src/scribe/services/backup_job.py b/src/scribe/services/backup_job.py new file mode 100644 index 00000000..4e197330 --- /dev/null +++ b/src/scribe/services/backup_job.py @@ -0,0 +1,188 @@ +"""The scheduled backup: when it runs, running it, and what each run did. + +ONE APScheduler job (services/scheduler.py) fires at the top of every hour and +reads the config at fire time: when backups are on and the hour is the chosen +one, it runs. Reading at fire time rather than scheduling the chosen hour +keeps the app's startup free of a database read (#4181 — a lifespan hook that +waits on the database is a worker that never serves), and a changed hour +needs no reschedule. + +A run exports the full backup, writes it as the dated tar +(services/backup_file.py) and delivers it (services/backup_destination.py). +Its outcome is kept as one entry in a bounded list — the newest +`KEEP_RUNS` — in the `backup_runs` admin setting, like DB maintenance keeps +its last run. A failure is logged as an error AND kept with its reason in +words, so the Settings page can say what went wrong without anyone reading +logs. +""" +from __future__ import annotations + +import asyncio +import json +import logging +import os +import time +from datetime import datetime, timedelta, timezone + +from apscheduler.triggers.cron import CronTrigger + +from scribe.services.backup_destination import ( + DestinationError, deliver, load_config, load_secret, +) +from scribe.services.scheduler import ScheduledJob +from scribe.services.settings import get_admin_setting, set_admin_setting + +logger = logging.getLogger(__name__) + +RUNS_SETTING = "backup_runs" +KEEP_RUNS = 30 +# A daily backup that hasn't succeeded for this long is overdue: a day plus +# room for one slow run, so a backup that lands at 02:40 instead of 02:05 +# does not flicker the page into a warning. +STALE_AFTER = timedelta(hours=26) +# The whole run's deadline (rule 156: a wait with no deadline is a bug). The +# S3 upload has its own 15-minute timeout, but a folder on a hung NFS mount +# has none, and a run that never returns holds the lock — every later backup +# would be skipped as "already running", silently, forever. +RUN_DEADLINE = timedelta(minutes=60) + +_LOCK = asyncio.Lock() +_TASKS: set[asyncio.Task] = set() + + +class BackupBusy(Exception): + """A backup is already running.""" + + +def is_running() -> bool: + return _LOCK.locked() + + +async def list_runs() -> list[dict]: + """The kept runs, newest first.""" + raw = await get_admin_setting(RUNS_SETTING, "") + try: + runs = json.loads(raw) if raw else [] + except ValueError: + return [] + return [r for r in runs if isinstance(r, dict)] if isinstance(runs, list) else [] + + +async def _record(run: dict) -> None: + runs = [run, *await list_runs()][:KEEP_RUNS] + await set_admin_setting(RUNS_SETTING, json.dumps(runs)) + + +def summarise(runs: list[dict], *, enabled: bool, now: datetime) -> dict: + """What the page leads with: when the last good backup was, and whether + backups are overdue. Pure, so the stale rule is tested on its own.""" + last_ok = next((r for r in runs if r.get("status") == "ok"), None) + last_ok_at = _parse(last_ok.get("finished_at")) if last_ok else None + if not enabled: + overdue = False + elif last_ok_at is None: + # On, and never succeeded: overdue once the first scheduled chance + # has passed. Before any run at all, there is nothing to be late for. + overdue = bool(runs) + else: + overdue = now - last_ok_at > STALE_AFTER + return { + "last_run": runs[0] if runs else None, + "last_success": last_ok, + "overdue": overdue, + } + + +def _parse(value) -> datetime | None: + try: + return datetime.fromisoformat(value) + except (TypeError, ValueError): + return None + + +async def run_backup(trigger: str) -> dict: + """Run one backup now and keep its outcome. Raises BackupBusy when one is + already running; never raises for a failed backup — that is the outcome. + """ + if _LOCK.locked(): + raise BackupBusy() + async with _LOCK: + started = datetime.now(timezone.utc) + clock = time.monotonic() + run: dict = {"started_at": started.isoformat(), "trigger": trigger} + try: + config = await load_config() + secret, _ = await load_secret() + run["destination"] = config.destination + delivered = await asyncio.wait_for( + _export_and_deliver(config, secret, started), RUN_DEADLINE.total_seconds(), + ) + run.update(status="ok", file=delivered.name, + size_bytes=delivered.size_bytes, where=delivered.where) + except (TimeoutError, asyncio.TimeoutError): + run.update(status="failed", error=( + f"The backup didn't finish within {int(RUN_DEADLINE.total_seconds() // 60)} " + "minutes and was abandoned. The destination may be hanging." + )) + except DestinationError as exc: + run.update(status="failed", error=str(exc)) + except Exception as exc: + logger.exception("backup run failed") + run.update(status="failed", error=f"The backup failed: {exc}") + run["finished_at"] = datetime.now(timezone.utc).isoformat() + run["duration_s"] = round(time.monotonic() - clock, 1) + if run["status"] == "ok": + logger.info("backup: wrote %s (%d bytes) to %s in %.1fs", + run["file"], run["size_bytes"], run["where"], run["duration_s"]) + else: + logger.error("backup: %s", run["error"]) + try: + await _record(run) + except Exception: + # The backup itself is done (or failed) either way; losing its + # entry in the history must not turn into an unhandled error in a + # scheduler thread. The log line above still says what happened. + logger.exception("backup: could not keep this run in the history") + return run + + +async def _export_and_deliver(config, secret, started): + from scribe.services.backup import export_full_backup + payload = await export_full_backup() + return await deliver( + payload, config, secret, started, + app_version=os.environ.get("APP_VERSION", "dev"), + ) + + +def start_backup_now() -> None: + """Start a manual backup in the background (the request returns at once; + the page reads progress from `is_running` and the run list). Raises + BackupBusy rather than queueing a second run behind the first.""" + if _LOCK.locked(): + raise BackupBusy() + task = asyncio.get_running_loop().create_task(run_backup("manual")) + _TASKS.add(task) + task.add_done_callback(_TASKS.discard) + + +async def _hourly_check() -> None: + config = await load_config() + if not config.enabled or datetime.now(timezone.utc).hour != config.hour: + return + try: + await run_backup("scheduled") + except BackupBusy: + logger.info("backup: a run is already going; the scheduled one is skipped") + + +_JOB = ScheduledJob("backup_hourly_check", _hourly_check, label="Backup") + + +def start_backup_scheduler(loop: asyncio.AbstractEventLoop) -> None: + _JOB.start(loop, CronTrigger(minute=0, timezone="UTC"), + describe="hourly; runs at the configured UTC hour") + + +def stop_backup_scheduler() -> None: + _JOB.stop() diff --git a/tests/test_services_backup_job.py b/tests/test_services_backup_job.py new file mode 100644 index 00000000..a54b8c82 --- /dev/null +++ b/tests/test_services_backup_job.py @@ -0,0 +1,223 @@ +"""The scheduled backup job (services/backup_job.py) and the admin surface +over it. No database: admin settings are an in-memory dict, and the export +and delivery are stand-ins, so what is under test is the job's own contract — +it records every outcome, keeps a bounded history, runs only at its hour, +and never runs twice at once. +""" +import asyncio +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +import pytest + +from scribe.services import backup_destination, backup_job +from scribe.services.backup_destination import ( + BackupConfig, Delivered, DestinationError, apply_changes, +) + +NOW = datetime(2026, 10, 15, 2, 0, 0, tzinfo=timezone.utc) + + +class _Settings(dict): + async def get(self, key, default=""): + return self.get_value(key, default) + + def get_value(self, key, default): + return dict.get(self, key, default) + + async def set(self, key, value): + self[key] = value + + +def _patched(settings, deliver=None, config=BackupConfig(enabled=True)): + async def load_config(): + return config + + async def load_secret(): + return (None, False) + + async def export(): + return {"version": 25, "users": []} + + async def delivered(payload, config, secret, at, **kw): + return Delivered("scribe-20261015T020000Z.tar", 1234, "/backups") + + return [ + patch.object(backup_job, "get_admin_setting", settings.get), + patch.object(backup_job, "set_admin_setting", settings.set), + patch.object(backup_job, "load_config", load_config), + patch.object(backup_job, "load_secret", load_secret), + patch.object(backup_job, "deliver", deliver or delivered), + patch("scribe.services.backup.export_full_backup", export), + ] + + +def _run(coro, patches): + for p in patches: + p.start() + try: + return asyncio.run(coro) + finally: + for p in reversed(patches): + p.stop() + + +def test_a_good_run_is_kept_with_its_file_and_size(): + settings = _Settings() + run = _run(backup_job.run_backup("manual"), _patched(settings)) + assert run["status"] == "ok" + assert (run["file"], run["size_bytes"], run["trigger"]) == ( + "scribe-20261015T020000Z.tar", 1234, "manual", + ) + runs = _run(backup_job.list_runs(), _patched(settings)) + assert runs == [run] + + +def test_a_failed_run_is_kept_with_its_reason_in_words(): + async def refuse(*a, **kw): + raise DestinationError("/backups isn't there.") + + settings = _Settings() + run = _run(backup_job.run_backup("scheduled"), _patched(settings, deliver=refuse)) + assert run["status"] == "failed" + assert run["error"] == "/backups isn't there." + + +def test_an_unexpected_error_is_a_failed_run_not_a_crash(): + async def boom(*a, **kw): + raise RuntimeError("disk on fire") + + settings = _Settings() + run = _run(backup_job.run_backup("scheduled"), _patched(settings, deliver=boom)) + assert run["status"] == "failed" + assert "disk on fire" in run["error"] + + +def test_a_run_that_hangs_is_abandoned_at_the_deadline(): + async def hang(*a, **kw): + await asyncio.sleep(10) + + settings = _Settings() + with patch.object(backup_job, "RUN_DEADLINE", timedelta(seconds=0.05)): + run = _run(backup_job.run_backup("scheduled"), _patched(settings, deliver=hang)) + assert run["status"] == "failed" + assert "didn't finish" in run["error"] + assert not backup_job.is_running() + + +def test_the_history_keeps_only_the_newest_runs(): + settings = _Settings() + + async def many(): + for _ in range(backup_job.KEEP_RUNS + 5): + await backup_job.run_backup("scheduled") + return await backup_job.list_runs() + + runs = _run(many(), _patched(settings)) + assert len(runs) == backup_job.KEEP_RUNS + assert runs[0]["started_at"] >= runs[-1]["started_at"] + + +def test_a_second_run_while_one_is_going_is_refused(): + started = asyncio.Event() + release = asyncio.Event() + + async def slow(*a, **kw): + started.set() + await release.wait() + return Delivered("x.tar", 1, "/backups") + + async def both(): + first = asyncio.create_task(backup_job.run_backup("manual")) + await started.wait() + with pytest.raises(backup_job.BackupBusy): + await backup_job.run_backup("manual") + release.set() + return await first + + run = _run(both(), _patched(_Settings(), deliver=slow)) + assert run["status"] == "ok" + + +@pytest.mark.parametrize("enabled,hour,ran", [ + (True, NOW.hour, True), + (True, (NOW.hour + 1) % 24, False), + (False, NOW.hour, False), +]) +def test_the_hourly_tick_runs_only_when_on_and_at_its_hour(enabled, hour, ran): + calls = [] + + async def fake_run(trigger): + calls.append(trigger) + + class _Now(datetime): + @classmethod + def now(cls, tz=None): + return NOW + + async def config(): + return BackupConfig(enabled=enabled, hour=hour) + + with patch.object(backup_job, "load_config", config), \ + patch.object(backup_job, "run_backup", fake_run), \ + patch.object(backup_job, "datetime", _Now): + asyncio.run(backup_job._hourly_check()) + assert calls == (["scheduled"] if ran else []) + + +# ── The page's verdict ─────────────────────────────────────────────────────── + +def _ok(at): + return {"status": "ok", "finished_at": at.isoformat()} + + +def test_a_recent_success_is_not_overdue(): + s = backup_job.summarise([_ok(NOW - timedelta(hours=3))], enabled=True, now=NOW) + assert s["overdue"] is False + assert s["last_success"]["status"] == "ok" + + +def test_failures_after_an_old_success_are_overdue(): + runs = [{"status": "failed", "finished_at": NOW.isoformat()}, + _ok(NOW - timedelta(hours=30))] + s = backup_job.summarise(runs, enabled=True, now=NOW) + assert s["overdue"] is True + assert s["last_run"]["status"] == "failed" + + +def test_off_is_never_overdue_and_nothing_run_yet_is_not_late(): + assert backup_job.summarise([], enabled=False, now=NOW)["overdue"] is False + assert backup_job.summarise([], enabled=True, now=NOW)["overdue"] is False + failed_only = [{"status": "failed", "finished_at": NOW.isoformat()}] + assert backup_job.summarise(failed_only, enabled=True, now=NOW)["overdue"] is True + + +# ── What an admin sends ────────────────────────────────────────────────────── + +def test_changes_apply_on_top_of_the_current_config(): + config = apply_changes(BackupConfig(), {"enabled": True, "hour": "5", "s3_secret": "x"}) + assert (config.enabled, config.hour, config.destination) == (True, 5, "folder") + + +@pytest.mark.parametrize("data,said", [ + ({"hour": 24}, "from 0 to 23"), + ({"hour": "two"}, "whole number"), + ({"destination": "ftp"}, "folder or an S3 bucket"), + ({"enabled": "yes"}, "true or false"), +]) +def test_a_typed_mistake_is_refused_with_the_reason(data, said): + with pytest.raises(ValueError, match=said): + apply_changes(BackupConfig(), data) + + +def test_routes_and_scheduler_are_wired(): + from scribe.app import create_app + from scribe.routes import admin as admin_routes + + for name in ("get_backups", "update_backups", "check_backups_destination", "run_backup_now"): + assert callable(getattr(admin_routes, name)) + rules = {r.rule for r in create_app().url_map.iter_rules()} + assert {"/api/admin/backups", "/api/admin/backups/check", "/api/admin/backups/run"} <= rules + assert callable(backup_job.start_backup_scheduler) + assert callable(backup_job.stop_backup_scheduler) + assert backup_destination.SECRET_SETTING == "backup_s3_secret"