feat: restore a new install from the off-site bucket, database first, then every file
CI and images / lint (push) Successful in 2s
CI and images / extension-version (push) Successful in 1s
CI and images / extension-test (push) Successful in 17s
CI and images / frontend-build (push) Successful in 19s
CI and images / backend-lint-and-test (push) Successful in 45s
CI and images / integration (push) Failing after 2m26s
CI and images / sign-extension (push) Skipped
CI and images / build-web (push) Skipped
CI and images / smoke-web (push) Skipped
CI and images / promote (push) Skipped
CI and images / refresh-dev (push) Skipped
CI and images / unattended-failure (push) Skipped
CI and images / build-agent (push) Skipped

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>
This commit is contained in:
2026-11-07 14:34:47 -05:00
co-authored by Claude Opus 5.5
parent d465198c4a
commit 375e7e2eda
19 changed files with 1619 additions and 49 deletions
+2 -3
View File
@@ -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
# that can read your stored credentials; a database restored without it needs
# every credential re-entered by hand.
# Backups leave the key and the stored logins out on purpose: after restoring
# onto a new install, sign in to each account again.
CURATOR_BOOTSTRAP_NEW_KEY=1
+35 -3
View File
@@ -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
only thing that can read your stored credentials. A database restored without it
needs every credential entered again by hand.
**Your account logins are sealed with that key, and no backup carries either.**
FabledCurator's backups (below) leave the key and the logins out on purpose, so
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
+97 -1
View File
@@ -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:
+119
View File
@@ -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())
+191
View File
@@ -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
+34 -6
View File
@@ -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
+71 -19
View File
@@ -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(
[
"psql", libpq, "-c",
"DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public;",
],
capture_output=True, check=True, timeout=120,
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_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(
["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,
)
+6
View File
@@ -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)
+304 -7
View File
@@ -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()
+6 -4
View File
@@ -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),
)
)
@@ -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',
+163 -2
View File
@@ -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)
@@ -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) {
+34 -3
View File
@@ -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,
+110
View File
@@ -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")
+88
View File
@@ -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"]
+297
View File
@@ -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
+35
View File
@@ -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>"
+23 -1
View File
@@ -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 --------------------------------------------------