From 396219e776aa9a115900d2b7bfd9fb5c1cfde115 Mon Sep 17 00:00:00 2001 From: Alex Blank Date: Tue, 28 Jul 2026 10:52:03 +0200 Subject: [PATCH] fix(v2): validate durable execution event replay --- backend/src/backup_tool/api/app.py | 3 +- tests/integration/test_sources_jobs.py | 110 ++++++++++++++++++++++++- 2 files changed, 109 insertions(+), 4 deletions(-) diff --git a/backend/src/backup_tool/api/app.py b/backend/src/backup_tool/api/app.py index e94e42b..0271e4a 100644 --- a/backend/src/backup_tool/api/app.py +++ b/backend/src/backup_tool/api/app.py @@ -81,8 +81,7 @@ async def execution_events_stream( ).all() for event in events: yield ( - f"id: {event.revision}\\nevent: execution\\n" - f"data: {json.dumps(event.payload)}\\n\\n" + f"id: {event.revision}\nevent: execution\ndata: {json.dumps(event.payload)}\n\n" ) last_revision = event.revision execution = await stream_db.get(Execution, execution_id) diff --git a/tests/integration/test_sources_jobs.py b/tests/integration/test_sources_jobs.py index f2a4f15..c4ec748 100644 --- a/tests/integration/test_sources_jobs.py +++ b/tests/integration/test_sources_jobs.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json from datetime import UTC, datetime, timedelta from pathlib import Path @@ -13,9 +14,11 @@ from backup_tool.execution import ( complete_cancellation, enqueue, heartbeat, + record_event, recover_stale, request_cancellation, ) +from backup_tool.worker import Worker PASSWORD = "correct-horse-battery-staple" @@ -234,12 +237,12 @@ async def test_execution_sse_replays_later_redacted_revision(tmp_path: Path) -> stream = execution_events_stream(app.state.sessions, execution_id, None) first = await anext(stream) - first_id = first.split("\\n", 1)[0].removeprefix("id: ") + 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 second.split("\n", 1)[0].removeprefix("id: ") != first_id assert "cancelled" in second replay = execution_events_stream(app.state.sessions, execution_id, first_id) @@ -248,6 +251,109 @@ async def test_execution_sse_replays_later_redacted_revision(tmp_path: Path) -> await replay.aclose() +@pytest.mark.asyncio +async def test_execution_events_preserve_progress_replay_and_recovery_order(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="ordered-events-repository", + root=str(repositories / "ordered"), + format_version=1, + compression="none", + encryption="none", + ) + source = Source( + name="ordered-events-source", + kind="local", + public_config={"root": str(source_root)}, + secret_refs=[], + ) + db.add_all([repository, source]) + await db.flush() + job = Job( + name="ordered-events-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 + + worker = Worker(settings, owner="ordering-worker") + assert await worker.run_once() + async with app.state.sessions() as db: + execution = await db.get(Execution, execution_id) + assert execution is not None and execution.state == "running" + execution.progress = {"files": 1} + await record_event(db, execution) + execution.progress = {"files": 2} + await record_event(db, execution) + await db.commit() + execution.lease_expires_at = datetime.now(UTC) - timedelta(seconds=1) + await db.commit() + assert await recover_stale(db) == 1 + execution = await db.get(Execution, execution_id) + assert execution is not None and execution.state == "queued" + assert await claim(db, execution_id, "cancelling-worker") is not None + assert await request_cancellation(db, execution_id) is not None + assert await complete_cancellation(db, execution_id, "cancelling-worker") + + def parse_frame(frame: str) -> tuple[int, dict[str, object]]: + try: + identifier, _, data = frame.partition("\n") + revision = int(identifier.removeprefix("id: ")) + payload = json.loads(data.split("data: ", 1)[1].strip()) + except (IndexError, TypeError, ValueError, json.JSONDecodeError) as error: + pytest.fail(f"Invalid SSE frame: {error}") + return revision, payload + + replay = execution_events_stream(app.state.sessions, execution_id, "0") + payloads: list[dict[str, object]] = [] + revisions: list[int] = [] + async for frame in replay: + revision, payload = parse_frame(frame) + revisions.append(revision) + payloads.append(payload) + assert revisions == list(range(1, len(revisions) + 1)) + progress_values = [payload["progress"] for payload in payloads if payload["progress"]] + assert progress_values[:2] == [{"files": 1}, {"files": 2}] + assert payloads[-1]["state"] == "cancelled" + + reconnect = execution_events_stream(app.state.sessions, execution_id, str(revisions[-2])) + replayed = await anext(reconnect) + replayed_revision, replayed_payload = parse_frame(replayed) + assert replayed_revision == revisions[-1] + assert replayed_payload["state"] == "cancelled" + await reconnect.aclose() + + @pytest.mark.asyncio async def test_local_source_rejects_unallowlisted_root(tmp_path: Path) -> None: allowed = tmp_path / "allowed"