diff --git a/alembic/versions/0121_backup_blob.py b/alembic/versions/0121_backup_blob.py new file mode 100644 index 0000000..373e48d --- /dev/null +++ b/alembic/versions/0121_backup_blob.py @@ -0,0 +1,46 @@ +"""backup_blob: which library files are already in the off-site bucket. + +One row per (destination, sha256) sent. The media backup's work list is every +image and attachment sha with no row for the current destination, so a run is +incremental and resumable by construction, and pointing backups at a new +bucket starts that bucket's copy from the beginning. + +backup_run.last_progress_at is the heartbeat of a run that spans hours; the +stall sweep measures a media run's silence from it, not from its start. + +Revision ID: 0121 +Revises: 0120 +Create Date: 2026-11-07 + +""" +import sqlalchemy as sa +from alembic import op + +revision = "0121" +down_revision = "0120" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "backup_blob", + sa.Column("target", sa.Text(), nullable=False), + sa.Column("sha256", sa.String(64), nullable=False), + sa.Column("key", sa.Text(), nullable=False), + sa.Column("size_bytes", sa.BigInteger(), nullable=False), + sa.Column( + "uploaded_at", sa.DateTime(timezone=True), nullable=False, + server_default=sa.text("now()"), + ), + sa.PrimaryKeyConstraint("target", "sha256", name=op.f("pk_backup_blob")), + ) + op.add_column( + "backup_run", + sa.Column("last_progress_at", sa.DateTime(timezone=True), nullable=True), + ) + + +def downgrade() -> None: + op.drop_column("backup_run", "last_progress_at") + op.drop_table("backup_blob") diff --git a/backend/app/api/system_backup.py b/backend/app/api/system_backup.py index 2d96809..1cacb77 100644 --- a/backend/app/api/system_backup.py +++ b/backend/app/api/system_backup.py @@ -20,7 +20,7 @@ system_backup_bp = Blueprint( "system_backup", __name__, url_prefix="/api/system/backup", ) -_KINDS = frozenset({"db", "images"}) +_KINDS = frozenset({"db", "images", "media"}) _TAG_MAX_LEN = 64 _BACKUP_SETTINGS_FIELDS = ( "backup_db_nightly_enabled", @@ -110,6 +110,38 @@ async def trigger_images_backup(): return jsonify({"status": "dispatched"}), 202 +@system_backup_bp.route("/media", methods=["POST"]) +async def trigger_media_backup(): + """Send the images and attachments not yet in the off-site bucket. Refused + when no destination is set up; a run already going is left to finish.""" + from ..services import backup_destination as dest + + async with get_session() as session: + row = await ImportSettings.load(session) + if not dest.describe(row)["configured"]: + return _bad("no_destination", + detail="set up an off-site destination before sending files") + from ..tasks.backup import backup_media_task + backup_media_task.delay(triggered_by="manual") + return jsonify({"status": "dispatched"}), 202 + + +@system_backup_bp.route("/media", methods=["GET"]) +async def media_coverage(): + """How much of the library is in the off-site bucket: files and bytes, + sent and total, each file counted once however many posts hold it.""" + from ..services import backup_destination as dest + from ..services import backup_media + + async with get_session() as session: + row = await ImportSettings.load(session) + f = dest.fields_of(row) + configured = dest.describe(row)["configured"] + tid = backup_media.destination_id(f.endpoint, f.bucket) if configured else None + counts = await session.run_sync(lambda s: backup_media.coverage(s, tid)) + return jsonify({"configured": configured, **counts}) + + @system_backup_bp.route("/runs", methods=["GET"]) async def list_runs(): try: diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 63313ed..d463b1e 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -302,6 +302,10 @@ def make_celery() -> Celery: "task": "backend.app.tasks.backup.prune_backups", "schedule": 86400.0, # daily }, + "reconcile-media-backup": { + "task": "backend.app.tasks.backup.reconcile_media_backup", + "schedule": 604800.0, # weekly: list the bucket, re-send what's missing + }, # Audit 2026-06-02 — three new per-entity recovery sweeps. # Each runs every 5 min like the other recover_stalled_* # sweeps; each is a no-op when nothing is stuck. diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index b890bfd..33e530a 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -4,6 +4,7 @@ from .app_setting import AppSetting from .artist import Artist from .artist_membership_suggestion import ArtistMembershipSuggestion from .artist_visit import ArtistVisit +from .backup_blob import BackupBlob from .backup_run import BackupRun from .base import Base from .character_prototype import CcipPrototypeState, CharacterPrototype @@ -55,6 +56,7 @@ __all__ = [ "Artist", "ArtistMembershipSuggestion", "ArtistVisit", + "BackupBlob", "BackupRun", "Source", "Credential", diff --git a/backend/app/models/backup_blob.py b/backend/app/models/backup_blob.py new file mode 100644 index 0000000..b42556e --- /dev/null +++ b/backend/app/models/backup_blob.py @@ -0,0 +1,26 @@ +"""backup_blob: a library file already sent to the off-site bucket. + +One row per (destination, sha256). `target` names the destination as +"/", so pointing backups at another bucket leaves these rows +inert and that bucket's copy starts from the beginning. `key` is where the +file sits in the bucket (`media/ab/cd/`). +""" + +from datetime import datetime + +from sqlalchemy import BigInteger, DateTime, String, Text, func +from sqlalchemy.orm import Mapped, mapped_column + +from .base import Base + + +class BackupBlob(Base): + __tablename__ = "backup_blob" + + target: Mapped[str] = mapped_column(Text, primary_key=True) + sha256: Mapped[str] = mapped_column(String(64), primary_key=True) + key: Mapped[str] = mapped_column(Text, nullable=False) + size_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False) + uploaded_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), + ) diff --git a/backend/app/models/backup_run.py b/backend/app/models/backup_run.py index d5e54cd..cac6a04 100644 --- a/backend/app/models/backup_run.py +++ b/backend/app/models/backup_run.py @@ -1,6 +1,7 @@ """FC-3h: backup_run — operator-facing artifact record for a backup run. -One row per backup attempt (kind='db' or 'images'). Lifecycle +One row per backup attempt (kind='db', 'images', or 'media' for the off-site +upload of images and attachments, which spans many task chunks). Lifecycle tracking (started_at/finished_at/duration_ms/exception text) lives in task_run from FC-3i — this row records artifact metadata: file paths, sizes, tag (retention protection), and restore lineage via @@ -62,6 +63,11 @@ class BackupRun(Base): ) # Self-referential FK, unindexed until 0089 (#3300): SET NULL has to find # the rows pointing at a deleted run before it can null them. + # Heartbeat of a run that spans hours (the media backup): set as it makes + # progress, and what the stall sweep measures its silence from. + last_progress_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True, + ) restored_from_id: Mapped[int | None] = mapped_column( ForeignKey("backup_run.id", ondelete="SET NULL"), nullable=True, index=True, diff --git a/backend/app/services/backup_media.py b/backend/app/services/backup_media.py new file mode 100644 index 0000000..aa9966c --- /dev/null +++ b/backend/app/services/backup_media.py @@ -0,0 +1,202 @@ +"""The off-site copy of the library's files: every image and attachment, sent +once each to the backup bucket as a content-addressed object. + +Layout in the bucket, beside the database dumps under `database/`: + + media/ab/cd/ + +`ab` and `cd` are the first four hex digits of the file's sha256, which keeps +any one prefix small, and the extension keeps the files readable in the bucket +without FabledCurator. The sha256 is the one recorded at import, so a file +uploads exactly once however many rows or posts point at it, and a run is +incremental: its work list is every sha with no `backup_blob` row for the +current destination. A file removed from the library stays in the bucket until +the bucket's own retention removes it; FabledCurator never deletes there. + +The bucket checks every object against that recorded sha256, so a file whose +bytes changed on disk since import is refused and reported, never stored under +the wrong name. +""" +from __future__ import annotations + +import re +import time +from dataclasses import dataclass +from pathlib import Path + +from sqlalchemy import delete, select, text +from sqlalchemy.dialects.postgresql import insert + +from ..models import BackupBlob +from .backup_s3 import ( + Pacer, + S3Error, + S3Target, + list_objects, + object_present, + put_file, + put_file_in_parts, +) + +PREFIX = "media/" +# Files above this size go in parts (S3 refuses a single PUT over 5 GiB, and a +# part that fails is resent alone instead of the whole file). +MULTIPART_OVER = 1 << 30 +# The bucket is listed this many keys at a time during a reconcile, with a +# short rest between pages. +LIST_PAGE = 1000 +LIST_REST_SECONDS = 0.2 +_EXT = re.compile(r"^\.[a-z0-9]{1,10}$") + + +@dataclass(frozen=True) +class Blob: + sha256: str + path: str + size_bytes: int + mime: str | None + + +def destination_id(endpoint: str, bucket: str) -> str: + """How backup_blob names a destination: one copy per endpoint and bucket.""" + return f"{endpoint.rstrip('/')}/{bucket}" + + +def target_id(target: S3Target) -> str: + return destination_id(target.endpoint, target.bucket) + + +def blob_key(sha256: str, path: str) -> str: + """The bucket key for a file: `media/ab/cd/`. An extension + that isn't short and plain is left off, so a key is always safe.""" + ext = Path(path).suffix.lower() + return f"{PREFIX}{sha256[:2]}/{sha256[2:4]}/{sha256}{ext if _EXT.match(ext) else ''}" + + +# Every library file once per sha256: images and attachments together. An +# attachment's sha can repeat across posts, and an image extracted from an +# archive is its own file with its own sha. +_LIBRARY = """ + SELECT sha256, path, size_bytes, mime FROM image_record + UNION ALL + SELECT sha256, path, size_bytes, mime FROM post_attachment +""" + + +def pending(session, tid: str, *, after: str = "", limit: int = 500) -> list[Blob]: + """The next `limit` files not yet in the bucket, in sha order after + `after`, so a run walks the library once even when some files fail.""" + rows = session.execute(text(f""" + SELECT DISTINCT ON (lib.sha256) lib.sha256, lib.path, lib.size_bytes, lib.mime + FROM ({_LIBRARY}) AS lib + WHERE lib.sha256 > :after + AND NOT EXISTS ( + SELECT 1 FROM backup_blob b WHERE b.target = :tid AND b.sha256 = lib.sha256 + ) + ORDER BY lib.sha256, lib.path + LIMIT :limit + """), {"after": after, "tid": tid, "limit": limit}).all() + return [Blob(r.sha256, r.path, r.size_bytes, r.mime) for r in rows] + + +def coverage(session, tid: str | None) -> dict: + """How much of the library is in the bucket: files and bytes, sent and + total, each sha counted once.""" + row = session.execute(text(f""" + WITH lib AS ( + SELECT DISTINCT ON (sha256) sha256, size_bytes FROM ({_LIBRARY}) AS u + ORDER BY sha256 + ) + SELECT count(*) AS files_total, + coalesce(sum(lib.size_bytes), 0) AS bytes_total, + count(b.sha256) AS files_sent, + coalesce(sum(lib.size_bytes) FILTER (WHERE b.sha256 IS NOT NULL), 0) + AS bytes_sent + FROM lib + LEFT JOIN backup_blob b ON b.sha256 = lib.sha256 AND b.target = :tid + """), {"tid": tid or ""}).one() + return { + "files_total": row.files_total, "bytes_total": int(row.bytes_total), + "files_sent": row.files_sent, "bytes_sent": int(row.bytes_sent), + } + + +def send(target: S3Target, blob: Blob, images_root: Path, pacer: Pacer | None) -> str: + """Upload one file to its key and return the key. Raises S3Error (or + OSError) with the reason when it can't be sent.""" + path = Path(blob.path) + if not path.is_absolute(): + path = images_root / path + if not path.is_file(): + raise S3Error(f"{path} is not on disk.") + size = path.stat().st_size + if size != blob.size_bytes: + raise S3Error(f"{path} is {size} bytes on disk but {blob.size_bytes} when imported.") + key = blob_key(blob.sha256, blob.path) + content_type = blob.mime or "application/octet-stream" + try: + if size > MULTIPART_OVER: + put_file_in_parts(target, key, path, sha256=blob.sha256, + content_type=content_type, pacer=pacer) + else: + put_file(target, key, path, content_type=content_type, + payload_sha256=blob.sha256, pacer=pacer) + except S3Error: + if object_present(target, key, size): + return key # already there from an earlier run + raise + return key + + +def record(session, tid: str, blobs: list[tuple[Blob, str]]) -> None: + """Note files as sent: (blob, key) pairs. A row already there is kept.""" + if blobs: + session.execute(insert(BackupBlob).values([ + {"target": tid, "sha256": b.sha256, "key": key, "size_bytes": b.size_bytes} + for b, key in blobs + ]).on_conflict_do_nothing()) + + +def reconcile(session, target: S3Target) -> dict: + """Check backup_blob against what the bucket actually holds under + `media/`. A row whose object is missing or the wrong size is dropped, so + the next run sends that file again; an object the bucket holds for a + library file with no row (a restored database, a lost commit) is recorded, + so it isn't sent twice. Returns the counts.""" + tid = target_id(target) + held: dict[str, int] = {} + token = None + while True: + page = list_objects(target, PREFIX, LIST_PAGE, continuation=token) + held.update((o.key, o.size) for o in page.objects) + token = page.next_token + if not token: + break + time.sleep(LIST_REST_SECONDS) + + rows = session.execute( + select(BackupBlob.sha256, BackupBlob.key, BackupBlob.size_bytes) + .where(BackupBlob.target == tid) + ).all() + gone = [r.sha256 for r in rows if held.get(r.key) != r.size_bytes] + for at in range(0, len(gone), 1000): + session.execute(delete(BackupBlob).where( + BackupBlob.target == tid, BackupBlob.sha256.in_(gone[at:at + 1000]), + )) + kept = {r.sha256 for r in rows} - set(gone) + + found = [] + for r in session.execute(text(f""" + SELECT DISTINCT ON (sha256) sha256, path, size_bytes, mime + FROM ({_LIBRARY}) AS lib ORDER BY sha256, path + """)).all(): + if r.sha256 in kept: + continue + blob = Blob(r.sha256, r.path, r.size_bytes, r.mime) + key = blob_key(blob.sha256, blob.path) + if held.get(key) == blob.size_bytes: + found.append((blob, key)) + for at in range(0, len(found), 1000): + record(session, tid, found[at:at + 1000]) + session.commit() + return {"listed": len(held), "dropped": len(gone), "adopted": len(found)} diff --git a/backend/app/services/backup_s3.py b/backend/app/services/backup_s3.py index ce09aaa..6b438fc 100644 --- a/backend/app/services/backup_s3.py +++ b/backend/app/services/backup_s3.py @@ -166,6 +166,7 @@ def _says(status: int, reason: str, content: bytes) -> str: "InvalidAccessKeyId": "the access key isn't known there", "NoSuchBucket": "there is no such bucket", "AccessDenied": "the key isn't allowed to do that", + "XAmzContentSHA256Mismatch": "the file's bytes don't match its recorded checksum", } said = hints.get(code) or message or reason if not code and 400 <= status < 500: @@ -190,10 +191,13 @@ def file_sha256(path: Path) -> str: def put_file(target: S3Target, key: str, path: Path, *, content_type: str = "application/octet-stream", payload_sha256: str | None = None, + pacer: Pacer | None = None, now: datetime | None = None) -> None: """Upload one file as one object, streamed from disk. `payload_sha256` is the file's sha256 when the caller already knows it (the library stores - every file's), which saves reading the file twice.""" + every file's), which saves reading the file twice; the bucket checks the + bytes against it, so a file that changed on disk is refused rather than + stored under the wrong name.""" digest = payload_sha256 or file_sha256(path) size = path.stat().st_size url = target.url(key) @@ -203,7 +207,8 @@ def put_file(target: S3Target, key: str, path: Path, *, conn = _connect(target, url, TRANSFER_TIMEOUT_SECONDS) try: with open(path, "rb") as f: - conn.request("PUT", _request_target(url), body=f, headers=headers) + conn.request("PUT", _request_target(url), + body=pacer.file_chunks(f) if pacer else f, headers=headers) response = conn.getresponse() content = response.read() except (OSError, http.client.HTTPException) as exc: @@ -307,15 +312,23 @@ class Pacer: self.sent = 0 self.started = time.monotonic() + def _wait(self, n: int) -> None: + if self.rate: + self.sent += n + ahead = self.sent / self.rate - (time.monotonic() - self.started) + if ahead > 0: + time.sleep(ahead) + 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) + self._wait(len(piece)) + yield piece + + def file_chunks(self, f) -> Iterator[bytes]: + while piece := f.read(CHUNK): + self._wait(len(piece)) yield piece @@ -455,3 +468,38 @@ class Upload: _send(self.target, "DELETE", self._url()) except Exception: # noqa: BLE001 — best effort, see above pass + + +def put_file_in_parts(target: S3Target, key: str, path: Path, *, sha256: str, + content_type: str = "application/octet-stream", + pacer: Pacer | None = None) -> int: + """Upload a large file as a multipart object, read from disk a part at a + time. The whole file's sha256 is computed on the way and must equal + `sha256` before the upload is completed; otherwise it is aborted, so a file + that changed on disk is never stored under its old name. Returns the + number of bytes sent.""" + upload = Upload(target, key, content_type=content_type, pacer=pacer) + digest = hashlib.sha256() + try: + with open(path, "rb") as f: + while part := f.read(PART_SIZE): + digest.update(part) + upload.add(part) + if digest.hexdigest() != sha256: + raise S3Error(f"{path} no longer matches its recorded checksum; not stored.") + upload.complete() + except BaseException: + upload.abort() + raise + return upload.size + + +def object_present(target: S3Target, key: str, size: int) -> bool: + """Whether `key` is already in the bucket at `size` bytes. Used when a put + is refused: an append-only bucket refuses to replace an object that is + already there, which for a content-addressed key means the work is done.""" + try: + page = list_objects(target, prefix=key, limit=1) + except S3Error: + return False + return any(o.key == key and o.size == size for o in page.objects) diff --git a/backend/app/tasks/backup.py b/backend/app/tasks/backup.py index 3ca675c..13f32ed 100644 --- a/backend/app/tasks/backup.py +++ b/backend/app/tasks/backup.py @@ -8,6 +8,7 @@ BackupRun. from __future__ import annotations import logging +import time from datetime import UTC, datetime from pathlib import Path @@ -18,7 +19,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_destination, backup_service +from ..services import backup_destination, backup_media, backup_s3, backup_service from ._sync_engine import sync_session_factory as _sync_session_factory log = logging.getLogger(__name__) @@ -267,7 +268,9 @@ def restore_images_task(self, *, source_backup_run_id: int) -> dict: def prune_backups() -> dict: """Daily Beat. Per-kind retention from ImportSettings. - Returns {"db_deleted": N, "images_deleted": M, "files_unlinked": K}. + 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. 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 @@ -276,12 +279,13 @@ def prune_backups() -> dict: Audit 2026-06-02 added the dedicated sweep.) """ SessionLocal = _sync_session_factory() - counts = {"db_deleted": 0, "images_deleted": 0, "files_unlinked": 0} + counts = {"db_deleted": 0, "images_deleted": 0, "media_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), ): candidates = session.execute( select(BackupRun) @@ -325,4 +329,168 @@ def backup_db_nightly() -> dict: if now_hour != configured_hour: return {"skipped": f"hour={now_hour} != configured={configured_hour}"} res = backup_db_task.delay(triggered_by="nightly") - return {"dispatched": res.id} + out = {"dispatched": res.id} + # With an off-site destination, the night's backup is the whole library: + # the database dump, then any images and attachments not yet in the bucket. + # Both run on the long-maintenance lane, one after the other. + with SessionLocal() as session: + try: + offsite = backup_destination.saved_target(ImportSettings.load_sync(session)) + except Exception: # noqa: BLE001 — the media run reports it itself + offsite = True + if offsite is not None: + out["media_dispatched"] = backup_media_task.delay(triggered_by="nightly").id + return out + + +# ── Off-site copy of the library's files ───────────────────────────────────── + +# Each task invocation sends files for this long, then queues the next chunk +# of the same run, so other long-maintenance jobs get their turn in between. +MEDIA_CHUNK_SECONDS = 45 * 60 +MEDIA_RUNS_KEPT = 60 +# A failed file is listed on the run, up to this many; the count covers all. +MEDIA_FAILURES_LISTED = 20 + + +def _offsite_manifest(target) -> dict: + return {"endpoint": target.endpoint, "bucket": target.bucket, + "key": backup_media.PREFIX} + + +@celery.task( + name="backend.app.tasks.backup.backup_media_task", + bind=True, + autoretry_for=(OperationalError, DBAPIError), + retry_backoff=30, retry_backoff_max=300, max_retries=2, + # A chunk stops starting files after MEDIA_CHUNK_SECONDS; the limits leave + # room for the file in flight, a multi-gigabyte archive at a low upload + # limit included. + soft_time_limit=21600, time_limit=23400, +) +def backup_media_task(self, *, run_id: int | None = None, + triggered_by: str = "manual") -> dict: + """Send the library's images and attachments that aren't in the off-site + bucket yet, one chunk of a run. Called without `run_id` it starts a run + (unless one is already going); each chunk queues the next until nothing is + left. A file that can't be sent is listed on the run and tried again by + the next run. Returns {'backup_run_id': N, ...}.""" + SessionLocal = _sync_session_factory() + now = datetime.now(UTC) + with SessionLocal() as session: + if run_id is None: + going = session.execute( + select(BackupRun.id).where(BackupRun.kind == "media") + .where(BackupRun.status == "running").limit(1) + ).scalar_one_or_none() + if going is not None: + return {"backup_run_id": going, "skipped": "a media backup is already running"} + 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": ""}) + session.add(row) + session.commit() + run_id = row.id + else: + row = session.get(BackupRun, run_id) + if row is None or row.status != "running": + return {"backup_run_id": run_id, "skipped": "run is no longer running"} + + try: + 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 None: + raise backup_s3.S3Error("No off-site destination is set up.") + more = _send_media_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: + backup_media_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 = "ok" + row.finished_at = datetime.now(UTC) + if m.get("failed"): + first = (m.get("failures") or [{}])[0] + row.error = (f"{m['failed']} file(s) weren't sent; the next run tries them again. " + f"First: {first.get('path')}: {first.get('error')}")[:2000] + session.commit() + return {"backup_run_id": run_id, "continues": False} + + +def _send_media_chunk(SessionLocal, run_id: int, target, pacer) -> bool: + """Send files until the chunk's time is up or none are left. Progress is + committed after every file, so a chunk that dies loses at most the file + in flight. Returns True when files remain for another chunk.""" + tid = backup_media.target_id(target) + deadline = time.monotonic() + MEDIA_CHUNK_SECONDS + with SessionLocal() as session: + row = session.get(BackupRun, run_id) + m = dict(row.manifest or {}) + m["offsite"] = _offsite_manifest(target) + cursor = m.get("cursor") or "" + while True: + batch = backup_media.pending(session, tid, after=cursor) + # End the read before any upload starts: a transaction left open + # across a long transfer holds back vacuum for its whole length. + session.commit() + if not batch: + m["cursor"] = cursor + row.manifest = dict(m) + row.last_progress_at = datetime.now(UTC) + session.commit() + return False + for blob in batch: + if time.monotonic() > deadline: + m["cursor"] = cursor + row.manifest = dict(m) + session.commit() + return True + try: + key = backup_media.send(target, blob, IMAGES_ROOT, pacer) + except (backup_s3.S3Error, OSError) as exc: + m["failed"] = m.get("failed", 0) + 1 + if len(m.get("failures") or []) < MEDIA_FAILURES_LISTED: + m["failures"] = [*(m.get("failures") or []), + {"path": blob.path, "error": str(exc)[:300]}] + else: + backup_media.record(session, tid, [(blob, key)]) + m["sent"] = m.get("sent", 0) + 1 + row.size_bytes = (row.size_bytes or 0) + blob.size_bytes + cursor = blob.sha256 + m["cursor"] = cursor + row.manifest = dict(m) + row.last_progress_at = datetime.now(UTC) + session.commit() + + +@celery.task( + name="backend.app.tasks.backup.reconcile_media_backup", + soft_time_limit=1800, time_limit=2100, +) +def reconcile_media_backup() -> dict: + """Weekly Beat. Compare what the off-site bucket holds under `media/` with + what FabledCurator believes it sent: a missing object is sent again by + the next run, and an object already there is not sent twice. Skipped + while a media run is going, and when no destination is set up.""" + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + target = backup_destination.saved_target(ImportSettings.load_sync(session)) + if target is None: + return {"skipped": "no destination"} + going = session.execute( + select(BackupRun.id).where(BackupRun.kind == "media") + .where(BackupRun.status == "running").limit(1) + ).scalar_one_or_none() + if going is not None: + return {"skipped": "a media backup is running"} + return backup_media.reconcile(session, target) diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 4c4c1a9..572d667 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -842,10 +842,18 @@ 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. .where( or_( and_(BackupRun.kind == "db", BackupRun.started_at < db_cutoff), - and_(BackupRun.kind != "db", BackupRun.started_at < slow_cutoff), + and_( + BackupRun.kind == "media", + func.coalesce(BackupRun.last_progress_at, BackupRun.started_at) + < slow_cutoff, + ), + and_(BackupRun.kind.not_in(["db", "media"]), + BackupRun.started_at < slow_cutoff), ) ) .values(status="error", finished_at=now, error=msg) diff --git a/frontend/src/components/settings/BackupCard.vue b/frontend/src/components/settings/BackupCard.vue index be8f6dd..74024c7 100644 --- a/frontend/src/components/settings/BackupCard.vue +++ b/frontend/src/components/settings/BackupCard.vue @@ -7,8 +7,8 @@

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. + own retention thins old dumps. Images and attachments go to the same + bucket's media/ folder, below.

pg_dump for the database (fast, nightly-schedulable); @@ -48,7 +48,8 @@ />

- Keep last applies to this list and to any dumps still on this server. + Nightly sends the database, then any new images and attachments. Keep + last applies to this list and to any dumps still on this server.

+ + + + ( store.destination?.configured ? store.destination : null )) +const coverage = computed(() => store.mediaCoverage) +const coveragePct = computed(() => { + const c = coverage.value + return c && c.files_total ? (100 * c.files_sent) / c.files_total : 0 +}) +const runningMedia = computed(() => store.mediaRuns.find((r) => r.status === 'running')) +const mediaRunning = computed(() => Boolean(runningMedia.value)) +const runningSent = computed(() => runningMedia.value?.manifest?.sent || 0) + +const mediaTriggering = ref(false) const dbTriggering = ref(false) const imagesTriggering = ref(false) const confirmOpen = ref(false) @@ -134,10 +179,19 @@ const confirmRunId = ref(0) const confirmDescription = ref('') let pollId = null +let polls = 0 +// Coverage counts every library file, so it refreshes every sixth poll +// (30 s) rather than with the run lists. +const COVERAGE_EVERY = 6 function pollOnce() { if (document.hidden) return store.loadRuns('db') - store.loadRuns('images') + if (offsite.value) { + store.loadRuns('media') + if (polls++ % COVERAGE_EVERY === 0) store.loadMediaCoverage() + } else { + store.loadRuns('images') + } } onMounted(() => { @@ -151,7 +205,7 @@ onUnmounted(() => { }) async function onTrigger(kind) { - const triggeringRef = kind === 'db' ? dbTriggering : imagesTriggering + const triggeringRef = { db: dbTriggering, images: imagesTriggering, media: mediaTriggering }[kind] triggeringRef.value = true try { await store.triggerBackup(kind) diff --git a/frontend/src/components/settings/BackupDestinationCard.vue b/frontend/src/components/settings/BackupDestinationCard.vue index acb9dd0..a6953f4 100644 --- a/frontend/src/components/settings/BackupDestinationCard.vue +++ b/frontend/src/components/settings/BackupDestinationCard.vue @@ -84,7 +84,8 @@ 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. + Database backups and the library's images and attachments go to this bucket. + Set when they run under Backups below.

diff --git a/frontend/src/stores/backup.js b/frontend/src/stores/backup.js index af099a3..e177c20 100644 --- a/frontend/src/stores/backup.js +++ b/frontend/src/stores/backup.js @@ -8,13 +8,15 @@ export const useBackupStore = defineStore('backup', () => { const dbRuns = ref([]) const imagesRuns = ref([]) + const mediaRuns = ref([]) const settings = ref(null) - const loading = ref({ dbRuns: false, imagesRuns: false, settings: false }) + const loading = ref({ dbRuns: false, imagesRuns: false, mediaRuns: false, settings: false }) const lastError = ref(null) + const RUNS = { db: dbRuns, images: imagesRuns, media: mediaRuns } async function loadRuns(kind) { - const targetRef = kind === 'db' ? dbRuns : imagesRuns - const loadingKey = kind === 'db' ? 'dbRuns' : 'imagesRuns' + const targetRef = RUNS[kind] + const loadingKey = `${kind}Runs` loading.value[loadingKey] = true lastError.value = null try { @@ -125,8 +127,21 @@ export const useBackupStore = defineStore('backup', () => { return api.post('/api/system/backup/destination/test', { body: fields }) } + // How much of the library is in the off-site bucket: + // {configured, files_total, files_sent, bytes_total, bytes_sent}. + const mediaCoverage = ref(null) + + async function loadMediaCoverage() { + try { + mediaCoverage.value = await api.get('/api/system/backup/media') + } catch (e) { + lastError.value = e.message + } + } + return { - dbRuns, imagesRuns, settings, loading, lastError, + dbRuns, imagesRuns, mediaRuns, mediaCoverage, loadMediaCoverage, + settings, loading, lastError, loadRuns, triggerBackup, restore, deleteRun, setTag, loadSettings, patchSettings, destination, loadDestination, saveDestination, removeDestination, testDestination, diff --git a/tests/test_backup_media.py b/tests/test_backup_media.py new file mode 100644 index 0000000..60c4c69 --- /dev/null +++ b/tests/test_backup_media.py @@ -0,0 +1,250 @@ +"""The off-site copy of the library's files (services/backup_media.py and the +backup_media_task chunk loop): each image and attachment goes to the bucket +once, by sha256, the work list is what isn't there yet, and a file that can't +be sent is reported and tried again rather than stopping the run.""" +from datetime import UTC, datetime, timedelta + +import pytest +from sqlalchemy import select + +from backend.app.models import BackupBlob, BackupRun, ImageRecord, PostAttachment +from backend.app.services import backup_media +from backend.app.services.backup_s3 import ListPage, S3Error, S3Target, StoredObject + +pytestmark = pytest.mark.integration + +TARGET = S3Target("http://archives.example:8081", "curator", "AK", "SK") +TID = "http://archives.example:8081/curator" + + +def _sha(c: str) -> str: + return c * 64 + + +def _image(db_sync, tmp_path, c, body=b"img", ext=".jpg"): + path = tmp_path / f"{c}{ext}" + path.write_bytes(body) + db_sync.add(ImageRecord(path=str(path), sha256=_sha(c), size_bytes=len(body), + mime="image/jpeg", origin="imported_filesystem")) + return path + + +def _attachment(db_sync, tmp_path, c, body=b"zip", name="set.zip"): + path = tmp_path / f"{c}-{name}" + path.write_bytes(body) + db_sync.add(PostAttachment(sha256=_sha(c), path=str(path), original_filename=name, + ext=".zip", size_bytes=len(body))) + return path + + +# ── keys and the work list ────────────────────────────────────────────────── + + +def test_keys_are_content_addressed_under_media(): + sha = "abcd" + "0" * 60 + assert backup_media.blob_key(sha, "/images/x/Pic.JPG") == f"media/ab/cd/{sha}.jpg" + assert backup_media.blob_key(sha, "/images/x/noext") == f"media/ab/cd/{sha}" + assert backup_media.blob_key(sha, "/images/x/a.we ird") == f"media/ab/cd/{sha}" + + +def test_the_work_list_is_every_file_once_that_the_bucket_lacks(db_sync, tmp_path): + _image(db_sync, tmp_path, "a") + _image(db_sync, tmp_path, "b") + _attachment(db_sync, tmp_path, "c") + # The same archive held by a second post is one file to send. + db_sync.add(PostAttachment(sha256=_sha("c"), path=str(tmp_path / "c-copy.zip"), + original_filename="set.zip", ext=".zip", size_bytes=3)) + db_sync.add(BackupBlob(target=TID, sha256=_sha("b"), key="media/bb", size_bytes=3)) + # A row for another bucket doesn't count for this one. + db_sync.add(BackupBlob(target="http://other/bucket", sha256=_sha("a"), key="k", + size_bytes=3)) + db_sync.commit() + + shas = [b.sha256 for b in backup_media.pending(db_sync, TID)] + assert shas == [_sha("a"), _sha("c")] + assert [b.sha256 for b in backup_media.pending(db_sync, TID, after=_sha("a"))] == [_sha("c")] + + cov = backup_media.coverage(db_sync, TID) + assert cov == {"files_total": 3, "bytes_total": 9, "files_sent": 1, "bytes_sent": 3} + + +# ── sending one file ──────────────────────────────────────────────────────── + + +@pytest.fixture +def puts(monkeypatch): + sent = [] + monkeypatch.setattr(backup_media, "put_file", + lambda t, key, path, **kw: sent.append(("whole", key, kw))) + monkeypatch.setattr(backup_media, "put_file_in_parts", + lambda t, key, path, **kw: sent.append(("parts", key, kw))) + return sent + + +def test_a_file_is_signed_with_its_recorded_sha(db_sync, tmp_path, puts): + path = _image(db_sync, tmp_path, "a") + blob = backup_media.Blob(_sha("a"), str(path), 3, "image/jpeg") + key = backup_media.send(TARGET, blob, tmp_path, None) + assert key == f"media/aa/aa/{_sha('a')}.jpg" + (how, sent_key, kw) = puts[0] + assert (how, sent_key) == ("whole", key) + assert kw["payload_sha256"] == _sha("a") and kw["content_type"] == "image/jpeg" + + +def test_a_large_file_goes_in_parts(db_sync, tmp_path, puts, monkeypatch): + monkeypatch.setattr(backup_media, "MULTIPART_OVER", 2) + path = _attachment(db_sync, tmp_path, "c") + backup_media.send(TARGET, backup_media.Blob(_sha("c"), str(path), 3, None), tmp_path, None) + assert puts[0][0] == "parts" + assert puts[0][2]["sha256"] == _sha("c") + + +def test_a_missing_or_changed_file_is_refused_before_sending(db_sync, tmp_path, puts): + gone = backup_media.Blob(_sha("a"), str(tmp_path / "gone.jpg"), 3, None) + with pytest.raises(S3Error, match="not on disk"): + backup_media.send(TARGET, gone, tmp_path, None) + path = _image(db_sync, tmp_path, "b", body=b"longer now") + changed = backup_media.Blob(_sha("b"), str(path), 3, None) + with pytest.raises(S3Error, match="when imported"): + backup_media.send(TARGET, changed, tmp_path, None) + assert puts == [] + + +def test_a_refusal_for_an_object_already_there_counts_as_sent(db_sync, tmp_path, monkeypatch): + def refused(*a, **kw): + raise S3Error("HTTP 403 AccessDenied: append-only.") + monkeypatch.setattr(backup_media, "put_file", refused) + monkeypatch.setattr(backup_media, "object_present", lambda t, key, size: True) + path = _image(db_sync, tmp_path, "a") + key = backup_media.send(TARGET, backup_media.Blob(_sha("a"), str(path), 3, None), + tmp_path, None) + assert key.startswith("media/aa/aa/") + + +# ── the run ───────────────────────────────────────────────────────────────── + + +@pytest.fixture +def run_env(monkeypatch, tmp_path): + monkeypatch.setattr("backend.app.tasks.backup.IMAGES_ROOT", tmp_path) + monkeypatch.setattr( + "backend.app.services.backup_destination.saved_target", lambda settings: TARGET, + ) + queued = [] + from backend.app.tasks import backup as tasks + monkeypatch.setattr(tasks.backup_media_task, "delay", lambda **kw: queued.append(kw)) + return queued + + +def _send_all_but(bad_sha): + def send(target, blob, images_root, pacer): + if blob.sha256 == bad_sha: + raise S3Error("HTTP 503: busy.") + return backup_media.blob_key(blob.sha256, blob.path) + return send + + +def test_a_run_sends_what_is_missing_and_reports_what_failed( + db_sync, tmp_path, run_env, monkeypatch, +): + from backend.app.tasks.backup import backup_media_task + for c in "abc": + _image(db_sync, tmp_path, c) + db_sync.commit() + monkeypatch.setattr(backup_media, "send", _send_all_but(_sha("b"))) + + result = backup_media_task(triggered_by="manual") + assert result["continues"] is False + row = db_sync.get(BackupRun, result["backup_run_id"]) + db_sync.refresh(row) + assert (row.kind, row.status, row.size_bytes) == ("media", "ok", 6) + assert row.manifest["sent"] == 2 and row.manifest["failed"] == 1 + assert row.manifest["offsite"]["bucket"] == "curator" + assert "1 file(s) weren't sent" in row.error + assert row.manifest["failures"][0]["error"] == "HTTP 503: busy." + sent = set(db_sync.execute(select(BackupBlob.sha256)).scalars()) + assert sent == {_sha("a"), _sha("c")} + + # The next run tries only the file that failed. + monkeypatch.setattr(backup_media, "send", _send_all_but(None)) + again = backup_media_task(triggered_by="manual") + row = db_sync.get(BackupRun, again["backup_run_id"]) + db_sync.refresh(row) + assert row.manifest["sent"] == 1 and row.error is None + + +def test_a_run_out_of_time_queues_its_next_chunk(db_sync, tmp_path, run_env, monkeypatch): + from backend.app.tasks import backup as tasks + for c in "ab": + _image(db_sync, tmp_path, c) + db_sync.commit() + monkeypatch.setattr(backup_media, "send", _send_all_but(None)) + monkeypatch.setattr(tasks, "MEDIA_CHUNK_SECONDS", -1) + + result = tasks.backup_media_task(triggered_by="manual") + assert result["continues"] is True + assert run_env == [{"run_id": result["backup_run_id"]}] + row = db_sync.get(BackupRun, result["backup_run_id"]) + db_sync.refresh(row) + assert row.status == "running" and row.last_progress_at is not None + + +def test_a_second_run_waits_for_the_one_going(db_sync, run_env): + from backend.app.tasks.backup import backup_media_task + now = datetime.now(UTC) + db_sync.add(BackupRun(kind="media", status="running", triggered_by="nightly", + started_at=now, manifest={})) + db_sync.commit() + assert "already running" in backup_media_task(triggered_by="manual")["skipped"] + + +def test_without_a_destination_the_run_fails_with_why(db_sync, monkeypatch): + from backend.app.tasks.backup import backup_media_task + monkeypatch.setattr( + "backend.app.services.backup_destination.saved_target", lambda settings: None, + ) + with pytest.raises(S3Error): + backup_media_task(triggered_by="manual") + row = db_sync.execute(select(BackupRun).where(BackupRun.kind == "media")).scalar_one() + assert row.status == "error" and "No off-site destination" in row.error + + +def test_a_long_media_run_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="media", status="running", triggered_by="nightly", + started_at=long_ago, last_progress_at=datetime.now(UTC), manifest={}) + silent = BackupRun(kind="media", status="running", triggered_by="nightly", + 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) == ("running", "error") + + +# ── reconcile ─────────────────────────────────────────────────────────────── + + +def test_reconcile_resends_what_the_bucket_lost_and_adopts_what_it_has( + db_sync, tmp_path, monkeypatch, +): + _image(db_sync, tmp_path, "a") # row, but the bucket lost it + _image(db_sync, tmp_path, "b") # in the bucket, no row + _image(db_sync, tmp_path, "c") # neither + key_a = backup_media.blob_key(_sha("a"), "x.jpg") + key_b = backup_media.blob_key(_sha("b"), "x.jpg") + db_sync.add(BackupBlob(target=TID, sha256=_sha("a"), key=key_a, size_bytes=3)) + db_sync.commit() + pages = iter([ + ListPage([StoredObject("media/zz/other", 1, "")], "next"), + ListPage([StoredObject(key_b, 3, "")], None), + ]) + monkeypatch.setattr(backup_media, "list_objects", lambda *a, **kw: next(pages)) + monkeypatch.setattr(backup_media.time, "sleep", lambda s: None) + + assert backup_media.reconcile(db_sync, TARGET) == { + "listed": 2, "dropped": 1, "adopted": 1, + } + assert set(db_sync.execute(select(BackupBlob.sha256)).scalars()) == {_sha("b")} diff --git a/tests/test_backup_s3.py b/tests/test_backup_s3.py index ba1107d..e4d03b8 100644 --- a/tests/test_backup_s3.py +++ b/tests/test_backup_s3.py @@ -322,3 +322,42 @@ def test_the_pacer_holds_bytes_to_the_limit(monkeypatch): 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) + + +def test_a_large_file_is_sent_in_parts_and_checked_against_its_sha(tmp_path, monkeypatch): + monkeypatch.setattr(s3, "PART_SIZE", 4) + f = tmp_path / "big.zip" + f.write_bytes(b"0123456789") + with _Server(_bucket()) as srv: + sent = s3.put_file_in_parts(srv.target(), "media/x.zip", f, + sha256=hashlib.sha256(b"0123456789").hexdigest()) + assert sent == 10 + assert [r["body"] for r in srv.seen if r["method"] == "PUT"] == [b"0123", b"4567", b"89"] + assert srv.seen[-1]["method"] == "POST" # completed + + +def test_a_file_that_changed_since_import_is_not_completed(tmp_path, monkeypatch): + monkeypatch.setattr(s3, "PART_SIZE", 4) + f = tmp_path / "big.zip" + f.write_bytes(b"0123456789") + with _Server(_bucket()) as srv: + with pytest.raises(s3.S3Error, match="recorded checksum"): + s3.put_file_in_parts(srv.target(), "media/x.zip", f, sha256="0" * 64) + assert srv.seen[-1]["method"] == "DELETE" # aborted, nothing stored + + +def test_a_paced_put_sends_the_same_bytes(tmp_path): + f = tmp_path / "a.bin" + f.write_bytes(b"z" * (s3.CHUNK + 5)) + with _Server(lambda h: (200, b"")) as srv: + s3.put_file(srv.target(), "a.bin", f, pacer=s3.Pacer(0)) + assert srv.seen[0]["body"] == b"z" * (s3.CHUNK + 5) + + +def test_presence_is_the_exact_key_at_the_exact_size(): + listing = (b"media/a.jpg3" + b"") + with _Server(lambda h: (200, listing)) as srv: + assert s3.object_present(srv.target(), "media/a.jpg", 3) + assert not s3.object_present(srv.target(), "media/a.jpg", 4) + assert not s3.object_present(srv.target(), "media/a.jp", 3)