feat: keep live snapshots usable through refresh failures (#149)
CI / lint (pull_request) Successful in 11s
CI / build-frontend (pull_request) Successful in 4s

This commit is contained in:
timmy committed 2026-08-06 23:54:55 +00:00
1 parent db0354a6e5
commit 8ee7dbf79e
4 files changed
+238 -2

No files matched your search

+69 -1
View File
@@ -57,10 +57,14 @@ NOTIFICATION_PAGE_TIMEOUT_SECONDS = 5.0
BULK_NOTIFICATION_CONCURRENCY = 5
BULK_NOTIFICATION_DEADLINE_SECONDS = 6.0
LIVE_SNAPSHOT_FRESHNESS_SECONDS = 8.0
LIVE_SNAPSHOT_RETRY_BASE_SECONDS = 5.0
LIVE_SNAPSHOT_RETRY_MAX_SECONDS = 60.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
_live_snapshot_failure_count = 0
_live_snapshot_retry_at: float | None = None
_read_notification_ids: set[int] = set()
@@ -292,12 +296,52 @@ async def _build_live_snapshot_before_deadline() -> dict:
return await _build_live_snapshot()
def _record_live_snapshot_failure() -> None:
global _live_snapshot_failure_count, _live_snapshot_retry_at
_live_snapshot_failure_count += 1
delay = min(
LIVE_SNAPSHOT_RETRY_MAX_SECONDS,
LIVE_SNAPSHOT_RETRY_BASE_SECONDS * (2 ** (_live_snapshot_failure_count - 1)),
)
_live_snapshot_retry_at = time.monotonic() + delay
def _merge_live_snapshot(previous: dict | None, refreshed: dict) -> dict:
if previous is None:
return refreshed
merged = dict(refreshed)
sections = dict(refreshed.get("sections") or {})
for section in ("context", "events", "notifications"):
if sections.get(section) == "fresh":
continue
if previous.get(section) is not None:
merged[section] = previous[section]
sections[section] = "stale"
if section == "notifications":
merged["notification_pagination"] = previous.get(
"notification_pagination"
)
merged["sections"] = sections
return merged
async def _refresh_live_snapshot() -> dict:
global _live_snapshot_value, _live_snapshot_created_at
result = await _build_live_snapshot_before_deadline()
global _live_snapshot_failure_count, _live_snapshot_retry_at
try:
result = await _build_live_snapshot_before_deadline()
except Exception:
_record_live_snapshot_failure()
raise
result = _merge_live_snapshot(_live_snapshot_value, result)
result = _without_read_notifications(result)
_live_snapshot_value = result
_live_snapshot_created_at = time.monotonic()
if any(state != "fresh" for state in result.get("sections", {}).values()):
_record_live_snapshot_failure()
else:
_live_snapshot_failure_count = 0
_live_snapshot_retry_at = None
return result
@@ -322,11 +366,19 @@ def _live_snapshot_payload(value: dict, *, stale: bool, revalidating: bool) -> d
if _live_snapshot_created_at is not None
else 0.0
)
retry_in_seconds = (
max(0.0, _live_snapshot_retry_at - time.monotonic())
if _live_snapshot_retry_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,
"degraded": _live_snapshot_retry_at is not None,
"last_refresh_failed": _live_snapshot_retry_at is not None,
"retry_in_seconds": math.ceil(retry_in_seconds),
}
return payload
@@ -374,6 +426,22 @@ 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_retry_at is not None
):
if now < _live_snapshot_retry_at:
return JSONResponse(
_live_snapshot_payload(
_live_snapshot_value, stale=True, revalidating=False
)
)
_start_live_snapshot_refresh()
return JSONResponse(
_live_snapshot_payload(
_live_snapshot_value, stale=True, revalidating=True
)
)
if (
_live_snapshot_value is not None
and _live_snapshot_created_at is not None