stackchain/stackchain-dashboard
2
0 Fork 0
Code Issues 9 Pull Requests 1 Actions Packages Projects Releases 713 Wiki Activity

feat: isolate live section refresh backoff (#167)
CI / lint (pull_request) Successful in 15s
Details
CI / build-frontend (pull_request) Successful in 5s
Details

Browse Source
This commit is contained in:
timmy committed 2026-08-07 04:53:42 +00:00
1 parent 5fe0b28747
commit 4870b58c5b
5 files changed
+262 -100
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/main.py
+140 -81
View File
@@ -1,8 +1,10 @@
import asyncio
import math
import time
from collections.abc import Awaitable
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any
from fastapi import FastAPI, HTTPException, Path as PathParam, Query
from fastapi.responses import JSONResponse
@@ -65,8 +67,17 @@ 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
LIVE_SNAPSHOT_SECTIONS = ("context", "events", "notifications")
_live_section_created_at: dict[str, float | None] = {
section: None for section in LIVE_SNAPSHOT_SECTIONS
}
_live_section_failure_count: dict[str, int] = {
section: 0 for section in LIVE_SNAPSHOT_SECTIONS
}
_live_section_retry_at: dict[str, float | None] = {
section: None for section in LIVE_SNAPSHOT_SECTIONS
}
_live_snapshot_refreshing_sections: set[str] = set()
_read_notification_ids: set[int] = set()
@@ -337,19 +348,41 @@ async def _load_context_for_user(user_data: dict) -> dict:
return _context_payload(user_data, repo_data, issues_data, prs_data)
async def _build_live_snapshot() -> dict:
user_data = await current_user()
if not isinstance(user_data, dict) or not user_data.get("login"):
raise ContextPayloadError("Gitea current-user response was invalid")
context_result, events_result, notifications_result = await asyncio.gather(
_load_context_for_user(user_data),
activity_events(user_data),
notifications(),
return_exceptions=True,
async def _build_live_snapshot(sections: set[str] | None = None) -> dict:
requested = set(sections or LIVE_SNAPSHOT_SECTIONS)
results: dict[str, object] = {}
user_data: dict | None = None
if requested & {"context", "events"}:
try:
user_data = await current_user()
if not isinstance(user_data, dict) or not user_data.get("login"):
raise ContextPayloadError("Gitea current-user response was invalid")
except Exception as exc:
for section in requested & {"context", "events"}:
results[section] = exc
loads: dict[str, Awaitable[Any]] = {}
if "context" in requested and "context" not in results:
assert user_data is not None
loads["context"] = _load_context_for_user(user_data)
if "events" in requested and "events" not in results:
assert user_data is not None
loads["events"] = activity_events(user_data)
if "notifications" in requested:
loads["notifications"] = notifications()
if loads:
loaded = await asyncio.gather(*loads.values(), return_exceptions=True)
results.update(zip(loads, loaded))
context_result = results.get("context")
events_result = results.get("events")
notifications_result = results.get("notifications")
context_ok = "context" in requested and not isinstance(context_result, BaseException)
events_ok = "events" in requested and not isinstance(events_result, BaseException)
notifications_ok = (
"notifications" in requested
and not isinstance(notifications_result, BaseException)
)
context_ok = not isinstance(context_result, BaseException)
events_ok = not isinstance(events_result, BaseException)
notifications_ok = not isinstance(notifications_result, BaseException)
notification_items = notifications_result
notification_pagination = None
if notifications_ok and isinstance(notifications_result, dict):
@@ -366,64 +399,88 @@ async def _build_live_snapshot() -> dict:
"notifications": notification_items if notifications_ok else None,
"notification_pagination": notification_pagination if notifications_ok else None,
"sections": {
"context": "fresh" if context_ok else "temporarily unavailable",
"events": "fresh" if events_ok else "temporarily unavailable",
"notifications": "fresh" if notifications_ok else "temporarily unavailable",
section: "fresh" if ok else "temporarily unavailable"
for section, ok in (
("context", context_ok),
("events", events_ok),
("notifications", notifications_ok),
)
if section in requested
},
}
async def _build_live_snapshot_before_deadline() -> dict:
async def _build_live_snapshot_before_deadline(sections: set[str]) -> dict:
async with asyncio.timeout(CONTEXT_TIMEOUT_SECONDS):
return await _build_live_snapshot()
return await _build_live_snapshot(sections)
def _record_live_snapshot_failure() -> None:
global _live_snapshot_failure_count, _live_snapshot_retry_at
_live_snapshot_failure_count += 1
def _record_live_section_failure(section: str) -> None:
_live_section_failure_count[section] += 1
delay = min(
LIVE_SNAPSHOT_RETRY_MAX_SECONDS,
LIVE_SNAPSHOT_RETRY_BASE_SECONDS * (2 ** (_live_snapshot_failure_count - 1)),
LIVE_SNAPSHOT_RETRY_BASE_SECONDS
* (2 ** (_live_section_failure_count[section] - 1)),
)
_live_snapshot_retry_at = time.monotonic() + delay
_live_section_retry_at[section] = time.monotonic() + delay
def _due_live_sections(now: float) -> set[str]:
due = set()
for section in LIVE_SNAPSHOT_SECTIONS:
retry_at = _live_section_retry_at[section]
if retry_at is not None and now < retry_at:
continue
created_at = _live_section_created_at[section]
if created_at is None or now - created_at >= LIVE_SNAPSHOT_FRESHNESS_SECONDS:
due.add(section)
return due
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"
merged = dict(previous)
sections = dict(previous.get("sections") or {})
for section, state in (refreshed.get("sections") or {}).items():
if state == "fresh":
merged[section] = refreshed.get(section)
sections[section] = "fresh"
if section == "notifications":
merged["notification_pagination"] = previous.get(
merged["notification_pagination"] = refreshed.get(
"notification_pagination"
)
elif previous.get(section) is not None:
sections[section] = "stale"
else:
merged[section] = None
sections[section] = "temporarily unavailable"
merged["sections"] = sections
return merged
async def _refresh_live_snapshot() -> dict:
async def _refresh_live_snapshot(sections: set[str]) -> dict:
global _live_snapshot_value, _live_snapshot_created_at
global _live_snapshot_failure_count, _live_snapshot_retry_at
try:
result = await _build_live_snapshot_before_deadline()
refreshed = await _build_live_snapshot_before_deadline(sections)
except Exception:
_record_live_snapshot_failure()
for section in sections:
_record_live_section_failure(section)
raise
result = _merge_live_snapshot(_live_snapshot_value, result)
now = time.monotonic()
refreshed_states = refreshed.get("sections") or {}
for section in sections:
if refreshed_states.get(section) == "fresh":
_live_section_created_at[section] = now
_live_section_failure_count[section] = 0
_live_section_retry_at[section] = None
else:
_record_live_section_failure(section)
result = _merge_live_snapshot(_live_snapshot_value, refreshed)
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
successful_times = [value for value in _live_section_created_at.values() if value is not None]
_live_snapshot_created_at = max(successful_times) if successful_times else None
return result
@@ -433,34 +490,47 @@ def _consume_live_snapshot_failure(task: asyncio.Task) -> None:
task.exception()
def _start_live_snapshot_refresh() -> asyncio.Task:
def _start_live_snapshot_refresh(sections: set[str]) -> asyncio.Task:
global _live_snapshot_task
global _live_snapshot_refreshing_sections
if _live_snapshot_task is None or _live_snapshot_task.done():
_live_snapshot_task = asyncio.create_task(_refresh_live_snapshot())
_live_snapshot_refreshing_sections = set(sections)
_live_snapshot_task = asyncio.create_task(_refresh_live_snapshot(sections))
_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
)
retry_in_seconds = (
max(0.0, _live_snapshot_retry_at - time.monotonic())
if _live_snapshot_retry_at is not None
else 0.0
)
now = time.monotonic()
section_freshness = {}
for section in LIVE_SNAPSHOT_SECTIONS:
created_at = _live_section_created_at[section]
retry_at = _live_section_retry_at[section]
age = max(0.0, now - created_at) if created_at is not None else 0.0
retry_in = max(0.0, retry_at - now) if retry_at is not None else 0.0
section_freshness[section] = {
"age_seconds": round(age, 3),
"stale": (value.get("sections") or {}).get(section) != "fresh"
or created_at is None
or age >= LIVE_SNAPSHOT_FRESHNESS_SECONDS,
"revalidating": revalidating
and section in _live_snapshot_refreshing_sections,
"degraded": retry_at is not None,
"retry_in_seconds": math.ceil(retry_in),
}
ages = [item["age_seconds"] for item in section_freshness.values()]
retries = [item["retry_in_seconds"] for item in section_freshness.values() if item["retry_in_seconds"]]
degraded = any(item["degraded"] for item in section_freshness.values())
payload["freshness"] = {
"age_seconds": round(age, 3),
"age_seconds": max(ages, default=0.0),
"fresh_for_seconds": LIVE_SNAPSHOT_FRESHNESS_SECONDS,
"stale": stale,
"stale": stale or any(item["stale"] for item in section_freshness.values()),
"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),
"degraded": degraded,
"last_refresh_failed": degraded,
"retry_in_seconds": min(retries, default=0),
"sections": section_freshness,
}
return payload
@@ -508,33 +578,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()
due_sections = _due_live_sections(now)
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
and now - _live_snapshot_created_at < LIVE_SNAPSHOT_FRESHNESS_SECONDS
and not due_sections
):
return JSONResponse(
_live_snapshot_payload(
_live_snapshot_value, stale=False, revalidating=False
_live_snapshot_value,
stale=any(
state != "fresh"
for state in _live_snapshot_value.get("sections", {}).values()
),
revalidating=False,
)
)
task = _start_live_snapshot_refresh()
task = _start_live_snapshot_refresh(due_sections)
if _live_snapshot_value is not None:
return JSONResponse(
_live_snapshot_payload(