From ea7bd15721fde7a50c88fa4081f68b16145f2fb9 Mon Sep 17 00:00:00 2001 From: timmy Date: Thu, 6 Aug 2026 21:32:58 +0000 Subject: [PATCH] feat: retain live snapshots while revalidating (#139) --- frontend/index.html | 3 + src/main.py | 116 +++++++++++++++++++++++--- tests/test_live_snapshot.py | 48 ++++++++++- tests/test_notification_read.py | 61 ++++++++++++++ tests/test_shared_context_snapshot.py | 10 ++- 5 files changed, 225 insertions(+), 13 deletions(-) diff --git a/frontend/index.html b/frontend/index.html index 5b97ca7..8d584a4 100644 --- a/frontend/index.html +++ b/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'); + } } diff --git a/src/main.py b/src/main.py index 4804676..846f1d4 100644 --- a/src/main.py +++ b/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"}) diff --git a/tests/test_live_snapshot.py b/tests/test_live_snapshot.py index da6c76a..ea58959 100644 --- a/tests/test_live_snapshot.py +++ b/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 diff --git a/tests/test_notification_read.py b/tests/test_notification_read.py index 6a3cf9a..21e17c4 100644 --- a/tests/test_notification_read.py +++ b/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}] diff --git a/tests/test_shared_context_snapshot.py b/tests/test_shared_context_snapshot.py index 5a2eb4f..f78787a 100644 --- a/tests/test_shared_context_snapshot.py +++ b/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 \ No newline at end of file + 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 -- 2.43.0