fix(v2): validate durable execution event replay

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"id: {event.revision}\nevent: execution\ndata: {json.dumps(event.payload)}\n\n"
-                    f"data: {json.dumps(event.payload)}\\n\\n"
                 )
                 last_revision = event.revision
             execution = await stream_db.get(Execution, execution_id)

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"