316 lines
13 KiB
Python
316 lines
13 KiB
Python
"""Single-node durable worker role.
|
|
|
|
The worker owns leases and performs repository I/O outside the API process.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import importlib
|
|
import signal
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import cast
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from backup_tool.config import Settings
|
|
from backup_tool.db.engine import create_engine
|
|
from backup_tool.db.models import Backup, Execution, Job, Repository, Restore, Source
|
|
from backup_tool.execution import (
|
|
claim,
|
|
complete_cancellation,
|
|
heartbeat,
|
|
record_event,
|
|
recover_stale,
|
|
transition,
|
|
)
|
|
from backup_tool.faults import FaultInjector, NoFault
|
|
from backup_tool.gc import process_retention_gc
|
|
from backup_tool.notifications.dispatcher import dispatch_one, recover_notification_leases
|
|
from backup_tool.notifications.events import emit_event
|
|
from backup_tool.observability.logging import configure_logging, log_event
|
|
from backup_tool.repository import reconcile_key_rotations
|
|
from backup_tool.security.secrets import EnvelopeCipher
|
|
|
|
snapshot = importlib.import_module("backup_tool.snapshot")
|
|
SnapshotError = snapshot.SnapshotError
|
|
SnapshotIntegrityError = snapshot.SnapshotIntegrityError
|
|
|
|
|
|
class Worker:
|
|
def __init__(
|
|
self,
|
|
settings: Settings,
|
|
*,
|
|
owner: str | None = None,
|
|
fault_injector: FaultInjector | None = None,
|
|
) -> None:
|
|
self.settings = settings
|
|
self.owner = owner or f"worker-{uuid4()}"
|
|
self.fault_injector = fault_injector or NoFault()
|
|
self._stopping = asyncio.Event()
|
|
self.engine = create_engine(settings)
|
|
self.sessions = async_sessionmaker(self.engine, expire_on_commit=False)
|
|
self.cipher = EnvelopeCipher.from_file(settings.master_key_file)
|
|
self._execution_turns = 0
|
|
self._next_maintenance_at: datetime | None = None
|
|
|
|
async def startup(self) -> int:
|
|
async with self.sessions() as db:
|
|
rotations = await reconcile_key_rotations(self.settings, db)
|
|
publications = cast(int, await snapshot.reconcile_publications(self.settings, db))
|
|
restored = cast(int, await snapshot.reconcile_restores(self.settings, db))
|
|
recovered_executions = await recover_stale(db)
|
|
recovered_deliveries = await recover_notification_leases(db)
|
|
maintenance = await process_retention_gc(db)
|
|
self._next_maintenance_at = datetime.now(UTC) + timedelta(seconds=60)
|
|
return (
|
|
rotations
|
|
+ publications
|
|
+ restored
|
|
+ recovered_executions
|
|
+ recovered_deliveries
|
|
+ maintenance.tombstoned
|
|
)
|
|
|
|
async def _run_restore(self, db: AsyncSession) -> bool:
|
|
restore_id = await db.scalar(
|
|
select(Restore.id)
|
|
.where(Restore.state == "queued")
|
|
.order_by(Restore.created_at)
|
|
.limit(1)
|
|
)
|
|
if restore_id is None:
|
|
return False
|
|
result = await db.execute(
|
|
update(Restore)
|
|
.where(Restore.id == restore_id, Restore.state == "queued")
|
|
.values(state="running")
|
|
)
|
|
if getattr(result, "rowcount", 0) != 1:
|
|
await db.rollback()
|
|
return False
|
|
await db.commit()
|
|
restore = await db.get(Restore, restore_id)
|
|
if restore is None:
|
|
return False
|
|
try:
|
|
backup = await db.get(Backup, restore.backup_id)
|
|
if backup is None or backup.integrity != "verified" or backup.tombstoned_at is not None:
|
|
raise SnapshotError("backup is unavailable")
|
|
execution = await db.get(Execution, backup.execution_id)
|
|
if execution is None:
|
|
raise SnapshotError("backup execution is unavailable")
|
|
job = await db.get(Job, execution.job_id)
|
|
if job is None:
|
|
raise SnapshotError("backup job is unavailable")
|
|
repository = await db.get(Repository, job.repository_id)
|
|
if repository is None:
|
|
raise SnapshotError("backup repository is unavailable")
|
|
restore.result = await snapshot.restore_full_snapshot(
|
|
self.settings, db, restore, backup, repository
|
|
)
|
|
restore.state = "committed"
|
|
await emit_event(
|
|
db,
|
|
"restore.committed",
|
|
correlation_id=restore.id,
|
|
resource={"restore_id": restore.id, "backup_id": restore.backup_id},
|
|
payload={"dry_run": restore.dry_run, "outcome": "committed"},
|
|
deduplication_key=f"restore:{restore.id}:committed",
|
|
)
|
|
await db.commit()
|
|
except SnapshotIntegrityError:
|
|
await db.rollback()
|
|
failed = await db.get(Restore, restore_id)
|
|
if failed is not None:
|
|
corrupted_backup = await db.get(Backup, failed.backup_id)
|
|
if corrupted_backup is not None:
|
|
corrupted_backup.integrity = "corrupt"
|
|
failed.state = "failed"
|
|
failed.result = {"reason": "restore_failed"}
|
|
await emit_event(
|
|
db,
|
|
"restore.failed",
|
|
correlation_id=failed.id,
|
|
resource={"restore_id": failed.id, "backup_id": failed.backup_id},
|
|
payload={"reason_code": "restore_failed"},
|
|
deduplication_key=f"restore:{failed.id}:failed",
|
|
)
|
|
await db.commit()
|
|
except SnapshotError:
|
|
await db.rollback()
|
|
failed = await db.get(Restore, restore_id)
|
|
if failed is not None:
|
|
failed.state = "failed"
|
|
failed.result = {"reason": "restore_failed"}
|
|
await emit_event(
|
|
db,
|
|
"restore.failed",
|
|
correlation_id=failed.id,
|
|
resource={"restore_id": failed.id, "backup_id": failed.backup_id},
|
|
payload={"reason_code": "restore_failed"},
|
|
deduplication_key=f"restore:{failed.id}:failed",
|
|
)
|
|
await db.commit()
|
|
return True
|
|
|
|
async def _run_maintenance(self, db: AsyncSession) -> None:
|
|
now = datetime.now(UTC)
|
|
if self._next_maintenance_at is None or now >= self._next_maintenance_at:
|
|
await process_retention_gc(db, now)
|
|
self._next_maintenance_at = now + timedelta(seconds=60)
|
|
|
|
async def run_once(self) -> bool:
|
|
if self._stopping.is_set():
|
|
return False
|
|
async with self.sessions() as db:
|
|
# Retention/GC is a worker responsibility, but is rate-limited so it
|
|
# cannot turn a sustained backup queue into a metadata polling loop.
|
|
await self._run_maintenance(db)
|
|
# Never let an always-nonempty execution queue starve due notifications.
|
|
if self._execution_turns >= 1 and await dispatch_one(
|
|
db, self.settings, self.cipher, self.owner
|
|
):
|
|
self._execution_turns = 0
|
|
return True
|
|
if self._stopping.is_set():
|
|
return False
|
|
execution_id = await db.scalar(
|
|
select(Execution.id)
|
|
.where(Execution.state == "queued")
|
|
.order_by(Execution.created_at)
|
|
.limit(1)
|
|
)
|
|
if execution_id is None:
|
|
if await self._run_restore(db):
|
|
return True
|
|
return await dispatch_one(db, self.settings, self.cipher, self.owner)
|
|
execution = await claim(db, execution_id, self.owner)
|
|
if execution is None:
|
|
return False
|
|
self._execution_turns += 1
|
|
# 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
|
|
now = datetime.now(UTC)
|
|
result = 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)
|
|
)
|
|
if getattr(result, "rowcount", 0) == 1:
|
|
await db.refresh(execution)
|
|
await record_event(db, execution)
|
|
await db.commit()
|
|
await heartbeat(db, execution.id, self.owner)
|
|
failure_detail: str | None = None
|
|
failure_reason = "transient_io"
|
|
try:
|
|
job = await db.get(Job, execution.job_id)
|
|
if job is None:
|
|
raise SnapshotError("execution job is unavailable")
|
|
source = await db.get(Source, job.source_id)
|
|
repository = await db.get(Repository, job.repository_id)
|
|
if source is None or repository is None:
|
|
raise SnapshotError("execution source or repository is unavailable")
|
|
backup = await snapshot.publish_full_snapshot(
|
|
self.settings,
|
|
db,
|
|
execution,
|
|
job,
|
|
source,
|
|
repository,
|
|
self.fault_injector,
|
|
cipher=self.cipher,
|
|
)
|
|
await db.flush()
|
|
await emit_event(
|
|
db,
|
|
"backup.committed",
|
|
correlation_id=execution.id,
|
|
resource={
|
|
"execution_id": execution.id,
|
|
"job_id": job.id,
|
|
"repository_id": repository.id,
|
|
"backup_id": backup.id,
|
|
},
|
|
payload={"integrity": backup.integrity, "effective_mode": job.requested_mode},
|
|
deduplication_key=f"backup:{backup.id}:committed",
|
|
)
|
|
await emit_event(
|
|
db,
|
|
"backup.verification_succeeded",
|
|
correlation_id=execution.id,
|
|
resource={"execution_id": execution.id, "backup_id": backup.id},
|
|
payload={"integrity": backup.integrity},
|
|
deduplication_key=f"backup:{backup.id}:verified",
|
|
)
|
|
execution.state = transition("running", "verifying")
|
|
await record_event(db, execution)
|
|
self.fault_injector.hit("metadata.before_commit")
|
|
await db.commit()
|
|
self.fault_injector.hit("metadata.after_commit")
|
|
execution.state = transition("verifying", "committed")
|
|
execution.completed_at = datetime.now(UTC)
|
|
execution.lease_owner = None
|
|
execution.lease_expires_at = None
|
|
await record_event(db, execution)
|
|
await db.commit()
|
|
snapshot.finalize_publication(Path(repository.root), execution.id)
|
|
except SnapshotError as error:
|
|
await db.rollback()
|
|
failure_detail = str(error)
|
|
failure_reason = error.reason_code
|
|
if failure_detail is not None:
|
|
failed = await db.get(Execution, execution_id)
|
|
if failed is None or failed.state not in {"preparing", "running", "verifying"}:
|
|
return True
|
|
failed.state = "failed"
|
|
failed.reason_code = failure_reason
|
|
failed.operator_message = failure_detail
|
|
failed.completed_at = datetime.now(UTC)
|
|
failed.lease_owner = None
|
|
failed.lease_expires_at = None
|
|
await record_event(db, failed)
|
|
await db.commit()
|
|
return True
|
|
|
|
async def run(self) -> None:
|
|
await self.startup()
|
|
while not self._stopping.is_set():
|
|
if not await self.run_once():
|
|
with contextlib.suppress(TimeoutError):
|
|
await asyncio.wait_for(self._stopping.wait(), timeout=0.25)
|
|
await self.engine.dispose()
|
|
|
|
def stop(self) -> None:
|
|
log_event("role_stopping", role="worker")
|
|
self._stopping.set()
|
|
|
|
|
|
def run_worker(settings: Settings) -> int:
|
|
configure_logging("worker", settings.log_level)
|
|
worker = Worker(settings)
|
|
loop = asyncio.new_event_loop()
|
|
for sig in (signal.SIGINT, signal.SIGTERM):
|
|
with contextlib.suppress(NotImplementedError):
|
|
loop.add_signal_handler(sig, worker.stop)
|
|
try:
|
|
loop.run_until_complete(worker.run())
|
|
finally:
|
|
loop.close()
|
|
log_event("role_stopped", role="worker")
|
|
return 0
|