From 4870b58c5b9f3972bb258a4eacdf621490838118 Mon Sep 17 00:00:00 2001 From: timmy Date: Fri, 7 Aug 2026 04:53:42 +0000 Subject: [PATCH] feat: isolate live section refresh backoff (#167) --- frontend/index.html | 26 ++- src/main.py | 221 ++++++++++++++++---------- tests/test_live_snapshot.py | 100 ++++++++++-- tests/test_notification_read.py | 4 +- tests/test_shared_context_snapshot.py | 11 ++ 5 files changed, 262 insertions(+), 100 deletions(-) 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