From 3e3187d597c62916bddc97d0b2caee1de45297c7 Mon Sep 17 00:00:00 2001 From: Spark Date: Mon, 12 Oct 2026 13:53:59 -0400 Subject: [PATCH] =?UTF-8?q?feat(perf):=20boucles=20par=20d=C3=A9p=C3=B4t?= =?UTF-8?q?=20parall=C3=A9lis=C3=A9es=20et=20cache=20stale-while-revalidat?= =?UTF-8?q?e=20(chantier=20A)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _parallel_map : asyncio.gather borné (16) pour dashboard, /health, /forecast, /triage et le mur — le patron de /focus, généralisé. Avant : la page attendait la somme des allers- retours Gitea de tous les dépôts, en séquentiel. - Client Gitea : les listes (dépôts, issues, jalons, étoiles) passent en stale-while-revalidate — une entrée périmée est servie immédiatement et rafraîchie en tâche de fond. Les mutations vident le cache et bumpent une génération qui désarme les rafraîchissements en vol. - /forecast calcule des lignes de jalons tous projets (écart ETA/échéance) ; /health trie par sévérité et expose la pastille ; le détail santé reçoit la liste des projets. --- app/gitea_client.py | 123 ++++++++++++++-------- app/main.py | 251 ++++++++++++++++++++++++++++++++------------ 2 files changed, 260 insertions(+), 114 deletions(-) diff --git a/app/gitea_client.py b/app/gitea_client.py index d9f64f8..a3b2c40 100644 --- a/app/gitea_client.py +++ b/app/gitea_client.py @@ -17,8 +17,10 @@ Changes from the 0.1.0 client: - .env loading targets the project root only (no hardcoded home path). """ +import asyncio import logging import time +from collections.abc import Awaitable, Callable from pathlib import Path from typing import Any @@ -60,6 +62,8 @@ class GiteaClient: ) self._user: dict | None = None self._cache: dict[str, tuple[float, Any]] = {} + self._refreshing: set[str] = set() + self._cache_gen = 0 async def aclose(self) -> None: await self.client.aclose() @@ -105,24 +109,62 @@ class GiteaClient: page += 1 return items - def _cache_get(self, key: str) -> Any | None: - if self.cache_ttl <= 0: - return None - hit = self._cache.get(key) - if hit and (time.monotonic() - hit[0]) < self.cache_ttl: - return hit[1] - return None - def _cache_put(self, key: str, value: Any) -> None: if self.cache_ttl > 0: self._cache[key] = (time.monotonic(), value) def _cache_clear(self) -> None: self._cache.clear() + # Invalidate in-flight background refreshes: a refresh that + # started before a mutation must not resurrect stale data. + self._cache_gen += 1 + + async def _cached_list( + self, key: str, fetcher: Callable[[], Awaitable[Any]] + ) -> Any: + """Stale-while-revalidate cache for list endpoints (0.19.0). + + Fresh (< TTL): serve the cache. Stale: serve the cached + value IMMEDIATELY and refresh it in the background (once + per key) — the user never waits on Gitea for a page when + any earlier answer exists. Absent: fetch and wait. + Mutations clear the cache (see _cache_clear), so writes are + never hidden by a stale entry. + """ + if self.cache_ttl <= 0: + return await fetcher() + hit = self._cache.get(key) + if hit is not None: + if (time.monotonic() - hit[0]) < self.cache_ttl: + return hit[1] + if key not in self._refreshing: + self._refreshing.add(key) + asyncio.create_task( + self._refresh_entry(key, fetcher, self._cache_gen) + ) + return hit[1] + value = await fetcher() + self._cache_put(key, value) + return value + + async def _refresh_entry( + self, + key: str, + fetcher: Callable[[], Awaitable[Any]], + generation: int, + ) -> None: + try: + value = await fetcher() + if generation == self._cache_gen: + self._cache_put(key, value) + except Exception as exc: + logger.debug("background refresh failed for %s: %s", key, exc) + finally: + self._refreshing.discard(key) def invalidate_cache(self) -> None: """Drop every cached list response (called on webhook events).""" - self._cache.clear() + self._cache_clear() # ── API ──────────────────────────────────────────────── @@ -141,20 +183,21 @@ class GiteaClient: return self._user async def search_repos(self) -> list[dict]: - """All repos visible to the token owner (paginated, cached).""" - cached = self._cache_get("search_repos") - if cached is not None: - return cached - params: dict[str, Any] = {} - try: - user = await self.get_user() - if user and user.get("id"): - params["uid"] = user["id"] - except Exception as exc: # no/invalid token → public search - logger.debug("get_user failed, falling back to public search: %s", exc) - repos = await self._get_all_pages("/repos/search", **params) - self._cache_put("search_repos", repos) - return repos + """All repos visible to the token owner (paginated, SWR-cached).""" + + async def _fetch() -> list[dict]: + params: dict[str, Any] = {} + try: + user = await self.get_user() + if user and user.get("id"): + params["uid"] = user["id"] + except Exception as exc: # no/invalid token → public search + logger.debug( + "get_user failed, falling back to public search: %s", exc + ) + return await self._get_all_pages("/repos/search", **params) + + return await self._cached_list("search_repos", _fetch) async def get_repo(self, owner: str, repo: str) -> dict: return await self._get(f"/repos/{owner}/{repo}") @@ -178,14 +221,12 @@ class GiteaClient: (see app.main._is_issue). """ key = f"issues:{owner}/{repo}:{state}" - cached = self._cache_get(key) - if cached is not None: - return cached - issues = await self._get_all_pages( - f"/repos/{owner}/{repo}/issues", state=state + return await self._cached_list( + key, + lambda: self._get_all_pages( + f"/repos/{owner}/{repo}/issues", state=state + ), ) - self._cache_put(key, issues) - return issues async def get_all_issues( self, repos: list[dict], state: str = "open" @@ -308,25 +349,19 @@ class GiteaClient: self, owner: str, repo: str, state: str = "open" ) -> list[dict]: key = f"milestones:{owner}/{repo}:{state}" - cached = self._cache_get(key) - if cached is not None: - return cached - milestones = await self._get_all_pages( - f"/repos/{owner}/{repo}/milestones", state=state + return await self._cached_list( + key, + lambda: self._get_all_pages( + f"/repos/{owner}/{repo}/milestones", state=state + ), ) - self._cache_put(key, milestones) - return milestones async def list_starred(self) -> list[dict]: """Repos starred by the token user (issue #20: pinning rides on Gitea stars — the forge stays the source of truth).""" - key = "user:starred" - cached = self._cache_get(key) - if cached is not None: - return cached - repos = await self._get_all_pages("/user/starred") - self._cache_put(key, repos) - return repos + return await self._cached_list( + "user:starred", lambda: self._get_all_pages("/user/starred") + ) async def set_starred(self, owner: str, repo: str, starred: bool) -> None: """Star / unstar a repo for the token user.""" diff --git a/app/main.py b/app/main.py index 51a3341..0aa9d21 100644 --- a/app/main.py +++ b/app/main.py @@ -378,6 +378,27 @@ def _parse_datetime(value: str | None) -> datetime | None: return None +async def _parallel_map(items, fn, limit: int = 16) -> list: + """Run an async fn over items with bounded concurrency (0.19.0). + + Cross-project pages used to await each repo in sequence, so the + page took the SUM of every Gitea round trip. Gathering with a + per-call semaphore (created here, bound to the current loop) + cuts that to roughly the slowest single repo, without letting + a big instance stampede Gitea. Exceptions come back in place — + callers decide what a failed repo means for them. + """ + semaphore = asyncio.Semaphore(limit) + + async def _run(item): + async with semaphore: + return await fn(item) + + return await asyncio.gather( + *(_run(item) for item in items), return_exceptions=True + ) + + def classify_repos( repos: list[dict], now: datetime | None = None ) -> tuple[list[dict], list[dict], list[dict]]: @@ -585,7 +606,7 @@ async def dashboard(request: Request): except Exception as exc: # stars are decoration, never fatal (#20) logger.debug("starred list failed: %s", exc) starred_names = set() - for repo_info in active: + async def _enrich_active(repo_info: dict) -> None: full_name = repo_info["full_name"] owner, _, repo = full_name.partition("/") repo_info["forecast_label"] = _repo_forecast_label( @@ -599,6 +620,8 @@ async def dashboard(request: Request): logger.debug("health dot failed for %s: %s", full_name, exc) repo_info["health_dot"] = None repo_info["starred"] = full_name in starred_names + + await _parallel_map(active, _enrich_active) for repo_info in inactive: repo_info["starred"] = repo_info["full_name"] in starred_names activity_by_repo = get_project_activity() @@ -1013,19 +1036,18 @@ async def triage_picker(request: Request): """Triage mode (issue #11): pick a project by its backlog size.""" repos = await gitea.search_repos() active, _inactive, _archived = classify_repos(repos) - projects = [] - for repo_info in active: + + async def _one_triage(repo_info: dict) -> dict | None: owner, _, name = repo_info["full_name"].partition("/") - try: - all_items = await gitea.get_repo_issues(owner, name, state="all") - except Exception: - continue + all_items = await gitea.get_repo_issues(owner, name, state="all") columns, _display, _state = _board_columns_for(owner, name, all_items) backlog = len(columns["backlog"]) if backlog: - projects.append( - {"full_name": repo_info["full_name"], "backlog": backlog} - ) + return {"full_name": repo_info["full_name"], "backlog": backlog} + return None + + results = await _parallel_map(active, _one_triage) + projects = [r for r in results if isinstance(r, dict)] projects.sort(key=lambda p: -p["backlog"]) return render(request, "triage.html", projects=projects) @@ -1554,30 +1576,52 @@ async def health_index(request: Request): """Flow health, global summary (issue #9).""" repos = await gitea.search_repos() active, _inactive, _archived = classify_repos(repos) - projects = [] - for repo_info in active: + + async def _one_health(repo_info: dict) -> dict: owner, _, name = repo_info["full_name"].partition("/") - try: - projects.append(await _project_health(owner, name)) - except Exception as exc: - logger.warning("Health failed for %s: %s", repo_info["full_name"], exc) + return await _project_health(owner, name) + + results = await _parallel_map(active, _one_health) + projects = [] + for repo_info, result in zip(active, results): + if isinstance(result, Exception): + logger.warning( + "Health failed for %s: %s", repo_info["full_name"], result + ) + continue + result["dot"] = _health_dot(result) + projects.append(result) + severity_rank = {"red": 0, "orange": 1, "green": 2} + projects.sort( + key=lambda p: ( + severity_rank.get(p["dot"]["level"], 3), + -len(p["zombies"]), + ) + ) return render(request, "health.html", projects=projects) @app.get("/health/{owner}/{repo}", response_class=HTMLResponse) async def health_project(request: Request, owner: str, repo: str): - """Flow health of one project: dwell per column + zombies.""" + """Flow health of one project: dwell per column + zombies. + + Passes the active project list so the page can offer a project + switcher — no more back-to-the-index round trip (0.19.0, C). + """ repos = await gitea.search_repos() - known = {r["full_name"] for r in repos} - if f"{owner}/{repo}" not in known: + active, _inactive, _archived = classify_repos(repos) + project_names = [r["full_name"] for r in active] + if f"{owner}/{repo}" not in {r["full_name"] for r in repos}: return render( request, "health_project.html", health=None, full_name=f"{owner}/{repo}", + project_names=project_names, ) health = await _project_health(owner, repo) return render( request, "health_project.html", health=health, full_name=f"{owner}/{repo}", + project_names=project_names, ) @@ -1604,11 +1648,29 @@ async def standup_page(request: Request): @app.get("/forecast", response_class=HTMLResponse) async def forecast_page(request: Request): - """Forecasts (issue #7): real-pace ETAs per project + milestone.""" + """Forecasts (issue #7; portfolio table in 0.19.0, chantier C). + + One row per open milestone across every active project, + sorted by risk (largest ETA-vs-deadline gap first) — the + cross-project question the old per-project card wall could + not answer without reading everything. Per-project pace + (weekly bars + backlog ETA) stays as a compact second table. + """ repos = await gitea.search_repos() active, _inactive, _archived = classify_repos(repos) - projects = [] - for repo_info in active: + + async def _fetch_milestones(repo_info: dict) -> list[dict]: + owner, _, name = repo_info["full_name"].partition("/") + try: + return await gitea.get_milestones(owner, name) + except Exception: + return [] + + ms_results = await _parallel_map(active, _fetch_milestones) + + rows: list[dict] = [] + projects: list[dict] = [] + for repo_info, ms_list in zip(active, ms_results): full = repo_info["full_name"] owner, _, name = full.partition("/") weekly = forecast_service.weekly_closures( @@ -1623,50 +1685,100 @@ async def forecast_page(request: Request): } for b in weekly ] + stats = forecast_service.throughput_stats(weekly) remaining = repo_info.get("open_issues_count") or 0 est = forecast_service.estimate(remaining, weekly) - milestones = [] - try: - ms_list = await gitea.get_milestones(owner, name) - except Exception: - ms_list = [] - for ms in ms_list: - if ms.get("state") not in (None, "open"): - continue - rem = ms.get("open_issues") or 0 - if rem == 0 and not (ms.get("closed_issues") or 0): - continue - m_est = forecast_service.estimate(rem, weekly) - verdict = None - due_raw = ms.get("due_on") - if m_est and not m_est.get("done") and due_raw: - due = _parse_datetime(str(due_raw)) - if due is not None: - verdict = ( - "⚠️ en retard à ce rythme" - if m_est["central"] > due.date() - else "✅ dans les temps à ce rythme" - ) - milestones.append( - { - "title": ms.get("title", ""), - "due_on": (due_raw or "")[:10], - "remaining": rem, - "eta": forecast_service.eta_label(m_est), - "verdict": verdict, - } - ) projects.append( { "full_name": full, "bars": bars, - "stats": forecast_service.throughput_stats(weekly), + "stats": stats, "remaining": remaining, "eta_label": forecast_service.eta_label(est), - "milestones": milestones, } ) - return render(request, "forecast.html", projects=projects) + if isinstance(ms_list, Exception): + ms_list = [] + next_title = None + next_due = None + for ms in ms_list: + if ms.get("state") not in (None, "open"): + continue + rem = ms.get("open_issues") or 0 + closed_n = ms.get("closed_issues") or 0 + if rem == 0 and not closed_n: + continue + total = rem + closed_n + m_est = forecast_service.estimate(rem, weekly) + due_raw = ms.get("due_on") + due_dt = _parse_datetime(str(due_raw)) if due_raw else None + due_date = due_dt.date() if due_dt else None + gap_days = None + verdict = "noeta" + if due_date is None: + verdict = "nodue" + elif m_est and not m_est.get("done"): + gap_days = (m_est["central"] - due_date).days + verdict = "late" if gap_days > 0 else "ontime" + elif m_est and m_est.get("done"): + verdict = "ontime" + rows.append( + { + "project": full, + "owner": owner, + "repo": name, + "title": ms.get("title", ""), + "due_label": ( + forecast_service.french_date(due_date, short=True) + if due_date else "—" + ), + "due_sort": due_date.isoformat() if due_date else "9999", + "remaining": rem, + "closed": closed_n, + "total": total, + "pct": round(100 * closed_n / total) if total else 0, + "eta": forecast_service.eta_label(m_est), + "gap_days": gap_days, + "verdict": verdict, + "throughput": stats["avg"] if stats else None, + "anchor": False, + } + ) + if ( + due_date is not None + and (next_due is None or due_date < next_due) + ): + next_due, next_title = due_date, ms.get("title", "") + for row in rows: + if ( + row["project"] == full + and row["title"] == next_title + and next_title is not None + ): + row["anchor"] = True + + def _risk_key(row: dict): + # Late first (worst gap first), then dated on-time by due + # date, then no-ETA, then no due date. + order = {"late": 0, "ontime": 1, "noeta": 2, "nodue": 3} + return ( + order.get(row["verdict"], 4), + -(row["gap_days"] or 0) if row["verdict"] == "late" else 0, + row["due_sort"], + row["project"], + row["title"], + ) + + rows.sort(key=_risk_key) + summary = { + "late": sum(1 for r in rows if r["verdict"] == "late"), + "ontime": sum(1 for r in rows if r["verdict"] == "ontime"), + "total": len(rows), + } + return render( + request, "forecast.html", + rows=rows, projects=projects, summary=summary, + ) @app.get("/focus", response_class=HTMLResponse) @@ -1739,15 +1851,14 @@ async def _collect_wall_data(only_repo: str | None) -> list[dict]: closed_today.get(event["repo_full_name"], 0) + 1 ) - slides = [] - for repo in repos: + async def _one_slide(repo: dict) -> dict | None: full_name = repo["full_name"] owner, _, name = full_name.partition("/") try: items = await gitea.get_repo_issues(owner, name, state="open") except Exception as exc: logger.debug("wall: issues failed for %s: %s", full_name, exc) - continue + return None open_issues = [i for i in items if _is_issue(i)] _adopt_label_columns(full_name, open_issues, get_board_state(full_name)) board_state = get_board_state(full_name) @@ -1769,17 +1880,17 @@ async def _collect_wall_data(only_repo: str | None) -> list[dict]: "aging": days is not None and days >= aging_days, } ) - slides.append( - { - "full_name": full_name, - "counts": {col: len(columns[col]) for col in KANBAN_COLUMNS}, - "open_total": len(open_issues), - "cards": cards[:9], - "cards_total": len(cards), - "closed_today": closed_today.get(full_name, 0), - } - ) - return slides + return { + "full_name": full_name, + "counts": {col: len(columns[col]) for col in KANBAN_COLUMNS}, + "open_total": len(open_issues), + "cards": cards[:9], + "cards_total": len(cards), + "closed_today": closed_today.get(full_name, 0), + } + + results = await _parallel_map(repos, _one_slide) + return [r for r in results if isinstance(r, dict)] @app.get("/wall", response_class=HTMLResponse)