feat(v2): enqueue and lease durable executions

backend/src/backup_tool/execution.py

@@ -1,6 +1,13 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import select, update
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backup_tool.db.models import Execution, Job
 
 ACTIVE_STATES = frozenset({"queued", "preparing", "running", "verifying", "cancelling"})
 TERMINAL_STATES = frozenset({"committed", "cancelled", "failed"})
@@ -21,8 +28,97 @@ class TransitionError(ValueError):
     code = "invalid_transition"
 
 
+class EnqueueError(ValueError):
+    def __init__(self, code: str, detail: str, active_execution_id: str | None = None):
+        super().__init__(detail)
+        self.code = code
+        self.active_execution_id = active_execution_id
+
+
 def transition(current: str, target: str) -> str:
     """Validate and return one durable execution-state transition."""
     if target not in _ALLOWED.get(current, frozenset()):
         raise TransitionError(f"Cannot transition execution from {current!r} to {target!r}.")
     return target
+
+
+async def enqueue(db: AsyncSession, job_id: str, trigger: str = "manual") -> Execution:
+    """Create exactly one active execution for an enabled active job."""
+    job = await db.get(Job, job_id)
+    if job is None:
+        raise EnqueueError("resource_not_found", "Job was not found.")
+    if job.state != "active" or not job.enabled:
+        raise EnqueueError("job_disabled", "Job is disabled or archived.")
+    job_identifier = job.id
+    execution = Execution(job_id=job_identifier, trigger=trigger, progress={})
+    db.add(execution)
+    try:
+        await db.commit()
+    except IntegrityError as error:
+        await db.rollback()
+        existing = await db.scalar(
+            select(Execution.id).where(
+                Execution.job_id == job_identifier, Execution.state.in_(ACTIVE_STATES)
+            )
+        )
+        raise EnqueueError(
+            "execution_active", "Job already has an active execution.", existing
+        ) from error
+    await db.refresh(execution)
+    return execution
+
+
+async def claim(
+    db: AsyncSession, execution_id: str, owner: str, lease_seconds: int = 60
+) -> Execution | None:
+    """Atomically lease a queued execution; a stale lease may be recovered."""
+    now = datetime.now(UTC)
+    expires = now + timedelta(seconds=lease_seconds)
+    result = await db.execute(
+        update(Execution)
+        .where(
+            Execution.id == execution_id,
+            Execution.state.in_({"queued", "preparing"}),
+            (Execution.lease_expires_at.is_(None)) | (Execution.lease_expires_at < now),
+        )
+        .values(state="preparing", lease_owner=owner, lease_expires_at=expires, heartbeat_at=now)
+    )
+    if getattr(result, "rowcount", 0) != 1:
+        await db.rollback()
+        return None
+    await db.commit()
+    return await db.get(Execution, execution_id)
+
+
+async def heartbeat(
+    db: AsyncSession, execution_id: str, owner: str, lease_seconds: int = 60
+) -> bool:
+    now = datetime.now(UTC)
+    result = await db.execute(
+        update(Execution)
+        .where(
+            Execution.id == execution_id,
+            Execution.lease_owner == owner,
+            Execution.state.in_(ACTIVE_STATES),
+        )
+        .values(heartbeat_at=now, lease_expires_at=now + timedelta(seconds=lease_seconds))
+    )
+    await db.commit()
+    return getattr(result, "rowcount", 0) == 1
+
+
+async def recover_stale(db: AsyncSession) -> int:
+    now = datetime.now(UTC)
+    result = await db.execute(
+        update(Execution)
+        .where(Execution.state.in_({"preparing", "running"}), Execution.lease_expires_at < now)
+        .values(
+            state="queued",
+            lease_owner=None,
+            lease_expires_at=None,
+            heartbeat_at=None,
+            reason_code="worker_lost",
+        )
+    )
+    await db.commit()
+    return getattr(result, "rowcount", 0) or 0