feat(perf): boucles par dépôt parallélisées et cache stale-while-revalidate (chantier A)

- _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.
This commit is contained in:
Spark
2026-10-12 13:53:59 -04:00
parent 593f076555
commit 3e3187d597
2 changed files with 260 additions and 114 deletions
+79 -44
View File
@@ -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."""
+181 -70
View File
@@ -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)