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>

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
-    it makes, so a backup leaves room on the link for everything else."""
+    """Holds the bytes a job moves to an average rate, across every request
+    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([
-                "pg_dump", "--no-owner", "--no-acl", "-Fc",
-                "-f", str(tmp_path), _libpq_url(db_url),
-            ]),
+            _gently([*_PG_DUMP, "-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:
-    """Wipe public schema, then load from the custom-format dump. Raises on
-    subprocess failure; partial-restore state is the caller's concern."""
-    libpq = _libpq_url(db_url)
-    subprocess.run(
+class RestoreFailed(Exception):
+    """psql or pg_restore exited with an error; the message is its last line."""
+
+
+# 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
+# other query behind it.
+_RESTORE_LOCK_TIMEOUT = "60s"
+# pg_restore of a large library (vector tables, rebuilt indexes) runs well past
+# 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, "-c",
-            "DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public;",
+            "psql", _libpq_url(db_url), "-v", "ON_ERROR_STOP=1", "-c",
+            f"{_END_OTHER_CONNECTIONS} DROP SCHEMA IF EXISTS public CASCADE; "
+            "CREATE SCHEMA public;",
         ],
-        capture_output=True, check=True, timeout=120,
+        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(
-        ["pg_restore", "--no-owner", "--no-acl", "-d", libpq, str(sql_path)],
-        capture_output=True, check=True,
-        timeout=_DB_SUBPROCESS_TIMEOUT_S,
+    _run_reporting(
+        _gently(["pg_restore", "--no-owner", "--no-acl", "--exit-on-error",
+                 "-d", _libpq_url(db_url), str(sql_path)]),
+        _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
-    is trimmed, to the latest MEDIA_RUNS_KEPT.
+    "restore_deleted": R, "files_unlinked": K}. Media and restore runs leave
+    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
-            # runs from its last progress, not from its start.
+            # A media run or an off-site restore spans many chunks and hours on
+            # 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),
                 )
             )