diff --git a/backend/src/backup_tool/api/app.py b/backend/src/backup_tool/api/app.py index 723620d..8eac08d 100644 --- a/backend/src/backup_tool/api/app.py +++ b/backend/src/backup_tool/api/app.py @@ -1,6 +1,7 @@ import asyncio import base64 import hashlib +import json from collections.abc import AsyncIterator from datetime import UTC, datetime, timedelta from pathlib import Path @@ -8,7 +9,7 @@ from typing import Annotated, Any, cast from fastapi import Depends, FastAPI, Header, Request, Response from fastapi.exceptions import RequestValidationError -from fastapi.responses import JSONResponse +from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel, Field from sqlalchemy import desc, select from sqlalchemy.exc import IntegrityError @@ -21,6 +22,7 @@ from backup_tool.db.engine import SchemaNotCurrentError, assert_schema_current, from backup_tool.db.models import ( ApiToken, AuditEvent, + Execution, IdempotencyRecord, Job, Repository, @@ -29,7 +31,13 @@ from backup_tool.db.models import ( Source, User, ) -from backup_tool.execution import EnqueueError, enqueue +from backup_tool.execution import ( + EnqueueError, + enqueue, + public_event, + request_cancellation, + retry, +) from backup_tool.repository import ( RepositoryError, initialize, @@ -49,6 +57,10 @@ from backup_tool.security.redaction import redact from backup_tool.security.secrets import EnvelopeCipher +async def single_execution_event(event: dict[str, object]) -> AsyncIterator[str]: + yield f"event: execution\\ndata: {json.dumps(event)}\\n\\n" + + class Problem(Exception): def __init__(self, status: int, code: str, detail: str): self.status = status @@ -777,6 +789,56 @@ def create_app(settings: Settings) -> FastAPI: "attempt": execution.attempt, } + @app.get("/api/v2/executions/{execution_id}") + async def get_execution( + execution_id: str, + db: Annotated[AsyncSession, Depends(session)], + _: Annotated[tuple[User, set[str], bool], Depends(actor)], + ) -> dict[str, object]: + execution = await db.get(Execution, execution_id) + if execution is None: + raise Problem(404, "resource_not_found", "Execution was not found.") + return public_event(execution) + + @app.post("/api/v2/executions/{execution_id}/cancel", status_code=202) + async def cancel_execution( + execution_id: str, + db: Annotated[AsyncSession, Depends(session)], + _: Annotated[tuple[User, set[str], bool], Depends(require)], + ) -> dict[str, object]: + execution = await request_cancellation(db, execution_id) + if execution is None: + raise Problem(409, "cancellation_not_allowed", "Execution cannot be cancelled.") + return public_event(execution) + + @app.post("/api/v2/executions/{execution_id}/retry", status_code=202) + async def retry_execution( + execution_id: str, + db: Annotated[AsyncSession, Depends(session)], + _: Annotated[tuple[User, set[str], bool], Depends(require)], + ) -> dict[str, object]: + try: + execution = await retry(db, execution_id) + except EnqueueError as error: + raise Problem(409, error.code, str(error)) from error + if execution is None: + raise Problem(404, "resource_not_found", "Execution was not found.") + return public_event(execution) + + @app.get("/api/v2/executions/{execution_id}/events") + async def execution_events( + execution_id: str, + db: Annotated[AsyncSession, Depends(session)], + _: Annotated[tuple[User, set[str], bool], Depends(actor)], + ) -> StreamingResponse: + execution = await db.get(Execution, execution_id) + if execution is None: + raise Problem(404, "resource_not_found", "Execution was not found.") + + return StreamingResponse( + single_execution_event(public_event(execution)), media_type="text/event-stream" + ) + @app.get("/api/v2/audit") async def list_audit( db: Annotated[AsyncSession, Depends(session)], diff --git a/backend/src/backup_tool/execution.py b/backend/src/backup_tool/execution.py index d64bd38..4f4fc0d 100644 --- a/backend/src/backup_tool/execution.py +++ b/backend/src/backup_tool/execution.py @@ -11,6 +11,7 @@ from backup_tool.db.models import Execution, Job ACTIVE_STATES = frozenset({"queued", "preparing", "running", "verifying", "cancelling"}) TERMINAL_STATES = frozenset({"committed", "cancelled", "failed"}) +TRANSIENT_REASONS = frozenset({"worker_lost", "timeout", "unavailable", "transient_io"}) _ALLOWED: Mapping[str, frozenset[str]] = { "queued": frozenset({"preparing", "cancelled", "failed"}), @@ -107,6 +108,71 @@ async def heartbeat( return getattr(result, "rowcount", 0) == 1 +async def request_cancellation(db: AsyncSession, execution_id: str) -> Execution | None: + result = await db.execute( + update(Execution) + .where( + Execution.id == execution_id, + Execution.state.in_({"queued", "preparing", "running"}), + ) + .values(state="cancelling", reason_code="cancellation_requested") + ) + if getattr(result, "rowcount", 0) != 1: + await db.rollback() + return None + await db.commit() + return await db.get(Execution, execution_id) + + +async def complete_cancellation(db: AsyncSession, execution_id: str, owner: str) -> bool: + result = await db.execute( + update(Execution) + .where( + Execution.id == execution_id, + Execution.lease_owner == owner, + Execution.state == "cancelling", + ) + .values( + state="cancelled", + completed_at=datetime.now(UTC), + lease_owner=None, + lease_expires_at=None, + ) + ) + await db.commit() + return getattr(result, "rowcount", 0) == 1 + + +async def retry(db: AsyncSession, execution_id: str) -> Execution | None: + execution = await db.get(Execution, execution_id) + if execution is None: + return None + if execution.state != "failed" or execution.reason_code not in TRANSIENT_REASONS: + raise EnqueueError("retry_not_allowed", "Execution failure is not retryable.") + execution.state = "queued" + execution.attempt += 1 + execution.lease_owner = None + execution.lease_expires_at = None + execution.heartbeat_at = None + execution.reason_code = None + execution.operator_message = None + await db.commit() + await db.refresh(execution) + return execution + + +def public_event(execution: Execution) -> dict[str, object]: + """Return redacted progress suitable for polling or SSE.""" + progress = {k: v for k, v in execution.progress.items() if k not in {"path", "secret", "token"}} + return { + "id": execution.id, + "state": execution.state, + "attempt": execution.attempt, + "reason_code": execution.reason_code, + "progress": progress, + } + + async def recover_stale(db: AsyncSession) -> int: now = datetime.now(UTC) result = await db.execute( diff --git a/tests/integration/test_sources_jobs.py b/tests/integration/test_sources_jobs.py index cb4130c..f6c096d 100644 --- a/tests/integration/test_sources_jobs.py +++ b/tests/integration/test_sources_jobs.py @@ -6,12 +6,15 @@ import httpx import pytest from backup_tool.api.app import create_app from backup_tool.config import Settings +from backup_tool.db.models import Execution PASSWORD = "correct-horse-battery-staple" async def login(client: httpx.AsyncClient) -> dict[str, str]: - response = await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD}) + response = await client.post( + "/api/v2/setup", json={"username": "admin", "password": PASSWORD} + ) assert response.status_code == 201 return {"X-CSRF-Token": client.cookies["backup_tool_csrf"]} @@ -46,7 +49,9 @@ async def test_local_source_probe_archive_and_repository_targeted_job( async with app.state.engine.begin() as connection: await connection.run_sync(Base.metadata.create_all) transport = httpx.ASGITransport(app=app) - async with httpx.AsyncClient(transport=transport, base_url="https://test") as client: + async with httpx.AsyncClient( + transport=transport, base_url="https://test" + ) as client: headers = await login(client) repository = await client.post( "/api/v2/repositories", @@ -94,6 +99,31 @@ async def test_local_source_probe_archive_and_repository_targeted_job( ) assert duplicate.status_code == 409 assert duplicate.json()["code"] == "execution_active" + polled = await client.get(f"/api/v2/executions/{execution.json()['id']}", headers=headers) + assert polled.status_code == 200 + assert polled.json()["state"] == "queued" + cancellation = await client.post( + f"/api/v2/executions/{execution.json()['id']}/cancel", headers=headers + ) + assert cancellation.status_code == 202 + assert cancellation.json()["state"] == "cancelling" + async with app.state.sessions() as db: + stored = await db.get(Execution, execution.json()["id"]) + assert stored is not None + stored.state = "failed" + stored.reason_code = "timeout" + await db.commit() + retried = await client.post( + f"/api/v2/executions/{execution.json()['id']}/retry", headers=headers + ) + assert retried.status_code == 202 + assert retried.json()["id"] == execution.json()["id"] + assert retried.json()["attempt"] == 2 + stream = await client.get( + f"/api/v2/executions/{execution.json()['id']}/events", headers=headers + ) + assert stream.status_code == 200 + assert "event: execution" in stream.text archived = await client.delete(f"/api/v2/sources/{source_id}", headers=headers) assert archived.status_code == 204 assert (