From 772482d79479f28e863b7a1f98900fb3c96e007c Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 6 Nov 2026 21:11:12 -0500 Subject: [PATCH] feat: with an off-site destination set up, database backups stream to it The nightly database backup (and "Run DB backup now") now sends pg_dump's output straight to the off-site bucket as a multipart upload, with no copy staged on this server. The object is named database/fabledcurator-.dump and appears only once pg_dump has exited cleanly and every part has landed; a failed dump aborts the upload and records pg_dump's last line. Without a destination, dumps are written to /images/_backups as before. - backup_s3: Upload (start / add part / complete / abort), with a busy or silent bucket rested and asked again (15 s, 60 s, Retry-After obeyed), and a Pacer that holds uploads to the new bandwidth limit. - Settings: "Upload limit (Mbit/s)" on the destination card, 200 by default, 0 for none (migration 0120). - pg_dump runs under nice (and ionice where installed), locally too. - Backups card says where dumps go; the runs table gains "Saved to" and shows a failed run's error. Off-site rows have no restore/tag/delete here: restoring from the bucket is a later step (#6880). Task #6878, milestone 586. Co-Authored-By: Claude Opus 5.5 --- alembic/versions/0120_backup_upload_limit.py | 30 +++ backend/app/api/system_backup.py | 20 +- backend/app/models/import_settings.py | 4 + backend/app/services/backup_destination.py | 1 + backend/app/services/backup_s3.py | 177 +++++++++++++++++- backend/app/services/backup_service.py | 82 +++++++- backend/app/tasks/backup.py | 47 +++-- .../src/components/settings/BackupCard.vue | 20 +- .../settings/BackupDestinationCard.vue | 20 +- .../components/settings/BackupRunsTable.vue | 18 +- tests/test_api_system_backup.py | 45 ++++- tests/test_backup_s3.py | 121 +++++++++++- tests/test_backup_service.py | 134 ++++++++++++- tests/test_tasks_backup.py | 59 +++++- 14 files changed, 749 insertions(+), 29 deletions(-) create mode 100644 alembic/versions/0120_backup_upload_limit.py diff --git a/alembic/versions/0120_backup_upload_limit.py b/alembic/versions/0120_backup_upload_limit.py new file mode 100644 index 0000000..395909d --- /dev/null +++ b/alembic/versions/0120_backup_upload_limit.py @@ -0,0 +1,30 @@ +"""import_settings: a bandwidth limit for uploads to the off-site backup bucket. + +In megabits per second; 0 means no limit. Every install starts at 200 Mbit/s, +which leaves room on a gigabit link for everything else the server does. + +Revision ID: 0120 +Revises: 0119 +Create Date: 2026-11-07 + +""" +import sqlalchemy as sa +from alembic import op + +revision = "0120" +down_revision = "0119" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "import_settings", + sa.Column( + "backup_upload_limit_mbps", sa.Integer(), nullable=False, server_default="200", + ), + ) + + +def downgrade() -> None: + op.drop_column("import_settings", "backup_upload_limit_mbps") diff --git a/backend/app/api/system_backup.py b/backend/app/api/system_backup.py index a2b84c9..2d96809 100644 --- a/backend/app/api/system_backup.py +++ b/backend/app/api/system_backup.py @@ -49,6 +49,9 @@ def _row_to_dict(r: BackupRun) -> dict: "error": r.error, "restored_from_id": r.restored_from_id, "manifest": r.manifest or {}, + # Where an off-site dump went ({endpoint, bucket, key}); None for a + # backup kept on this server. + "offsite": (r.manifest or {}).get("offsite"), } @@ -180,6 +183,12 @@ async def trigger_restore(run_id: int): "not_restorable", detail=f"source backup status={row.status!r}; only 'ok' rows are restorable", ) + if (row.manifest or {}).get("offsite"): + return _bad( + "not_restorable", + detail=("this dump is kept in the off-site bucket, not on this server, " + "so it can't be restored from here"), + ) expected = f"restore-{row.kind}-{row.id}" if supplied != expected: return _bad( @@ -250,7 +259,8 @@ async def get_destination(): @system_backup_bp.route("/destination", methods=["PATCH"]) async def patch_destination(): """Save the destination. Body {endpoint, bucket, access_key, region, - secret_key}; an empty field clears it. A missing or empty secret_key + secret_key, upload_limit_mbps}; an empty field clears it, and a missing + upload_limit_mbps keeps the saved limit. A missing or empty secret_key keeps the saved one, so the screen can save the other fields without the secret being typed again.""" from ..services import backup_destination as dest @@ -264,9 +274,17 @@ async def patch_destination(): secret = body.get("secret_key") if secret is not None and not isinstance(secret, str): return _bad("invalid_destination", detail="secret_key must be text") + limit = body.get("upload_limit_mbps") + if limit is not None and ( + not isinstance(limit, int) or isinstance(limit, bool) or not 0 <= limit <= 100_000 + ): + return _bad("invalid_destination", + detail="upload_limit_mbps must be a whole number from 0 (no limit) up") async with get_session() as session: row = await ImportSettings.load(session) + if limit is not None: + row.backup_upload_limit_mbps = limit row.backup_s3_endpoint = fields.endpoint row.backup_s3_bucket = fields.bucket row.backup_s3_access_key = fields.access_key diff --git a/backend/app/models/import_settings.py b/backend/app/models/import_settings.py index 2f3efea..434a7d8 100644 --- a/backend/app/models/import_settings.py +++ b/backend/app/models/import_settings.py @@ -111,6 +111,10 @@ class ImportSettings(Base): String(64), nullable=False, default="us-east-1", server_default="us-east-1", ) + # Upload bandwidth to that bucket, in megabits per second; 0 = no limit. + backup_upload_limit_mbps: Mapped[int] = mapped_column( + Integer, nullable=False, default=200, server_default="200", + ) # FC-6.3 series continuation matcher. enabled gates the rescan; threshold is # the weighted-score cut-off (0..1) above which a pending suggestion is made. diff --git a/backend/app/services/backup_destination.py b/backend/app/services/backup_destination.py index fb4d261..9c39162 100644 --- a/backend/app/services/backup_destination.py +++ b/backend/app/services/backup_destination.py @@ -114,6 +114,7 @@ def describe(settings) -> dict: "access_key": f.access_key or "", "region": f.region, "secret_set": bool(settings.backup_s3_secret_sealed), + "upload_limit_mbps": settings.backup_upload_limit_mbps, "configured": bool( f.endpoint and f.bucket and f.access_key and settings.backup_s3_secret_sealed ), diff --git a/backend/app/services/backup_s3.py b/backend/app/services/backup_s3.py index 7ed2bb4..ce09aaa 100644 --- a/backend/app/services/backup_s3.py +++ b/backend/app/services/backup_s3.py @@ -1,6 +1,7 @@ """A small S3 client for sending backups off the box: AWS Signature Version 4, path-style addressing (endpoint/bucket/key), and the calls a backup needs — -put an object, read one back, list a prefix. +put an object (whole, or in parts when its size isn't known up front), read +one back, list a prefix. FabledCurator puts, lists and reads; it never deletes. The bucket keeps the history and thins it on its own policy, so a key that may only add and read @@ -21,7 +22,9 @@ import http.client import os import socket import ssl +import time import xml.etree.ElementTree as ET +from collections.abc import Iterator from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path @@ -280,3 +283,175 @@ def list_objects(target: S3Target, prefix: str = "", limit: int = 1000, *, objects.append(StoredObject(fields.get("Key", ""), int(fields.get("Size") or 0), fields.get("LastModified", ""))) return ListPage(objects, next_token) + + +# ── Multipart: one object sent in parts ───────────────────────────────────── + +# Each part is held in memory while it is sent, so this bounds the memory a +# streamed upload uses. S3 allows 10,000 parts, which at this size is 640 GiB +# for one object, and every part but the last must be at least 5 MiB. +PART_SIZE = 64 * CHUNK +# A part the bucket couldn't take (it didn't answer, or said it was busy) is +# sent again after a rest that grows each time: 15 s, then 60 s. +PART_RETRY_WAITS = (15, 60) +_BUSY = frozenset({429, 500, 502, 503, 504}) +_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.""" + + 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 chunks(self, data: bytes) -> Iterator[memoryview]: + view = memoryview(data) + for at in range(0, len(view), CHUNK): + piece = view[at:at + CHUNK] + if self.rate: + self.sent += len(piece) + ahead = self.sent / self.rate - (time.monotonic() - self.started) + if ahead > 0: + time.sleep(ahead) + yield piece + + +def _send(target: S3Target, method: str, url: str, body: bytes = b"", + headers: dict[str, str] | None = None, *, timeout: float = TIMEOUT_SECONDS, + pacer: Pacer | None = None, + now: datetime | None = None) -> tuple[int, str, http.client.HTTPMessage, bytes]: + """Sign and send one request with an in-memory body, paced when a pacer + is given. Returns (status, reason, headers, content).""" + digest = hashlib.sha256(body).hexdigest() + signed = sign(method, url, {"Content-Length": str(len(body)), **(headers or {})}, + digest, target.access_key, target.secret_key, target.region, + now or datetime.now(UTC)) + conn = _connect(target, url, timeout) + try: + conn.request(method, _request_target(url), + body=pacer.chunks(body) if pacer and body else body, headers=signed) + response = conn.getresponse() + content = response.read() + except (OSError, http.client.HTTPException) as exc: + raise _reach_error(target, exc) from None + finally: + conn.close() + return response.status, response.reason, response.headers, content + + +def _xml_field(content: bytes, name: str) -> str | None: + try: + root = ET.fromstring(content) + except ET.ParseError: + return None + for item in root.iter(): + if item.tag.rsplit("}", 1)[-1] == name: + return item.text + return None + + +def _is_error_body(content: bytes) -> bool: + try: + return bool(content) and ET.fromstring(content).tag.rsplit("}", 1)[-1] == "Error" + except ET.ParseError: + return False + + +def _retry_after(headers: http.client.HTTPMessage | None, default: int) -> int: + try: + asked = int((headers or {}).get("Retry-After") or default) + except ValueError: + asked = default + return max(0, min(asked, _MAX_RETRY_AFTER)) + + +class Upload: + """One object written as a series of parts, for data whose size isn't known + when it starts, such as a database dump streamed from pg_dump. + + The object appears in the bucket only when `complete()` succeeds. Until + then the parts are invisible, so a stream that fails partway leaves + nothing behind that looks like a finished backup. `abort()` asks the + bucket to drop the parts at once; if the abort is refused or never + arrives, S3 services expire an unfinished upload on their own (Fabled + Archives after a day).""" + + def __init__(self, target: S3Target, key: str, *, + content_type: str = "application/octet-stream", + pacer: Pacer | None = None): + self.target = target + self.key = key + self.size = 0 + self.pacer = pacer + self._etags: list[str] = [] + status, reason, _, content = _send( + target, "POST", f"{target.url(key)}?uploads", headers={"Content-Type": content_type}, + ) + upload_id = _xml_field(content, "UploadId") if status == 200 else None + if not upload_id: + raise S3Error(f"Starting the upload of {key} refused: " + f"{_says(status, reason, content)}") + self.upload_id = upload_id + + def _url(self, **params: str) -> str: + query = urlencode({**params, "uploadId": self.upload_id}, quote_via=quote) + return f"{self.target.url(self.key)}?{query}" + + @property + def parts(self) -> int: + return len(self._etags) + + def add(self, data: bytes) -> None: + """Send the next part, resting and sending it again when the bucket + doesn't answer or says it is busy.""" + n = len(self._etags) + 1 + for wait in (*PART_RETRY_WAITS, None): + headers = None + try: + status, reason, headers, content = _send( + self.target, "PUT", self._url(partNumber=str(n)), data, + timeout=TRANSFER_TIMEOUT_SECONDS, pacer=self.pacer, + ) + except S3Error as exc: + if wait is None: + raise S3Error(f"Part {n} of {self.key}: {exc}") from None + time.sleep(wait) + continue + etag = headers.get("ETag") if status == 200 else None + if etag: + self._etags.append(etag) + self.size += len(data) + return + if status not in _BUSY or wait is None: + raise S3Error(f"Part {n} of {self.key} refused: " + f"{_says(status, reason, content)}") + time.sleep(_retry_after(headers, wait)) + + def complete(self) -> None: + """Join the parts into the object. S3 can refuse a completion with an + error body under a 200 status, so the body is read too.""" + if not self._etags: + raise S3Error(f"Nothing was sent for {self.key}.") + body = "" + "".join( + f"{n}{etag}" + for n, etag in enumerate(self._etags, start=1) + ) + "" + status, reason, _, content = _send( + self.target, "POST", self._url(), body.encode(), + headers={"Content-Type": "application/xml"}, timeout=TRANSFER_TIMEOUT_SECONDS, + ) + if status != 200 or _is_error_body(content): + raise S3Error(f"Finishing the upload of {self.key} refused: " + f"{_says(status, reason, content)}") + + def abort(self) -> None: + """Drop the parts sent so far. Never raises: it runs while another + failure is already being reported, and the bucket expires the parts on + its own if this doesn't reach it.""" + try: + _send(self.target, "DELETE", self._url()) + except Exception: # noqa: BLE001 — best effort, see above + pass diff --git a/backend/app/services/backup_service.py b/backend/app/services/backup_service.py index 3c1bced..0fcb16f 100644 --- a/backend/app/services/backup_service.py +++ b/backend/app/services/backup_service.py @@ -4,7 +4,9 @@ Two independent backup kinds: - 'db' — pg_dump only; fast; nightly via Beat (settings-gated) - 'images' — tar+zstd of /images; slow; manual trigger only -Files live under /_backups/. Each backup writes: +With an off-site destination set up, a database dump streams straight to that +bucket instead (backup_db_offsite). Otherwise files live under +/_backups/. Each local backup writes: fc__.{sql|tar.zst} — the artifact fc__.json — manifest (kind/tag/triggered_by) @@ -19,6 +21,7 @@ import os import shutil import subprocess import tempfile +import time from datetime import UTC, datetime from pathlib import Path @@ -153,10 +156,10 @@ 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), - ], + ]), _DB_SUBPROCESS_TIMEOUT_S, ) shutil.move(str(tmp_path), str(sql_path)) @@ -177,6 +180,79 @@ def backup_db( } +class DumpFailed(Exception): + """pg_dump exited with an error; the message is its last line of output.""" + + +def offsite_dump_key(ts: str) -> str: + """Where a database dump lands in the off-site bucket. Dated names sit + side by side under one prefix, which is what lets the bucket's retention + thin them by age.""" + return f"database/fabledcurator-{ts}.dump" + + +def _gently(cmd: list[str]) -> list[str]: + """`cmd` at low CPU priority, and low disk priority where `ionice` is + installed, so a long dump leaves the server responsive.""" + io = ["ionice", "-c", "2", "-n", "7"] if shutil.which("ionice") else [] + return ["nice", "-n", "10", *io, *cmd] + + +def _last_line(f) -> str | None: + f.seek(0) + lines = f.read().decode(errors="replace").strip().splitlines() + return lines[-1].strip() if lines else None + + +def backup_db_offsite(*, db_url: str, target, limit_mbps: int | None = None) -> dict: + """Stream `pg_dump` straight into the off-site bucket, in parts, with no + copy staged on this server. The object appears only once pg_dump has + exited cleanly and every part has arrived; a dump that fails partway + leaves nothing in the bucket, and raises DumpFailed with pg_dump's own + last line. Returns a dict for the caller to record in BackupRun.""" + from .backup_s3 import PART_SIZE, Pacer, Upload + + ts = _now_ts() + key = offsite_dump_key(ts) + deadline = time.monotonic() + _DB_SUBPROCESS_TIMEOUT_S + with tempfile.TemporaryFile() as err: + proc = subprocess.Popen( + _gently(["pg_dump", "--no-owner", "--no-acl", "-Fc", _libpq_url(db_url)]), + stdout=subprocess.PIPE, stderr=err, + ) + upload = None + try: + upload = Upload(target, key, pacer=Pacer(limit_mbps)) + while part := proc.stdout.read(PART_SIZE): + if time.monotonic() > deadline: + raise subprocess.TimeoutExpired(proc.args, _DB_SUBPROCESS_TIMEOUT_S) + upload.add(part) + proc.stdout.close() + proc.wait(timeout=_KILL_REAP_GRACE_S * 6) + if proc.returncode != 0: + raise DumpFailed(_last_line(err) or f"pg_dump exited with {proc.returncode}") + upload.complete() + except BaseException: + if proc.poll() is None: + proc.kill() + try: + proc.wait(timeout=_KILL_REAP_GRACE_S) + except subprocess.TimeoutExpired: + pass # unkillable (D-state): abandon the reap, fail fast + if upload is not None: + upload.abort() + raise + return { + "kind": "db", + "ts": ts, + "key": key, + "bucket": target.bucket, + "endpoint": target.endpoint, + "parts": upload.parts, + "size_bytes": upload.size, + } + + def backup_images( *, images_root: Path, tag: str | None = None, triggered_by: str = "manual", diff --git a/backend/app/tasks/backup.py b/backend/app/tasks/backup.py index 77ad1f0..3ca675c 100644 --- a/backend/app/tasks/backup.py +++ b/backend/app/tasks/backup.py @@ -18,7 +18,7 @@ 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_service +from ..services import backup_destination, backup_service from ._sync_engine import sync_session_factory as _sync_session_factory log = logging.getLogger(__name__) @@ -49,7 +49,12 @@ def _mark_failed(session, row: BackupRun, exc: BaseException) -> None: ) def backup_db_task(self, *, tag: str | None = None, triggered_by: str = "manual") -> dict: - """Create one DB backup. Returns {'backup_run_id': N}.""" + """Create one DB backup. Returns {'backup_run_id': N}. + + With an off-site destination set up, the dump streams to that bucket and + nothing is written here; the row's manifest says where it went + (`offsite`: endpoint, bucket, key). Without one, the dump is written to + /images/_backups.""" SessionLocal = _sync_session_factory() cfg = get_config() now = datetime.now(UTC) @@ -64,10 +69,19 @@ def backup_db_task(self, *, tag: str | None = None, run_id = row.id try: - result = backup_service.backup_db( - db_url=cfg.database_url_sync, images_root=IMAGES_ROOT, - tag=tag, triggered_by=triggered_by, - ) + with SessionLocal() as session: + settings = ImportSettings.load_sync(session) + target = backup_destination.saved_target(settings) + limit_mbps = settings.backup_upload_limit_mbps + if target is not None: + result = backup_service.backup_db_offsite( + db_url=cfg.database_url_sync, target=target, limit_mbps=limit_mbps, + ) + else: + result = backup_service.backup_db( + db_url=cfg.database_url_sync, images_root=IMAGES_ROOT, + tag=tag, triggered_by=triggered_by, + ) except (SoftTimeLimitExceeded, Exception) as exc: with SessionLocal() as session: row = session.get(BackupRun, run_id) @@ -79,12 +93,23 @@ def backup_db_task(self, *, tag: str | None = None, row = session.get(BackupRun, run_id) row.status = "ok" row.finished_at = datetime.now(UTC) - row.sql_path = result["sql_path"] row.size_bytes = result["size_bytes"] - row.manifest = { - "manifest_path": result["manifest_path"], - "ts": result["ts"], - } + if target is not None: + row.manifest = { + "ts": result["ts"], + "offsite": { + "endpoint": result["endpoint"], + "bucket": result["bucket"], + "key": result["key"], + "parts": result["parts"], + }, + } + else: + row.sql_path = result["sql_path"] + row.manifest = { + "manifest_path": result["manifest_path"], + "ts": result["ts"], + } session.commit() return {"backup_run_id": run_id} diff --git a/frontend/src/components/settings/BackupCard.vue b/frontend/src/components/settings/BackupCard.vue index 0e2783e..be8f6dd 100644 --- a/frontend/src/components/settings/BackupCard.vue +++ b/frontend/src/components/settings/BackupCard.vue @@ -4,11 +4,18 @@ title="Backups" blurb="Database + image backups, schedules and restore." > -

+

+ Database dumps stream to the {{ offsite.bucket }} bucket + at {{ offsite.endpoint }}, with no copy kept on this server; the bucket's + own retention thins old dumps. The images backup below still writes a + tar+zstd to /images/_backups/, by hand only. +

+

pg_dump for the database (fast, nightly-schedulable); tar+zstd for the images (slow, manual only). Files live in /images/_backups/. Tag a backup to protect it - from autoprune. + from autoprune. Set up an off-site destination above to send + database dumps off this server instead.

+

+ Keep last applies to this list and to any dumps still on this server. +

store.settings || { backup_images_keep_last_n: 3, }) +// The off-site destination, once it is fully set up: database dumps go there. +const offsite = computed(() => ( + store.destination?.configured ? store.destination : null +)) + const dbTriggering = ref(false) const imagesTriggering = ref(false) const confirmOpen = ref(false) @@ -127,6 +142,7 @@ function pollOnce() { onMounted(() => { store.loadSettings() + if (!store.destination) store.loadDestination().catch(() => {}) pollOnce() pollId = setInterval(pollOnce, 5000) }) diff --git a/frontend/src/components/settings/BackupDestinationCard.vue b/frontend/src/components/settings/BackupDestinationCard.vue index 8b1a13d..acb9dd0 100644 --- a/frontend/src/components/settings/BackupDestinationCard.vue +++ b/frontend/src/components/settings/BackupDestinationCard.vue @@ -51,6 +51,12 @@ :hint="secretSet ? 'A saved secret is never shown again.' : ''" persistent-hint class="mb-3" :disabled="busy" /> +
No destination is set up, so nothing is sent off this server yet.

+

+ Database backups go to this bucket. Set when they run under Backups below. +

@@ -106,7 +115,11 @@ import MaintenanceTile from '../common/MaintenanceTile.vue' const store = useBackupStore() const FIELDS = ['endpoint', 'bucket', 'access_key', 'region'] -const form = reactive({ endpoint: '', bucket: '', access_key: '', region: 'us-east-1', secret_key: '' }) +const DEFAULT_LIMIT = 200 +const form = reactive({ + endpoint: '', bucket: '', access_key: '', region: 'us-east-1', secret_key: '', + upload_limit_mbps: DEFAULT_LIMIT, +}) const saved = ref(null) const loadError = ref('') const saving = ref(false) @@ -122,11 +135,13 @@ const dirty = computed(() => ( !saved.value || form.secret_key.trim() !== '' || FIELDS.some((f) => (form[f] || '').trim() !== (saved.value[f] || '')) + || form.upload_limit_mbps !== saved.value.upload_limit_mbps )) function apply(dest) { saved.value = dest for (const f of FIELDS) form[f] = dest[f] || (f === 'region' ? 'us-east-1' : '') + form.upload_limit_mbps = dest.upload_limit_mbps ?? DEFAULT_LIMIT form.secret_key = '' } @@ -134,6 +149,8 @@ function fields() { const out = {} for (const f of FIELDS) out[f] = form[f] if (form.secret_key.trim()) out.secret_key = form.secret_key + const limit = Number(form.upload_limit_mbps) + if (Number.isInteger(limit) && limit >= 0) out.upload_limit_mbps = limit return out } @@ -195,4 +212,5 @@ async function onRemove() { .fc-dest-row { display: flex; gap: 12px; flex-wrap: wrap; } .fc-dest-row > :first-child { flex: 1 1 200px; } .fc-dest-region { flex: 0 1 160px; } +.fc-dest-limit { max-width: 260px; } diff --git a/frontend/src/components/settings/BackupRunsTable.vue b/frontend/src/components/settings/BackupRunsTable.vue index 14b4e5c..8d778d4 100644 --- a/frontend/src/components/settings/BackupRunsTable.vue +++ b/frontend/src/components/settings/BackupRunsTable.vue @@ -4,6 +4,7 @@ When Status + Saved to Duration Size Tag @@ -20,6 +21,15 @@ {{ statusIcon(r.status) }} {{ r.status }} +
{{ r.error }}
+ + + + This server {{ formatDuration(r) }} {{ formatBytes(r.size_bytes) }} @@ -30,7 +40,8 @@ — - + — + - No backups yet. + No backups yet. @@ -97,4 +108,7 @@ function formatRelative(iso) { diff --git a/tests/test_api_system_backup.py b/tests/test_api_system_backup.py index d0d82a4..1d87a80 100644 --- a/tests/test_api_system_backup.py +++ b/tests/test_api_system_backup.py @@ -232,6 +232,27 @@ async def test_restore_rejects_non_ok_status(client, db): assert body["error"] == "not_restorable" +@pytest.mark.asyncio +async def test_an_off_site_dump_is_not_restored_from_this_screen(client, db): + db.add(BackupRun( + kind="db", status="ok", tag=None, triggered_by="nightly", + started_at=datetime.now(UTC), finished_at=datetime.now(UTC), + manifest={"offsite": {"endpoint": "http://archives.example:8081", + "bucket": "curator", "key": "database/x.dump"}}, + )) + await db.commit() + run = (await (await client.get("/api/system/backup/runs?kind=db&limit=1")).get_json())[ + "runs"][0] + assert run["offsite"]["bucket"] == "curator" + + resp = await client.post( + f"/api/system/backup/runs/{run['id']}/restore", + json={"confirm": f"restore-db-{run['id']}"}, + ) + assert resp.status_code == 400 + assert (await resp.get_json())["error"] == "not_restorable" + + # --- DELETE /runs/ ---------------------------------------------- @@ -352,7 +373,7 @@ async def test_destination_saves_and_never_returns_the_secret(client, db_sync, _ assert body == { "endpoint": "http://archives.example:8081", "bucket": "curator", "access_key": "AKEXAMPLE", "region": "us-east-1", - "secret_set": True, "configured": True, + "secret_set": True, "configured": True, "upload_limit_mbps": 200, } assert "s3cr3t-example" not in str(await (await client.get( "/api/system/backup/destination")).get_json()) @@ -398,7 +419,7 @@ async def test_destination_remove_forgets_everything(client, _sealing_key): resp = await client.delete("/api/system/backup/destination") assert await resp.get_json() == { "endpoint": "", "bucket": "", "access_key": "", "region": "us-east-1", - "secret_set": False, "configured": False, + "secret_set": False, "configured": False, "upload_limit_mbps": 200, } @@ -455,3 +476,23 @@ async def test_destination_test_reports_an_unreachable_bucket_as_a_result( "ok": False, "message": "Couldn't reach http://archives.example:8081: nothing is listening there.", } + + +@pytest.mark.asyncio +async def test_the_upload_limit_saves_and_a_missing_one_keeps_it(client, _sealing_key): + resp = await client.patch( + "/api/system/backup/destination", json={**_DEST, "upload_limit_mbps": 50}, + ) + assert (await resp.get_json())["upload_limit_mbps"] == 50 + resp = await client.patch("/api/system/backup/destination", json=_DEST) + assert (await resp.get_json())["upload_limit_mbps"] == 50 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("limit", [-1, 2.5, "fast", True]) +async def test_an_upload_limit_that_isnt_a_whole_number_is_refused(client, _sealing_key, limit): + resp = await client.patch( + "/api/system/backup/destination", json={**_DEST, "upload_limit_mbps": limit}, + ) + assert resp.status_code == 400 + assert "upload_limit_mbps" in (await resp.get_json())["detail"] diff --git a/tests/test_backup_s3.py b/tests/test_backup_s3.py index d62ccfd..ba1107d 100644 --- a/tests/test_backup_s3.py +++ b/tests/test_backup_s3.py @@ -1,6 +1,7 @@ """The backup destination's S3 client (services/backup_s3.py): signatures match AWS's published Signature Version 4 examples, and the calls send and read what S3 expects, over a real HTTP connection to a local stand-in server.""" +import hashlib import socket import threading from datetime import UTC, datetime @@ -55,7 +56,7 @@ def test_paths_are_path_style_and_keys_are_escaped(): class _Server: """A local HTTP server that records each request and answers with - `reply(handler) -> (status, body)`.""" + `reply(handler) -> (status, body)` or `(status, body, headers)`.""" def __init__(self, reply): self.seen = [] @@ -69,13 +70,15 @@ class _Server: "headers": {k.lower(): v for k, v in self.headers.items()}, "body": self.rfile.read(length) if length else b"", }) - status, body = reply(self) + status, body, *extra = reply(self) self.send_response(status) + for name, value in (extra[0] if extra else {}).items(): + self.send_header(name, value) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) - do_GET = do_PUT = _serve + do_GET = do_PUT = do_POST = do_DELETE = _serve def log_message(self, *args): pass @@ -207,3 +210,115 @@ def test_a_refused_get_says_so_and_writes_nothing(tmp_path): s3.get_file(srv.target(), "b.dump", out) assert not out.exists() assert not (tmp_path / "b.dump.partial").exists() + + +# ── Multipart ─────────────────────────────────────────────────────────────── + +_STARTED = b"up1" + + +def _bucket(parts_reply=None, complete=(200, b"")): + """A stand-in bucket for one multipart upload: starts it, takes parts (each + answered by `parts_reply(n)`, an ETag by default) and completes it.""" + def reply(h): + if h.command == "POST" and h.path.endswith("?uploads"): + return 200, _STARTED + if h.command == "PUT": + n = int(h.path.split("partNumber=")[1].split("&")[0]) + return parts_reply(n) if parts_reply else (200, b"", {"ETag": f'"e{n}"'}) + if h.command == "POST": + return complete + return 204, b"" + return reply + + +def test_an_upload_sends_its_parts_then_joins_them_in_order(): + with _Server(_bucket()) as srv: + up = s3.Upload(srv.target(), "database/x.dump") + up.add(b"first,") + up.add(b"second") + up.complete() + start, one, two, done = srv.seen + assert (start["method"], start["path"]) == ("POST", "/curator/database/x.dump?uploads") + assert one["path"] == "/curator/database/x.dump?partNumber=1&uploadId=up1" + assert two["body"] == b"second" + assert two["headers"]["x-amz-content-sha256"] == hashlib.sha256(b"second").hexdigest() + assert done["path"] == "/curator/database/x.dump?uploadId=up1" + assert done["body"].index(b'"e1"') < done["body"].index(b'"e2"') + assert (up.parts, up.size) == (2, len(b"first,second")) + + +def test_a_refused_start_says_why(): + with _Server(lambda h: (403, b"AccessDenied")) as srv: + with pytest.raises(s3.S3Error, match="the key isn't allowed"): + s3.Upload(srv.target(), "database/x.dump") + + +def test_a_completion_refused_under_a_200_is_a_failure(): + error = (200, b"InternalErrordisk full") + with _Server(_bucket(complete=error)) as srv: + up = s3.Upload(srv.target(), "database/x.dump") + up.add(b"data") + with pytest.raises(s3.S3Error, match="Finishing the upload"): + up.complete() + + +def test_nothing_sent_is_not_completed(): + with _Server(_bucket()) as srv: + up = s3.Upload(srv.target(), "database/x.dump") + with pytest.raises(s3.S3Error, match="Nothing was sent"): + up.complete() + + +def test_a_busy_bucket_gets_the_part_again_after_a_rest(monkeypatch): + rests = [] + monkeypatch.setattr(s3.time, "sleep", rests.append) + answers = iter([(503, b"", {"Retry-After": "7"}), (200, b"", {"ETag": '"e1"'})]) + with _Server(_bucket(parts_reply=lambda n: next(answers))) as srv: + up = s3.Upload(srv.target(), "database/x.dump") + up.add(b"data") + assert rests == [7] + assert up.parts == 1 + + +def test_a_part_refused_outright_is_not_sent_again(monkeypatch): + monkeypatch.setattr(s3.time, "sleep", lambda s: pytest.fail("rested on a refusal")) + refused = (403, b"AccessDenied") + with _Server(_bucket(parts_reply=lambda n: refused)) as srv: + up = s3.Upload(srv.target(), "database/x.dump") + with pytest.raises(s3.S3Error, match="Part 1"): + up.add(b"data") + + +def test_a_part_that_never_lands_gives_up_after_the_rests(monkeypatch): + rests = [] + monkeypatch.setattr(s3.time, "sleep", rests.append) + with _Server(_bucket(parts_reply=lambda n: (503, b""))) as srv: + up = s3.Upload(srv.target(), "database/x.dump") + with pytest.raises(s3.S3Error, match="Part 1"): + up.add(b"data") + assert rests == list(s3.PART_RETRY_WAITS) + + +def test_abort_asks_the_bucket_to_drop_the_parts_and_never_raises(): + with _Server(_bucket()) as srv: + up = s3.Upload(srv.target(), "database/x.dump") + up.abort() + assert srv.seen[-1]["method"] == "DELETE" + assert srv.seen[-1]["path"] == "/curator/database/x.dump?uploadId=up1" + up.abort() # the server is gone now: still no exception + + +def test_the_pacer_holds_bytes_to_the_limit(monkeypatch): + clock = [0.0] + monkeypatch.setattr(s3.time, "monotonic", lambda: clock[0]) + monkeypatch.setattr(s3.time, "sleep", lambda secs: clock.__setitem__(0, clock[0] + secs)) + pacer = s3.Pacer(8) # 8 Mbit/s = 1,000,000 bytes/s + data = b"x" * (3 * s3.CHUNK) + assert b"".join(pacer.chunks(data)) == data + assert clock[0] == pytest.approx(len(data) / 1_000_000) + + +def test_no_limit_never_waits(monkeypatch): + monkeypatch.setattr(s3.time, "sleep", lambda secs: pytest.fail("waited without a limit")) + assert b"".join(s3.Pacer(0).chunks(b"y" * (2 * s3.CHUNK + 1))) == b"y" * (2 * s3.CHUNK + 1) diff --git a/tests/test_backup_service.py b/tests/test_backup_service.py index 2c2d323..48bfabc 100644 --- a/tests/test_backup_service.py +++ b/tests/test_backup_service.py @@ -5,6 +5,7 @@ without external binaries. The real subprocess behavior is exercised implicitly via the Celery task tests in test_tasks_backup.py. """ import json +import re import subprocess from pathlib import Path @@ -28,7 +29,7 @@ def fake_subprocess(monkeypatch): stderr = b"" def _write_sentinel(cmd): - if cmd[0] == "pg_dump": + if "pg_dump" in cmd: i = cmd.index("-f") Path(cmd[i + 1]).write_bytes(b"-- fake pg_dump\n") elif cmd[0] == "tar" and "-cf" in cmd: @@ -87,7 +88,7 @@ def test_backup_db_strips_sqlalchemy_psycopg_driver(tmp_path, fake_subprocess): db_url="postgresql+psycopg://u:p@h/d", images_root=tmp_path, ) cmd = fake_subprocess[0] - assert cmd[0] == "pg_dump" + assert "pg_dump" in cmd assert cmd[-1].startswith("postgresql://") assert "+psycopg" not in cmd[-1] @@ -284,3 +285,132 @@ def test_run_bounded_fails_fast_when_unkillable(monkeypatch): with pytest.raises(subprocess.TimeoutExpired): backup_service._run_bounded(["pg_dump"], 1) assert killed["n"] == 1 + + +# --- backup_db_offsite: pg_dump streamed to the bucket --------------- + + +class _Dump: + """A pg_dump stand-in: `output` on stdout, `said` on stderr, exiting with + `code` once its output has been read.""" + + def __init__(self, output=b"", said=b"", code=0): + self.output, self.said, self.code = output, said, code + self.cmd = None + self.killed = False + + def __call__(self, cmd, stdout=None, stderr=None, **kwargs): + import io + self.cmd = list(cmd) + self.args = cmd + self.stdout = io.BytesIO(self.output) + stderr.write(self.said) + self.returncode = None + return self + + def poll(self): + return self.returncode + + def wait(self, timeout=None): + if self.returncode is None: + self.returncode = -9 if self.killed else self.code + return self.returncode + + def kill(self): + self.killed = True + + +class _Bucket: + """Records what an Upload was asked to do instead of sending it.""" + + def __init__(self, fail_on_part=None): + self.fail_on_part = fail_on_part + self.uploads = [] + + def __call__(self, target, key, **kwargs): + from backend.app.services.backup_s3 import S3Error + bucket = self + + class _Upload: + def __init__(self): + self.key, self.received = key, [] + self.completed = self.aborted = False + bucket.uploads.append(self) + + @property + def parts(self): + return len(self.received) + + @property + def size(self): + return sum(len(p) for p in self.received) + + def add(self, data): + if bucket.fail_on_part == len(self.received) + 1: + raise S3Error("HTTP 503: busy.") + self.received.append(data) + + def complete(self): + self.completed = True + + def abort(self): + self.aborted = True + + return _Upload() + + +@pytest.fixture +def offsite(monkeypatch): + from backend.app.services import backup_s3 + monkeypatch.setattr(backup_s3, "PART_SIZE", 4) + bucket = _Bucket() + monkeypatch.setattr(backup_s3, "Upload", bucket) + target = backup_s3.S3Target("http://archives.example:8081", "curator", "AK", "SK") + return bucket, target + + +def test_a_dump_streams_to_the_bucket_in_parts_with_nothing_kept_here(offsite, monkeypatch): + bucket, target = offsite + dump = _Dump(output=b"0123456789") + monkeypatch.setattr("subprocess.Popen", dump) + result = backup_service.backup_db_offsite( + db_url="postgresql+psycopg://u:p@h/d", target=target, limit_mbps=50, + ) + (upload,) = bucket.uploads + assert upload.received == [b"0123", b"4567", b"89"] + assert upload.completed and not upload.aborted + assert re.fullmatch(r"database/fabledcurator-\d{8}T\d{6}Z\.dump", result["key"]) + assert result["key"] == upload.key + assert (result["bucket"], result["size_bytes"], result["parts"]) == ("curator", 10, 3) + assert dump.cmd[0] == "nice" and "pg_dump" in dump.cmd and "-Fc" in dump.cmd + assert "-f" not in dump.cmd # stdout, not a file + assert dump.cmd[-1] == "postgresql://u:p@h/d" + + +def test_a_failed_dump_leaves_nothing_in_the_bucket_and_says_why(offsite, monkeypatch): + bucket, target = offsite + said = b"pg_dump: dumping contents of table x\npg_dump: error: connection lost\n" + monkeypatch.setattr("subprocess.Popen", _Dump(output=b"012345", said=said, code=1)) + with pytest.raises(backup_service.DumpFailed, match="^pg_dump: error: connection lost$"): + backup_service.backup_db_offsite(db_url="postgresql://u@h/d", target=target) + (upload,) = bucket.uploads + assert upload.aborted and not upload.completed + + +def test_a_bucket_that_stops_taking_parts_stops_the_dump(offsite, monkeypatch): + from backend.app.services.backup_s3 import S3Error + bucket, target = offsite + bucket.fail_on_part = 2 + dump = _Dump(output=b"0123456789") + monkeypatch.setattr("subprocess.Popen", dump) + with pytest.raises(S3Error): + backup_service.backup_db_offsite(db_url="postgresql://u@h/d", target=target) + assert dump.killed + (upload,) = bucket.uploads + assert upload.aborted and not upload.completed + + +def test_dump_names_sort_by_time(): + assert backup_service.offsite_dump_key("20261107T030000Z") == ( + "database/fabledcurator-20261107T030000Z.dump" + ) diff --git a/tests/test_tasks_backup.py b/tests/test_tasks_backup.py index c179b41..0557a52 100644 --- a/tests/test_tasks_backup.py +++ b/tests/test_tasks_backup.py @@ -33,7 +33,7 @@ def fake_subprocess_and_images_root(monkeypatch, tmp_path): stderr = b"" def _sentinel(cmd): - if cmd[0] == "pg_dump": + if "pg_dump" in cmd: i = cmd.index("-f") Path(cmd[i + 1]).write_bytes(b"-- fake pg_dump\n") elif cmd[0] == "tar" and "-cf" in cmd: @@ -128,6 +128,63 @@ async def test_backup_db_task_persists_tag(db_sync): assert tag == "pre-cutover" +@pytest.fixture +def archives(monkeypatch): + """An off-site destination is set up; the dump itself is faked.""" + from backend.app.services.backup_s3 import S3Target + target = S3Target("http://archives.example:8081", "curator", "AK", "SK") + monkeypatch.setattr( + "backend.app.services.backup_destination.saved_target", lambda settings: target, + ) + sent = [] + + def _offsite(*, db_url, target, limit_mbps): + sent.append({"target": target, "limit_mbps": limit_mbps}) + return { + "kind": "db", "ts": "20261107T030000Z", + "key": "database/fabledcurator-20261107T030000Z.dump", + "bucket": target.bucket, "endpoint": target.endpoint, + "parts": 3, "size_bytes": 1234, + } + monkeypatch.setattr("backend.app.services.backup_service.backup_db_offsite", _offsite) + return sent + + +@pytest.mark.asyncio +async def test_with_a_destination_the_dump_goes_off_site_and_nothing_is_written_here( + db_sync, archives, tmp_path, +): + from backend.app.tasks.backup import backup_db_task + run_id = backup_db_task.delay(triggered_by="nightly").get()["backup_run_id"] + row = db_sync.get(BackupRun, run_id) + assert row.status == "ok" + assert row.sql_path is None + assert row.size_bytes == 1234 + assert row.manifest["offsite"] == { + "endpoint": "http://archives.example:8081", "bucket": "curator", + "key": "database/fabledcurator-20261107T030000Z.dump", "parts": 3, + } + assert archives[0]["limit_mbps"] == 200 # the setting's default + assert not (tmp_path / "_backups").exists() + + +@pytest.mark.asyncio +async def test_an_off_site_dump_that_fails_is_recorded_with_why(db_sync, monkeypatch, archives): + from backend.app.services.backup_service import DumpFailed + from backend.app.tasks.backup import backup_db_task + + def _fails(**kwargs): + raise DumpFailed("pg_dump: error: connection lost") + monkeypatch.setattr("backend.app.services.backup_service.backup_db_offsite", _fails) + with pytest.raises(DumpFailed): + backup_db_task.delay().get() + row = db_sync.execute( + select(BackupRun).where(BackupRun.kind == "db").order_by(BackupRun.id.desc()).limit(1) + ).scalar_one() + assert row.status == "error" + assert row.error == "DumpFailed: pg_dump: error: connection lost" + + # --- backup_images_task ---------------------------------------------