feat: retain live snapshots while revalidating (#139)

frontend/index.html

@@ -617,6 +617,9 @@ textarea { resize: vertical; min-height: 120px; }
     } else {
       setEventStreamStatus('Update failed · showing last activity');
     }
+    if (snapshot.freshness?.revalidating) {
+      setEventStreamStatus('Refreshing · showing recent snapshot');
+    }
   }
 
 

src/main.py

@@ -1,5 +1,6 @@
 import asyncio
 import math
+import time
 from contextlib import asynccontextmanager
 from pathlib import Path
 
@@ -51,8 +52,12 @@ EVENT_STREAM_TIMEOUT_SECONDS = 5.0
 READINESS_TIMEOUT_SECONDS = 5.0
 REVIEW_DETAIL_TIMEOUT_SECONDS = 5.0
 NOTIFICATION_MUTATION_TIMEOUT_SECONDS = 5.0
+LIVE_SNAPSHOT_FRESHNESS_SECONDS = 8.0
 FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
 _live_snapshot_task: asyncio.Task | None = None
+_live_snapshot_value: dict | None = None
+_live_snapshot_created_at: float | None = None
+_read_notification_ids: set[int] = set()
 
 
 class ContextPayloadError(ValueError):
@@ -268,18 +273,110 @@ async def _build_live_snapshot_before_deadline() -> dict:
         return await _build_live_snapshot()
 
 
-@app.get("/api/v1/live")
-async def live_snapshot() -> JSONResponse:
-    """Return a fresh, section-aware snapshot; join only an active identical load."""
+async def _refresh_live_snapshot() -> dict:
+    global _live_snapshot_value, _live_snapshot_created_at
+    result = await _build_live_snapshot_before_deadline()
+    result = _without_read_notifications(result)
+    _live_snapshot_value = result
+    _live_snapshot_created_at = time.monotonic()
+    return result
+
+
+def _consume_live_snapshot_failure(task: asyncio.Task) -> None:
+    if task.cancelled():
+        return
+    task.exception()
+
+
+def _start_live_snapshot_refresh() -> asyncio.Task:
     global _live_snapshot_task
     if _live_snapshot_task is None or _live_snapshot_task.done():
-        _live_snapshot_task = asyncio.create_task(
-            _build_live_snapshot_before_deadline()
+        _live_snapshot_task = asyncio.create_task(_refresh_live_snapshot())
+        _live_snapshot_task.add_done_callback(_consume_live_snapshot_failure)
+    return _live_snapshot_task
+
+
+def _live_snapshot_payload(value: dict, *, stale: bool, revalidating: bool) -> dict:
+    payload = dict(value)
+    age = (
+        max(0.0, time.monotonic() - _live_snapshot_created_at)
+        if _live_snapshot_created_at is not None
+        else 0.0
+    )
+    payload["freshness"] = {
+        "age_seconds": round(age, 3),
+        "fresh_for_seconds": LIVE_SNAPSHOT_FRESHNESS_SECONDS,
+        "stale": stale,
+        "revalidating": revalidating,
+    }
+    return payload
+
+
+def _remove_notification_from_live_snapshot(thread_id: int) -> None:
+    global _live_snapshot_value, _read_notification_ids
+    _read_notification_ids = _read_notification_ids | {thread_id}
+    if _live_snapshot_value is None:
+        return
+    retained_notifications = _live_snapshot_value.get("notifications")
+    if not isinstance(retained_notifications, list):
+        return
+    updated = dict(_live_snapshot_value)
+    updated["notifications"] = [
+        notification
+        for notification in retained_notifications
+        if not isinstance(notification, dict) or notification.get("id") != thread_id
+    ]
+    _live_snapshot_value = updated
+
+
+def _without_read_notifications(snapshot: dict) -> dict:
+    global _read_notification_ids
+    snapshot_notifications = snapshot.get("notifications")
+    if not isinstance(snapshot_notifications, list):
+        return snapshot
+    returned_ids = {
+        notification.get("id")
+        for notification in snapshot_notifications
+        if isinstance(notification, dict)
+    }
+    updated = dict(snapshot)
+    updated["notifications"] = [
+        notification
+        for notification in snapshot_notifications
+        if not isinstance(notification, dict)
+        or notification.get("id") not in _read_notification_ids
+    ]
+    _read_notification_ids = _read_notification_ids.intersection(returned_ids)
+    return updated
+
+
+@app.get("/api/v1/live")
+async def live_snapshot() -> JSONResponse:
+    """Return a freshness-bounded snapshot and share identical upstream loads."""
+    global _live_snapshot_task, _live_snapshot_value, _live_snapshot_created_at
+    now = time.monotonic()
+    if (
+        _live_snapshot_value is not None
+        and _live_snapshot_created_at is not None
+        and now - _live_snapshot_created_at < LIVE_SNAPSHOT_FRESHNESS_SECONDS
+    ):
+        return JSONResponse(
+            _live_snapshot_payload(
+                _live_snapshot_value, stale=False, revalidating=False
+            )
+        )
+    task = _start_live_snapshot_refresh()
+    if _live_snapshot_value is not None:
+        return JSONResponse(
+            _live_snapshot_payload(
+                _live_snapshot_value, stale=True, revalidating=True
+            )
         )
-    task = _live_snapshot_task
     try:
         result = await asyncio.shield(task)
-        return JSONResponse(result)
+        return JSONResponse(
+            _live_snapshot_payload(result, stale=False, revalidating=False)
+        )
     except TimeoutError:
         return JSONResponse(
             {"error": f"Gitea live snapshot timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"},
@@ -291,9 +388,7 @@ async def live_snapshot() -> JSONResponse:
             {"error": "Gitea live snapshot is temporarily unavailable"},
             status_code=503,
         )
-    finally:
-        if task.done() and _live_snapshot_task is task:
-            _live_snapshot_task = None
+
 
 
 @app.get("/api/v1/events")
@@ -347,6 +442,7 @@ async def read_notification(thread_id: int = PathParam(gt=0)) -> JSONResponse:
             {"error": "The update could not be marked read. Please retry."},
             status_code=503,
         )
+    _remove_notification_from_live_snapshot(thread_id)
     return JSONResponse({"id": thread_id, "status": "read"})
 
 

tests/test_live_snapshot.py

@@ -9,8 +9,12 @@ from src import main
 @pytest.fixture(autouse=True)
 def reset_live_snapshot_task():
     main._live_snapshot_task = None
+    main._live_snapshot_value = None
+    main._live_snapshot_created_at = None
     yield
     main._live_snapshot_task = None
+    main._live_snapshot_value = None
+    main._live_snapshot_created_at = None
 
 
 def payload(response):
@@ -147,7 +151,7 @@ async def test_live_snapshot_keeps_fresh_activity_when_work_fails(monkeypatch):
 
 
 @pytest.mark.anyio
-async def test_live_snapshot_coalesces_only_simultaneous_requests(monkeypatch):
+async def test_live_snapshot_reuses_completed_snapshot_inside_freshness_window(monkeypatch):
     user_calls = 0
     release = asyncio.Event()
 
@@ -183,7 +187,47 @@ async def test_live_snapshot_coalesces_only_simultaneous_requests(monkeypatch):
     await asyncio.gather(first, second)
     await main.live_snapshot()
 
-    assert user_calls == 2
+    assert user_calls == 1
+
+
+@pytest.mark.anyio
+async def test_stale_snapshot_returns_immediately_while_one_refresh_revalidates(monkeypatch):
+    now = 100.0
+    refresh_started = asyncio.Event()
+    release_refresh = asyncio.Event()
+    builds = 0
+
+    async def snapshot():
+        nonlocal builds
+        builds += 1
+        if builds == 2:
+            refresh_started.set()
+            await release_refresh.wait()
+        return {
+            "context": {"generation": builds},
+            "events": [],
+            "notifications": [],
+            "sections": {},
+        }
+
+    monkeypatch.setattr(main.time, "monotonic", lambda: now)
+    monkeypatch.setattr(main, "_build_live_snapshot", snapshot)
+
+    first = payload(await main.live_snapshot())
+    now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
+    stale_request = asyncio.create_task(main.live_snapshot())
+    await refresh_started.wait()
+    await asyncio.sleep(0)
+    returned_immediately = stale_request.done()
+    release_refresh.set()
+    stale = payload(await stale_request)
+
+    assert returned_immediately is True
+    assert first["context"]["generation"] == 1
+    assert stale["context"]["generation"] == 1
+    assert stale["freshness"]["stale"] is True
+    assert stale["freshness"]["revalidating"] is True
+    assert builds == 2
 
 
 @pytest.mark.anyio

tests/test_notification_read.py

@@ -1,3 +1,5 @@
+import asyncio
+
 import httpx
 import pytest
 
@@ -45,3 +47,62 @@ async def test_mark_notification_read_api_is_bounded_and_never_cacheable(monkeyp
     assert response.headers["cache-control"] == "no-store"
     assert invalid.status_code == 422
     assert marked == [42]
+
+
+@pytest.mark.anyio
+async def test_mark_notification_read_removes_thread_from_retained_live_snapshot(monkeypatch):
+    async def mark(_thread_id):
+        return None
+
+    monkeypatch.setattr(main, "mark_notification_read", mark)
+    monkeypatch.setattr(
+        main,
+        "_live_snapshot_value",
+        {
+            "context": {},
+            "events": [],
+            "notifications": [{"id": 42}, {"id": 43}],
+            "sections": {"notifications": "fresh"},
+        },
+    )
+
+    response = await main.read_notification(42)
+
+    assert response.status_code == 200
+    assert main._live_snapshot_value is not None
+    assert main._live_snapshot_value["notifications"] == [{"id": 43}]
+
+
+@pytest.mark.anyio
+async def test_inflight_refresh_cannot_restore_a_notification_marked_read(monkeypatch):
+    refresh_started = asyncio.Event()
+    release_refresh = asyncio.Event()
+
+    async def mark(_thread_id):
+        return None
+
+    async def stale_upstream_snapshot():
+        refresh_started.set()
+        await release_refresh.wait()
+        return {
+            "context": {},
+            "events": [],
+            "notifications": [{"id": 42}, {"id": 43}],
+            "sections": {"notifications": "fresh"},
+        }
+
+    monkeypatch.setattr(main, "mark_notification_read", mark)
+    monkeypatch.setattr(
+        main, "_build_live_snapshot_before_deadline", stale_upstream_snapshot
+    )
+    monkeypatch.setattr(main, "_live_snapshot_task", None)
+    monkeypatch.setattr(main, "_live_snapshot_value", {"notifications": [{"id": 42}]})
+
+    refresh = main._start_live_snapshot_refresh()
+    await refresh_started.wait()
+    await main.read_notification(42)
+    release_refresh.set()
+    await refresh
+
+    assert main._live_snapshot_value is not None
+    assert main._live_snapshot_value["notifications"] == [{"id": 43}]

tests/test_shared_context_snapshot.py

@@ -26,4 +26,12 @@ async def test_one_live_snapshot_updates_work_and_activity_on_one_timer():
     assert "renderContextSnapshot(snapshot.context)" in html
     assert "paintEventStream(snapshot.events)" in html
     assert "loadEventStream" not in html
-    assert "5000" not in html
+    assert "5000" not in html
+
+
+@pytest.mark.anyio
+async def test_dashboard_announces_when_recent_snapshot_is_revalidating():
+    html = await dashboard()
+
+    assert "snapshot.freshness?.revalidating" in html
+    assert "Refreshing · showing recent snapshot" in html