"""FC-3h: backup/restore Celery tasks. All tasks live on the maintenance queue (per celery_app.task_routes). task_run lifecycle tracking is automatic via FC-3i signals — these tasks just record the operator-facing artifact metadata into BackupRun. """ from __future__ import annotations import logging import time from datetime import UTC, datetime from pathlib import Path from celery.exceptions import SoftTimeLimitExceeded from sqlalchemy import select from sqlalchemy.exc import DBAPIError, OperationalError from ..celery_app import celery from ..config import get_config from ..models import BackupRun, ImportSettings from ..services import backup_destination, backup_media, backup_s3, backup_service from ._sync_engine import sync_session_factory as _sync_session_factory log = logging.getLogger(__name__) IMAGES_ROOT = Path("/images") def _mark_failed(session, row: BackupRun, exc: BaseException) -> None: """Flip a BackupRun row from running/restoring to error with a truncated error message and finished_at. Caller already holds the session open.""" row.status = "error" row.error = f"{type(exc).__name__}: {exc}"[:2000] row.finished_at = datetime.now(UTC) session.add(row) session.commit() @celery.task( name="backend.app.tasks.backup.backup_db_task", bind=True, autoretry_for=(OperationalError, DBAPIError), retry_backoff=10, retry_backoff_max=120, max_retries=2, # A pg_dump can't be chunked; the 12-min limit timed out once the DB grew # (operator-flagged 2026-06-07). 30/35 min gives real headroom. (A long # backup still briefly holds the concurrency-1 maintenance lane — the # structural fix is a dedicated lane for the long one-shots.) soft_time_limit=1800, time_limit=2100, ) def backup_db_task(self, *, tag: str | None = None, triggered_by: str = "manual") -> dict: """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) with SessionLocal() as session: row = BackupRun( kind="db", status="running", tag=tag, triggered_by=triggered_by, started_at=now, manifest={}, ) session.add(row) session.commit() session.refresh(row) run_id = row.id 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 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) if row is not None: _mark_failed(session, row, exc) raise with SessionLocal() as session: row = session.get(BackupRun, run_id) row.status = "ok" row.finished_at = datetime.now(UTC) row.size_bytes = result["size_bytes"] 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} @celery.task( name="backend.app.tasks.backup.backup_images_task", bind=True, autoretry_for=(OperationalError, DBAPIError), retry_backoff=30, retry_backoff_max=300, max_retries=1, soft_time_limit=21600, time_limit=23400, ) def backup_images_task(self, *, tag: str | None = None, triggered_by: str = "manual") -> dict: """Create one images backup. Same shape as backup_db_task; uses tar_path instead of sql_path.""" SessionLocal = _sync_session_factory() now = datetime.now(UTC) with SessionLocal() as session: row = BackupRun( kind="images", status="running", tag=tag, triggered_by=triggered_by, started_at=now, manifest={}, ) session.add(row) session.commit() session.refresh(row) run_id = row.id try: result = backup_service.backup_images( 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) if row is not None: _mark_failed(session, row, exc) raise with SessionLocal() as session: row = session.get(BackupRun, run_id) row.status = "ok" row.finished_at = datetime.now(UTC) row.tar_path = result["tar_path"] row.size_bytes = result["size_bytes"] row.manifest = { "manifest_path": result["manifest_path"], "ts": result["ts"], } session.commit() return {"backup_run_id": run_id} @celery.task( name="backend.app.tasks.backup.restore_db_task", bind=True, max_retries=0, # NEVER auto-retry a half-applied restore. soft_time_limit=1200, time_limit=1800, ) def restore_db_task(self, *, source_backup_run_id: int) -> dict: """Restore from a previous DB backup. Inserts a NEW BackupRun row (kind='db', status='restoring') linked to the source via restored_from_id; flips to 'restored' on success or 'error' on failure. Operator sees the restore as a row in the dashboard.""" SessionLocal = _sync_session_factory() cfg = get_config() now = datetime.now(UTC) with SessionLocal() as session: src = session.get(BackupRun, source_backup_run_id) if src is None or src.kind != "db" or not src.sql_path: raise ValueError( f"BackupRun id={source_backup_run_id} is not a valid DB backup" ) marker = BackupRun( kind="db", status="restoring", triggered_by="restore", started_at=now, restored_from_id=src.id, manifest={"source_sql_path": src.sql_path}, ) session.add(marker) session.commit() session.refresh(marker) marker_id = marker.id sql_path = src.sql_path try: backup_service.restore_db( db_url=cfg.database_url_sync, sql_path=Path(sql_path), ) except (SoftTimeLimitExceeded, Exception) as exc: with SessionLocal() as session: row = session.get(BackupRun, marker_id) if row is not None: _mark_failed(session, row, exc) raise with SessionLocal() as session: row = session.get(BackupRun, marker_id) row.status = "restored" row.finished_at = datetime.now(UTC) session.commit() return {"backup_run_id": marker_id} @celery.task( name="backend.app.tasks.backup.restore_images_task", bind=True, max_retries=0, soft_time_limit=21600, time_limit=23400, ) def restore_images_task(self, *, source_backup_run_id: int) -> dict: """Mirrors restore_db_task; uses backup_service.restore_images.""" SessionLocal = _sync_session_factory() now = datetime.now(UTC) with SessionLocal() as session: src = session.get(BackupRun, source_backup_run_id) if src is None or src.kind != "images" or not src.tar_path: raise ValueError( f"BackupRun id={source_backup_run_id} is not a valid images backup" ) marker = BackupRun( kind="images", status="restoring", triggered_by="restore", started_at=now, restored_from_id=src.id, manifest={"source_tar_path": src.tar_path}, ) session.add(marker) session.commit() session.refresh(marker) marker_id = marker.id tar_path = src.tar_path try: backup_service.restore_images( images_root=IMAGES_ROOT, tar_path=Path(tar_path), ) except (SoftTimeLimitExceeded, Exception) as exc: with SessionLocal() as session: row = session.get(BackupRun, marker_id) if row is not None: _mark_failed(session, row, exc) raise with SessionLocal() as session: row = session.get(BackupRun, marker_id) row.status = "restored" row.finished_at = datetime.now(UTC) session.commit() return {"backup_run_id": marker_id} @celery.task( name="backend.app.tasks.backup.prune_backups", soft_time_limit=300, time_limit=600, ) def prune_backups() -> dict: """Daily Beat. Per-kind retention from ImportSettings. Returns {"db_deleted": N, "images_deleted": M, "media_deleted": L, "files_unlinked": K}. Media runs leave no files here; only their history is trimmed, to the latest MEDIA_RUNS_KEPT. 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 'error' first. (Earlier docstring claimed the FC-3i TaskRun sweep handled those, but TaskRun cleanup never touched BackupRun rows. Audit 2026-06-02 added the dedicated sweep.) """ SessionLocal = _sync_session_factory() counts = {"db_deleted": 0, "images_deleted": 0, "media_deleted": 0, "files_unlinked": 0} 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) .where(BackupRun.kind == kind) .where(BackupRun.tag.is_(None)) .where(BackupRun.status.in_(["ok", "error"])) .order_by(BackupRun.started_at.desc()) .offset(keep) ).scalars().all() for row in candidates: result = backup_service.unlink_artifact_files( sql_path=row.sql_path, tar_path=row.tar_path, manifest_path=(row.manifest or {}).get("manifest_path"), ) counts["files_unlinked"] += sum( 1 for v in result.values() if v ) session.delete(row) counts[f"{kind}_deleted"] += 1 session.commit() return counts @celery.task( name="backend.app.tasks.backup.backup_db_nightly", soft_time_limit=60, time_limit=120, ) def backup_db_nightly() -> dict: """Hourly tick. Dispatches a real backup ONLY if the configured UTC hour matches and the nightly setting is enabled. Returns either {'skipped': ''} or {'dispatched': ''}.""" SessionLocal = _sync_session_factory() with SessionLocal() as session: s = ImportSettings.load_sync(session) nightly_enabled = s.backup_db_nightly_enabled configured_hour = s.backup_db_nightly_hour_utc if not nightly_enabled: return {"skipped": "nightly disabled"} now_hour = datetime.now(UTC).hour if now_hour != configured_hour: return {"skipped": f"hour={now_hour} != configured={configured_hour}"} res = backup_db_task.delay(triggered_by="nightly") 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)