feat: keep Find Work instant during refresh (#213)

src/gitea_proxy.py

@@ -11,6 +11,7 @@ GITEA_URL = os.getenv("GITEA_URL", "http://127.0.0.1:3000").rstrip("/")
 GITEA_TOKEN = os.getenv("GITEA_TOKEN", "")
 REVIEW_DIFF_MAX_BYTES = 64 * 1024
 REVIEW_DIFF_MAX_LINES = 400
+AVAILABLE_ISSUE_PAGE_CONCURRENCY = 3
 _client: httpx.AsyncClient | None = None
 
 
@@ -328,8 +329,8 @@ async def available_issue_snapshot(max_pages: int = 10, upstream_limit: int = 50
     """Load, filter, and globally rank a bounded snapshot of available issues."""
     items: list[dict] = []
     seen_ids: set[Any] = set()
-    upstream_total: int | None = None
-    for upstream_page in range(1, max_pages + 1):
+
+    async def load_page(upstream_page: int) -> tuple[list[Any], int | None]:
         response = await _get_client().get(
             "/api/v1/repos/issues/search",
             headers=_auth(),
@@ -342,21 +343,41 @@ async def available_issue_snapshot(max_pages: int = 10, upstream_limit: int = 50
         payload = response.json()
         if not isinstance(payload, list):
             raise ValueError("Gitea available issue search response was not a list")
-        if upstream_total is None:
-            try:
-                upstream_total = max(0, int(response.headers["X-Total-Count"]))
-            except (KeyError, TypeError, ValueError):
-                upstream_total = None
+        try:
+            total = max(0, int(response.headers["X-Total-Count"]))
+        except (KeyError, TypeError, ValueError):
+            total = None
+        return payload, total
+
+    first_payload, upstream_total = await load_page(1)
+    pages: list[list[Any]] = [first_payload]
+    if upstream_total is not None:
+        page_count = min(max_pages, max(1, (upstream_total + upstream_limit - 1) // upstream_limit))
+        semaphore = asyncio.Semaphore(AVAILABLE_ISSUE_PAGE_CONCURRENCY)
+
+        async def load_bounded(page: int) -> list[Any]:
+            async with semaphore:
+                payload, _ = await load_page(page)
+                return payload
+
+        if page_count > 1:
+            pages.extend(await asyncio.gather(*(
+                load_bounded(page) for page in range(2, page_count + 1)
+            )))
+    else:
+        previous = first_payload
+        for page in range(2, max_pages + 1):
+            if not previous or len(previous) < upstream_limit:
+                break
+            previous, _ = await load_page(page)
+            pages.append(previous)
+
+    for payload in pages:
         for raw_item in payload:
             item = _normalize_available_issue(raw_item)
             if item is not None and item["id"] not in seen_ids:
                 seen_ids.add(item["id"])
                 items.append(item)
-        loaded = upstream_page * upstream_limit
-        if not payload or len(payload) < upstream_limit or (
-            upstream_total is not None and loaded >= upstream_total
-        ):
-            break
 
     priority = {"p0", "priority-high", "critical"}
     items.sort(key=lambda item: (item["repository"], item["number"] or 0))

src/main.py

@@ -76,6 +76,7 @@ LIVE_SNAPSHOT_FRESHNESS_SECONDS = 8.0
 LIVE_SNAPSHOT_RETRY_BASE_SECONDS = 5.0
 LIVE_SNAPSHOT_RETRY_MAX_SECONDS = 60.0
 AVAILABLE_ISSUE_SNAPSHOT_FRESHNESS_SECONDS = 15.0
+AVAILABLE_ISSUE_SNAPSHOT_RETRY_SECONDS = 5.0
 FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
 _live_snapshot_task: asyncio.Task | None = None
 _live_snapshot_value: dict | None = None
@@ -107,6 +108,7 @@ _idempotency_ledger = IdempotencyLedger(
 _available_issue_snapshot_task: asyncio.Task | None = None
 _available_issue_snapshot_value: list[dict] | None = None
 _available_issue_snapshot_created_at: float | None = None
+_available_issue_snapshot_retry_at: float | None = None
 
 
 class ContextPayloadError(ValueError):
@@ -600,13 +602,24 @@ async def paged_work(
 
 async def _refresh_available_issue_snapshot() -> list[dict]:
     global _available_issue_snapshot_value, _available_issue_snapshot_created_at
+    global _available_issue_snapshot_retry_at
     result = await gitea_proxy.available_issue_snapshot()
     _available_issue_snapshot_value = result
     _available_issue_snapshot_created_at = time.monotonic()
+    _available_issue_snapshot_retry_at = None
     return result
 
 
-async def _available_issue_snapshot() -> tuple[list[dict], bool]:
+def _observe_available_issue_refresh(task: asyncio.Task) -> None:
+    global _available_issue_snapshot_retry_at
+    if not task.cancelled():
+        if task.exception() is not None:
+            _available_issue_snapshot_retry_at = (
+                time.monotonic() + AVAILABLE_ISSUE_SNAPSHOT_RETRY_SECONDS
+            )
+
+
+async def _available_issue_snapshot() -> tuple[list[dict], bool, bool, bool]:
     global _available_issue_snapshot_task
     now = time.monotonic()
     if (
@@ -615,16 +628,25 @@ async def _available_issue_snapshot() -> tuple[list[dict], bool]:
         and now - _available_issue_snapshot_created_at
         < AVAILABLE_ISSUE_SNAPSHOT_FRESHNESS_SECONDS
     ):
-        return _available_issue_snapshot_value, False
+        return _available_issue_snapshot_value, False, False, False
+    if (
+        _available_issue_snapshot_value is not None
+        and _available_issue_snapshot_retry_at is not None
+        and now < _available_issue_snapshot_retry_at
+    ):
+        return _available_issue_snapshot_value, True, False, True
     if _available_issue_snapshot_task is None or _available_issue_snapshot_task.done():
         _available_issue_snapshot_task = asyncio.create_task(
             _refresh_available_issue_snapshot()
         )
+        _available_issue_snapshot_task.add_done_callback(_observe_available_issue_refresh)
+    if _available_issue_snapshot_value is not None:
+        return _available_issue_snapshot_value, True, True, False
     try:
-        return await asyncio.shield(_available_issue_snapshot_task), False
+        return await asyncio.shield(_available_issue_snapshot_task), False, False, False
     except Exception:
         if _available_issue_snapshot_value is not None:
-            return _available_issue_snapshot_value, True
+            return _available_issue_snapshot_value, True, False, True
         raise
 
 
@@ -642,7 +664,7 @@ def _available_issue_page(items: list[dict], page: int, limit: int = 50) -> dict
 @app.get("/api/v1/available-issues")
 async def available_issues(page: int = Query(default=1, ge=1, le=100)) -> JSONResponse:
     try:
-        items, stale = await asyncio.wait_for(
+        items, stale, revalidating, refresh_failed = await asyncio.wait_for(
             _available_issue_snapshot(), timeout=WORK_PAGE_TIMEOUT_SECONDS
         )
     except Exception:
@@ -654,6 +676,10 @@ async def available_issues(page: int = Query(default=1, ge=1, le=100)) -> JSONRe
     result = _available_issue_page(items, page)
     if stale:
         result["stale"] = True
+    if revalidating:
+        result["revalidating"] = True
+    if refresh_failed:
+        result["refresh_failed"] = True
     return JSONResponse(result)