feat: restore a new install from the off-site bucket, database first, then every file

Settings → Maintenance → Backups gains "Restore from the bucket": pick one of
the bucket's dated dumps, type its confirmation, and FabledCurator saves the
current database to the bucket, replaces it with the dump, migrates it to this
version, then fetches every image and attachment back to the path its row
records. Each file is checked against its sha256 on arrival and moved into
place only then; a file already on disk is left alone, so "Fetch missing
files" repairs a library that lost files without touching the database.

- services/backup_restore.py: list/find/fetch dumps, migrate, keep this
  install's destination (the dump's sealed secret can't be opened here),
  library_files work list, restore_file (one fetch per sha, copies for
  duplicate paths, refuses paths outside /images).
- restore_offsite_task: the database phase, then 45-minute file chunks with a
  heartbeat; a failed load leaves an empty migrated database and says why.
  The dump's own run row is marked ok, other runs then going "Interrupted".
- backup_s3.get_file: sha/size check on arrival, ObjectMissing for a 404,
  download pacing.
- backup_service: every dump leaves the credential table's rows out;
  restore_db reports pg_restore's last line, stops at the first error, ends
  other connections and waits at most 60 s for locks.
- Backups and orphan repair wait while a restore runs; the stall sweep judges
  restores by heartbeat; restore runs are pruned to 20.
- CLI: python -m backend.app.scripts.restore list|database|files.
- README: backup and restore section; logins are in no backup.
- Tests: a round trip through an in-process S3 bucket (tests/fake_s3.py) with
  real pg_dump/pg_restore, plus file, run, API and get_file cases.

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

.env.example

@@ -56,9 +56,8 @@ SECRET_KEY=
 # LINE. Leaving it set disarms the protection permanently, on an instance that
 # by then has credentials worth protecting.
 #
-# BACK UP ./images/secrets/ ALONGSIDE YOUR DATABASE. The key is the only thing
+# Backups leave the key and the stored logins out on purpose: after restoring
-# that can read your stored credentials; a database restored without it needs
+# onto a new install, sign in to each account again.
-# every credential re-entered by hand.
 CURATOR_BOOTSTRAP_NEW_KEY=1
 
 

README.md

@@ -114,9 +114,10 @@ that cannot decrypt anything, leaving an instance that looks healthy while every
 paywalled download fails. Making you say so once, on an empty install, is the
 price of that not happening silently later.
 
-**Which means: back up `./images/secrets/` alongside your database.** It is the
+**Your account logins are sealed with that key, and no backup carries either.**
-only thing that can read your stored credentials. A database restored without it
+FabledCurator's backups (below) leave the key and the logins out on purpose, so
-needs every credential entered again by hand.
+a stolen backup holds no way into your accounts. After a restore, sign in to
+each account again through the browser extension.
 
 A few other things are worth knowing about the first few minutes:
 
@@ -177,6 +178,37 @@ Run it for a burst of tagging, stop it to get your card back. It deploys from
 `agent/docker-compose.yml`, not the main stack — see
 [agent/README.md](agent/README.md).
 
+## Backing up and restoring
+
+Backups live under **Settings → Maintenance → Backups**. With nothing else set
+up, the database is dumped nightly into `./images/_backups`, which is on the
+same disk as what it protects.
+
+To keep a copy off the machine, set up an off-site destination on the same
+screen: any S3 bucket (Fabled Archives, MinIO, AWS) and a key that may add and
+read objects. Each night FabledCurator then sends the database as a dated dump
+under `database/`, and every image and attachment once, named by its
+fingerprint, under `media/`. It never deletes from the bucket; the bucket's own
+retention thins old dumps.
+
+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
+and checked against its fingerprint as it arrives; progress shows on the same
+screen. **Fetch missing files** does only the second half, for a library that
+lost files but kept its database.
+
+If the web page isn't reachable, the same restore runs from a shell:
+
+```bash
+docker compose exec web python -m backend.app.scripts.restore list
+docker compose exec web python -m backend.app.scripts.restore database --dump latest --yes
+```
+
+On an install with no destination saved yet, add `--endpoint`, `--bucket` and
+`--access-key`, and put the secret key in `FC_BACKUP_SECRET_KEY` (or type it at
+the prompt).
+
 ## Upgrading
 
 ```bash

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"})
+_KINDS = frozenset({"db", "images", "media", "restore"})
 _TAG_MAX_LEN = 64
 _BACKUP_SETTINGS_FIELDS = (
     "backup_db_nightly_enabled",
@@ -142,6 +142,102 @@ async def media_coverage():
     return jsonify({"configured": configured, **counts})
 
 
+# ── Restoring from the off-site bucket ──────────────────────────────────────
+
+
+async def _saved_target():
+    """(target, None) for the saved destination, or (None, error response)."""
+    from ..services import backup_destination as dest
+    from ..services.backup_s3 import S3Error
+
+    async with get_session() as session:
+        row = await ImportSettings.load(session)
+        try:
+            target = dest.saved_target(row)
+        except S3Error as exc:
+            return None, _bad("destination_unreadable", detail=str(exc))
+    if target is None:
+        return None, _bad("no_destination",
+                          detail="set up an off-site destination before restoring from it")
+    return target, None
+
+
+async def _busy_reason() -> str | None:
+    from ..tasks.backup import backup_busy
+
+    async with get_session() as session:
+        return await session.run_sync(backup_busy)
+
+
+def restore_token(ts: str) -> str:
+    """What the person types to confirm replacing the database with the dump
+    taken at `ts`."""
+    return f"restore-offsite-{ts}"
+
+
+@system_backup_bp.route("/offsite/dumps", methods=["GET"])
+async def list_offsite_dumps():
+    """The database dumps in the off-site bucket, newest first, each with
+    the confirmation a restore of it asks for."""
+    import asyncio
+
+    from ..services import backup_restore
+    from ..services.backup_s3 import S3Error
+
+    target, err = await _saved_target()
+    if err:
+        return err
+    try:
+        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
+    ]})
+
+
+@system_backup_bp.route("/offsite/restore", methods=["POST"])
+async def restore_from_offsite():
+    """Replace this install's database with a dump from the bucket, then
+    fetch the library's files back. Body: {key, confirm}, where confirm is
+    the dump's token from /offsite/dumps. Refused while a backup or another
+    restore is going."""
+    body = await request.get_json(silent=True) or {}
+    key = body.get("key")
+    if not isinstance(key, str) or not key.startswith("database/"):
+        return _bad("invalid_key", detail="key must be a dump's key from /offsite/dumps")
+    ts = key.removeprefix("database/fabledcurator-").removesuffix(".dump")
+    expected = restore_token(ts)
+    if body.get("confirm") != expected:
+        return _bad("confirm_mismatch", detail=f"confirm must equal {expected!r}",
+                    expected=expected)
+    _, err = await _saved_target()
+    if err:
+        return err
+    busy = await _busy_reason()
+    if busy:
+        return _bad("busy", detail=f"{busy}; try again when it has finished", status=409)
+    from ..tasks.backup import restore_offsite_task
+    restore_offsite_task.delay(dump_key=key)
+    return jsonify({"status": "dispatched"}), 202
+
+
+@system_backup_bp.route("/offsite/fetch-files", methods=["POST"])
+async def fetch_missing_files():
+    """Fetch every library file that is missing from disk back from the
+    bucket, leaving the database as it is."""
+    _, err = await _saved_target()
+    if err:
+        return err
+    busy = await _busy_reason()
+    if busy:
+        return _bad("busy", detail=f"{busy}; try again when it has finished", status=409)
+    from ..tasks.backup import restore_offsite_task
+    restore_offsite_task.delay(files_only=True)
+    return jsonify({"status": "dispatched"}), 202
+
+
 @system_backup_bp.route("/runs", methods=["GET"])
 async def list_runs():
     try:

backend/app/scripts/restore.py

@@ -0,0 +1,119 @@
+"""Restore FabledCurator from its off-site bucket, from a shell in the
+container. The Backup screen does the same; this is for an install whose page
+you can't reach yet, or for scripting a rebuild.
+
+    python -m backend.app.scripts.restore list
+    python -m backend.app.scripts.restore database --dump latest --yes
+    python -m backend.app.scripts.restore files
+
+On a brand-new install with no destination saved, name it on the command
+line: --endpoint, --bucket and --access-key, with the secret key in the
+FC_BACKUP_SECRET_KEY environment variable (or typed at the prompt). It is
+checked by listing the bucket, then saved as this install's destination,
+sealed, exactly as the settings page saves it.
+
+`database` replaces the database with the chosen dump here, then hands the
+files to the worker; their progress shows on the Backup screen. `files`
+fetches only what is missing from disk and leaves the database alone.
+"""
+from __future__ import annotations
+
+import argparse
+import getpass
+import os
+import sys
+
+from ..models import ImportSettings
+from ..services import backup_destination, backup_restore
+from ..services.backup_s3 import S3Error
+from ..tasks._sync_engine import sync_session_factory
+
+
+def _human(n: int) -> str:
+    for unit in ("B", "KB", "MB", "GB", "TB"):
+        if n < 1000 or unit == "TB":
+            return f"{n:.0f} {unit}" if unit == "B" else f"{n:.1f} {unit}"
+        n /= 1000
+    return str(n)
+
+
+def _destination(args) -> object:
+    """The target to restore from: the one named on the command line, saved
+    first, or the one already saved."""
+    session_factory = sync_session_factory()
+    if args.endpoint or args.bucket or args.access_key:
+        fields, why = backup_destination.clean_fields({
+            "endpoint": args.endpoint, "bucket": args.bucket,
+            "access_key": args.access_key, "region": args.region,
+        })
+        if fields is None:
+            raise SystemExit(f"Destination not usable: {why}.")
+        secret = os.environ.get("FC_BACKUP_SECRET_KEY") or getpass.getpass("Secret key: ")
+        target = backup_destination.target_for(fields, secret.strip() or None)
+        if target is None:
+            raise SystemExit("Name the address, bucket, access key and secret key.")
+        check = backup_destination.check_target(target)
+        print(check["message"])
+        if not check["ok"]:
+            raise SystemExit(1)
+        with session_factory() as session:
+            settings = ImportSettings.load_sync(session)
+            limit = settings.backup_upload_limit_mbps
+            backup_restore.keep_destination(session, target, limit)
+        print("Saved as this install's off-site destination.")
+        return target
+    with session_factory() as session:
+        target = backup_destination.saved_target(ImportSettings.load_sync(session))
+    if target is None:
+        raise SystemExit("No off-site destination is saved; name one with "
+                         "--endpoint, --bucket and --access-key.")
+    return target
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(prog="python -m backend.app.scripts.restore",
+                                     description=__doc__.split("\n\n")[0])
+    parser.add_argument("action", choices=("list", "database", "files"))
+    parser.add_argument("--dump", default="latest",
+                        help="the dump's key from `list`, or latest (the default)")
+    parser.add_argument("--yes", action="store_true",
+                        help="confirm replacing this install's database")
+    parser.add_argument("--endpoint")
+    parser.add_argument("--bucket")
+    parser.add_argument("--access-key")
+    parser.add_argument("--region", default=None)
+    args = parser.parse_args(argv)
+
+    try:
+        target = _destination(args)
+        if args.action == "list":
+            for d in backup_restore.list_dumps(target):
+                print(f"{d.key}  {_human(d.size_bytes)}")
+            return 0
+
+        from ..tasks.backup import restore_offsite_task
+        if args.action == "files":
+            restore_offsite_task.delay(files_only=True)
+            print("Queued: the worker fetches every missing file; watch the Backup screen.")
+            return 0
+
+        dump = backup_restore.find_dump(target, args.dump)
+        if not args.yes:
+            print(f"This replaces this install's database with {dump.key} "
+                  f"({_human(dump.size_bytes)}). Run again with --yes to go ahead.")
+            return 2
+        print(f"Restoring {dump.key}; this takes a while for a large library…")
+        result = restore_offsite_task(dump_key=dump.key)
+        if result.get("skipped"):
+            print(f"Not started: {result['skipped']}.")
+            return 1
+        print("The database is restored. The worker is fetching the library's files; "
+              "watch the Backup screen.")
+        return 0
+    except S3Error as exc:
+        print(f"Couldn't restore: {exc}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

backend/app/services/backup_restore.py

@@ -0,0 +1,191 @@
+"""Restoring from the off-site bucket: the database from one of its dated
+dumps, then every library file the restored rows name, fetched back to the
+path each row records.
+
+The database comes first because it is the map: each image and attachment row
+holds its file's path and sha256, which is also the file's name in the bucket
+(`media/ab/cd/<sha256><ext>`). A file is checked against that sha256 on
+arrival and only then moved into place, so a restore never leaves a wrong or
+half-written file in the library. A file already on disk at the right size is
+left alone, which makes a restore safe to run again and lets it bring back
+only what a damaged library is missing.
+"""
+from __future__ import annotations
+
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+
+from sqlalchemy import select, text
+
+from ..models import BackupBlob, ImportSettings
+from . import backup_destination
+from .backup_media import _LIBRARY, blob_key
+from .backup_s3 import Pacer, S3Error, S3Target, get_file, list_objects
+
+DUMP_PREFIX = "database/"
+_DUMP_NAME = re.compile(r"^database/fabledcurator-(\d{8}T\d{6}Z)\.dump$")
+_ALEMBIC_INI = Path(__file__).resolve().parents[3] / "alembic.ini"
+_MIGRATE_TIMEOUT_S = 1800
+
+
+@dataclass(frozen=True)
+class Dump:
+    key: str
+    ts: str           # 20261107T134258Z, UTC
+    size_bytes: int
+
+
+def list_dumps(target: S3Target) -> list[Dump]:
+    """Every database dump in the bucket, newest first."""
+    dumps, token = [], None
+    while True:
+        page = list_objects(target, DUMP_PREFIX, 1000, continuation=token)
+        for o in page.objects:
+            if m := _DUMP_NAME.match(o.key):
+                dumps.append(Dump(o.key, m.group(1), o.size))
+        token = page.next_token
+        if not token:
+            break
+    return sorted(dumps, key=lambda d: d.ts, reverse=True)
+
+
+def find_dump(target: S3Target, key: str | None) -> Dump:
+    """The dump at `key`, or the newest when `key` is None or "latest"."""
+    dumps = list_dumps(target)
+    if not dumps:
+        raise S3Error(f"The {target.bucket} bucket holds no database dumps.")
+    if key in (None, "", "latest"):
+        return dumps[0]
+    for d in dumps:
+        if d.key == key:
+            return d
+    raise S3Error(f"{key} is not among the dumps in the {target.bucket} bucket.")
+
+
+def fetch_dump(target: S3Target, dump: Dump, into: Path, pacer: Pacer | None) -> Path:
+    """Download a dump into the folder `into`, checked against its listed
+    size. pg_restore reads it from there."""
+    into.mkdir(parents=True, exist_ok=True)
+    path = into / Path(dump.key).name
+    get_file(target, dump.key, path, size=dump.size_bytes, pacer=pacer)
+    return path
+
+
+def migrate() -> 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."""
+    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,
+        )
+        if proc.returncode != 0:
+            err.seek(0)
+            lines = err.read().decode(errors="replace").strip().splitlines()
+            raise S3Error("The restored database couldn't be brought up to this "
+                          f"version: {lines[-1] if lines else 'alembic failed'}")
+
+
+def keep_destination(session, target: S3Target, limit_mbps: int | None) -> None:
+    """Write this install's destination into the restored settings. The
+    dump's own copy of the secret is sealed with the old server's credential
+    key, which never leaves that server, so it can't be opened here."""
+    settings = ImportSettings.load_sync(session)
+    settings.backup_s3_endpoint = target.endpoint
+    settings.backup_s3_bucket = target.bucket
+    settings.backup_s3_access_key = target.access_key
+    settings.backup_s3_region = target.region
+    settings.backup_s3_secret_sealed = backup_destination.seal(target.secret_key)
+    settings.backup_upload_limit_mbps = limit_mbps
+    session.commit()
+
+
+# ── files ───────────────────────────────────────────────────────────────────
+
+
+@dataclass(frozen=True)
+class LibraryFile:
+    sha256: str
+    paths: tuple[str, ...]     # every row's path for this sha, sorted
+    size_bytes: int
+
+
+def library_files(session, *, after: str = "", limit: int = 200) -> list[LibraryFile]:
+    """The next `limit` distinct files the library names, in sha order after
+    `after`, each with every path a row gives it."""
+    rows = session.execute(text(f"""
+        SELECT lib.sha256, array_agg(DISTINCT lib.path ORDER BY lib.path) AS paths,
+               max(lib.size_bytes) AS size_bytes
+        FROM ({_LIBRARY}) AS lib
+        WHERE lib.sha256 > :after
+        GROUP BY lib.sha256
+        ORDER BY lib.sha256
+        LIMIT :limit
+    """), {"after": after, "limit": limit}).all()
+    return [LibraryFile(r.sha256, tuple(r.paths), r.size_bytes) for r in rows]
+
+
+def sent_keys(session, tid: str, shas: list[str]) -> dict[str, str]:
+    """The bucket key recorded for each sha that was sent to this destination."""
+    if not shas:
+        return {}
+    return dict(session.execute(
+        select(BackupBlob.sha256, BackupBlob.key)
+        .where(BackupBlob.target == tid, BackupBlob.sha256.in_(shas))
+    ).all())
+
+
+def _resolve(path: str, images_root: Path) -> Path:
+    """Where a row's path lands on this server. A path outside the library
+    folder is refused: a restore writes only inside it."""
+    p = Path(path)
+    if not p.is_absolute():
+        p = images_root / p
+    root = Path(*images_root.parts)
+    if ".." in p.parts or not (p == root or root in p.parents):
+        raise S3Error(f"{path} is outside {images_root}, so it isn't restored there.")
+    return p
+
+
+def _present(p: Path, size: int) -> bool:
+    try:
+        return p.is_file() and p.stat().st_size == size
+    except OSError:
+        return False
+
+
+def restore_file(target: S3Target, f: LibraryFile, key: str | None,
+                 images_root: Path, pacer: Pacer | None) -> str:
+    """Put one file back at every path the library gives it. Returns
+    "present" (nothing was missing), "copied" (a path already on disk
+    supplied the others) or "restored" (fetched from the bucket). Raises
+    ObjectMissing when the bucket doesn't hold it, S3Error or OSError
+    otherwise."""
+    places = [_resolve(p, images_root) for p in f.paths]
+    have = [p for p in places if _present(p, f.size_bytes)]
+    need = [p for p in places if p not in have]
+    if not need:
+        return "present"
+    outcome = "copied"
+    if not have:
+        first = need.pop(0)
+        first.parent.mkdir(parents=True, exist_ok=True)
+        get_file(target, key or blob_key(f.sha256, f.paths[0]), first,
+                 sha256=f.sha256, size=f.size_bytes, pacer=pacer)
+        have.append(first)
+        outcome = "restored"
+    for p in need:
+        p.parent.mkdir(parents=True, exist_ok=True)
+        partial = p.with_name(p.name + ".partial")
+        shutil.copyfile(have[0], partial)
+        partial.replace(p)
+    return outcome
+

backend/app/services/backup_s3.py

@@ -220,29 +220,55 @@ def put_file(target: S3Target, key: str, path: Path, *,
                       f"{_says(response.status, response.reason, content)}")
 
 
+class ObjectMissing(S3Error):
+    """The bucket holds nothing at the key asked for."""
+
+
 def get_file(target: S3Target, key: str, path: Path, *,
+             sha256: str | None = None, size: int | None = None,
+             pacer: Pacer | None = None,
              now: datetime | None = None) -> None:
     """Download one object to `path`, streamed to disk. The bytes land under a
     `.partial` name and are renamed into place when complete, so a broken
-    transfer never leaves a truncated file at `path`."""
+    transfer never leaves a truncated file at `path`. With `sha256` (and
+    `size`) given, the bytes are checked on arrival and a copy that doesn't
+    match is thrown away and reported. Raises ObjectMissing when the bucket
+    has no such object."""
     url = target.url(key)
     headers = sign("GET", url, {}, EMPTY_SHA256, target.access_key, target.secret_key,
                    target.region, now or datetime.now(UTC))
     partial = path.with_name(path.name + ".partial")
+    digest = hashlib.sha256()
+    got = 0
     conn = _connect(target, url, TRANSFER_TIMEOUT_SECONDS)
     try:
         conn.request("GET", _request_target(url), headers=headers)
         response = conn.getresponse()
+        if response.status == 404:
+            response.read()
+            raise ObjectMissing(f"{key} is not in the {target.bucket} bucket.")
         if response.status != 200:
             raise S3Error(f"Reading {key} refused: "
                           f"{_says(response.status, response.reason, response.read())}")
         with open(partial, "wb") as f:
             while chunk := response.read(CHUNK):
+                if pacer:
+                    pacer.take(len(chunk))
+                digest.update(chunk)
+                got += len(chunk)
                 f.write(chunk)
+        if size is not None and got != size:
+            raise S3Error(f"{key} arrived as {got} bytes, not the {size} expected.")
+        if sha256 and digest.hexdigest() != sha256:
+            raise S3Error(f"{key} arrived with a different fingerprint "
+                          f"({digest.hexdigest()[:12]}…), so it wasn't kept.")
         os.replace(partial, path)
     except (OSError, http.client.HTTPException) as exc:
         partial.unlink(missing_ok=True)
         raise _reach_error(target, exc) from None
+    except S3Error:
+        partial.unlink(missing_ok=True)
+        raise
     finally:
         conn.close()
 
@@ -304,15 +330,17 @@ _MAX_RETRY_AFTER = 300
 
 
 class Pacer:
-    """Holds the bytes a job sends to an average rate, across every request
+    """Holds the bytes a job moves to an average rate, across every request
-    it makes, so a backup leaves room on the link for everything else."""
+    it makes, so a backup or a restore leaves room on the link for everything
+    else."""
 
     def __init__(self, mbps: int | None):
         self.rate = mbps * 1_000_000 / 8 if mbps else None   # bytes per second
         self.sent = 0
         self.started = time.monotonic()
 
-    def _wait(self, n: int) -> None:
+    def take(self, n: int) -> None:
+        """Count `n` bytes moved, pausing while the job is ahead of its rate."""
         if self.rate:
             self.sent += n
             ahead = self.sent / self.rate - (time.monotonic() - self.started)
@@ -323,12 +351,12 @@ class Pacer:
         view = memoryview(data)
         for at in range(0, len(view), CHUNK):
             piece = view[at:at + CHUNK]
-            self._wait(len(piece))
+            self.take(len(piece))
             yield piece
 
     def file_chunks(self, f) -> Iterator[bytes]:
         while piece := f.read(CHUNK):
-            self._wait(len(piece))
+            self.take(len(piece))
             yield piece
 
 

backend/app/services/backup_service.py

@@ -65,6 +65,14 @@ _IMAGES_SUBPROCESS_TIMEOUT_S = 21000       # ~5.8 hr, under the 6-hr images soft
 _KILL_REAP_GRACE_S = 10
 
 
+# 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"]
+
+
 def _run_bounded(cmd: list[str], timeout: int) -> None:
     """subprocess.run(check=True, timeout) whose reaper can't itself hang.
 
@@ -156,10 +164,7 @@ def backup_db(
     tmp_path = Path(tmp_name)
     try:
         _run_bounded(
-            _gently([
+            _gently([*_PG_DUMP, "-f", str(tmp_path), _libpq_url(db_url)]),
-                "pg_dump", "--no-owner", "--no-acl", "-Fc",
-                "-f", str(tmp_path), _libpq_url(db_url),
-            ]),
             _DB_SUBPROCESS_TIMEOUT_S,
         )
         shutil.move(str(tmp_path), str(sql_path))
@@ -217,7 +222,7 @@ def backup_db_offsite(*, db_url: str, target, limit_mbps: int | None = None) ->
     deadline = time.monotonic() + _DB_SUBPROCESS_TIMEOUT_S
     with tempfile.TemporaryFile() as err:
         proc = subprocess.Popen(
-            _gently(["pg_dump", "--no-owner", "--no-acl", "-Fc", _libpq_url(db_url)]),
+            _gently([*_PG_DUMP, _libpq_url(db_url)]),
             stdout=subprocess.PIPE, stderr=err,
         )
         upload = None
@@ -289,22 +294,69 @@ def backup_images(
     }
 
 
-def restore_db(*, db_url: str, sql_path: Path) -> None:
+class RestoreFailed(Exception):
-    """Wipe public schema, then load from the custom-format dump. Raises on
+    """psql or pg_restore exited with an error; the message is its last line."""
-    subprocess failure; partial-restore state is the caller's concern."""
+
-    libpq = _libpq_url(db_url)
+
-    subprocess.run(
+# A restore waits this long for other connections to let go of the tables it
-        [
+# replaces, then gives up with Postgres' own reason instead of queueing every
-            "psql", libpq, "-c",
+# other query behind it.
-            "DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public;",
+_RESTORE_LOCK_TIMEOUT = "60s"
-        ],
+# pg_restore of a large library (vector tables, rebuilt indexes) runs well past
-        capture_output=True, check=True, timeout=120,
+# a dump's length; it is bounded on its own.
+_RESTORE_SUBPROCESS_TIMEOUT_S = 3 * 3600
+
+
+def _run_reporting(cmd: list[str], timeout: int, env: dict | None = None) -> None:
+    """Run `cmd`; raise RestoreFailed with the last line it printed when it
+    exits non-zero."""
+    with tempfile.TemporaryFile() as err:
+        proc = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=err, timeout=timeout,
+                              env=env)
+        if proc.returncode != 0:
+            raise RestoreFailed(_last_line(err) or f"{cmd[0]} exited with {proc.returncode}")
+
+
+# Every other connection to this database: the web server's, the other
+# workers'. Ended before the tables are dropped, so none holds a lock the drop
+# waits on, and again once the restore is in, so none keeps statements it
+# prepared against the old tables. Each reconnects on its next query.
+_END_OTHER_CONNECTIONS = (
+    "SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
+    "WHERE datname = current_database() AND pid <> pg_backend_pid();"
 )
+
+
+def end_other_connections(*, db_url: str) -> None:
+    _run_reporting(["psql", _libpq_url(db_url), "-v", "ON_ERROR_STOP=1",
+                    "-c", _END_OTHER_CONNECTIONS], 60)
+
+
+def wipe_db(*, db_url: str) -> None:
+    """Drop and recreate the public schema: an empty database, ready for a
+    dump or for `alembic upgrade head`."""
+    _run_reporting(
+        [
+            "psql", _libpq_url(db_url), "-v", "ON_ERROR_STOP=1", "-c",
+            f"{_END_OTHER_CONNECTIONS} DROP SCHEMA IF EXISTS public CASCADE; "
+            "CREATE SCHEMA public;",
+        ],
+        120,
+        # Set for the session, not as a SET inside the same -c string.
+        env={**os.environ, "PGOPTIONS": f"-c lock_timeout={_RESTORE_LOCK_TIMEOUT}"},
+    )
+
+
+def restore_db(*, db_url: str, sql_path: Path) -> None:
+    """Wipe the public schema, then load the custom-format dump. pg_restore
+    stops at the first error, which RestoreFailed carries; the database is
+    then partly loaded, and the caller decides what to put back."""
+    wipe_db(db_url=db_url)
     # Custom-format (-Fc) dumps are restored with pg_restore, not psql.
-    subprocess.run(
+    _run_reporting(
-        ["pg_restore", "--no-owner", "--no-acl", "-d", libpq, str(sql_path)],
+        _gently(["pg_restore", "--no-owner", "--no-acl", "--exit-on-error",
-        capture_output=True, check=True,
+                 "-d", _libpq_url(db_url), str(sql_path)]),
-        timeout=_DB_SUBPROCESS_TIMEOUT_S,
+        _RESTORE_SUBPROCESS_TIMEOUT_S,
     )
 
 

backend/app/tasks/admin.py

@@ -74,6 +74,12 @@ def prune_missing_file_records_task(self) -> dict:
     thumbs: list[str] = []
     last_id = 0
     with SessionLocal() as session:
+        # While a restore is fetching the library back, files are missing
+        # because they haven't arrived yet, not because they are gone.
+        from .backup import restore_going
+        if restore_going(session):
+            return {"checked": 0, "missing": 0, "deleted": 0,
+                    "skipped": "a restore is fetching the library's files"}
         while True:
             rows = session.execute(
                 select(ImageRecord.id, ImageRecord.path, ImageRecord.thumbnail_path)

backend/app/tasks/backup.py

@@ -9,17 +9,24 @@ from __future__ import annotations
 
 import logging
 import time
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 from celery.exceptions import SoftTimeLimitExceeded
-from sqlalchemy import select
+from sqlalchemy import select, text, update
 from sqlalchemy.exc import DBAPIError, OperationalError
 
 from ..celery_app import celery
 from ..config import get_config
 from ..models import BackupRun, ImportSettings
-from ..services import backup_destination, backup_media, backup_s3, backup_service
+from ..services import (
+    backup_destination,
+    backup_media,
+    backup_restore,
+    backup_s3,
+    backup_service,
+)
+from ._sync_engine import get_sync_engine
 from ._sync_engine import sync_session_factory as _sync_session_factory
 
 log = logging.getLogger(__name__)
@@ -60,6 +67,9 @@ def backup_db_task(self, *, tag: str | None = None,
     cfg = get_config()
     now = datetime.now(UTC)
     with SessionLocal() as session:
+        # A restore replaces the database this would dump.
+        if restore_going(session):
+            return {"skipped": "a restore is going"}
         row = BackupRun(
             kind="db", status="running", tag=tag,
             triggered_by=triggered_by, started_at=now, manifest={},
@@ -269,8 +279,9 @@ def prune_backups() -> dict:
     """Daily Beat. Per-kind retention from ImportSettings.
 
     Returns {"db_deleted": N, "images_deleted": M, "media_deleted": L,
-    "files_unlinked": K}. Media runs leave no files here; only their history
+    "restore_deleted": R, "files_unlinked": K}. Media and restore runs leave
-    is trimmed, to the latest MEDIA_RUNS_KEPT.
+    no files here; only their history is trimmed, to the latest
+    MEDIA_RUNS_KEPT and RESTORE_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
@@ -279,19 +290,21 @@ def prune_backups() -> dict:
     Audit 2026-06-02 added the dedicated sweep.)
     """
     SessionLocal = _sync_session_factory()
-    counts = {"db_deleted": 0, "images_deleted": 0, "media_deleted": 0, "files_unlinked": 0}
+    counts = {"db_deleted": 0, "images_deleted": 0, "media_deleted": 0,
+              "restore_deleted": 0, "files_unlinked": 0}
     with SessionLocal() as session:
         s = ImportSettings.load_sync(session)
         for kind, keep in (
             ("db", s.backup_db_keep_last_n),
             ("images", s.backup_images_keep_last_n),
             ("media", MEDIA_RUNS_KEPT),
+            ("restore", RESTORE_RUNS_KEPT),
         ):
             candidates = session.execute(
                 select(BackupRun)
                 .where(BackupRun.kind == kind)
                 .where(BackupRun.tag.is_(None))
-                .where(BackupRun.status.in_(["ok", "error"]))
+                .where(BackupRun.status.in_(["ok", "error", "restored"]))
                 .order_by(BackupRun.started_at.desc())
                 .offset(keep)
             ).scalars().all()
@@ -385,6 +398,9 @@ def backup_media_task(self, *, run_id: int | None = None,
             ).scalar_one_or_none()
             if going is not None:
                 return {"backup_run_id": going, "skipped": "a media backup is already running"}
+            # During a restore the library's files are still arriving.
+            if restore_going(session):
+                return {"skipped": "a restore is going"}
             row = BackupRun(kind="media", status="running", triggered_by=triggered_by,
                             started_at=now, last_progress_at=now, size_bytes=0,
                             manifest={"sent": 0, "failed": 0, "failures": [], "cursor": ""})
@@ -494,3 +510,284 @@ def reconcile_media_backup() -> dict:
         if going is not None:
             return {"skipped": "a media backup is running"}
         return backup_media.reconcile(session, target)
+
+
+# ── Restoring from the off-site bucket ───────────────────────────────────────
+
+# The file phase fetches for this long per task, then queues its next chunk,
+# like a media run.
+RESTORE_CHUNK_SECONDS = 45 * 60
+RESTORE_RUNS_KEPT = 20
+RESTORE_FAILURES_LISTED = 20
+# A run over files already on disk notes its place this often, not per file.
+_RESTORE_NOTE_SECONDS = 5
+_RESTORE_COUNTS = ("restored", "copied", "present", "missing", "failed")
+
+
+def backup_busy(session) -> str | None:
+    """Why a backup or a restore can't start right now, or None."""
+    row = session.execute(
+        select(BackupRun.kind).where(BackupRun.status.in_(["running", "restoring"])).limit(1)
+    ).first()
+    if row is None:
+        return None
+    return "a restore is going" if row.kind == "restore" else f"a {row.kind} backup is running"
+
+
+def restore_going(session) -> bool:
+    return session.execute(
+        select(BackupRun.id).where(BackupRun.kind == "restore")
+        .where(BackupRun.status == "restoring").limit(1)
+    ).first() is not None
+
+
+def _library_is_empty(session) -> bool:
+    return session.execute(text(
+        "SELECT NOT EXISTS (SELECT 1 FROM image_record) "
+        "AND NOT EXISTS (SELECT 1 FROM post_attachment)"
+    )).scalar()
+
+
+def _note(session, row: BackupRun, m: dict) -> None:
+    row.manifest = dict(m)
+    row.last_progress_at = datetime.now(UTC)
+    session.commit()
+
+
+@celery.task(
+    name="backend.app.tasks.backup.restore_offsite_task",
+    bind=True,
+    max_retries=0,  # a half-applied restore is never retried on its own
+    soft_time_limit=21600, time_limit=23400,
+)
+def restore_offsite_task(self, *, dump_key: str | None = None, files_only: bool = False,
+                         run_id: int | None = None) -> dict:
+    """Restore this install from the off-site bucket, one chunk at a time.
+
+    Started without `run_id`, it replaces the database with the dump at
+    `dump_key` (the newest when None), then fetches every library file the
+    restored rows name. With `files_only`, the database stays and only the
+    missing files are fetched, which also repairs a library that lost files.
+    Each chunk queues the next; the run's row counts files restored from the
+    bucket, copied from a duplicate already on disk, already present, missing
+    from the bucket, and failed. Returns {'backup_run_id': N, ...}."""
+    SessionLocal = _sync_session_factory()
+    if run_id is not None:
+        return _restore_files(SessionLocal, run_id)
+
+    now = datetime.now(UTC)
+    with SessionLocal() as session:
+        busy = backup_busy(session)
+        if busy:
+            return {"skipped": busy}
+        settings = ImportSettings.load_sync(session)
+        limit_mbps = settings.backup_upload_limit_mbps
+        row = BackupRun(kind="restore", status="restoring", triggered_by="restore",
+                        started_at=now, last_progress_at=now, size_bytes=0,
+                        manifest={"phase": "files" if files_only else "database",
+                                  "cursor": "", "failures": [],
+                                  **dict.fromkeys(_RESTORE_COUNTS, 0)})
+        session.add(row)
+        session.commit()
+        run_id = row.id
+        try:
+            target = backup_destination.saved_target(settings)
+            if target is None:
+                raise backup_s3.S3Error("No off-site destination is set up.")
+        except backup_s3.S3Error as exc:
+            _mark_failed(session, row, exc)
+            raise
+        row.manifest = {**row.manifest, "offsite": _offsite_manifest(target)}
+        session.commit()
+
+    if files_only:
+        return _restore_files(SessionLocal, run_id)
+    run_id = _restore_database(SessionLocal, run_id, target, limit_mbps, dump_key)
+    # The files follow as chunks of their own, on the worker.
+    restore_offsite_task.delay(run_id=run_id)
+    return {"backup_run_id": run_id, "continues": True}
+
+
+def _restore_database(SessionLocal, run_id: int, target, limit_mbps, dump_key) -> int:
+    """Replace the database with a dump from the bucket. Returns the run's id
+    in the restored database, where its row is written again."""
+    cfg = get_config()
+    pacer = backup_s3.Pacer(limit_mbps)
+    staged = None
+    try:
+        dump = backup_restore.find_dump(target, dump_key)
+        with SessionLocal() as session:
+            row = session.get(BackupRun, run_id)
+            m = dict(row.manifest)
+            m["dump"] = {"key": dump.key, "ts": dump.ts, "size_bytes": dump.size_bytes}
+            # The database about to be replaced goes to the bucket first, so
+            # the restore can itself be undone. An empty one isn't worth a
+            # 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)
+                m["saved_before"] = saved["key"]
+            _note(session, row, m)
+            started_at, manifest = row.started_at, dict(m)
+        staged = backup_restore.fetch_dump(target, dump, IMAGES_ROOT / "_backups", pacer)
+    except (SoftTimeLimitExceeded, Exception) as exc:
+        if staged is not None:
+            staged.unlink(missing_ok=True)
+        with SessionLocal() as session:
+            row = session.get(BackupRun, run_id)
+            if row is not None:
+                _mark_failed(session, row, exc)
+        raise
+
+    # From here on the database is being replaced; every pooled connection
+    # still points at the tables being dropped.
+    get_sync_engine().dispose()
+    try:
+        backup_service.restore_db(db_url=cfg.database_url_sync, sql_path=staged)
+        backup_restore.migrate()
+        backup_service.end_other_connections(db_url=cfg.database_url_sync)
+    except (SoftTimeLimitExceeded, Exception) as exc:
+        _leave_empty_install(SessionLocal, cfg, target, limit_mbps, started_at, manifest, exc)
+        raise
+    finally:
+        staged.unlink(missing_ok=True)
+    get_sync_engine().dispose()
+
+    with SessionLocal() as session:
+        backup_restore.keep_destination(session, target, limit_mbps)
+        # The dump holds its own run as still going: it is the dump just
+        # restored, so it finished. Any other run going then is over.
+        dump_at = datetime.strptime(manifest["dump"]["ts"], "%Y%m%dT%H%M%SZ").replace(tzinfo=UTC)
+        own = session.execute(
+            select(BackupRun).where(BackupRun.kind == "db", BackupRun.status == "running")
+            # The name's time is truncated to the second, after the row began.
+            .where(BackupRun.started_at.between(dump_at - timedelta(minutes=2),
+                                                dump_at + timedelta(seconds=1)))
+            .order_by(BackupRun.started_at.desc()).limit(1)
+        ).scalar_one_or_none()
+        if own is not None:
+            own.status = "ok"
+            own.manifest = {"ts": manifest["dump"]["ts"], "offsite": {
+                **manifest["offsite"], "key": manifest["dump"]["key"]}}
+            own.size_bytes = manifest["dump"]["size_bytes"]
+        session.execute(
+            update(BackupRun).where(BackupRun.status.in_(["running", "restoring"]))
+            .values(status="error", finished_at=datetime.now(UTC),
+                    error="Interrupted: this run was going when the restored dump was taken.")
+        )
+        row = BackupRun(kind="restore", status="restoring", triggered_by="restore",
+                        started_at=started_at, last_progress_at=datetime.now(UTC),
+                        size_bytes=0, manifest={**manifest, "phase": "files"})
+        session.add(row)
+        session.commit()
+        return row.id
+
+
+def _leave_empty_install(SessionLocal, cfg, target, limit_mbps, started_at, manifest,
+                         exc: BaseException) -> None:
+    """After a failed load, put back an empty database at this version, so the
+    install starts and can try again, and record why it stopped."""
+    try:
+        backup_service.wipe_db(db_url=cfg.database_url_sync)
+        backup_restore.migrate()
+        backup_service.end_other_connections(db_url=cfg.database_url_sync)
+        get_sync_engine().dispose()
+        with SessionLocal() as session:
+            backup_restore.keep_destination(session, target, limit_mbps)
+            session.add(BackupRun(
+                kind="restore", status="error", triggered_by="restore",
+                started_at=started_at, finished_at=datetime.now(UTC), manifest=manifest,
+                error=(f"The restore stopped while loading the database: {exc}. "
+                       "The database is empty again; choose a dump and restore again.")[:2000],
+            ))
+            session.commit()
+    except Exception:  # noqa: BLE001 — the original failure is the one raised
+        log.exception("restore: the database couldn't be reset after a failed load")
+
+
+def _restore_files(SessionLocal, run_id: int) -> dict:
+    """One chunk of a restore's file phase; queues the next while files remain."""
+    with SessionLocal() as session:
+        row = session.get(BackupRun, run_id)
+        if row is None or row.status != "restoring":
+            return {"backup_run_id": run_id, "skipped": "the restore is no longer going"}
+        settings = ImportSettings.load_sync(session)
+        limit_mbps = settings.backup_upload_limit_mbps
+        try:
+            target = backup_destination.saved_target(settings)
+            if target is None:
+                raise backup_s3.S3Error("No off-site destination is set up.")
+        except backup_s3.S3Error as exc:
+            _mark_failed(session, row, exc)
+            raise
+    try:
+        more = _fetch_files_chunk(SessionLocal, run_id, target, backup_s3.Pacer(limit_mbps))
+    except (SoftTimeLimitExceeded, Exception) as exc:
+        with SessionLocal() as session:
+            row = session.get(BackupRun, run_id)
+            if row is not None:
+                _mark_failed(session, row, exc)
+        raise
+
+    if more:
+        restore_offsite_task.delay(run_id=run_id)
+        return {"backup_run_id": run_id, "continues": True}
+    with SessionLocal() as session:
+        row = session.get(BackupRun, run_id)
+        m = dict(row.manifest or {})
+        row.status = "restored"
+        row.finished_at = datetime.now(UTC)
+        missed = m.get("missing", 0) + m.get("failed", 0)
+        if missed:
+            first = (m.get("failures") or [{}])[0]
+            row.error = (f"{m.get('missing', 0)} file(s) aren't in the bucket and "
+                         f"{m.get('failed', 0)} couldn't be restored. "
+                         f"First: {first.get('path')}: {first.get('error')}")[:2000]
+        session.commit()
+    # Thumbnails aren't backed up: they are made again from the restored files.
+    from .thumbnail import backfill_thumbnails
+    backfill_thumbnails.delay()
+    return {"backup_run_id": run_id, "continues": False}
+
+
+def _fetch_files_chunk(SessionLocal, run_id: int, target, pacer) -> bool:
+    """Put library files back until the chunk's time is up or none are left.
+    Returns True when files remain for another chunk."""
+    tid = backup_media.target_id(target)
+    deadline = time.monotonic() + RESTORE_CHUNK_SECONDS
+    with SessionLocal() as session:
+        row = session.get(BackupRun, run_id)
+        m = dict(row.manifest or {})
+        cursor = m.get("cursor") or ""
+        noted = time.monotonic()
+        while True:
+            batch = backup_restore.library_files(session, after=cursor)
+            keys = backup_restore.sent_keys(session, tid, [f.sha256 for f in batch])
+            # End the read before any download starts.
+            session.commit()
+            if not batch:
+                m["cursor"] = cursor
+                _note(session, row, m)
+                return False
+            for f in batch:
+                if time.monotonic() > deadline:
+                    m["cursor"] = cursor
+                    _note(session, row, m)
+                    return True
+                try:
+                    outcome = backup_restore.restore_file(
+                        target, f, keys.get(f.sha256), IMAGES_ROOT, pacer)
+                except (backup_s3.S3Error, OSError) as exc:
+                    outcome = "missing" if isinstance(exc, backup_s3.ObjectMissing) else "failed"
+                    if len(m.get("failures") or []) < RESTORE_FAILURES_LISTED:
+                        m["failures"] = [*(m.get("failures") or []),
+                                         {"path": f.paths[0], "error": str(exc)[:300]}]
+                m[outcome] = m.get(outcome, 0) + 1
+                if outcome in ("restored", "copied"):
+                    row.size_bytes = (row.size_bytes or 0) + f.size_bytes
+                cursor = f.sha256
+                m["cursor"] = cursor
+                if outcome != "present" or time.monotonic() - noted > _RESTORE_NOTE_SECONDS:
+                    _note(session, row, m)
+                    noted = time.monotonic()

backend/app/tasks/maintenance.py

@@ -176,6 +176,8 @@ TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
     # One media-backup chunk stops starting files after 45 min but may finish
     # a large file in flight, up to its 6.5 h hard limit.
     "backend.app.tasks.backup.backup_media_task": 420,
+    # A restore's database load, or one chunk of its file phase, likewise.
+    "backend.app.tasks.backup.restore_offsite_task": 420,
     # Library audit scans the full library — 2h hard limit.
     "backend.app.tasks.library_audit.scan_library_for_rule": 130,
     # External file-host fetches (mega/gdrive/film packs) can run to the task's
@@ -845,17 +847,17 @@ def recover_stalled_backup_runs() -> int:
             update(BackupRun)
             .where(BackupRun.status.in_(["running", "restoring"]))
             # db backups/restores are fast (40-min window); images run hours (7h).
-            # A media run spans many chunks and hours on purpose, so its window
+            # A media run or an off-site restore spans many chunks and hours on
-            # runs from its last progress, not from its start.
+            # purpose, so its window runs from its last progress, not its start.
             .where(
                 or_(
                     and_(BackupRun.kind == "db", BackupRun.started_at < db_cutoff),
                     and_(
-                        BackupRun.kind == "media",
+                        BackupRun.kind.in_(["media", "restore"]),
                         func.coalesce(BackupRun.last_progress_at, BackupRun.started_at)
                         < slow_cutoff,
                     ),
-                    and_(BackupRun.kind.not_in(["db", "media"]),
+                    and_(BackupRun.kind.not_in(["db", "media", "restore"]),
                          BackupRun.started_at < slow_cutoff),
                 )
             )

frontend/src/components/modal/DestructiveConfirmModal.vue

@@ -89,6 +89,7 @@ const kindLabel = computed(() => ({
   artist: 'artist',
   tag: 'tag',
   'images-selection': 'image selection',
+  offsite: 'from the off-site bucket',
 }[props.kind] || props.kind))
 const alertType = computed(
   () => props.action === 'restore' ? 'warning' : 'error',

frontend/src/components/settings/BackupCard.vue

@@ -96,6 +96,87 @@
           A run is going: {{ runningSent.toLocaleString() }} files sent so far.
         </span>
         <BackupRunsTable :runs="store.mediaRuns" />
+
+        <v-divider class="my-4" />
+
+        <!-- Off-site: restoring this install from the bucket -->
+        <h3 class="fc-section-title">Restore from the bucket</h3>
+        <p class="fc-muted text-body-2 mb-2">
+          Replace this install's database with one of the bucket's dumps, then
+          fetch every image and attachment back to where it belongs, each
+          checked against its fingerprint as it arrives. The database being
+          replaced is saved to the bucket first, so a restore can be undone.
+          Account logins aren't in the backup: sign in to each again afterwards.
+        </p>
+        <p class="fc-muted text-body-2 mb-2">
+          To repair a library that lost files, fetch only what's missing from
+          disk; the database stays as it is.
+        </p>
+        <div class="fc-settings-row mb-3">
+          <v-btn
+            color="warning" variant="tonal" rounded="pill"
+            prepend-icon="mdi-database-arrow-down"
+            :disabled="anyGoing" :loading="dumpsLoading"
+            @click="openDumps"
+          >Choose a dump…</v-btn>
+          <v-btn
+            variant="tonal" rounded="pill"
+            prepend-icon="mdi-cloud-download-outline"
+            :disabled="anyGoing" :loading="fetchTriggering"
+            @click="onFetchMissing"
+          >Fetch missing files</v-btn>
+        </div>
+        <v-alert
+          v-if="restoreDispatched || runningRestore"
+          type="info" variant="tonal" density="compact" class="mb-3"
+        >
+          <template v-if="runningRestore?.manifest?.phase === 'files'">
+            Fetching files: {{ restoreLine(runningRestore.manifest) }}.
+          </template>
+          <template v-else>
+            Restoring the database. While it loads, this page may stop answering
+            for a few minutes; it comes back on its own.
+          </template>
+        </v-alert>
+        <BackupRunsTable :runs="store.restoreRuns" empty-text="No restores yet." />
+
+        <v-dialog v-model="dumpsOpen" max-width="560">
+          <v-card>
+            <v-card-title>Dumps in {{ offsite.bucket }}</v-card-title>
+            <v-card-text>
+              <v-alert
+                v-if="dumpsError" type="warning" variant="tonal" density="compact" class="mb-3"
+              >{{ dumpsError }}</v-alert>
+              <v-table v-else density="compact">
+                <thead>
+                  <tr><th>Taken</th><th class="text-right">Size</th><th /></tr>
+                </thead>
+                <tbody>
+                  <tr v-for="d in dumps" :key="d.key">
+                    <td :title="d.key">{{ dumpWhen(d.ts) }}</td>
+                    <td class="text-right" style="font-variant-numeric: tabular-nums;">
+                      {{ humanBytes(d.size_bytes) }}
+                    </td>
+                    <td class="text-right">
+                      <v-btn size="small" variant="text" color="warning" @click="onPickDump(d)">
+                        Restore…
+                      </v-btn>
+                    </td>
+                  </tr>
+                  <tr v-if="!dumps.length">
+                    <td colspan="3" class="text-center fc-muted py-4">
+                      The bucket holds no database dumps yet.
+                    </td>
+                  </tr>
+                </tbody>
+              </v-table>
+            </v-card-text>
+            <v-card-actions>
+              <v-spacer />
+              <v-btn variant="text" @click="dumpsOpen = false">Close</v-btn>
+            </v-card-actions>
+          </v-card>
+        </v-dialog>
       </template>
 
       <template v-else>
@@ -132,18 +213,20 @@
       :kind="confirmKind"
       :run-id="confirmRunId"
       :description="confirmDescription"
+      :expected-token-override="confirmToken"
       @confirm="onConfirmSubmit"
     />
   </MaintenanceTile>
 </template>
 
 <script setup>
-import { computed, onMounted, onUnmounted, ref } from 'vue'
+import { computed, onMounted, onUnmounted, ref, watch } from 'vue'
 
 import { useBackupStore } from '../../stores/backup.js'
 import BackupConfirmModal from '../modal/DestructiveConfirmModal.vue'
 import MaintenanceTile from '../common/MaintenanceTile.vue'
 import { humanBytes } from '../../utils/bytes.js'
+import { formatDateTime } from '../../utils/date.js'
 import BackupRunsTable from './BackupRunsTable.vue'
 
 const store = useBackupStore()
@@ -177,6 +260,79 @@ const confirmAction = ref('restore')
 const confirmKind = ref('db')
 const confirmRunId = ref(0)
 const confirmDescription = ref('')
+const confirmToken = ref('')
+
+// Restoring from the bucket.
+const runningRestore = computed(() => store.restoreRuns.find((r) => r.status === 'restoring'))
+const anyGoing = computed(() => Boolean(
+  runningRestore.value || mediaRunning.value
+  || store.dbRuns.some((r) => r.status === 'running'),
+))
+const restoreDispatched = ref(false)
+const dumpsOpen = ref(false)
+const dumpsLoading = ref(false)
+const dumpsError = ref('')
+const dumps = ref([])
+const pickedDump = ref(null)
+const fetchTriggering = ref(false)
+// The notice that a restore was asked for stays up until its run is over.
+watch(runningRestore, (now, before) => {
+  if (before && !now) restoreDispatched.value = false
+})
+
+function restoreLine(m) {
+  const n = (k) => (m?.[k] || 0).toLocaleString()
+  return `${n('restored')} restored, ${n('copied')} copied from a duplicate, `
+    + `${n('present')} already there, ${n('missing')} not in the bucket, ${n('failed')} failed`
+}
+
+// 20261107T134258Z → the reader's local date and time.
+function dumpWhen(ts) {
+  const iso = ts.replace(
+    /^(\d{4})(\d{2})(\d{2})T(\d{2})(\d{2})(\d{2})Z$/, '$1-$2-$3T$4:$5:$6Z',
+  )
+  return formatDateTime(iso)
+}
+
+async function openDumps() {
+  dumpsLoading.value = true
+  dumpsError.value = ''
+  try {
+    dumps.value = await store.listOffsiteDumps()
+  } catch (e) {
+    dumps.value = []
+    dumpsError.value = e.message
+  } finally {
+    dumpsLoading.value = false
+    dumpsOpen.value = true
+  }
+}
+
+function onPickDump(d) {
+  pickedDump.value = d
+  dumpsOpen.value = false
+  confirmAction.value = 'restore'
+  confirmKind.value = 'offsite'
+  confirmRunId.value = ''
+  confirmToken.value = d.confirm
+  confirmDescription.value = (
+    `Replaces this install's database with the dump taken ${dumpWhen(d.ts)}, `
+    + 'then fetches the library\'s files from the bucket.'
+  )
+  confirmOpen.value = true
+}
+
+async function onFetchMissing() {
+  fetchTriggering.value = true
+  try {
+    await store.fetchMissingFiles()
+    pollOnce()
+  } catch {
+    // store.lastError is shown above.
+  } finally {
+    fetchTriggering.value = false
+  }
+}
 
 let pollId = null
 let polls = 0
@@ -188,6 +344,7 @@ function pollOnce() {
   store.loadRuns('db')
   if (offsite.value) {
     store.loadRuns('media')
+    store.loadRuns('restore')
     if (polls++ % COVERAGE_EVERY === 0) store.loadMediaCoverage()
   } else {
     store.loadRuns('images')
@@ -223,6 +380,7 @@ function _openConfirm(action, run) {
   confirmAction.value = action
   confirmKind.value = run.kind
   confirmRunId.value = run.id
+  confirmToken.value = ''
   confirmDescription.value = (
     action === 'restore'
       ? `Source artifact: ${run.kind === 'db' ? run.sql_path : run.tar_path}`
@@ -246,7 +404,10 @@ async function onTag(run) {
 
 async function onConfirmSubmit(token) {
   try {
-    if (confirmAction.value === 'restore') {
+    if (confirmKind.value === 'offsite') {
+      await store.restoreFromOffsite(pickedDump.value.key, token)
+      restoreDispatched.value = true
+    } else if (confirmAction.value === 'restore') {
       await store.restore(confirmRunId.value, token)
     } else {
       await store.deleteRun(confirmRunId.value, token)

frontend/src/components/settings/BackupRunsTable.vue

@@ -58,7 +58,7 @@
         </td>
       </tr>
       <tr v-if="!runs.length">
-        <td colspan="7" class="text-center fc-muted py-4">No backups yet.</td>
+        <td colspan="7" class="text-center fc-muted py-4">{{ emptyText }}</td>
       </tr>
     </tbody>
   </v-table>
@@ -68,7 +68,10 @@
 import { formatRelative as fmtRelative } from '../../utils/date.js'
 import KebabMenu from '../common/KebabMenu.vue'
 
-defineProps({ runs: { type: Array, default: () => [] } })
+defineProps({
+  runs: { type: Array, default: () => [] },
+  emptyText: { type: String, default: 'No backups yet.' },
+})
 defineEmits(['restore', 'delete', 'tag'])
 
 function statusIcon(s) {

frontend/src/stores/backup.js

@@ -9,10 +9,13 @@ export const useBackupStore = defineStore('backup', () => {
   const dbRuns = ref([])
   const imagesRuns = ref([])
   const mediaRuns = ref([])
+  const restoreRuns = ref([])
   const settings = ref(null)
-  const loading = ref({ dbRuns: false, imagesRuns: false, mediaRuns: false, settings: false })
+  const loading = ref({
+    dbRuns: false, imagesRuns: false, mediaRuns: false, restoreRuns: false, settings: false,
+  })
   const lastError = ref(null)
-  const RUNS = { db: dbRuns, images: imagesRuns, media: mediaRuns }
+  const RUNS = { db: dbRuns, images: imagesRuns, media: mediaRuns, restore: restoreRuns }
 
   async function loadRuns(kind) {
     const targetRef = RUNS[kind]
@@ -139,8 +142,36 @@ export const useBackupStore = defineStore('backup', () => {
     }
   }
 
+  // Restoring from the bucket. Dumps arrive newest first, each with the
+  // `confirm` text a restore of it asks for.
+  async function listOffsiteDumps() {
+    const body = await api.get('/api/system/backup/offsite/dumps')
+    return body.dumps || []
+  }
+
+  async function restoreFromOffsite(key, confirm) {
+    lastError.value = null
+    try {
+      await api.post('/api/system/backup/offsite/restore', { body: { key, confirm } })
+    } catch (e) {
+      lastError.value = e.message
+      throw e
+    }
+  }
+
+  async function fetchMissingFiles() {
+    lastError.value = null
+    try {
+      await api.post('/api/system/backup/offsite/fetch-files', { body: {} })
+    } catch (e) {
+      lastError.value = e.message
+      throw e
+    }
+  }
+
   return {
-    dbRuns, imagesRuns, mediaRuns, mediaCoverage, loadMediaCoverage,
+    dbRuns, imagesRuns, mediaRuns, restoreRuns, mediaCoverage, loadMediaCoverage,
+    listOffsiteDumps, restoreFromOffsite, fetchMissingFiles,
     settings, loading, lastError,
     loadRuns, triggerBackup, restore, deleteRun, setTag,
     loadSettings, patchSettings,

tests/fake_s3.py

@@ -0,0 +1,110 @@
+"""An in-process S3 bucket for round-trip tests: put, multipart upload,
+get, and ListObjectsV2, kept in memory. It doesn't check signatures; the
+signing has its own tests against AWS's published vectors. Like Fabled
+Archives, it refuses to replace an object that is already there.
+
+    with FakeS3() as s3:
+        target = s3.target()
+        ...
+        s3.objects  # {key: bytes}
+"""
+from __future__ import annotations
+
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import parse_qs, unquote, urlsplit
+from xml.sax.saxutils import escape
+
+from backend.app.services.backup_s3 import S3Target
+
+BUCKET = "curator"
+
+
+class FakeS3:
+    def __init__(self):
+        self.objects: dict[str, bytes] = {}
+        self.uploads: dict[str, dict[int, bytes]] = {}
+        outer = self
+
+        class Handler(BaseHTTPRequestHandler):
+            def _parts(self):
+                u = urlsplit(self.path)
+                path = unquote(u.path).lstrip("/")
+                bucket, _, key = path.partition("/")
+                query = parse_qs(u.query, keep_blank_values=True)   # `?uploads` has no value
+                return bucket, key, {k: v[0] for k, v in query.items()}
+
+            def _reply(self, status, body=b"", headers=None):
+                self.send_response(status)
+                for name, value in (headers or {}).items():
+                    self.send_header(name, value)
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                if self.command != "HEAD":
+                    self.wfile.write(body)
+
+            def _body(self):
+                length = int(self.headers.get("Content-Length") or 0)
+                return self.rfile.read(length) if length else b""
+
+            def do_PUT(self):
+                _, key, q = self._parts()
+                body = self._body()
+                if "uploadId" in q:
+                    outer.uploads[q["uploadId"]][int(q["partNumber"])] = body
+                    return self._reply(200, headers={"ETag": f'"p{q["partNumber"]}"'})
+                if key in outer.objects:
+                    return self._reply(403, b"<Error><Code>AccessDenied</Code></Error>")
+                outer.objects[key] = body
+                return self._reply(200)
+
+            def do_POST(self):
+                _, key, q = self._parts()
+                self._body()
+                if "uploads" in q:
+                    upload_id = f"u{len(outer.uploads) + 1}"
+                    outer.uploads[upload_id] = {}
+                    return self._reply(200, (
+                        "<InitiateMultipartUploadResult>"
+                        f"<UploadId>{upload_id}</UploadId>"
+                        "</InitiateMultipartUploadResult>").encode())
+                parts = outer.uploads.pop(q["uploadId"])
+                outer.objects[key] = b"".join(parts[n] for n in sorted(parts))
+                return self._reply(200, b"<CompleteMultipartUploadResult/>")
+
+            def do_DELETE(self):
+                _, _, q = self._parts()
+                outer.uploads.pop(q.get("uploadId", ""), None)
+                return self._reply(204)
+
+            def do_GET(self):
+                _, key, q = self._parts()
+                if q.get("list-type") == "2":
+                    prefix = q.get("prefix", "")
+                    items = "".join(
+                        f"<Contents><Key>{escape(k)}</Key><Size>{len(v)}</Size>"
+                        "<LastModified>2026-01-01T00:00:00Z</LastModified></Contents>"
+                        for k, v in sorted(outer.objects.items()) if k.startswith(prefix)
+                    )
+                    return self._reply(200, f"<ListBucketResult>{items}</ListBucketResult>".encode())
+                if key not in outer.objects:
+                    return self._reply(404, b"<Error><Code>NoSuchKey</Code></Error>")
+                return self._reply(200, outer.objects[key])
+
+            def log_message(self, *args):
+                pass
+
+        self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
+
+    def __enter__(self):
+        self.thread.start()
+        return self
+
+    def __exit__(self, *exc):
+        self.httpd.shutdown()
+        self.httpd.server_close()
+
+    def target(self) -> S3Target:
+        host, port = self.httpd.server_address
+        return S3Target(f"http://{host}:{port}", BUCKET, "AK", "SK")

tests/test_api_system_backup.py

@@ -496,3 +496,91 @@ async def test_an_upload_limit_that_isnt_a_whole_number_is_refused(client, _seal
     )
     assert resp.status_code == 400
     assert "upload_limit_mbps" in (await resp.get_json())["detail"]
+
+
+# ── restoring from the off-site bucket ──────────────────────────────────────
+
+
+@pytest.fixture
+def _saved_bucket(monkeypatch):
+    from backend.app.services import backup_destination
+    from backend.app.services.backup_s3 import S3Target
+
+    target = S3Target("http://archives.example:8081", "curator", "AK", "SK")
+    monkeypatch.setattr(backup_destination, "saved_target", lambda settings: target)
+    dispatched = []
+    monkeypatch.setattr("backend.app.tasks.backup.restore_offsite_task.delay",
+                        lambda **kw: dispatched.append(kw))
+    return dispatched
+
+
+_DUMP = "database/fabledcurator-20261107T134258Z.dump"
+
+
+@pytest.mark.asyncio
+async def test_the_bucket_dumps_are_listed_with_their_confirmation(
+    client, _saved_bucket, monkeypatch,
+):
+    from backend.app.services import backup_restore
+
+    monkeypatch.setattr(backup_restore, "list_dumps", lambda target: [
+        backup_restore.Dump(_DUMP, "20261107T134258Z", 3222868924),
+    ])
+    resp = await client.get("/api/system/backup/offsite/dumps")
+    assert resp.status_code == 200
+    assert (await resp.get_json())["dumps"] == [{
+        "key": _DUMP, "ts": "20261107T134258Z", "size_bytes": 3222868924,
+        "confirm": "restore-offsite-20261107T134258Z",
+    }]
+
+
+@pytest.mark.asyncio
+async def test_a_restore_needs_the_dump_typed_back(client, _saved_bucket):
+    resp = await client.post("/api/system/backup/offsite/restore",
+                             json={"key": _DUMP, "confirm": "yes"})
+    assert resp.status_code == 400
+    assert (await resp.get_json())["expected"] == "restore-offsite-20261107T134258Z"
+    assert _saved_bucket == []
+
+    resp = await client.post("/api/system/backup/offsite/restore", json={
+        "key": _DUMP, "confirm": "restore-offsite-20261107T134258Z"})
+    assert resp.status_code == 202
+    assert _saved_bucket == [{"dump_key": _DUMP}]
+
+
+@pytest.mark.asyncio
+async def test_a_restore_waits_while_a_backup_runs(client, _saved_bucket, db_sync):
+    db_sync.add(BackupRun(kind="db", status="running", triggered_by="nightly",
+                          started_at=datetime.now(UTC), manifest={}))
+    db_sync.commit()
+    resp = await client.post("/api/system/backup/offsite/restore", json={
+        "key": _DUMP, "confirm": "restore-offsite-20261107T134258Z"})
+    assert resp.status_code == 409
+    assert "db backup is running" in (await resp.get_json())["detail"]
+    resp = await client.post("/api/system/backup/offsite/fetch-files", json={})
+    assert resp.status_code == 409
+    assert _saved_bucket == []
+
+
+@pytest.mark.asyncio
+async def test_fetching_missing_files_leaves_the_database(client, _saved_bucket):
+    resp = await client.post("/api/system/backup/offsite/fetch-files", json={})
+    assert resp.status_code == 202
+    assert _saved_bucket == [{"files_only": True}]
+
+
+@pytest.mark.asyncio
+async def test_restoring_needs_a_destination(client):
+    resp = await client.get("/api/system/backup/offsite/dumps")
+    assert resp.status_code == 400
+    assert (await resp.get_json())["error"] == "no_destination"
+
+
+@pytest.mark.asyncio
+async def test_restore_runs_are_listed_as_their_own_kind(client, db_sync):
+    db_sync.add(BackupRun(kind="restore", status="restoring", triggered_by="restore",
+                          started_at=datetime.now(UTC), manifest={"phase": "files"}))
+    db_sync.commit()
+    resp = await client.get("/api/system/backup/runs?kind=restore")
+    runs = (await resp.get_json())["runs"]
+    assert [r["status"] for r in runs] == ["restoring"]

tests/test_backup_restore.py

@@ -0,0 +1,297 @@
+"""Restoring from the off-site bucket (services/backup_restore.py and
+restore_offsite_task): the database from a dated dump, then every library
+file back at the path its row records, checked against its sha256 on arrival.
+The round trip at the bottom backs a seeded library up to an in-process S3
+bucket, wipes it, restores it, and compares every file."""
+import hashlib
+import shutil
+import time
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from sqlalchemy import func, select
+
+from backend.app.models import Artist, BackupRun, ImageRecord, Post, PostAttachment
+from backend.app.services import backup_restore
+from backend.app.services.backup_s3 import ListPage, ObjectMissing, S3Error, S3Target, StoredObject
+from tests.fake_s3 import FakeS3
+
+pytestmark = pytest.mark.integration
+
+TARGET = S3Target("http://archives.example:8081", "curator", "AK", "SK")
+
+
+def _sha(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def _file(root, rel, data):
+    p = root / rel
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_bytes(data)
+    return p
+
+
+def _image(db_sync, path, data, mime="image/jpeg"):
+    db_sync.add(ImageRecord(path=str(path), sha256=_sha(data), size_bytes=len(data),
+                            mime=mime, origin="imported_filesystem"))
+
+
+# ── dumps ───────────────────────────────────────────────────────────────────
+
+
+def test_dumps_are_listed_newest_first_and_other_keys_ignored(monkeypatch):
+    page = ListPage([
+        StoredObject("database/fabledcurator-20261101T034000Z.dump", 10, ""),
+        StoredObject("database/fabledcurator-20261107T134258Z.dump", 30, ""),
+        StoredObject("database/notes.txt", 1, ""),
+    ], None)
+    monkeypatch.setattr(backup_restore, "list_objects", lambda *a, **kw: page)
+    dumps = backup_restore.list_dumps(TARGET)
+    assert [d.ts for d in dumps] == ["20261107T134258Z", "20261101T034000Z"]
+    assert backup_restore.find_dump(TARGET, "latest").size_bytes == 30
+    assert backup_restore.find_dump(TARGET, dumps[1].key).ts == "20261101T034000Z"
+    with pytest.raises(S3Error, match="not among"):
+        backup_restore.find_dump(TARGET, "database/fabledcurator-1.dump")
+
+
+def test_an_empty_bucket_has_no_dump_to_restore(monkeypatch):
+    monkeypatch.setattr(backup_restore, "list_objects", lambda *a, **kw: ListPage([], None))
+    with pytest.raises(S3Error, match="no database dumps"):
+        backup_restore.find_dump(TARGET, None)
+
+
+# ── one file ────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def fetches(monkeypatch):
+    got = []
+
+    def get_file(target, key, path, *, sha256=None, size=None, pacer=None):
+        got.append(key)
+        path.write_bytes(b"x" * size)
+    monkeypatch.setattr(backup_restore, "get_file", get_file)
+    return got
+
+
+def test_a_file_on_disk_at_its_size_is_left_alone(tmp_path, fetches):
+    p = _file(tmp_path, "a/one.jpg", b"xxx")
+    f = backup_restore.LibraryFile("a" * 64, (str(p),), 3)
+    assert backup_restore.restore_file(TARGET, f, None, tmp_path, None) == "present"
+    assert fetches == []
+
+
+def test_a_missing_file_is_fetched_by_its_key_to_its_path(tmp_path, fetches):
+    sha = "abcd" + "0" * 60
+    f = backup_restore.LibraryFile(sha, (str(tmp_path / "bea/one.JPG"),), 4)
+    assert backup_restore.restore_file(TARGET, f, None, tmp_path, None) == "restored"
+    assert fetches == [f"media/ab/cd/{sha}.jpg"]
+    assert (tmp_path / "bea/one.JPG").read_bytes() == b"xxxx"
+
+
+def test_a_recorded_key_is_used_over_the_computed_one(tmp_path, fetches):
+    f = backup_restore.LibraryFile("e" * 64, (str(tmp_path / "x.png"),), 2)
+    backup_restore.restore_file(TARGET, f, "media/ee/ee/recorded.png", tmp_path, None)
+    assert fetches == ["media/ee/ee/recorded.png"]
+
+
+def test_one_fetch_fills_every_path_that_shares_the_file(tmp_path, fetches):
+    paths = (str(tmp_path / "p1/set.zip"), str(tmp_path / "p2/set.zip"))
+    f = backup_restore.LibraryFile("c" * 64, paths, 5)
+    assert backup_restore.restore_file(TARGET, f, None, tmp_path, None) == "restored"
+    assert len(fetches) == 1
+    assert (tmp_path / "p2/set.zip").read_bytes() == b"xxxxx"
+    # With one copy left on disk, the other is copied, not fetched.
+    (tmp_path / "p1/set.zip").unlink()
+    assert backup_restore.restore_file(TARGET, f, None, tmp_path, None) == "copied"
+    assert len(fetches) == 1
+
+
+def test_a_path_outside_the_library_is_refused(tmp_path, fetches):
+    f = backup_restore.LibraryFile("d" * 64, ("/etc/passwd",), 3)
+    with pytest.raises(S3Error, match="outside"):
+        backup_restore.restore_file(TARGET, f, None, tmp_path, None)
+    f = backup_restore.LibraryFile("d" * 64, (str(tmp_path / "../escape.jpg"),), 3)
+    with pytest.raises(S3Error, match="outside"):
+        backup_restore.restore_file(TARGET, f, None, tmp_path, None)
+    assert fetches == []
+
+
+def test_the_work_list_gives_each_file_every_path(db_sync, tmp_path):
+    _image(db_sync, tmp_path / "a.jpg", b"one")
+    artist = Artist(name="Bea", slug="bea")
+    db_sync.add(artist)
+    db_sync.flush()
+    post = Post(artist_id=artist.id, external_post_id="p1")
+    db_sync.add(post)
+    db_sync.flush()
+    for i, post_id in enumerate((None, post.id)):
+        db_sync.add(PostAttachment(post_id=post_id, sha256=_sha(b"zip"),
+                                   path=str(tmp_path / f"{i}/set.zip"),
+                                   original_filename="set.zip", ext=".zip", size_bytes=3))
+    db_sync.commit()
+    files = {f.sha256: f for f in backup_restore.library_files(db_sync)}
+    assert files[_sha(b"zip")].paths == (str(tmp_path / "0/set.zip"), str(tmp_path / "1/set.zip"))
+    assert files[_sha(b"one")].paths == (str(tmp_path / "a.jpg"),)
+
+
+# ── the file phase as a run ─────────────────────────────────────────────────
+
+
+@pytest.fixture
+def run_env(monkeypatch, tmp_path):
+    from backend.app.tasks import backup as tasks
+    monkeypatch.setattr(tasks, "IMAGES_ROOT", tmp_path)
+    monkeypatch.setattr(
+        "backend.app.services.backup_destination.saved_target", lambda settings: TARGET,
+    )
+    queued, thumbs = [], []
+    monkeypatch.setattr(tasks.restore_offsite_task, "delay", lambda **kw: queued.append(kw))
+    monkeypatch.setattr("backend.app.tasks.thumbnail.backfill_thumbnails.delay",
+                        lambda: thumbs.append(True))
+    return queued, thumbs
+
+
+def test_fetching_missing_files_counts_each_outcome(db_sync, tmp_path, run_env, monkeypatch):
+    from backend.app.tasks.backup import restore_offsite_task
+    _file(tmp_path, "here.jpg", b"here")
+    _image(db_sync, tmp_path / "here.jpg", b"here")
+    _image(db_sync, tmp_path / "gone.jpg", b"gone")
+    _image(db_sync, tmp_path / "lost.jpg", b"lost")
+    db_sync.commit()
+
+    def get_file(target, key, path, *, sha256=None, size=None, pacer=None):
+        if sha256 == _sha(b"lost"):
+            raise ObjectMissing(f"{key} is not in the curator bucket.")
+        path.write_bytes(b"gone")
+    monkeypatch.setattr(backup_restore, "get_file", get_file)
+
+    result = restore_offsite_task(files_only=True)
+    assert result["continues"] is False
+    row = db_sync.get(BackupRun, result["backup_run_id"])
+    db_sync.refresh(row)
+    m = row.manifest
+    assert (row.kind, row.status) == ("restore", "restored")
+    assert (m["restored"], m["present"], m["missing"], m["failed"]) == (1, 1, 1, 0)
+    assert "1 file(s) aren't in the bucket" in row.error
+    assert (tmp_path / "gone.jpg").read_bytes() == b"gone"
+    queued, thumbs = run_env
+    assert queued == [] and thumbs == [True]
+
+
+def test_a_restore_out_of_time_queues_its_next_chunk(db_sync, tmp_path, run_env, monkeypatch):
+    from backend.app.tasks import backup as tasks
+    _image(db_sync, tmp_path / "a.jpg", b"aa")
+    db_sync.commit()
+    monkeypatch.setattr(tasks, "RESTORE_CHUNK_SECONDS", -1)
+    result = tasks.restore_offsite_task(files_only=True)
+    queued, _ = run_env
+    assert result["continues"] is True
+    assert queued == [{"run_id": result["backup_run_id"]}]
+    row = db_sync.get(BackupRun, result["backup_run_id"])
+    db_sync.refresh(row)
+    assert row.status == "restoring" and row.last_progress_at is not None
+
+
+def test_a_restore_waits_for_a_backup_that_is_running(db_sync, run_env):
+    from backend.app.tasks.backup import restore_offsite_task
+    db_sync.add(BackupRun(kind="media", status="running", triggered_by="nightly",
+                          started_at=datetime.now(UTC), manifest={}))
+    db_sync.commit()
+    assert restore_offsite_task(files_only=True)["skipped"] == "a media backup is running"
+
+
+def test_backups_wait_while_a_restore_is_going(db_sync, monkeypatch):
+    from backend.app.tasks.backup import backup_db_task, backup_media_task
+    db_sync.add(BackupRun(kind="restore", status="restoring", triggered_by="restore",
+                          started_at=datetime.now(UTC), manifest={}))
+    db_sync.commit()
+    assert backup_db_task(triggered_by="nightly")["skipped"] == "a restore is going"
+    assert backup_media_task(triggered_by="nightly")["skipped"] == "a restore is going"
+
+
+def test_a_long_restore_is_judged_stalled_by_its_heartbeat(db_sync):
+    from backend.app.tasks.maintenance import recover_stalled_backup_runs
+    long_ago = datetime.now(UTC) - timedelta(hours=20)
+    alive = BackupRun(kind="restore", status="restoring", triggered_by="restore",
+                      started_at=long_ago, last_progress_at=datetime.now(UTC), manifest={})
+    silent = BackupRun(kind="restore", status="restoring", triggered_by="restore",
+                       started_at=long_ago, last_progress_at=long_ago, manifest={})
+    db_sync.add_all([alive, silent])
+    db_sync.commit()
+    assert recover_stalled_backup_runs() == 1
+    db_sync.refresh(alive)
+    db_sync.refresh(silent)
+    assert (alive.status, silent.status) == ("restoring", "error")
+
+
+# ── the round trip ──────────────────────────────────────────────────────────
+
+
+def test_a_library_comes_back_whole_from_the_bucket(db_sync, tmp_path, monkeypatch):
+    """Back the library up to a bucket, lose the files and change the
+    database, restore from the bucket: the rows are the backup's and every
+    file is back at its path with its bytes."""
+    from backend.app.services import backup_destination
+    from backend.app.tasks import backup as tasks
+
+    images = tmp_path / "images"
+    library = {
+        "bea/one.jpg": b"first picture",
+        "bea/two.png": b"second picture, a png",
+        "attachments/ab/set.zip": b"an archive of pictures" * 50,
+    }
+    for rel, data in library.items():
+        _file(images, rel, data)
+    _image(db_sync, images / "bea/one.jpg", library["bea/one.jpg"])
+    _image(db_sync, images / "bea/two.png", library["bea/two.png"], mime="image/png")
+    db_sync.add(PostAttachment(sha256=_sha(library["attachments/ab/set.zip"]),
+                               path=str(images / "attachments/ab/set.zip"),
+                               original_filename="set.zip", ext=".zip",
+                               size_bytes=len(library["attachments/ab/set.zip"])))
+    db_sync.commit()
+
+    monkeypatch.setattr(tasks, "IMAGES_ROOT", images)
+    monkeypatch.setattr(backup_destination, "seal", lambda secret: f"sealed:{secret}")
+    queued = []
+    monkeypatch.setattr(tasks.restore_offsite_task, "delay", lambda **kw: queued.append(kw))
+    monkeypatch.setattr("backend.app.tasks.thumbnail.backfill_thumbnails.delay", lambda: None)
+
+    with FakeS3() as s3:
+        monkeypatch.setattr(backup_destination, "saved_target", lambda settings: s3.target())
+        dumped = tasks.backup_db_task(triggered_by="manual")
+        tasks.backup_media_task(triggered_by="manual")
+        dump_key = [k for k in s3.objects if k.startswith("database/")][0]
+        assert len([k for k in s3.objects if k.startswith("media/")]) == 3
+
+        # Lose the files, and change the database after the dump.
+        shutil.rmtree(images)
+        db_sync.execute(ImageRecord.__table__.delete().where(
+            ImageRecord.sha256 == _sha(library["bea/two.png"])))
+        db_sync.add(Artist(name="Added later", slug="added-later"))
+        db_sync.commit()
+        time.sleep(1.1)   # the safety dump gets its own second-resolution name
+
+        first = tasks.restore_offsite_task(dump_key=dump_key)
+        assert queued == [{"run_id": first["backup_run_id"]}]
+        done = tasks.restore_offsite_task(run_id=first["backup_run_id"])
+        assert done["continues"] is False
+        dumps_now = [k for k in s3.objects if k.startswith("database/")]
+
+    # The database is the dump's: a fresh session, since the restore ended
+    # every connection that was open.
+    with tasks._sync_session_factory()() as s:
+        assert s.execute(select(func.count()).select_from(ImageRecord)).scalar() == 2
+        assert s.execute(select(Artist).where(Artist.slug == "added-later")).first() is None
+        row = s.get(BackupRun, done["backup_run_id"])
+        assert row.status == "restored" and row.error is None
+        assert row.manifest["restored"] == 3
+        assert row.manifest["dump"]["key"] == dump_key
+        # The database it replaced was saved to the bucket first.
+        assert row.manifest["saved_before"] in dumps_now and len(dumps_now) == 2
+        assert s.get(BackupRun, dumped["backup_run_id"]).status == "ok"
+        settings = s.execute(select(tasks.ImportSettings)).scalars().first()
+        assert settings.backup_s3_secret_sealed == "sealed:SK"
+    for rel, data in library.items():
+        assert (images / rel).read_bytes() == data

tests/test_backup_s3.py

@@ -212,6 +212,41 @@ def test_a_refused_get_says_so_and_writes_nothing(tmp_path):
     assert not (tmp_path / "b.dump.partial").exists()
 
 
+def test_a_missing_object_is_named_as_missing(tmp_path):
+    body = b"<Error><Code>NoSuchKey</Code></Error>"
+    with _Server(lambda h: (404, body)) as srv, pytest.raises(s3.ObjectMissing):
+        s3.get_file(srv.target(), "media/ab/cd/x.jpg", tmp_path / "x.jpg")
+
+
+def test_a_fetched_file_is_kept_only_when_its_fingerprint_matches(tmp_path):
+    good = b"picture bytes"
+    sha = hashlib.sha256(good).hexdigest()
+    with _Server(lambda h: (200, good)) as srv:
+        s3.get_file(srv.target(), "k", tmp_path / "ok.jpg", sha256=sha, size=len(good))
+    assert (tmp_path / "ok.jpg").read_bytes() == good
+
+    with _Server(lambda h: (200, b"other bytes!!")) as srv, \
+            pytest.raises(s3.S3Error, match="different fingerprint"):
+        s3.get_file(srv.target(), "k", tmp_path / "bad.jpg", sha256=sha, size=len(good))
+    assert not (tmp_path / "bad.jpg").exists()
+    assert not (tmp_path / "bad.jpg.partial").exists()
+
+
+def test_a_short_download_is_not_kept(tmp_path):
+    with _Server(lambda h: (200, b"abc")) as srv, pytest.raises(s3.S3Error, match="3 bytes"):
+        s3.get_file(srv.target(), "k", tmp_path / "x", size=10)
+    assert not (tmp_path / "x").exists()
+
+
+def test_a_paced_download_counts_its_bytes(tmp_path):
+    pacer = s3.Pacer(None)
+    taken = []
+    pacer.take = taken.append
+    with _Server(lambda h: (200, b"z" * 2500)) as srv:
+        s3.get_file(srv.target(), "k", tmp_path / "z", pacer=pacer)
+    assert sum(taken) == 2500
+
+
 # ── Multipart ───────────────────────────────────────────────────────────────
 
 _STARTED = b"<InitiateMultipartUploadResult><UploadId>up1</UploadId></InitiateMultipartUploadResult>"

tests/test_backup_service.py

@@ -172,11 +172,33 @@ def test_restore_db_drops_schema_then_loads(tmp_path, fake_subprocess):
     assert fake_subprocess[0][0] == "psql"
     assert "-c" in fake_subprocess[0]
     assert "DROP SCHEMA IF EXISTS public CASCADE" in fake_subprocess[0][-1]
-    assert fake_subprocess[1][0] == "pg_restore"
+    assert "pg_terminate_backend" in fake_subprocess[0][-1]
+    assert "pg_restore" in fake_subprocess[1] and "--exit-on-error" in fake_subprocess[1]
     assert "-d" in fake_subprocess[1]
     assert str(dump_path) in fake_subprocess[1]
 
 
+def test_a_failed_restore_says_what_pg_restore_said(tmp_path, monkeypatch):
+    class _Done:
+        def __init__(self, code):
+            self.returncode = code
+
+    def run(cmd, stderr=None, **kw):
+        if "pg_restore" in cmd:
+            stderr.write(b"pg_restore: processing data\npg_restore: error: disk full\n")
+            return _Done(1)
+        return _Done(0)
+
+    monkeypatch.setattr("subprocess.run", run)
+    with pytest.raises(backup_service.RestoreFailed, match="error: disk full"):
+        backup_service.restore_db(db_url="postgresql://u@h/d", sql_path=tmp_path / "x.dump")
+
+
+def test_every_dump_leaves_out_account_logins(tmp_path, fake_subprocess):
+    backup_service.backup_db(db_url="postgresql://u@h/d", images_root=tmp_path)
+    assert "--exclude-table-data=credential" in fake_subprocess[0]
+
+
 # --- restore_images --------------------------------------------------