diff --git a/backend/src/backup_tool/api/app.py b/backend/src/backup_tool/api/app.py index bd596db..5561300 100644 --- a/backend/src/backup_tool/api/app.py +++ b/backend/src/backup_tool/api/app.py @@ -2,7 +2,7 @@ import asyncio import base64 import hashlib import json -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator, AsyncIterator from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Annotated, Any, cast @@ -32,6 +32,7 @@ from backup_tool.db.models import ( User, ) from backup_tool.execution import ( + TERMINAL_STATES, EnqueueError, enqueue, public_event, @@ -57,8 +58,30 @@ 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" +def execution_event_id(execution: Execution) -> str: + """Stable monotonic cursor for an execution's current durable revision.""" + updated = execution.updated_at.astimezone(UTC).isoformat() + return f"{updated}:{execution.state}:{execution.attempt}" + + +async def execution_events_stream( + sessions: async_sessionmaker[AsyncSession], execution_id: str, last_event_id: str | None +) -> AsyncGenerator[str, None]: + """Poll durable execution state and yield each changed redacted revision as SSE.""" + last_sent = last_event_id + while True: + async with sessions() as stream_db: + execution = await stream_db.get(Execution, execution_id) + if execution is None: + return + event_id = execution_event_id(execution) + if event_id != last_sent: + payload = public_event(execution) + yield (f"id: {event_id}\\nevent: execution\\ndata: {json.dumps(payload)}\\n\\n") + last_sent = event_id + if execution.state in TERMINAL_STATES: + return + await asyncio.sleep(0.1) class Problem(Exception): @@ -838,14 +861,15 @@ def create_app(settings: Settings) -> FastAPI: execution_id: str, db: Annotated[AsyncSession, Depends(session)], identity: Annotated[tuple[User, set[str], bool], Depends(actor)], + last_event_id: Annotated[str | None, Header(alias="Last-Event-ID")] = None, ) -> StreamingResponse: enforce_scope(identity[1], "execution:read") - execution = await db.get(Execution, execution_id) - if execution is None: + if await db.get(Execution, execution_id) 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" + execution_events_stream(app.state.sessions, execution_id, last_event_id), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) @app.get("/api/v2/audit") diff --git a/tests/integration/test_sources_jobs.py b/tests/integration/test_sources_jobs.py index 4e1b65f..f2a4f15 100644 --- a/tests/integration/test_sources_jobs.py +++ b/tests/integration/test_sources_jobs.py @@ -5,12 +5,13 @@ from pathlib import Path import httpx import pytest -from backup_tool.api.app import create_app +from backup_tool.api.app import create_app, execution_events_stream from backup_tool.config import Settings -from backup_tool.db.models import Execution +from backup_tool.db.models import Execution, Repository, Source from backup_tool.execution import ( claim, complete_cancellation, + enqueue, heartbeat, recover_stale, request_cancellation, @@ -157,6 +158,10 @@ async def test_local_source_probe_archive_and_repository_targeted_job( assert retried.status_code == 202 assert retried.json()["id"] == execution.json()["id"] assert retried.json()["attempt"] == 2 + cancelled = await client.post( + f"/api/v2/executions/{execution.json()['id']}/cancel", headers=headers + ) + assert cancelled.status_code == 202 stream = await client.get( f"/api/v2/executions/{execution.json()['id']}/events", headers=headers ) @@ -172,6 +177,77 @@ async def test_local_source_probe_archive_and_repository_targeted_job( ).status_code == 409 +@pytest.mark.asyncio +async def test_execution_sse_replays_later_redacted_revision(tmp_path: Path) -> None: + source_root = tmp_path / "sources" + source_root.mkdir() + data_dir = tmp_path / "data" + data_dir.mkdir() + key = tmp_path / "master.key" + key.write_bytes(b"x" * 32) + key.chmod(0o600) + repositories = tmp_path / "repositories" + restore = tmp_path / "restore" + repositories.mkdir() + restore.mkdir() + settings = Settings( + data_dir=data_dir, + database_url=f"sqlite+aiosqlite:///{data_dir / 'db.sqlite'}", + repository_roots=(repositories,), + local_source_roots=(source_root,), + restore_roots=(restore,), + master_key_file=key, + ) + app = create_app(settings) + from backup_tool.db.models import Base, Job + + async with app.state.engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + async with app.state.sessions() as db: + repository = Repository( + name="stream-repository", + root=str(repositories / "stream"), + format_version=1, + compression="none", + encryption="none", + ) + source = Source( + name="stream-source", + kind="local", + public_config={"root": str(source_root)}, + secret_refs=[], + ) + db.add_all([repository, source]) + await db.flush() + job = Job( + name="stream-job", + source_id=source.id, + repository_id=repository.id, + requested_mode="full", + exclusions=[], + retention={}, + ) + db.add(job) + await db.commit() + execution = await enqueue(db, job.id) + execution_id = execution.id + + stream = execution_events_stream(app.state.sessions, execution_id, None) + first = await anext(stream) + first_id = first.split("\\n", 1)[0].removeprefix("id: ") + assert "queued" in first + async with app.state.sessions() as db: + assert await request_cancellation(db, execution_id) is not None + second = await anext(stream) + assert second.split("\\n", 1)[0].removeprefix("id: ") != first_id + assert "cancelled" in second + + replay = execution_events_stream(app.state.sessions, execution_id, first_id) + assert "cancelled" in await anext(replay) + await stream.aclose() + await replay.aclose() + + @pytest.mark.asyncio async def test_local_source_rejects_unallowlisted_root(tmp_path: Path) -> None: allowed = tmp_path / "allowed"