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-<UTC
time>.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 <noreply@anthropic.com>

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

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.

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
         ),

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 = "<CompleteMultipartUpload>" + "".join(
+            f"<Part><PartNumber>{n}</PartNumber><ETag>{etag}</ETag></Part>"
+            for n, etag in enumerate(self._etags, start=1)
+        ) + "</CompleteMultipartUpload>"
+        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

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 <images_root>/_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
+<images_root>/_backups/. Each local backup writes:
   fc_<kind>_<ts>.{sql|tar.zst}        — the artifact
   fc_<kind>_<ts>.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",

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}