Retain live snapshots while revalidating #140

Merged
rockachopa merged 1 commits from timmy/139-stale-while-revalidate-live-snapshot into main 2026-08-06 21:33:48 +00:00
5 changed files with 225 additions and 13 deletions

View File

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

View File

@ -1,5 +1,6 @@
import asyncio import asyncio
import math import math
import time
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from pathlib import Path from pathlib import Path
@ -51,8 +52,12 @@ EVENT_STREAM_TIMEOUT_SECONDS = 5.0
READINESS_TIMEOUT_SECONDS = 5.0 READINESS_TIMEOUT_SECONDS = 5.0
REVIEW_DETAIL_TIMEOUT_SECONDS = 5.0 REVIEW_DETAIL_TIMEOUT_SECONDS = 5.0
NOTIFICATION_MUTATION_TIMEOUT_SECONDS = 5.0 NOTIFICATION_MUTATION_TIMEOUT_SECONDS = 5.0
LIVE_SNAPSHOT_FRESHNESS_SECONDS = 8.0
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend" FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
_live_snapshot_task: asyncio.Task | None = None _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): class ContextPayloadError(ValueError):
@ -268,18 +273,110 @@ async def _build_live_snapshot_before_deadline() -> dict:
return await _build_live_snapshot() return await _build_live_snapshot()
@app.get("/api/v1/live") async def _refresh_live_snapshot() -> dict:
async def live_snapshot() -> JSONResponse: global _live_snapshot_value, _live_snapshot_created_at
"""Return a fresh, section-aware snapshot; join only an active identical load.""" 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 global _live_snapshot_task
if _live_snapshot_task is None or _live_snapshot_task.done(): if _live_snapshot_task is None or _live_snapshot_task.done():
_live_snapshot_task = asyncio.create_task( _live_snapshot_task = asyncio.create_task(_refresh_live_snapshot())
_build_live_snapshot_before_deadline() _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: try:
result = await asyncio.shield(task) result = await asyncio.shield(task)
return JSONResponse(result) return JSONResponse(
_live_snapshot_payload(result, stale=False, revalidating=False)
)
except TimeoutError: except TimeoutError:
return JSONResponse( return JSONResponse(
{"error": f"Gitea live snapshot timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"}, {"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"}, {"error": "Gitea live snapshot is temporarily unavailable"},
status_code=503, status_code=503,
) )
finally:
if task.done() and _live_snapshot_task is task:
_live_snapshot_task = None
@app.get("/api/v1/events") @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."}, {"error": "The update could not be marked read. Please retry."},
status_code=503, status_code=503,
) )
_remove_notification_from_live_snapshot(thread_id)
return JSONResponse({"id": thread_id, "status": "read"}) return JSONResponse({"id": thread_id, "status": "read"})

View File

@ -9,8 +9,12 @@ from src import main
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def reset_live_snapshot_task(): def reset_live_snapshot_task():
main._live_snapshot_task = None main._live_snapshot_task = None
main._live_snapshot_value = None
main._live_snapshot_created_at = None
yield yield
main._live_snapshot_task = None main._live_snapshot_task = None
main._live_snapshot_value = None
main._live_snapshot_created_at = None
def payload(response): def payload(response):
@ -147,7 +151,7 @@ async def test_live_snapshot_keeps_fresh_activity_when_work_fails(monkeypatch):
@pytest.mark.anyio @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 user_calls = 0
release = asyncio.Event() release = asyncio.Event()
@ -183,7 +187,47 @@ async def test_live_snapshot_coalesces_only_simultaneous_requests(monkeypatch):
await asyncio.gather(first, second) await asyncio.gather(first, second)
await main.live_snapshot() 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 @pytest.mark.anyio

View File

@ -1,3 +1,5 @@
import asyncio
import httpx import httpx
import pytest 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 response.headers["cache-control"] == "no-store"
assert invalid.status_code == 422 assert invalid.status_code == 422
assert marked == [42] 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}]

View File

@ -26,4 +26,12 @@ async def test_one_live_snapshot_updates_work_and_activity_on_one_timer():
assert "renderContextSnapshot(snapshot.context)" in html assert "renderContextSnapshot(snapshot.context)" in html
assert "paintEventStream(snapshot.events)" in html assert "paintEventStream(snapshot.events)" in html
assert "loadEventStream" not 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