From bb87b04ca8dad64bb99669db081e9d1c8b17b4e3 Mon Sep 17 00:00:00 2001
From: Alex Blank <alexblank@fastmail.com>
Date: Mon, 27 Jul 2026 22:48:08 +0200
Subject: [PATCH] fix(v2): preserve execution cancellation and retry safety

---
 backend/src/backup_tool/execution.py | 31 ++++++++++++++++++++++++----
 backend/src/backup_tool/worker.py    |  3 ++-
 2 files changed, 29 insertions(+), 5 deletions(-)

diff --git a/backend/src/backup_tool/execution.py b/backend/src/backup_tool/execution.py
index aaab7a5..9e412db 100644
--- a/backend/src/backup_tool/execution.py
+++ b/backend/src/backup_tool/execution.py
@@ -164,7 +164,18 @@ async def retry(db: AsyncSession, execution_id: str) -> Execution | None:
     execution.heartbeat_at = None
     execution.reason_code = None
     execution.operator_message = None
-    await db.commit()
+    try:
+        await db.commit()
+    except IntegrityError as error:
+        await db.rollback()
+        active_id = await db.scalar(
+            select(Execution.id).where(
+                Execution.job_id == execution.job_id, Execution.state.in_(ACTIVE_STATES)
+            )
+        )
+        raise EnqueueError(
+            "execution_active", "Job already has an active execution.", active_id
+        ) from error
     await db.refresh(execution)
     return execution
 
@@ -199,10 +210,22 @@ def public_event(execution: Execution) -> dict[str, object]:
 
 async def recover_stale(db: AsyncSession) -> int:
     now = datetime.now(UTC)
-    result = await db.execute(
+    cancelled = await db.execute(
+        update(Execution)
+        .where(Execution.state == "cancelling", Execution.lease_expires_at < now)
+        .values(
+            state="cancelled",
+            completed_at=now,
+            lease_owner=None,
+            lease_expires_at=None,
+            heartbeat_at=None,
+            reason_code="cancellation_requested",
+        )
+    )
+    recovered = await db.execute(
         update(Execution)
         .where(
-            Execution.state.in_({"preparing", "running", "verifying", "cancelling"}),
+            Execution.state.in_({"preparing", "running", "verifying"}),
             Execution.lease_expires_at < now,
         )
         .values(
@@ -214,4 +237,4 @@ async def recover_stale(db: AsyncSession) -> int:
         )
     )
     await db.commit()
-    return getattr(result, "rowcount", 0) or 0
+    return (getattr(cancelled, "rowcount", 0) or 0) + (getattr(recovered, "rowcount", 0) or 0)
diff --git a/backend/src/backup_tool/worker.py b/backend/src/backup_tool/worker.py
index 39d51b7..ae537e9 100644
--- a/backend/src/backup_tool/worker.py
+++ b/backend/src/backup_tool/worker.py
@@ -45,7 +45,8 @@ class Worker:
             execution = await claim(db, execution_id, self.owner)
             if execution is None:
                 return False
-            # M6 supplies source/repository work; this slice proves durable ownership.
+            # Reload after the claim: a control request can race the lease acquisition.
+            await db.refresh(execution)
             if execution.state == "cancelling":
                 await complete_cancellation(db, execution.id, self.owner)
                 return True