diff --git a/frontend/index.html b/frontend/index.html
index fb589ae..71c63be 100644
--- a/frontend/index.html
+++ b/frontend/index.html
@@ -1216,8 +1216,12 @@ textarea { resize: vertical; min-height: 120px; }
}
function renderLiveSnapshot(snapshot) {
- const notificationsFresh = Array.isArray(snapshot.notifications);
- if (notificationsFresh) {
+ const contextFreshness = snapshot.freshness?.sections?.context;
+ const eventsFreshness = snapshot.freshness?.sections?.events;
+ const notificationFreshness = snapshot.freshness?.sections?.notifications;
+ const hasNotifications = Array.isArray(snapshot.notifications);
+ const notificationsFresh = hasNotifications && !notificationFreshness?.stale;
+ if (hasNotifications) {
if (notificationPagination.page > 1) {
const byId = new Map(lastNotifications.map(item => [item.id, item]));
snapshot.notifications.forEach(item => byId.set(item.id, item));
@@ -1238,19 +1242,27 @@ textarea { resize: vertical; min-height: 120px; }
if (snapshot.context) {
snapshot.context.notifications = lastNotifications;
renderContextSnapshot(snapshot.context);
- if (!notificationsFresh) markNotificationsStale();
+ if (contextFreshness?.stale) markMyWorkStale();
+ else if (!notificationsFresh) markNotificationsStale();
} else handleContextError(new Error('Context section unavailable'));
- if (snapshot.events) {
- paintEventStream(snapshot.events);
+ if (Array.isArray(snapshot.events)) paintEventStream(snapshot.events);
+ if (eventsFreshness?.revalidating) {
+ setEventStreamStatus('Refreshing activity · showing last activity');
+ } else if (eventsFreshness?.stale || eventsFreshness?.degraded) {
+ const retrySeconds = Number(eventsFreshness.retry_in_seconds) || 0;
+ setEventStreamStatus('Activity refresh failed · showing last activity' +
+ (retrySeconds > 0 ? ' · retrying in ' + retrySeconds + 's' : ''));
+ } else if (Array.isArray(snapshot.events)) {
setEventStreamStatus('Updated ' + fmt(new Date()));
} else {
setEventStreamStatus('Update failed · showing last activity');
}
- if (snapshot.freshness?.degraded && !snapshot.freshness.revalidating) {
+ // Compatibility fallback for snapshots produced before section metadata.
+ if (!eventsFreshness && snapshot.freshness?.degraded && !snapshot.freshness.revalidating) {
const retrySeconds = Number(snapshot.freshness.retry_in_seconds) || 0;
setEventStreamStatus('Refresh failed · showing last known data' +
(retrySeconds > 0 ? ' · retrying in ' + retrySeconds + 's' : ''));
- } else if (snapshot.freshness?.revalidating) {
+ } else if (!eventsFreshness && snapshot.freshness?.revalidating) {
setEventStreamStatus('Refreshing · showing recent snapshot');
}
}
diff --git a/src/main.py b/src/main.py
index 853ff4e..1501acf 100644
--- a/src/main.py
+++ b/src/main.py
@@ -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(
diff --git a/tests/test_live_snapshot.py b/tests/test_live_snapshot.py
index 22d16e2..9ce5193 100644
--- a/tests/test_live_snapshot.py
+++ b/tests/test_live_snapshot.py
@@ -11,14 +11,30 @@ def reset_live_snapshot_task():
main._live_snapshot_task = None
main._live_snapshot_value = None
main._live_snapshot_created_at = None
- main._live_snapshot_failure_count = 0
- main._live_snapshot_retry_at = None
+ main._live_section_created_at = {
+ section: None for section in main.LIVE_SNAPSHOT_SECTIONS
+ }
+ main._live_section_failure_count = {
+ section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
+ }
+ main._live_section_retry_at = {
+ section: None for section in main.LIVE_SNAPSHOT_SECTIONS
+ }
+ main._live_snapshot_refreshing_sections = set()
yield
main._live_snapshot_task = None
main._live_snapshot_value = None
main._live_snapshot_created_at = None
- main._live_snapshot_failure_count = 0
- main._live_snapshot_retry_at = None
+ main._live_section_created_at = {
+ section: None for section in main.LIVE_SNAPSHOT_SECTIONS
+ }
+ main._live_section_failure_count = {
+ section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
+ }
+ main._live_section_retry_at = {
+ section: None for section in main.LIVE_SNAPSHOT_SECTIONS
+ }
+ main._live_snapshot_refreshing_sections = set()
def payload(response):
@@ -209,7 +225,7 @@ async def test_stale_snapshot_returns_immediately_while_one_refresh_revalidates(
release_refresh = asyncio.Event()
builds = 0
- async def snapshot():
+ async def snapshot(_sections=None):
nonlocal builds
builds += 1
if builds == 2:
@@ -247,7 +263,7 @@ async def test_failed_revalidation_enters_cooldown_and_keeps_last_snapshot(monke
now = 100.0
builds = 0
- async def snapshot():
+ async def snapshot(_sections=None):
nonlocal builds
builds += 1
if builds > 1:
@@ -288,7 +304,7 @@ async def test_partial_refresh_updates_fresh_sections_and_retains_failed_section
now = 100.0
builds = 0
- async def snapshot():
+ async def snapshot(_sections=None):
nonlocal builds
builds += 1
if builds == 1:
@@ -332,6 +348,70 @@ async def test_partial_refresh_updates_fresh_sections_and_retains_failed_section
assert result["freshness"]["revalidating"] is False
+@pytest.mark.anyio
+async def test_notification_cooldown_does_not_stop_due_work_and_activity_refreshes(monkeypatch):
+ now = 100.0
+ calls = {"context": 0, "events": 0, "notifications": 0}
+ event_refresh_times = []
+ notification_outage = False
+
+ async def user():
+ return {"id": 1, "login": "timmy"}
+
+ async def work():
+ calls["context"] += 1
+ return []
+
+ async def empty_work():
+ return []
+
+ async def events(_authenticated_user):
+ calls["events"] += 1
+ event_refresh_times.append(now)
+ return [{"generation": calls["events"]}]
+
+ async def updates():
+ calls["notifications"] += 1
+ if notification_outage:
+ raise ConnectionError("notifications unavailable")
+ return []
+
+ monkeypatch.setattr(main.time, "monotonic", lambda: now)
+ monkeypatch.setattr(main, "current_user", user)
+ monkeypatch.setattr(main, "repos", work)
+ monkeypatch.setattr(main, "issues", empty_work)
+ monkeypatch.setattr(main, "pull_requests", empty_work)
+ monkeypatch.setattr(main, "activity_events", events)
+ monkeypatch.setattr(main, "notifications", updates)
+
+ await main.live_snapshot()
+ notification_outage = True
+
+ for elapsed in (
+ main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1,
+ main.LIVE_SNAPSHOT_RETRY_BASE_SECONDS,
+ ):
+ now += elapsed
+ await main.live_snapshot()
+ assert main._live_snapshot_task is not None
+ await main._live_snapshot_task
+
+ # Notification backoff is now 10 seconds. The healthy sections become due
+ # one second before that cooldown expires and must refresh independently.
+ now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
+ response = payload(await main.live_snapshot())
+ assert main._live_snapshot_task is not None
+ await main._live_snapshot_task
+ refreshed = payload(await main.live_snapshot())
+
+ assert calls == {"context": 3, "events": 3, "notifications": 3}
+ assert event_refresh_times == [100.0, 109.0, 123.0]
+ assert response["freshness"]["sections"]["context"]["revalidating"] is True
+ assert refreshed["events"] == [{"generation": 3}]
+ assert refreshed["sections"]["notifications"] == "stale"
+ assert refreshed["freshness"]["sections"]["notifications"]["retry_in_seconds"] == 1
+
+
@pytest.mark.anyio
async def test_retry_window_starts_one_refresh_and_success_clears_degraded_state(monkeypatch):
now = 100.0
@@ -339,7 +419,7 @@ async def test_retry_window_starts_one_refresh_and_success_clears_degraded_state
retry_started = asyncio.Event()
release_retry = asyncio.Event()
- async def snapshot():
+ async def snapshot(_sections=None):
nonlocal builds
builds += 1
if builds == 1:
@@ -399,7 +479,7 @@ async def test_cancelling_one_waiter_does_not_cancel_the_shared_snapshot(monkeyp
release = asyncio.Event()
snapshot_calls = 0
- async def blocked_snapshot():
+ async def blocked_snapshot(_sections=None):
nonlocal snapshot_calls
snapshot_calls += 1
started.set()
@@ -437,7 +517,7 @@ async def test_shared_snapshot_deadline_cancels_upstream_work_for_all_waiters(mo
started = asyncio.Event()
cancelled = asyncio.Event()
- async def snapshot_that_exceeds_deadline():
+ async def snapshot_that_exceeds_deadline(_sections=None):
started.set()
try:
await asyncio.Event().wait()
diff --git a/tests/test_notification_read.py b/tests/test_notification_read.py
index 78532b6..ee924cd 100644
--- a/tests/test_notification_read.py
+++ b/tests/test_notification_read.py
@@ -214,7 +214,7 @@ async def test_inflight_refresh_cannot_restore_a_notification_marked_read(monkey
async def mark(_thread_id):
return None
- async def stale_upstream_snapshot():
+ async def stale_upstream_snapshot(_sections):
refresh_started.set()
await release_refresh.wait()
return {
@@ -231,7 +231,7 @@ async def test_inflight_refresh_cannot_restore_a_notification_marked_read(monkey
monkeypatch.setattr(main, "_live_snapshot_task", None)
monkeypatch.setattr(main, "_live_snapshot_value", {"notifications": [{"id": 42}]})
- refresh = main._start_live_snapshot_refresh()
+ refresh = main._start_live_snapshot_refresh(set(main.LIVE_SNAPSHOT_SECTIONS))
await refresh_started.wait()
await main.read_notification(42)
release_refresh.set()
diff --git a/tests/test_shared_context_snapshot.py b/tests/test_shared_context_snapshot.py
index 897551b..d093e1e 100644
--- a/tests/test_shared_context_snapshot.py
+++ b/tests/test_shared_context_snapshot.py
@@ -44,3 +44,14 @@ async def test_dashboard_announces_failed_refresh_and_retry_without_blanking_pan
assert "snapshot.freshness?.degraded" in html
assert "Refresh failed · showing last known data" in html
assert "snapshot.freshness.retry_in_seconds" in html
+
+
+@pytest.mark.anyio
+async def test_dashboard_reports_each_live_section_from_its_own_freshness():
+ html = await dashboard()
+
+ assert "snapshot.freshness?.sections?.notifications" in html
+ assert "snapshot.freshness?.sections?.events" in html
+ assert "const notificationsFresh = Array.isArray(snapshot.notifications)" not in html
+ assert "Unread updates unavailable · showing last known updates" in html
+ assert "Activity refresh failed · showing last activity" in html