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 }}
+
+
+
+ mdi-cloud-check-outline{{ r.offsite.bucket }}
+
+ 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 ---------------------------------------------