Add local backup retention: keep-local toggle, download, delete-local

Lets operators opt into keeping the local archive after upload (global
keep_local_backups setting), pull a remote archive back down into the
Proxmox dump directory for manual restore, and drop a locally-retained
copy without touching the remote object. Consolidates deletion so
retention, job-cascade delete, manual delete, and crash-recovery resume
all clean up local copies alongside remote ones.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

backend/app/backup_service.py

@@ -22,14 +22,7 @@ ACTIVE_STATES = {
     "deleting",
 }
 RUNNING_STATES = ACTIVE_STATES - {"queued"}
-RECOVERABLE_STATES = {
-    "pve_running",
-    "local_ready",
-    "uploading",
-    "remote_ready",
-    "local_deleting",
-    "deleting",
-}
+RECOVERABLE_STATES = (ACTIVE_STATES - {"queued"}) | {"local_downloading"}
 TERMINAL_STATES = {"completed", "failed", "deleted"}
 
 
@@ -298,10 +291,11 @@ async def process_backup(backup_id: str) -> None:
             backup = update_backup(backup_id, state="local_deleting")
 
         if backup["state"] == "local_deleting":
-            if backup.get("local_path"):
+            if not settings.get("keep_local_backups") and backup.get("local_path"):
                 local_path = Path(backup["local_path"])
                 if local_path.exists():
                     local_path.unlink()
+                backup = update_backup(backup_id, local_path=None)
             backup = update_backup(backup_id, state="completed", completed_at=utc_now())
             await notify("backup_completed", backup=backup)
             await apply_retention_for_job(backup["job_id"])
@@ -310,15 +304,26 @@ async def process_backup(backup_id: str) -> None:
         await notify("backup_failed", backup=failed)
 
 
-async def delete_remote_backup(backup: dict, reason: str) -> None:
-    if not backup.get("remote_path"):
-        update_backup(backup["id"], state="deleted", deleted_at=utc_now())
-        return
+async def delete_remote_backup(backup: dict, reason: str) -> dict:
     settings = get_settings()
     update_backup(backup["id"], state="deleting")
-    await _delete_remote_file_and_empty_parent(settings, backup["remote_path"])
-    deleted = update_backup(backup["id"], state="deleted", deleted_at=utc_now())
+    if backup.get("remote_path"):
+        exists = await asyncio.to_thread(
+            commands.rclone_object_exists,
+            settings["rclone_path"],
+            backup["remote_path"],
+        )
+        if exists:
+            await _delete_remote_file_and_empty_parent(settings, backup["remote_path"])
+        else:
+            await _remove_empty_remote_parent(settings, backup["remote_path"])
+    if backup.get("local_path"):
+        local_path = Path(backup["local_path"])
+        if local_path.exists() and local_path.is_file():
+            local_path.unlink()
+    deleted = update_backup(backup["id"], state="deleted", deleted_at=utc_now(), local_path=None)
     await notify("remote_deleted", backup=deleted, message=reason)
+    return deleted
 
 
 async def delete_backup_manually(backup_id: str, reason: str = "manual deletion") -> dict:
@@ -327,28 +332,61 @@ async def delete_backup_manually(backup_id: str, reason: str = "manual deletion"
         return purge_deleted_backup(backup_id)
     if backup["state"] in ACTIVE_STATES:
         raise RuntimeError(f"Backup {backup_id} is active in state {backup['state']} and cannot be deleted")
+    return await delete_remote_backup(backup, reason)
+
+
+def _archive_name_from_remote(remote_path: str) -> str:
+    return Path(remote_path.split(":", 1)[1]).name
+
+
+async def download_backup_locally(backup_id: str) -> dict:
+    backup = _backup_row(backup_id)
+    if backup["state"] != "completed" or not backup.get("remote_path"):
+        raise RuntimeError(f"Backup {backup_id} has no completed remote archive to download")
+    if backup.get("local_path") and Path(backup["local_path"]).exists():
+        return backup
+
+    now = utc_now()
+    with db() as conn:
+        cur = conn.execute(
+            "UPDATE backups SET state = 'local_downloading', updated_at = ? WHERE id = ? AND state = 'completed'",
+            (now, backup_id),
+        )
+        if cur.rowcount == 0:
+            raise RuntimeError(f"Backup {backup_id} is busy and cannot be downloaded right now")
 
     settings = get_settings()
-    if backup.get("remote_path") and backup["state"] != "deleted":
-        exists = await asyncio.to_thread(
-            commands.rclone_object_exists,
+    target = Path(settings["local_backup_dir"]) / _archive_name_from_remote(backup["remote_path"])
+    try:
+        await asyncio.to_thread(
+            commands.rclone_download,
             settings["rclone_path"],
             backup["remote_path"],
+            str(target),
         )
-        if exists:
-            update_backup(backup_id, state="deleting")
-            await _delete_remote_file_and_empty_parent(settings, backup["remote_path"])
-        else:
-            await _remove_empty_remote_parent(settings, backup["remote_path"])
+        if not target.exists():
+            raise RuntimeError("Download finished but local archive was not found")
+    except Exception:
+        update_backup(backup_id, state="completed")
+        raise
 
-    if backup.get("local_path"):
-        local_path = Path(backup["local_path"])
-        if local_path.exists() and local_path.is_file():
-            local_path.unlink()
+    downloaded = update_backup(backup_id, state="completed", local_path=str(target))
+    await notify("local_downloaded", backup=downloaded)
+    return downloaded
 
-    deleted = update_backup(backup_id, state="deleted", deleted_at=utc_now())
-    await notify("remote_deleted", backup=deleted, message=reason)
-    return deleted
+
+async def delete_local_copy(backup_id: str) -> dict:
+    backup = _backup_row(backup_id)
+    if backup["state"] != "completed":
+        raise RuntimeError(f"Backup {backup_id} is in state {backup['state']} and has no local copy to delete")
+    if not backup.get("local_path"):
+        return backup
+    local_path = Path(backup["local_path"])
+    if local_path.exists() and local_path.is_file():
+        local_path.unlink()
+    updated = update_backup(backup_id, local_path=None)
+    await notify("local_deleted", backup=updated)
+    return updated
 
 
 async def delete_job_with_backups(job_id: int) -> dict:
@@ -440,6 +478,12 @@ async def recover_unfinished() -> None:
         await notify("recovery_action", backup=backup, message=f"resuming state {backup['state']}")
         if backup["state"] == "deleting":
             await delete_remote_backup(backup, "recovery delete resume")
+        elif backup["state"] == "local_downloading":
+            update_backup(backup["id"], state="completed")
+            try:
+                await download_backup_locally(backup["id"])
+            except Exception as exc:
+                await notify("recovery_action", backup=backup, message=f"resume download failed: {exc}")
         else:
             await process_backup(backup["id"])