feat(perf): chantier B — instantanés SQLite et rafraîchisseur d'arrière-plan
- Table snapshots + écriture traversante du cache client, hydratation au démarrage (SWR : servi + rafraîchi en fond). - Boucle de rafraîchissement (120 s par défaut) : index, étoiles, issues all+open et jalons des projets actifs ; webhook = rafraîchissement immédiat du dépôt ; fraîcheur affichée.
This commit is contained in:
+104
-13
@@ -64,6 +64,10 @@ class GiteaClient:
|
||||
self._cache: dict[str, tuple[float, Any]] = {}
|
||||
self._refreshing: set[str] = set()
|
||||
self._cache_gen = 0
|
||||
# Snapshot bookkeeping (chantier B): cache key -> UTC time
|
||||
# the payload was fetched from Gitea, for the freshness
|
||||
# label and the persisted `snapshots` table.
|
||||
self._snapshot_times: dict[str, str] = {}
|
||||
|
||||
async def aclose(self) -> None:
|
||||
await self.client.aclose()
|
||||
@@ -112,6 +116,90 @@ class GiteaClient:
|
||||
def _cache_put(self, key: str, value: Any) -> None:
|
||||
if self.cache_ttl > 0:
|
||||
self._cache[key] = (time.monotonic(), value)
|
||||
self._persist_snapshot(key, value)
|
||||
|
||||
# ── Snapshots (chantier B, 0.22.0) ──────────────────────
|
||||
|
||||
def _persist_snapshot(self, key: str, value: Any) -> None:
|
||||
"""Write-through to the SQLite snapshots table.
|
||||
|
||||
Persistence must never break a read: any failure (DB
|
||||
locked, unserializable payload) is logged and swallowed.
|
||||
"""
|
||||
try:
|
||||
from app import models
|
||||
|
||||
models.save_snapshot(key, value)
|
||||
import datetime
|
||||
|
||||
self._snapshot_times[key] = datetime.datetime.now(
|
||||
datetime.timezone.utc
|
||||
).strftime("%Y-%m-%d %H:%M:%S")
|
||||
except Exception as exc:
|
||||
logger.debug("snapshot persist failed for %s: %s", key, exc)
|
||||
|
||||
def hydrate_from_snapshots(self) -> int:
|
||||
"""Preload the cache from persisted snapshots (startup).
|
||||
|
||||
Entries are inserted with an already-stale timestamp: the
|
||||
SWR layer serves them instantly AND schedules a background
|
||||
refresh, so the first page after a restart renders from
|
||||
local data (tens of ms) while Gitea data catches up.
|
||||
Returns the number of entries hydrated.
|
||||
"""
|
||||
if self.cache_ttl <= 0:
|
||||
return 0
|
||||
try:
|
||||
from app import models
|
||||
|
||||
snapshots = models.load_snapshots()
|
||||
except Exception as exc:
|
||||
logger.warning("snapshot hydration failed: %s", exc)
|
||||
return 0
|
||||
stale_at = time.monotonic() - self.cache_ttl - 1.0
|
||||
for key, (payload, updated_at) in snapshots.items():
|
||||
self._cache[key] = (stale_at, payload)
|
||||
self._snapshot_times[key] = updated_at
|
||||
if snapshots:
|
||||
logger.info("hydrated %d snapshot(s) from SQLite", len(snapshots))
|
||||
return len(snapshots)
|
||||
|
||||
def latest_snapshot_at(self) -> str | None:
|
||||
"""Freshest Gitea fetch time across tracked keys (UTC str)."""
|
||||
if not self._snapshot_times:
|
||||
return None
|
||||
return max(self._snapshot_times.values())
|
||||
|
||||
async def _force(self, key: str, fetcher: Callable[[], Awaitable[Any]]) -> Any:
|
||||
"""Fetch now, bypassing freshness (background refresher)."""
|
||||
value = await fetcher()
|
||||
self._cache_put(key, value)
|
||||
return value
|
||||
|
||||
async def refresh_starred(self) -> list[dict]:
|
||||
return await self._force(
|
||||
"user:starred", lambda: self._get_all_pages("/user/starred")
|
||||
)
|
||||
|
||||
async def refresh_repo_issues(
|
||||
self, owner: str, repo: str, state: str = "all"
|
||||
) -> list[dict]:
|
||||
return await self._force(
|
||||
f"issues:{owner}/{repo}:{state}",
|
||||
lambda: self._get_all_pages(
|
||||
f"/repos/{owner}/{repo}/issues", state=state
|
||||
),
|
||||
)
|
||||
|
||||
async def refresh_milestones(
|
||||
self, owner: str, repo: str
|
||||
) -> list[dict]:
|
||||
return await self._force(
|
||||
f"milestones:{owner}/{repo}:open",
|
||||
lambda: self._get_all_pages(
|
||||
f"/repos/{owner}/{repo}/milestones", state="open"
|
||||
),
|
||||
)
|
||||
|
||||
def _cache_clear(self) -> None:
|
||||
self._cache.clear()
|
||||
@@ -182,22 +270,25 @@ class GiteaClient:
|
||||
self._user = await self._get("/user")
|
||||
return self._user
|
||||
|
||||
async def _fetch_search_repos(self) -> 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)
|
||||
|
||||
async def search_repos(self) -> list[dict]:
|
||||
"""All repos visible to the token owner (paginated, SWR-cached)."""
|
||||
return await self._cached_list("search_repos", self._fetch_search_repos)
|
||||
|
||||
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 refresh_search_repos(self) -> list[dict]:
|
||||
"""Force-fetch the repo index (background refresher)."""
|
||||
return await self._force("search_repos", self._fetch_search_repos)
|
||||
|
||||
async def get_repo(self, owner: str, repo: str) -> dict:
|
||||
return await self._get(f"/repos/{owner}/{repo}")
|
||||
|
||||
@@ -11,6 +11,7 @@ and upgrades older databases in place (position column, notes
|
||||
updated_at, cleanup of rows carrying an unknown column name).
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
from contextlib import contextmanager
|
||||
@@ -91,6 +92,12 @@ def init_db() -> None:
|
||||
value TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS snapshots (
|
||||
key TEXT PRIMARY KEY,
|
||||
payload TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_events_repo
|
||||
ON board_events(repo_full_name, issue_number, created_at);
|
||||
"""
|
||||
@@ -292,6 +299,48 @@ def set_kv(key: str, value: str) -> None:
|
||||
)
|
||||
|
||||
|
||||
# ── Snapshots (chantier B, 0.22.0) ─────────────────────────
|
||||
# Persisted copies of the Gitea list responses the pages read
|
||||
# (repos, issues, milestones, starred). The client cache is
|
||||
# hydrated from these rows at startup, so even the first page
|
||||
# after a restart renders from local data while the background
|
||||
# refresher re-fetches from Gitea.
|
||||
|
||||
def save_snapshot(key: str, payload) -> None:
|
||||
with get_db() as conn:
|
||||
conn.execute(
|
||||
"INSERT INTO snapshots (key, payload, updated_at) "
|
||||
"VALUES (?, ?, datetime('now')) "
|
||||
"ON CONFLICT(key) DO UPDATE SET payload = excluded.payload, "
|
||||
"updated_at = excluded.updated_at",
|
||||
(key, json.dumps(payload)),
|
||||
)
|
||||
|
||||
|
||||
def load_snapshots() -> dict:
|
||||
"""All snapshots as {key: (payload, updated_at_utc_str)}."""
|
||||
with get_db() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT key, payload, updated_at FROM snapshots"
|
||||
).fetchall()
|
||||
out = {}
|
||||
for row in rows:
|
||||
try:
|
||||
out[row["key"]] = (json.loads(row["payload"]), row["updated_at"])
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
return out
|
||||
|
||||
|
||||
def latest_snapshot_at() -> str | None:
|
||||
"""UTC timestamp (SQLite format) of the freshest snapshot."""
|
||||
with get_db() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT MAX(updated_at) AS latest FROM snapshots"
|
||||
).fetchone()
|
||||
return row["latest"] if row else None
|
||||
|
||||
|
||||
# ── Event journal ──────────────────────────────────────────
|
||||
|
||||
EVENT_TYPES = ("created", "moved", "closed", "reopened", "commented", "rule")
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
"""Background snapshot refresher — chantier B (0.22.0).
|
||||
|
||||
The expensive part of Kanbante is the fan-out of Gitea list
|
||||
calls behind the portfolio pages (repo index, then issues and
|
||||
milestones for every active project). Since 0.19.0 those calls
|
||||
are parallel and stale-while-revalidate, which fixed repeat
|
||||
loads, but the FIRST page after a restart still paid seconds.
|
||||
|
||||
The refresher pays that cost instead of the user: every
|
||||
KANBANTE_SNAPSHOT_INTERVAL seconds it force-refreshes the hot
|
||||
keys (repo index, starred, issues + milestones of active
|
||||
repos). Every fetch write-throughs to the SQLite `snapshots`
|
||||
table, and the client cache is hydrated from that table at
|
||||
startup — so pages read local data from the very first request
|
||||
and only ever wait on Gitea when no snapshot exists yet.
|
||||
|
||||
Freshness contract, shown in the UI: Gitea-side data may lag
|
||||
by up to one interval; a Gitea webhook refreshes the touched
|
||||
repo immediately (see refresh_repo_snapshots); anything the
|
||||
user does locally (board journal) is always instant.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import datetime
|
||||
import logging
|
||||
|
||||
from app import models
|
||||
from app.config import get_settings
|
||||
|
||||
logger = logging.getLogger("kanbante.snapshots")
|
||||
|
||||
_REFRESH_CONCURRENCY = 8
|
||||
|
||||
|
||||
async def refresh_repo_snapshots(gitea, owner: str, repo: str) -> None:
|
||||
"""Refresh one repo's hot keys (webhooks call this).
|
||||
|
||||
Both issue states are refreshed: most pages read "all", but
|
||||
Focus/Triage/Mes issues read "open" — a snapshot set missing
|
||||
either leaves a slow first hit on the table.
|
||||
"""
|
||||
try:
|
||||
await gitea.refresh_repo_issues(owner, repo, "all")
|
||||
await gitea.refresh_repo_issues(owner, repo, "open")
|
||||
await gitea.refresh_milestones(owner, repo)
|
||||
except Exception as exc:
|
||||
logger.debug("repo snapshot refresh failed for %s/%s: %s", owner, repo, exc)
|
||||
|
||||
|
||||
async def refresh_snapshots(gitea, classify_repos) -> int:
|
||||
"""One full refresh cycle. Returns the active-repo count."""
|
||||
repos = await gitea.refresh_search_repos()
|
||||
try:
|
||||
await gitea.refresh_starred()
|
||||
except Exception as exc:
|
||||
logger.debug("starred refresh failed: %s", exc)
|
||||
active, _inactive, _archived = classify_repos(repos)
|
||||
semaphore = asyncio.Semaphore(_REFRESH_CONCURRENCY)
|
||||
|
||||
async def _one(repo: dict) -> None:
|
||||
owner, _, name = repo["full_name"].partition("/")
|
||||
async with semaphore:
|
||||
await refresh_repo_snapshots(gitea, owner, name)
|
||||
|
||||
await asyncio.gather(*(_one(repo) for repo in active))
|
||||
logger.info("snapshot refresh complete (%d active repos)", len(active))
|
||||
return len(active)
|
||||
|
||||
|
||||
async def snapshot_loop(gitea, classify_repos) -> None:
|
||||
"""Periodic refresher, started by the app lifespan.
|
||||
|
||||
Sleeps briefly first: hydration already serves the pages,
|
||||
the loop only has to keep the snapshots warm.
|
||||
"""
|
||||
interval = get_settings().snapshot_interval
|
||||
if interval <= 0:
|
||||
logger.info("snapshot refresher disabled (interval <= 0)")
|
||||
return
|
||||
await asyncio.sleep(10)
|
||||
while True:
|
||||
try:
|
||||
await refresh_snapshots(gitea, classify_repos)
|
||||
except Exception as exc:
|
||||
logger.warning("snapshot refresh cycle failed: %s", exc)
|
||||
await asyncio.sleep(interval)
|
||||
|
||||
|
||||
def freshness_label() -> str | None:
|
||||
"""« à l'instant » / « il y a N min » from the freshest snapshot.
|
||||
|
||||
Honest freshness display (chantier B contract): the sidebar
|
||||
shows how old the Gitea data on screen can be.
|
||||
"""
|
||||
try:
|
||||
latest = models.latest_snapshot_at()
|
||||
except Exception:
|
||||
return None
|
||||
if not latest:
|
||||
return None
|
||||
try:
|
||||
fetched = datetime.datetime.strptime(
|
||||
latest, "%Y-%m-%d %H:%M:%S"
|
||||
).replace(tzinfo=datetime.timezone.utc)
|
||||
except ValueError:
|
||||
return None
|
||||
age = (
|
||||
datetime.datetime.now(datetime.timezone.utc) - fetched
|
||||
).total_seconds()
|
||||
if age < 90:
|
||||
return "à l'instant"
|
||||
if age < 3600:
|
||||
return f"il y a {int(age // 60)} min"
|
||||
if age < 86400:
|
||||
return f"il y a {int(age // 3600)} h"
|
||||
return f"il y a {int(age // 86400)} j"
|
||||
Reference in New Issue
Block a user