feat(v2): add durable worker lifecycle service

backend/src/backup_tool/worker.py

@@ -0,0 +1,92 @@
+"""Single-node durable worker role.
+
+The worker owns leases; backup publishing is deliberately supplied by later M6 work.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import signal
+from datetime import UTC, datetime
+from uuid import uuid4
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from backup_tool.config import Settings
+from backup_tool.db.engine import create_engine
+from backup_tool.db.models import Execution
+from backup_tool.execution import claim, complete_cancellation, heartbeat, recover_stale, transition
+
+
+class Worker:
+    def __init__(self, settings: Settings, *, owner: str | None = None) -> None:
+        self.settings = settings
+        self.owner = owner or f"worker-{uuid4()}"
+        self._stopping = asyncio.Event()
+        self.engine = create_engine(settings)
+        self.sessions = async_sessionmaker(self.engine, expire_on_commit=False)
+
+    async def startup(self) -> int:
+        async with self.sessions() as db:
+            return await recover_stale(db)
+
+    async def run_once(self) -> bool:
+        async with self.sessions() as db:
+            execution_id = await db.scalar(
+                select(Execution.id)
+                .where(Execution.state == "queued")
+                .order_by(Execution.created_at)
+                .limit(1)
+            )
+            if execution_id is None:
+                return False
+            execution = await claim(db, execution_id, self.owner)
+            if execution is None:
+                return False
+            # M6 supplies source/repository work; this slice proves durable ownership.
+            if execution.state == "cancelling":
+                await complete_cancellation(db, execution.id, self.owner)
+                return True
+            now = datetime.now(UTC)
+            await db.execute(
+                update(Execution)
+                .where(
+                    Execution.id == execution.id,
+                    Execution.lease_owner == self.owner,
+                    Execution.lease_expires_at >= now,
+                    Execution.state == "preparing",
+                )
+                .values(state=transition("preparing", "running"), started_at=now)
+            )
+            await db.commit()
+            await heartbeat(db, execution.id, self.owner)
+            return True
+
+    async def run(self) -> None:
+        await self.startup()
+        while not self._stopping.is_set():
+            if not await self.run_once():
+                try:
+                    await asyncio.wait_for(self._stopping.wait(), timeout=0.25)
+                except TimeoutError:
+                    pass
+        await self.engine.dispose()
+
+    def stop(self) -> None:
+        self._stopping.set()
+
+
+def run_worker(settings: Settings) -> int:
+    worker = Worker(settings)
+    loop = asyncio.new_event_loop()
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        try:
+            loop.add_signal_handler(sig, worker.stop)
+        except NotImplementedError:
+            pass
+    try:
+        loop.run_until_complete(worker.run())
+    finally:
+        loop.close()
+    return 0