feat(6847): a rollback first backs up the state it replaces

backup_job.rollback_to_backup:
1. Checks the backup can apply here (backup.check_replace, which shares
   _replace_mode with replace_from_backup), refusing before anything is written.
2. Takes the backup lock, so no scheduled backup runs during the rollback.
3. Saves the current store with save_state_before_restore.
4. Replaces the store.

The safety file goes to the backup destination when backups are on and it
takes it. Otherwise it goes to DATA_DIR/pre-restore, with a note when the
destination refused. If neither takes it the rollback does not run. The
result names where the file went.

Milestone 582, step 3.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/grimoire/services/backup.py

@@ -2961,6 +2961,43 @@ async def _reset_sequences(session) -> None:
         ))
 
 
+def _refuse_unless_full(data: dict) -> None:
+    if (data.get("version") or 1) < 2 or data.get("scope", "full") != "full":
+        raise RestoreRefused(
+            "This file isn't a full backup of the instance (it is a single user's "
+            "export, or a very old format), so it can't put the whole instance back."
+        )
+
+
+async def _replace_mode(session, data: dict) -> str:
+    """"fresh" or "rollback", or RestoreRefused saying why neither applies."""
+    from grimoire.services.instance import get_instance_id
+
+    _refuse_unless_full(data)
+    if await _is_fresh(session):
+        return "fresh"
+    backup_instance = data.get("instance_id")
+    if not backup_instance:
+        raise RestoreRefused(
+            "This backup was made before backups recorded which instance they "
+            "came from, so it can only be restored into a new, empty instance."
+        )
+    if backup_instance != await get_instance_id(session):
+        raise RestoreRefused(
+            "This backup comes from a different Grimoire instance. Rolling back "
+            "matches records by id, and its ids belong to that instance, so it "
+            "can only be restored into a new, empty instance."
+        )
+    return "rollback"
+
+
+async def check_replace(data: dict) -> str:
+    """Whether `data` can replace this store, without changing anything:
+    "fresh" or "rollback", or RestoreRefused with the reason."""
+    async with async_session() as session:
+        return await _replace_mode(session, data)
+
+
 async def replace_from_backup(data: dict) -> dict:
     """Put this store back to a full backup, in one transaction.
 
@@ -2981,40 +3018,18 @@ async def replace_from_backup(data: dict) -> dict:
     """
     from sqlalchemy import update
 
-    from grimoire.services.instance import adopt_instance_id, get_instance_id
-
-    if (data.get("version") or 1) < 2 or data.get("scope", "full") != "full":
-        raise RestoreRefused(
-            "This file isn't a full backup of the instance (it is a single user's "
-            "export, or a very old format), so it can't put the whole instance back."
-        )
-    backup_instance = data.get("instance_id")
+    from grimoire.services.instance import adopt_instance_id
 
     async with async_session() as session:
-        fresh = await _is_fresh(session)
-        own_instance = await get_instance_id(session)
-        if not fresh:
-            if not backup_instance:
-                raise RestoreRefused(
-                    "This backup was made before backups recorded which instance they "
-                    "came from, so it can only be restored into a new, empty instance."
-                )
-            if backup_instance != own_instance:
-                raise RestoreRefused(
-                    "This backup comes from a different Grimoire instance. Rolling back "
-                    "matches records by id, and its ids belong to that instance, so it "
-                    "can only be restored into a new, empty instance."
-                )
-
+        mode = await _replace_mode(session, data)
         await _clear_for_rollback(session, data)
         stats = await _load_v2(session, data, _Maps(), keep_ids=True)
         await _reset_sequences(session)
         await session.execute(update(User).values(session_version=User.session_version + 1))
-        if fresh and backup_instance:
-            await adopt_instance_id(session, backup_instance)
+        if mode == "fresh" and data.get("instance_id"):
+            await adopt_instance_id(session, data["instance_id"])
         await session.commit()
 
-    mode = "fresh" if fresh else "rollback"
     logger.info("Replaced the store from a backup (%s): %s", mode, stats)
     return {"mode": mode, "restored": stats, "exported_at": data.get("exported_at")}
 

src/grimoire/services/backup_job.py

@@ -24,11 +24,14 @@ import os
 import time
 from datetime import datetime, timedelta, timezone
 
+from pathlib import Path
+
 from apscheduler.triggers.cron import CronTrigger
 
 from grimoire.services.backup_destination import (
     DestinationError, deliver, load_config, load_secret,
 )
+from grimoire.services.backup_file import write_archive
 from grimoire.services.scheduler import ScheduledJob
 from grimoire.services.settings import get_admin_setting, set_admin_setting
 
@@ -45,6 +48,9 @@ STALE_AFTER = timedelta(hours=26)
 # has none, and a run that never returns holds the lock — every later backup
 # would be skipped as "already running", silently, forever.
 RUN_DEADLINE = timedelta(minutes=60)
+# Under DATA_DIR: where a rollback keeps the backup of what it replaced when
+# no backup destination takes it.
+PRE_RESTORE_DIR = "pre-restore"
 
 _LOCK = asyncio.Lock()
 _TASKS: set[asyncio.Task] = set()
@@ -155,6 +161,57 @@ async def _export_and_deliver(config, secret, started):
     )
 
 
+async def save_state_before_restore() -> dict:
+    """Back up the store as it is now, before a rollback replaces it, and say
+    where the file went: the backup destination when backups are on and it
+    takes the file, otherwise DATA_DIR/pre-restore. Raises when neither takes
+    it, so a rollback never runs without a way back."""
+    from grimoire.config import Config
+    from grimoire.services.backup import export_full_backup
+
+    at = datetime.now(timezone.utc)
+    app_version = os.environ.get("APP_VERSION", "dev")
+    payload = await export_full_backup()
+    note = None
+    config = await load_config()
+    if config.enabled:
+        secret, _ = await load_secret()
+        try:
+            delivered = await asyncio.wait_for(
+                deliver(payload, config, secret, at, app_version=app_version),
+                RUN_DEADLINE.total_seconds(),
+            )
+            return {"file": delivered.name, "size_bytes": delivered.size_bytes,
+                    "where": delivered.where}
+        except (DestinationError, TimeoutError, asyncio.TimeoutError) as exc:
+            note = (f"The backup destination didn't take it ({exc or 'it timed out'}), "
+                    "so it was kept on the instance instead.")
+    folder = Path(Config.DATA_DIR) / PRE_RESTORE_DIR
+    path = await asyncio.to_thread(write_archive, payload, folder, at, app_version=app_version)
+    saved = {"file": path.name, "size_bytes": path.stat().st_size, "where": str(folder)}
+    if note:
+        saved["note"] = note
+    return saved
+
+
+async def rollback_to_backup(data: dict) -> dict:
+    """Roll the store back to a full backup, after backing up what it
+    replaces. Refuses (RestoreRefused) before writing anything when the backup
+    can't be applied here, and raises BackupBusy while a backup is running:
+    a backup taken during a rollback would mix the two states."""
+    from grimoire.services.backup import check_replace, replace_from_backup
+
+    await check_replace(data)
+    if _LOCK.locked():
+        raise BackupBusy()
+    async with _LOCK:
+        safety = await save_state_before_restore()
+        result = await replace_from_backup(data)
+    result["safety_backup"] = safety
+    logger.info("rollback: the state it replaced is in %s/%s", safety["where"], safety["file"])
+    return result
+
+
 def start_backup_now() -> None:
     """Start a manual backup in the background (the request returns at once;
     the page reads progress from `is_running` and the run list). Raises