293 lines
10 KiB
Python
293 lines
10 KiB
Python
"""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
|
|
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_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,
|
|
soft_time_limit=600, time_limit=720,
|
|
)
|
|
def backup_db_task(self, *, tag: str | None = None,
|
|
triggered_by: str = "manual") -> dict:
|
|
"""Create one DB backup. Returns {'backup_run_id': N}."""
|
|
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:
|
|
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.sql_path = result["sql_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.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, "files_unlinked": K}.
|
|
Tagged rows (tag IS NOT NULL) are never pruned.
|
|
Status='running' / 'restoring' rows are never pruned (recovery
|
|
sweep from FC-3i handles those via task_run).
|
|
"""
|
|
SessionLocal = _sync_session_factory()
|
|
counts = {"db_deleted": 0, "images_deleted": 0, "files_unlinked": 0}
|
|
with SessionLocal() as session:
|
|
s = session.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
for kind, keep in (
|
|
("db", s.backup_db_keep_last_n),
|
|
("images", s.backup_images_keep_last_n),
|
|
):
|
|
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': '<reason>'} or {'dispatched': '<task_id>'}."""
|
|
SessionLocal = _sync_session_factory()
|
|
with SessionLocal() as session:
|
|
s = session.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
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")
|
|
return {"dispatched": res.id}
|