"""Scheduled (recurring) stack backups to a destination. A background loop wakes every minute, runs any schedules whose ``next_run`` has passed and records status + the next run time. A schedule covers several stacks (or all of them); each is backed up on its own — its own file, upload, retention and status — one after the other, so one failing stack does not cost the others their backup. Times are stored as naive UTC to match SQLite's datetime handling. """ from __future__ import annotations import asyncio import json import logging import os from datetime import datetime, timedelta, timezone from sqlmodel import Session, select from database import engine from models.backup_destination import BackupDestination from models.backup_schedule import BackupSchedule, stacks_of from models.setting import EVENT_BACKUP_FAILED from models.stack import Stack from services import ( backup_destination_service as dest_service, backup_service, notify_service, ) logger = logging.getLogger("stackpilot.schedule") #: A stack that is busy (deploying, updating) when its turn comes is tried #: again after the others, up to this many more times, this far apart. BUSY_ATTEMPTS = 3 BUSY_WAIT_SECONDS = 120.0 #: Schedules with a run in progress (scheduler or "Run now"), so the same #: schedule never runs twice at once. _running: set[int] = set() def is_running(schedule_id: int) -> bool: return schedule_id in _running def resolve_stacks(session: Session, schedule: BackupSchedule) -> list[str]: if schedule.all_stacks: return sorted(s.id for s in session.exec(select(Stack)).all()) return stacks_of(schedule) def _now() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None) def compute_next_run( frequency: str, hour: int, minute: int, weekday: int, after: datetime ) -> datetime: """Next run strictly after ``after`` (naive UTC).""" if frequency == "hourly": nxt = after.replace(minute=minute % 60, second=0, microsecond=0) if nxt <= after: nxt += timedelta(hours=1) return nxt base = after.replace(hour=hour % 24, minute=minute % 60, second=0, microsecond=0) if frequency == "weekly": days_ahead = (weekday - base.weekday()) % 7 cand = base + timedelta(days=days_ahead) if cand <= after: cand += timedelta(days=7) return cand # daily (default) if base <= after: base += timedelta(days=1) return base def _prune(dest: BackupDestination, stack_id: str, keep: int) -> int: if keep <= 0: return 0 items = dest_service.list_backups(dest) pattern = backup_service.backup_name_pattern(stack_id) mine = [i for i in items if pattern.match(i["name"])] mine.sort(key=lambda x: x.get("modified") or 0, reverse=True) removed = 0 for old in mine[keep:]: try: dest_service.delete(dest, old["name"]) removed += 1 except dest_service.DestinationError as exc: logger.warning("retention delete failed for %s: %s", old["name"], exc) return removed def _problems(report: dict) -> list[str]: """What makes a backup that was written less than a success.""" problems = [f"not captured: {e}" for e in report.get("errors") or []] if report.get("restart_error"): problems.append(f"stack did not start again: {report['restart_error']}") return problems async def _alert(session: Session, title: str, body: str) -> None: try: await notify_service.notify(EVENT_BACKUP_FAILED, title, body, session) except Exception: # noqa: BLE001 pass class _Busy(Exception): pass async def _backup_stack( session: Session, schedule: BackupSchedule, dest: BackupDestination, stack_id: str ) -> str: """Back up, upload and rotate one stack. Returns its status line; raises _Busy when the stack is locked by another operation.""" stack = session.get(Stack, stack_id) if not stack: return "error: stack not found" path = None try: path, report = await backup_service.create_backup_ex( stack_id, stack.name, include_volumes=schedule.include_volumes, stop_first=schedule.stop_first, ) filename = backup_service.backup_filename(stack_id, schedule.include_volumes) await asyncio.to_thread(dest_service.upload, dest, path, filename) problems = _problems(report) if problems: # Retention only after a complete backup: an incomplete one must # not push the last good backup out. return f"warning: {'; '.join(problems)}" await asyncio.to_thread(_prune, dest, stack_id, schedule.keep) return "ok" except backup_service.StackBusyError as exc: raise _Busy(str(exc)) from exc except Exception as exc: # noqa: BLE001 logger.warning("Scheduled backup of %s failed: %s", stack_id, exc) return f"error: {exc}" finally: if path and os.path.exists(path): os.unlink(path) def _summarise(statuses: dict[str, str]) -> str: """One status line for the schedule from the per-stack ones.""" if not statuses: return "error: no stacks to back up" bad = {sid: st for sid, st in statuses.items() if st != "ok"} if not bad: return "ok" worst = "error" if any(not st.startswith("warning") for st in bad.values()) else "warning" if len(statuses) == 1: return next(iter(statuses.values())) detail = "; ".join(f"{sid}: {st}" for sid, st in sorted(bad.items())) return f"{worst}: {len(bad)} of {len(statuses)} stacks — {detail}" async def run_schedule(session: Session, schedule: BackupSchedule) -> dict: """Execute one schedule now. Updates status + next_run. Returns a summary.""" if schedule.id in _running: return {"ok": False, "error": "this schedule is already running"} _running.add(schedule.id) now = _now() statuses: dict[str, str] = {} dest = session.get(BackupDestination, schedule.destination_id) try: stack_ids = resolve_stacks(session, schedule) if not dest: statuses = {sid: f"error: destination {schedule.destination_id} not found" for sid in stack_ids} else: pending = list(stack_ids) for attempt in range(BUSY_ATTEMPTS + 1): busy: list[str] = [] for stack_id in pending: try: statuses[stack_id] = await _backup_stack(session, schedule, dest, stack_id) except _Busy as exc: busy.append(stack_id) statuses[stack_id] = f"error: {exc}" if not busy or attempt == BUSY_ATTEMPTS: break # Deploys and updates are short: wait, then take the busy # stacks again rather than skipping them until the next run. pending = busy await asyncio.sleep(BUSY_WAIT_SECONDS) schedule.last_status = _summarise(statuses)[:300] problems = {sid: st for sid, st in statuses.items() if st != "ok"} if problems or not statuses: await _alert( session, f"Scheduled backup to '{dest.name if dest else schedule.destination_id}' " + ("failed" if schedule.last_status.startswith("error") else "is incomplete"), "\n".join(f"{sid}: {st}" for sid, st in sorted(problems.items())) or schedule.last_status, ) logger.info("Scheduled backup %s: %s", schedule.id, schedule.last_status) return { "ok": schedule.last_status == "ok", "destination": dest.name if dest else None, "stacks": statuses, **({} if schedule.last_status == "ok" else {"error": schedule.last_status}), } finally: _running.discard(schedule.id) schedule.stack_status = json.dumps( {sid: {"status": st[:300], "at": now.isoformat()} for sid, st in statuses.items()} ) schedule.last_run = now schedule.next_run = compute_next_run( schedule.frequency, schedule.hour, schedule.minute, schedule.weekday, now ) session.add(schedule) session.commit() session.refresh(schedule) async def run_in_background(schedule_id: int) -> None: """Run a schedule with its own session, for "Run now": a run over many stacks takes far longer than a request may.""" with Session(engine) as session: schedule = session.get(BackupSchedule, schedule_id) if schedule: await run_schedule(session, schedule) async def run_due() -> None: now = _now() with Session(engine) as session: schedules = session.exec( select(BackupSchedule).where(BackupSchedule.enabled == True) # noqa: E712 ).all() due = [] for s in schedules: if s.next_run is None: s.next_run = compute_next_run(s.frequency, s.hour, s.minute, s.weekday, now) session.add(s) elif s.next_run <= now: due.append(s) session.commit() for s in due: if not is_running(s.id): await run_schedule(session, s) async def scheduler_loop() -> None: await asyncio.sleep(20) # let startup settle while True: try: await run_due() except Exception as exc: # noqa: BLE001 logger.warning("scheduler tick failed: %s", exc) await asyncio.sleep(60)