feat: test a database backup without touching the install

"Test a backup" on the Backups card loads a dump into a separate, temporary
database on the same Postgres server, the same way a restore does, checks every
table against the row counts recorded with the dump, brings the copy up to this
version, and drops it. The live database is only read.

Every dump now records its row counts, read in the snapshot pg_dump dumps from
(pg_export_snapshot + --snapshot), so the comparison is exact. Off-site the
counts sit beside the dump as <name>.counts.json; locally they go in the
manifest file. A dump taken before this is compared with the live database
instead.

A test cut off by a restart is ended, and its database dropped, when the long
maintenance lane starts. A test doesn't block backups; a restore and a second
test wait for it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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/<int:run_id>/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/<int:run_id>", methods=["DELETE"])
 async def delete_run(run_id: int):
     body = await request.get_json(silent=True) or {}

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)

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

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 `<dump name>.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.")

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)