diff --git a/backend/src/backup_tool/worker.py b/backend/src/backup_tool/worker.py new file mode 100644 index 0000000..c5c147b --- /dev/null +++ b/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