diff --git a/README.md b/README.md index e909d4c..d256596 100644 --- a/README.md +++ b/README.md @@ -197,6 +197,14 @@ goes again in the next run. When the newest database backup is more than a day old or failed, or the check found files missing, the **Backups** row on **Settings → System** and the dot beside the brand turn amber and say why. +**Test a backup** proves a dump restores without touching your install: the dump +is loaded into a separate, temporary database on the same Postgres server, +every table is checked against the row counts recorded as the dump was taken, +the copy is brought up to this version, and then it is deleted. It needs about +as much free space on the database's disk as the database itself, while it +runs. A dump taken before counts were recorded is shown beside the live +database instead. + To rebuild after losing the server, start a fresh install (first boot as above), enter the same destination, and under **Restore from the bucket** choose a dump. The database is replaced, then every file is fetched back to its place diff --git a/backend/app/api/system_backup.py b/backend/app/api/system_backup.py index 0fad222..3590527 100644 --- a/backend/app/api/system_backup.py +++ b/backend/app/api/system_backup.py @@ -20,7 +20,7 @@ system_backup_bp = Blueprint( "system_backup", __name__, url_prefix="/api/system/backup", ) -_KINDS = frozenset({"db", "images", "media", "restore", "reconcile"}) +_KINDS = frozenset({"db", "images", "media", "restore", "reconcile", "test"}) _TAG_MAX_LEN = 64 _BACKUP_SETTINGS_FIELDS = ( "backup_db_nightly_enabled", @@ -191,10 +191,52 @@ async def list_offsite_dumps(): dumps = await asyncio.to_thread(backup_restore.list_dumps, target) except S3Error as exc: return _bad("destination_unreachable", detail=str(exc), status=502) - return jsonify({"dumps": [ - {"key": d.key, "ts": d.ts, "size_bytes": d.size_bytes, "confirm": restore_token(d.ts)} - for d in dumps - ]}) + return jsonify({ + "dumps": [ + {"key": d.key, "ts": d.ts, "size_bytes": d.size_bytes, "counted": d.counted, + "confirm": restore_token(d.ts)} + for d in dumps + ], + # A test restore needs about this much free space on the database's + # disk while it runs. + "database_bytes": await _database_bytes(), + }) + + +async def _database_bytes() -> int: + from sqlalchemy import text + + async with get_session() as session: + return (await session.execute( + text("SELECT pg_database_size(current_database())"))).scalar_one() + + +async def _restore_test_blocked() -> str | None: + from ..tasks.backup import restore_test_blocked + + async with get_session() as session: + return await session.run_sync(restore_test_blocked) + + +@system_backup_bp.route("/offsite/test", methods=["POST"]) +async def restore_test_offsite(): + """Test a dump from the bucket without touching this install: it is + loaded into a temporary database, checked table by table, and dropped. + Body: {key}, a dump's key from /offsite/dumps, or {} for the newest. + Needs no confirmation; refused while a restore or another test is going.""" + body = await request.get_json(silent=True) or {} + key = body.get("key") + if key is not None and (not isinstance(key, str) or not key.startswith("database/")): + return _bad("invalid_key", detail="key must be a dump's key from /offsite/dumps") + _, err = await _saved_target() + if err: + return err + busy = await _restore_test_blocked() + if busy: + return _bad("busy", detail=f"{busy}; try again when it has finished", status=409) + from ..tasks.backup import restore_test_task + restore_test_task.delay(dump_key=key) + return jsonify({"status": "dispatched"}), 202 @system_backup_bp.route("/offsite/restore", methods=["POST"]) @@ -335,6 +377,25 @@ async def trigger_restore(run_id: int): return jsonify({"status": "dispatched", "kind": kind}), 202 +@system_backup_bp.route("/runs//test", methods=["POST"]) +async def restore_test_local(run_id: int): + """Test a database dump kept on this server, the same way as one in the + bucket. Needs no confirmation.""" + async with get_session() as session: + row = await session.get(BackupRun, run_id) + if row is None: + return _bad("not_found", status=404) + if row.kind != "db" or row.status != "ok" or not row.sql_path: + return _bad("not_testable", + detail="only a finished database dump on this server can be tested here") + busy = await _restore_test_blocked() + if busy: + return _bad("busy", detail=f"{busy}; try again when it has finished", status=409) + from ..tasks.backup import restore_test_task + restore_test_task.delay(source_backup_run_id=run_id) + return jsonify({"status": "dispatched"}), 202 + + @system_backup_bp.route("/runs/", methods=["DELETE"]) async def delete_run(run_id: int): body = await request.get_json(silent=True) or {} diff --git a/backend/app/services/backup_restore.py b/backend/app/services/backup_restore.py index 4407aec..d909c7d 100644 --- a/backend/app/services/backup_restore.py +++ b/backend/app/services/backup_restore.py @@ -12,6 +12,7 @@ only what a damaged library is missing. """ from __future__ import annotations +import os import re import shutil import subprocess @@ -38,19 +39,23 @@ class Dump: key: str ts: str # 20261107T134258Z, UTC size_bytes: int + counted: bool = False # its row counts sit beside it, for a test restore def list_dumps(target: S3Target) -> list[Dump]: """Every database dump in the bucket, newest first.""" - dumps, token = [], None + found, keys, token = [], set(), None while True: page = list_objects(target, DUMP_PREFIX, 1000, continuation=token) for o in page.objects: + keys.add(o.key) if m := _DUMP_NAME.match(o.key): - dumps.append(Dump(o.key, m.group(1), o.size)) + found.append((o.key, m.group(1), o.size)) token = page.next_token if not token: break + dumps = [Dump(k, ts, size, counted=k.removesuffix(".dump") + ".counts.json" in keys) + for k, ts, size in found] return sorted(dumps, key=lambda d: d.ts, reverse=True) @@ -76,16 +81,18 @@ def fetch_dump(target: S3Target, dump: Dump, into: Path, pacer: Pacer | None) -> return path -def migrate() -> None: +def migrate(db_name: str | None = None) -> None: """Bring a restored database up to this install's schema. A dump taken by an older FabledCurator gains the migrations it predates; one from a newer FabledCurator fails here, with alembic's reason, because this install - doesn't know its schema.""" + doesn't know its schema. `db_name` names another database on the same + server to bring up instead of this install's own.""" + env = {**os.environ, "DB_NAME": db_name} if db_name else None with tempfile.TemporaryFile() as err: proc = subprocess.run( [sys.executable, "-m", "alembic", "-c", str(_ALEMBIC_INI), "upgrade", "head"], cwd=_ALEMBIC_INI.parent, stdout=subprocess.DEVNULL, stderr=err, - timeout=_MIGRATE_TIMEOUT_S, + timeout=_MIGRATE_TIMEOUT_S, env=env, ) if proc.returncode != 0: err.seek(0) diff --git a/backend/app/services/backup_service.py b/backend/app/services/backup_service.py index ca4dabf..5d18771 100644 --- a/backend/app/services/backup_service.py +++ b/backend/app/services/backup_service.py @@ -17,14 +17,18 @@ lifecycle + soft/hard time limits + retention bookkeeping. from __future__ import annotations import json +import logging import os import shutil import subprocess import tempfile import time +from contextlib import contextmanager from datetime import UTC, datetime from pathlib import Path +log = logging.getLogger(__name__) + _BACKUPS_DIRNAME = "_backups" # Excluded from the images tarball, and each for its own reason (#4233, #4234): @@ -65,12 +69,83 @@ _IMAGES_SUBPROCESS_TIMEOUT_S = 21000 # ~5.8 hr, under the 6-hr images soft _KILL_REAP_GRACE_S = 10 +# Tables whose rows stay out of every dump. `credential` holds account logins, +# sealed with the key in /images/secrets: a restored install keeps the table, +# empty, and you sign in to each account again. +DATA_LEFT_OUT = ("credential",) + # Every database dump, local or off-site: custom format (-Fc, compressed and # restored with pg_restore), without ownership or grants so it loads into any -# install. The `credential` table's rows (account logins, sealed with the key -# in /images/secrets) stay out: a restored install keeps the table, empty, and -# you sign in to each account again. -_PG_DUMP = ["pg_dump", "--no-owner", "--no-acl", "-Fc", "--exclude-table-data=credential"] +# install. +_PG_DUMP = ["pg_dump", "--no-owner", "--no-acl", "-Fc", + *(f"--exclude-table-data={t}" for t in DATA_LEFT_OUT)] + + +def table_counts(conn) -> dict[str, int]: + """How many rows each table in the public schema holds, read on a psycopg + connection, in the order of the tables' names.""" + from psycopg import sql + + names = [r[0] for r in conn.execute( + "SELECT tablename FROM pg_tables WHERE schemaname = 'public' ORDER BY tablename")] + return { + n: conn.execute(sql.SQL("SELECT count(*) FROM {}").format(sql.Identifier(n))).fetchone()[0] + for n in names + } + + +def database_counts(conn) -> dict: + """{"alembic": the schema's revision, "tables": {name: rows}}.""" + has_version = conn.execute("SELECT to_regclass('public.alembic_version')").fetchone()[0] + alembic = (conn.execute("SELECT version_num FROM alembic_version").fetchone() + if has_version else None) + return {"alembic": alembic[0] if alembic else None, "tables": table_counts(conn)} + + +def read_counts(db_url: str) -> dict: + """database_counts for the database at `db_url`, as it is now.""" + import psycopg + + with psycopg.connect(_libpq_url(db_url)) as conn: + return database_counts(conn) + + +class DumpSnapshot: + """A read-only transaction whose snapshot pg_dump dumps from (it is handed + over with --snapshot), so rows counted here are exactly the rows the dump + holds. A test restore checks every table against these counts.""" + + def __init__(self, conn): + self.conn = conn + self.id = conn.execute("SELECT pg_export_snapshot()").fetchone()[0] + + def counts(self) -> dict | None: + """The counts, or None when they can't be read: the dump is good + without them, and a test restore then says none were recorded.""" + try: + return self._counts() + except Exception: # noqa: BLE001 — see above + log.exception("backup: the dump's row counts couldn't be read") + return None + + def _counts(self) -> dict: + counts = database_counts(self.conn) + for t in DATA_LEFT_OUT: + if t in counts["tables"]: + counts["tables"][t] = 0 # the dump holds the table, not its rows + return counts + + +@contextmanager +def dump_snapshot(db_url: str): + """Open the snapshot a dump is taken from. It stays open until the dump + is finished: pg_dump can only take it up while it is.""" + import psycopg + + with psycopg.connect(_libpq_url(db_url)) as conn: + conn.isolation_level = psycopg.IsolationLevel.REPEATABLE_READ + conn.read_only = True + yield DumpSnapshot(conn) def _run_bounded(cmd: list[str], timeout: int) -> None: @@ -128,7 +203,7 @@ def _file_size_or_none(path: Path) -> int | None: def _write_manifest( out_dir: Path, *, kind: str, ts: str, tag: str | None, triggered_by: str, - artifact_path: Path, + artifact_path: Path, counts: dict | None = None, ) -> Path: manifest = { "kind": kind, @@ -138,6 +213,8 @@ def _write_manifest( "created_at": datetime.now(UTC).isoformat(), "artifact_path": str(artifact_path), } + if counts is not None: + manifest["counts"] = counts mf = out_dir / f"fc_{kind}_{ts}.json" mf.write_text(json.dumps(manifest, indent=2)) return mf @@ -163,17 +240,20 @@ def backup_db( os.close(fd) tmp_path = Path(tmp_name) try: - _run_bounded( - _gently([*_PG_DUMP, "-f", str(tmp_path), _libpq_url(db_url)]), - _DB_SUBPROCESS_TIMEOUT_S, - ) + with dump_snapshot(db_url) as snap: + _run_bounded( + _gently([*_PG_DUMP, f"--snapshot={snap.id}", "-f", str(tmp_path), + _libpq_url(db_url)]), + _DB_SUBPROCESS_TIMEOUT_S, + ) + counts = snap.counts() shutil.move(str(tmp_path), str(sql_path)) finally: if tmp_path.exists(): tmp_path.unlink(missing_ok=True) manifest_path = _write_manifest( out_dir, kind="db", ts=ts, tag=tag, triggered_by=triggered_by, - artifact_path=sql_path, + artifact_path=sql_path, counts=counts, ) return { "kind": "db", @@ -182,6 +262,7 @@ def backup_db( "tar_path": None, "manifest_path": str(manifest_path), "size_bytes": _file_size_or_none(sql_path), + "counts": counts, } @@ -214,15 +295,24 @@ def backup_db_offsite(*, db_url: str, target, limit_mbps: int | None = None) -> copy staged on this server. The object appears only once pg_dump has exited cleanly and every part has arrived; a dump that fails partway leaves nothing in the bucket, and raises DumpFailed with pg_dump's own - last line. Returns a dict for the caller to record in BackupRun.""" + last line. Returns a dict for the caller to record in BackupRun, with + `counts`: each table's rows in the dump, for the caller to store beside + it.""" + ts = _now_ts() + with dump_snapshot(db_url) as snap: + result = _stream_dump(db_url, target, ts, limit_mbps, snap.id) + result["counts"] = snap.counts() + return result + + +def _stream_dump(db_url: str, target, ts: str, limit_mbps: int | None, snapshot: str) -> dict: from .backup_s3 import PART_SIZE, Pacer, Upload - ts = _now_ts() key = offsite_dump_key(ts) deadline = time.monotonic() + _DB_SUBPROCESS_TIMEOUT_S with tempfile.TemporaryFile() as err: proc = subprocess.Popen( - _gently([*_PG_DUMP, _libpq_url(db_url)]), + _gently([*_PG_DUMP, f"--snapshot={snapshot}", _libpq_url(db_url)]), stdout=subprocess.PIPE, stderr=err, ) upload = None diff --git a/backend/app/services/backup_verify.py b/backend/app/services/backup_verify.py new file mode 100644 index 0000000..d75ffc9 --- /dev/null +++ b/backend/app/services/backup_verify.py @@ -0,0 +1,162 @@ +"""Testing a database dump without touching this install. + +The dump is loaded into a separate, temporary database on the same Postgres +server, using the same steps a real restore takes. Every table is counted +there and compared, the copy is brought up to this version, and the copy is +then dropped. The live database is only read, to show its counts alongside. + +Every dump records how many rows each table held, counted in the same +snapshot pg_dump dumped from, so the comparison is exact. For a dump in the +bucket the counts sit beside it as `.counts.json`; for a dump on +this server they are in its manifest file. A dump taken before counts were +recorded can still be tested: it is then compared with the live database, +where a difference is what changed after the dump was taken, not a fault. + +The temporary database needs about as much free space on the database's disk +as the live one takes, and only for the length of the test. +""" +from __future__ import annotations + +import json +import tempfile +from collections.abc import Callable +from pathlib import Path + +from sqlalchemy.engine import make_url + +from . import backup_restore, backup_service +from .backup_s3 import ObjectMissing, S3Target, get_file, put_file + +SCRATCH_SUFFIX = "_restore_test" + + +def counts_key(dump_key: str) -> str: + """Where a dump's row counts sit in the bucket, beside the dump.""" + return dump_key.removesuffix(".dump") + ".counts.json" + + +def store_counts(target: S3Target, dump_key: str, counts: dict) -> None: + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "counts.json" + path.write_text(json.dumps(counts)) + put_file(target, counts_key(dump_key), path, content_type="application/json") + + +def fetch_counts(target: S3Target, dump_key: str) -> dict | None: + """The counts recorded with a dump in the bucket, or None for a dump that + has none.""" + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "counts.json" + try: + get_file(target, counts_key(dump_key), path) + except ObjectMissing: + return None + return json.loads(path.read_text()) + + +def local_counts(manifest_path: str | None) -> dict | None: + """The counts in a dump's manifest file on this server, or None.""" + try: + return json.loads(Path(manifest_path).read_text()).get("counts") + except (TypeError, OSError, ValueError): + return None + + +def scratch_name(db_url: str) -> str: + """The temporary database's name: the live one's, plus SCRATCH_SUFFIX.""" + return f"{make_url(db_url).database}{SCRATCH_SUFFIX}" + + +def scratch_url(db_url: str) -> str: + return make_url(db_url).set(database=scratch_name(db_url)).render_as_string( + hide_password=False) + + +def _on_server(db_url: str, statement) -> None: + """Run a CREATE or DROP DATABASE, which can't run inside a transaction.""" + import psycopg + + with psycopg.connect(backup_service._libpq_url(db_url), autocommit=True) as conn: + conn.execute(statement) + + +def drop_scratch(db_url: str) -> None: + """Drop the temporary database if it is there, ending any connection to it.""" + from psycopg import sql + + _on_server(db_url, sql.SQL("DROP DATABASE IF EXISTS {} WITH (FORCE)").format( + sql.Identifier(scratch_name(db_url)))) + + +def _create_scratch(db_url: str) -> None: + from psycopg import sql + + _on_server(db_url, sql.SQL("CREATE DATABASE {} TEMPLATE template0").format( + sql.Identifier(scratch_name(db_url)))) + + +def trial_restore(*, db_url: str, dump_path: Path, recorded: dict | None, + note: Callable[[str], None] = lambda phase: None) -> dict: + """Load the dump at `dump_path` into the temporary database, compare it, + bring it up to this version, and drop it. `note` is told each phase as it + starts: "loading", "comparing", "upgrading". Raises RestoreFailed (or + S3Error from the upgrade) when the dump doesn't load or can't be brought + up; the temporary database is dropped either way.""" + scratch = scratch_url(db_url) + if make_url(scratch).database == make_url(db_url).database: + raise ValueError("The test database would be the live one.") + drop_scratch(db_url) # one left by a test cut off by a restart + _create_scratch(db_url) + try: + note("loading") + backup_service.restore_db(db_url=scratch, sql_path=dump_path) + note("comparing") + restored = backup_service.read_counts(scratch) + live = backup_service.read_counts(db_url) + note("upgrading") + backup_restore.migrate(db_name=scratch_name(db_url)) + upgraded = backup_service.read_counts(scratch)["alembic"] + finally: + drop_scratch(db_url) + return {**compare(recorded, restored, live), + "alembic_dump": restored["alembic"], "alembic_now": upgraded} + + +def compare(recorded: dict | None, restored: dict, live: dict) -> dict: + """Each table's rows in the dump (as recorded), in the restored copy, and + in the live database. With counts recorded, every table must come back + with exactly its count; `mismatched` names those that didn't.""" + rec = (recorded or {}).get("tables") + names = sorted(set(restored["tables"]) | set(rec or {}) | set(live["tables"])) + tables = [ + {"table": n, "dump": rec.get(n) if rec is not None else None, + "restored": restored["tables"].get(n), "live": live["tables"].get(n)} + for n in names + ] + mismatched = [t["table"] for t in tables + if rec is not None and t["table"] in rec and t["restored"] != t["dump"]] + return { + "counted": rec is not None, + "tables": tables, + "mismatched": mismatched, + "rows_restored": sum(v for v in restored["tables"].values()), + } + + +def summary(result: dict) -> str: + """One sentence on how a test went.""" + n = len([t for t in result["tables"] if t["restored"] is not None]) + rows = f"{result['rows_restored']:,}" + if result["mismatched"]: + shown = ", ".join( + f"{t['table']} ({t['dump']:,} in the dump, " + + ("table missing)" if t["restored"] is None else f"{t['restored']:,} restored)") + for t in result["tables"] if t["table"] in result["mismatched"][:5] + ) + more = len(result["mismatched"]) - 5 + return (f"{len(result['mismatched'])} table(s) came back different: {shown}" + + (f", and {more} more" if more > 0 else "") + ".") + if result["counted"]: + return f"All {n} tables came back with the {rows} rows the dump recorded." + return (f"The dump loaded cleanly: {n} tables, {rows} rows. It was taken before " + "dumps recorded their counts, so it is shown beside the live database instead.") diff --git a/backend/app/tasks/backup.py b/backend/app/tasks/backup.py index 0e42aea..5a2561f 100644 --- a/backend/app/tasks/backup.py +++ b/backend/app/tasks/backup.py @@ -26,6 +26,7 @@ from ..services import ( backup_restore, backup_s3, backup_service, + backup_verify, ) from ._sync_engine import get_sync_engine from ._sync_engine import sync_session_factory as _sync_session_factory @@ -86,9 +87,7 @@ def backup_db_task(self, *, tag: str | None = None, target = backup_destination.saved_target(settings) limit_mbps = settings.backup_upload_limit_mbps if target is not None: - result = backup_service.backup_db_offsite( - db_url=cfg.database_url_sync, target=target, limit_mbps=limit_mbps, - ) + result = _dump_offsite(cfg, target, limit_mbps) else: result = backup_service.backup_db( db_url=cfg.database_url_sync, images_root=IMAGES_ROOT, @@ -280,10 +279,10 @@ def prune_backups() -> dict: """Daily Beat. Per-kind retention from ImportSettings. Returns {"db_deleted": N, "images_deleted": M, "media_deleted": L, - "restore_deleted": R, "reconcile_deleted": C, "files_unlinked": K}. Media, - restore and reconcile runs leave no files here; only their history is - trimmed, to the latest MEDIA_RUNS_KEPT, RESTORE_RUNS_KEPT and - RECONCILE_RUNS_KEPT. + "restore_deleted": R, "reconcile_deleted": C, "test_deleted": T, + "files_unlinked": K}. Media, restore, reconcile and test runs leave no + files here; only their history is trimmed, to the latest MEDIA_RUNS_KEPT, + RESTORE_RUNS_KEPT, RECONCILE_RUNS_KEPT and TEST_RUNS_KEPT. Tagged rows (tag IS NOT NULL) are never pruned. Status='running' / 'restoring' rows are never pruned — the recover_stalled_backup_runs sweep flips truly-stuck ones to @@ -293,7 +292,8 @@ def prune_backups() -> dict: """ SessionLocal = _sync_session_factory() counts = {"db_deleted": 0, "images_deleted": 0, "media_deleted": 0, - "restore_deleted": 0, "reconcile_deleted": 0, "files_unlinked": 0} + "restore_deleted": 0, "reconcile_deleted": 0, "test_deleted": 0, + "files_unlinked": 0} with SessionLocal() as session: s = ImportSettings.load_sync(session) for kind, keep in ( @@ -302,6 +302,7 @@ def prune_backups() -> dict: ("media", MEDIA_RUNS_KEPT), ("restore", RESTORE_RUNS_KEPT), ("reconcile", RECONCILE_RUNS_KEPT), + ("test", TEST_RUNS_KEPT), ): candidates = session.execute( select(BackupRun) @@ -371,6 +372,21 @@ MEDIA_FAILURES_LISTED = 20 RECONCILE_RUNS_KEPT = 12 +def _dump_offsite(cfg, target, limit_mbps) -> dict: + """Dump the database to the bucket, with its row counts beside it for a + test restore to check against. The dump stands without its counts, so a + failure to store them is logged, not raised.""" + result = backup_service.backup_db_offsite( + db_url=cfg.database_url_sync, target=target, limit_mbps=limit_mbps) + if result.get("counts"): + try: + backup_verify.store_counts(target, result["key"], result["counts"]) + except (backup_s3.S3Error, OSError): + log.warning("backup: the row counts for %s weren't stored", result["key"], + exc_info=True) + return result + + def _offsite_manifest(target) -> dict: return {"endpoint": target.endpoint, "bucket": target.bucket, "key": backup_media.PREFIX} @@ -439,6 +455,11 @@ def resume_offsite_runs() -> dict: ).scalars() if (r.manifest or {}).get("phase") == "files" ] + try: + tests = end_cut_off_tests() + except Exception: # noqa: BLE001 — resuming the runs below matters more + log.exception("couldn't end a test restore cut off by the restart") + tests = 0 for run_id in media: backup_media_task.delay(run_id=run_id, resume=True) for run_id in restores: @@ -446,7 +467,7 @@ def resume_offsite_runs() -> dict: if media or restores: log.info("resumed off-site runs after a restart: media %s, restore %s", media, restores) - return {"media": media, "restore": restores} + return {"media": media, "restore": restores, "tests_ended": tests} @celery.task( @@ -647,7 +668,11 @@ def backup_busy(session) -> str | None: ).first() if row is None: return None - return "a restore is going" if row.kind == "restore" else f"a {row.kind} backup is running" + if row.kind == "restore": + return "a restore is going" + if row.kind == "test": + return "a test restore is running" + return f"a {row.kind} backup is running" def restore_going(session) -> bool: @@ -657,6 +682,134 @@ def restore_going(session) -> bool: ).first() is not None +# ── Test restore ───────────────────────────────────────────────────────────── + +TEST_RUNS_KEPT = 20 + + +@celery.task( + name="backend.app.tasks.backup.restore_test_task", + bind=True, + max_retries=0, + # Downloading a dump, loading it and rebuilding its indexes; a restore of + # the database alone, with room to spare. + soft_time_limit=4 * 3600, time_limit=4 * 3600 + 900, +) +def restore_test_task(self, *, dump_key: str | None = None, + source_backup_run_id: int | None = None) -> dict: + """Test a database dump without touching this install: load it into a + temporary database, check every table against the counts recorded with + the dump, bring it up to this version, and drop it again + (services/backup_verify.py). Tests the bucket's dump at `dump_key` (the + newest when None), or, given `source_backup_run_id`, a dump on this + server. The run ends "ok" when every table matched, "error" with what + didn't otherwise. Returns {'backup_run_id': N, ...}.""" + SessionLocal = _sync_session_factory() + cfg = get_config() + now = datetime.now(UTC) + with SessionLocal() as session: + if why := restore_test_blocked(session): + return {"skipped": why} + row = BackupRun(kind="test", status="running", triggered_by="manual", + started_at=now, last_progress_at=now, manifest={"phase": "fetching"}) + session.add(row) + session.commit() + run_id = row.id + + def note(phase: str, **more) -> None: + with SessionLocal() as session: + r = session.get(BackupRun, run_id) + _note(session, r, {**r.manifest, "phase": phase, **more}) + + staged = offsite = None + try: + if source_backup_run_id is not None: + dump_path, recorded, source = _local_dump(SessionLocal, source_backup_run_id) + else: + with SessionLocal() as session: + settings = ImportSettings.load_sync(session) + target = backup_destination.saved_target(settings) + limit_mbps = settings.backup_upload_limit_mbps + if target is None: + raise backup_s3.S3Error("No off-site destination is set up.") + dump = backup_restore.find_dump(target, dump_key) + source = {"key": dump.key, "ts": dump.ts, "size_bytes": dump.size_bytes} + offsite = {**_offsite_manifest(target), "key": dump.key} + note("fetching", dump=source, offsite=offsite) + recorded = backup_verify.fetch_counts(target, dump.key) + staged = backup_restore.fetch_dump( + target, dump, IMAGES_ROOT / "_backups", backup_s3.Pacer(limit_mbps)) + dump_path = staged + note("loading", dump=source) + result = backup_verify.trial_restore( + db_url=cfg.database_url_sync, dump_path=dump_path, recorded=recorded, + note=note) + except (SoftTimeLimitExceeded, Exception) as exc: + with SessionLocal() as session: + r = session.get(BackupRun, run_id) + if r is not None: + _mark_failed(session, r, exc) + raise + finally: + if staged is not None: + staged.unlink(missing_ok=True) + + with SessionLocal() as session: + r = session.get(BackupRun, run_id) + r.status = "error" if result["mismatched"] else "ok" + r.error = backup_verify.summary(result) if result["mismatched"] else None + r.finished_at = datetime.now(UTC) + r.size_bytes = source.get("size_bytes") + r.manifest = {"phase": "done", "dump": source, "offsite": offsite, + "summary": backup_verify.summary(result), **result} + session.commit() + return {"backup_run_id": run_id, "mismatched": result["mismatched"]} + + +def _local_dump(SessionLocal, run_id: int) -> tuple[Path, dict | None, dict]: + """A dump on this server: its path, its recorded counts, and what the + test run says it tested.""" + with SessionLocal() as session: + src = session.get(BackupRun, run_id) + if src is None or src.kind != "db" or src.status != "ok" or not src.sql_path: + raise backup_s3.S3Error(f"Backup {run_id} isn't a database dump on this server.") + path = Path(src.sql_path) + if not path.is_file(): + raise backup_s3.S3Error(f"{path} is no longer on this server.") + recorded = backup_verify.local_counts((src.manifest or {}).get("manifest_path")) + ts = (src.manifest or {}).get("ts") + return path, recorded, {"run_id": run_id, "ts": ts, "size_bytes": src.size_bytes} + + +def restore_test_blocked(session) -> str | None: + """Why a test restore can't start right now, or None. A backup going + doesn't stop one: the test only reads this install's database.""" + if restore_going(session): + return "a restore is going" + going = session.execute( + select(BackupRun.id).where(BackupRun.kind == "test") + .where(BackupRun.status == "running").limit(1) + ).first() + return "a test restore is already running" if going else None + + +def end_cut_off_tests() -> int: + """End every test restore a restart cut short, and drop the temporary + database it left. Called when the long-maintenance lane starts, where the + tests run, so none is alive then.""" + with _sync_session_factory()() as session: + ended = session.execute( + update(BackupRun).where(BackupRun.kind == "test") + .where(BackupRun.status == "running") + .values(status="error", finished_at=datetime.now(UTC), + error="Interrupted: the server restarted during the test. Test again.") + ).rowcount or 0 + session.commit() + if ended: + backup_verify.drop_scratch(get_config().database_url_sync) + return ended + + def _library_is_empty(session) -> bool: return session.execute(text( "SELECT NOT EXISTS (SELECT 1 FROM image_record) " @@ -749,8 +902,7 @@ def _restore_database(SessionLocal, run_id: int, target, limit_mbps, dump_key) - # dump, and would sit in the list as the newest. if not _library_is_empty(session): session.commit() - saved = backup_service.backup_db_offsite( - db_url=cfg.database_url_sync, target=target, limit_mbps=limit_mbps) + saved = _dump_offsite(cfg, target, limit_mbps) m["saved_before"] = saved["key"] _note(session, row, m) started_at, manifest = row.started_at, dict(m) diff --git a/frontend/src/components/settings/BackupCard.vue b/frontend/src/components/settings/BackupCard.vue index c88eea5..495523e 100644 --- a/frontend/src/components/settings/BackupCard.vue +++ b/frontend/src/components/settings/BackupCard.vue @@ -78,10 +78,15 @@ @restore="onRestore" @delete="onDelete" @tag="onTag" + @test="(r) => testSection?.start({ runId: r.id })" /> + + + +