From 1c99f6a432155b97847b2597f121e0331b782491 Mon Sep 17 00:00:00 2001 From: 370025263 <370025263@qq.com> Date: Fri, 31 Jul 2026 11:04:23 +0800 Subject: [PATCH] perf(dashboard): collapse per-request full table scans on my page a4 made the my page the default landing, and its endpoints each ran several uncached full scans per request: trajectories via _traj_user_map, trajectories x watch_dirs via skill_lineage per skill chip, and full atom_adoption fetches in my_contributions plus the unindexed skill IN query in skill_main_producers. One first screen stacked a dozen full scans; the sqlite call gate then serialized every other dashboard endpoint behind them, which is why the whole panel felt slower. Fixes, no semantics changed: - _traj_user_map: 5s TTL + singleflight, one scan per request wave - skill_lineage: visible trajectories join extracted and cached the same way, shared across skill chips on a page - my_contributions: full adoption rows cached 5s, callers copy rows - registry DDL: index on atom_adoption(skill) so lineage and producer lookups stop scanning the whole table --- src/xskill/dashboard/console.py | 26 +++++- src/xskill/dashboard/explore.py | 57 ++++++++---- src/xskill/events.py | 39 +++++++-- src/xskill/pipeline/registry.py | 2 + tests/test_dashboard_my_scan_cache.py | 119 ++++++++++++++++++++++++++ 5 files changed, 217 insertions(+), 26 deletions(-) create mode 100644 tests/test_dashboard_my_scan_cache.py diff --git a/src/xskill/dashboard/console.py b/src/xskill/dashboard/console.py index fe0cf704..b21be48b 100644 --- a/src/xskill/dashboard/console.py +++ b/src/xskill/dashboard/console.py @@ -44,6 +44,28 @@ _reco_trigger_cache = SingleFlightTtlCache( ttl_seconds=_RECO_TRIGGER_TTL_SECONDS, max_entries=32) +_ADOPTION_ROWS_TTL_SECONDS = 5.0 +_adoption_rows_cache = SingleFlightTtlCache( + ttl_seconds=_ADOPTION_ROWS_TTL_SECONDS, max_entries=8) + + +def _adoption_rows(db_path: Optional[Path]) -> list[dict]: + """全量 atom_adoption 行（5s TTL + 单飞）。 + + 贡献归属判定要在 Python 侧按 atom_id 内嵌 traj_id 做包含匹配，无法下推 + SQL；每请求整表搬运随 adoption 体量线性放大。短窗缓存把一波请求收敛成 + 一次扫描。返回共享只读 list，调用方逐条 dict() 后再改。 + """ + def build() -> list[dict]: + with pooled_connection(db_path) as conn: + return [ + dict(r) for r in conn.execute( + "SELECT atom_id, skill, weightscore FROM atom_adoption", + ).fetchall() + ] + + return _adoption_rows_cache.get_or_build(str(db_path or ""), build) + def _team_ctx(): from xskill.team.server.api import team_context @@ -472,9 +494,7 @@ def my_contributions(ident=Depends(require_user)): (user,), ).fetchall() } - adoption = conn.execute( - "SELECT atom_id, skill, weightscore FROM atom_adoption" - ).fetchall() + adoption = _adoption_rows(db_path) # atom_id 内嵌 traj_id（atom__NNNN）——按包含判定归属 my_adopted = [dict(a) for a in adoption if any(stem in (a["atom_id"] or "") for stem in my_stems)] diff --git a/src/xskill/dashboard/explore.py b/src/xskill/dashboard/explore.py index 4d25143a..50815f13 100644 --- a/src/xskill/dashboard/explore.py +++ b/src/xskill/dashboard/explore.py @@ -16,7 +16,11 @@ dashboard_visible_trajectory_sql, pooled_connection, ) -from xskill.dashboard.metrics import _resolve_local_root, load_usage_records +from xskill.dashboard.metrics import ( + SingleFlightTtlCache, + _resolve_local_root, + load_usage_records, +) class TrajExplorer: @@ -141,6 +145,42 @@ def _atom_destinations(self, atom_id: str) -> list[dict]: return out +_VISIBLE_TRAJ_TTL_SECONDS = 5.0 +_visible_traj_cache = SingleFlightTtlCache( + ttl_seconds=_VISIBLE_TRAJ_TTL_SECONDS, max_entries=8) + + +def _visible_traj_info(db_path: Optional[Path]) -> dict[str, dict]: + """可见轨迹 stem → {user, model, root}（5s TTL + 单飞）。 + + 本映射要全量 JOIN ``trajectories × watch_dirs``；skill_lineage 被 + 「我的」页贡献图按 skill 逐个调用，逐调逐扫会随轨迹量线性放大。 + 短窗缓存把一页 N 个 skill 的请求收敛成一次扫描。返回共享只读 dict。 + """ + from xskill.config import get_registry_db_path + + def build() -> dict[str, dict]: + with pooled_connection(db_path) as conn: + wd_rows = conn.execute( + "SELECT t.filename fn, t.source_model model, w.label label," + " w.path wpath FROM trajectories t" + " JOIN watch_dirs w ON t.watch_dir_id=w.id" + f" WHERE {dashboard_visible_trajectory_sql('t', 'w')}" + ).fetchall() + db_dir = ( + Path(db_path).parent if db_path else get_registry_db_path().parent + ) + out: dict[str, dict] = {} + for r in wd_rows: + stem = r["fn"][:-3] if r["fn"].endswith(".md") else r["fn"] + out[stem] = {"user": r["label"] or "(local)", + "model": r["model"] or "", + "root": _resolve_local_root(r["wpath"], db_dir)} + return out + + return _visible_traj_cache.get_or_build(str(db_path or ""), build) + + def skill_lineage(skill_dir: Path, name: str, db_path: Optional[Path] = None) -> dict: """skill 血缘（图①下半区）：贡献原子（adoption 事件 + 在途 candidates） @@ -153,20 +193,7 @@ def skill_lineage(skill_dir: Path, name: str, adoption = conn.execute( "SELECT atom_id, weightscore, ts FROM atom_adoption WHERE skill=?" " ORDER BY ts", (name,)).fetchall() - wd_rows = conn.execute( - "SELECT t.filename fn, t.source_model model, w.label label," - " w.path wpath FROM trajectories t" - " JOIN watch_dirs w ON t.watch_dir_id=w.id" - f" WHERE {dashboard_visible_trajectory_sql('t', 'w')}" - ).fetchall() - from xskill.config import get_registry_db_path - db_dir = Path(db_path).parent if db_path else get_registry_db_path().parent - traj_info: dict[str, dict] = {} - for r in wd_rows: - stem = r["fn"][:-3] if r["fn"].endswith(".md") else r["fn"] - traj_info[stem] = {"user": r["label"] or "(local)", - "model": r["model"] or "", - "root": _resolve_local_root(r["wpath"], db_dir)} + traj_info = _visible_traj_info(db_path) from xskill.skill.candidates import load_candidates from xskill.dashboard.metrics import _traj_of_atom entries: dict[str, dict] = {} diff --git a/src/xskill/events.py b/src/xskill/events.py index 2d33dc82..2614f8a6 100644 --- a/src/xskill/events.py +++ b/src/xskill/events.py @@ -26,6 +26,8 @@ import json import logging +import threading +import time from pathlib import Path from typing import Optional @@ -59,15 +61,36 @@ def _traj_of_atom(atom_id: str) -> str: return stem if stem and idx.isdigit() else "" +_TRAJ_USER_TTL_SECONDS = 5.0 +_traj_user_cache: dict[str, tuple[float, dict[str, str]]] = {} +_traj_user_lock = threading.Lock() + + def _traj_user_map(db_path: Optional[Path] = None) -> dict[str, str]: - """filename stem → user_key。调用方批量算贡献人时只建一次。""" - with pooled_connection(db_path) as conn: - return { - (r["filename"][:-3] if r["filename"].endswith(".md") - else r["filename"]): (r["user_key"] or "") - for r in conn.execute( - "SELECT filename, user_key FROM trajectories").fetchall() - } + """filename stem → user_key（5s TTL + 单飞）。 + + 本映射要全表扫 ``trajectories``，而「我的」页一次首屏会经多个端点 + 各自调到这里；短窗缓存把一波请求收敛成一次扫描，锁内构建保证到期 + 瞬间只有一个线程真扫。返回的是缓存内共享 dict，调用方只读不改写。 + """ + key = str(db_path or "") + with _traj_user_lock: + hit = _traj_user_cache.get(key) + if hit and hit[0] > time.monotonic(): + return hit[1] + if len(_traj_user_cache) > 8: + _traj_user_cache.clear() + with pooled_connection(db_path) as conn: + mapping = { + (r["filename"][:-3] if r["filename"].endswith(".md") + else r["filename"]): (r["user_key"] or "") + for r in conn.execute( + "SELECT filename, user_key FROM trajectories").fetchall() + } + _traj_user_cache[key] = ( + time.monotonic() + _TRAJ_USER_TTL_SECONDS, mapping, + ) + return mapping def skill_contributors(skill: str, *, min_weight: int = CONTRIBUTOR_MIN_WEIGHT, diff --git a/src/xskill/pipeline/registry.py b/src/xskill/pipeline/registry.py index 4b927fd1..483e7a79 100644 --- a/src/xskill/pipeline/registry.py +++ b/src/xskill/pipeline/registry.py @@ -185,6 +185,8 @@ def dashboard_visible_trajectory_sql( was_new INTEGER -- 1=首次加入 0=覆盖 ); CREATE INDEX IF NOT EXISTS idx_atom_adopt ON atom_adoption(atom_id); +-- 血缘/主贡献人按 skill 过滤（skill_lineage、skill_main_producers 的 IN 查询） +CREATE INDEX IF NOT EXISTS idx_atom_adopt_skill ON atom_adoption(skill); CREATE TABLE IF NOT EXISTS canary_decision ( id INTEGER PRIMARY KEY AUTOINCREMENT, diff --git a/tests/test_dashboard_my_scan_cache.py b/tests/test_dashboard_my_scan_cache.py new file mode 100644 index 00000000..0e8875ad --- /dev/null +++ b/tests/test_dashboard_my_scan_cache.py @@ -0,0 +1,119 @@ +"""「我的」页全表扫收敛回归：三处 5s 单飞缓存 + atom_adoption(skill) 索引。 + +a4 线上事故：我的页端点每请求各自全表扫 trajectories / atom_adoption， +首屏叠出十余次全扫，SQLite gate 串行化拖慢全部面板接口。修复后同一波 +请求只扫一次（TTL 内共享只读结果），adoption 按 skill 过滤走索引。 +""" +from __future__ import annotations + +from pathlib import Path + +from xskill import events +from xskill.dashboard import console, explore +from xskill.pipeline.registry import get_connection + + +def _seed_db(db: Path) -> None: + conn = get_connection(db) + conn.execute( + "INSERT INTO watch_dirs(id,path,label,ecosystem)" + " VALUES(1,'/cc','alice','claude_code')", + ) + conn.execute( + "INSERT INTO trajectories(watch_dir_id,filename,status,user_key)" + " VALUES(1,'traj_1.md','done','alice')", + ) + conn.execute( + "INSERT INTO trajectories(watch_dir_id,filename,status,user_key)" + " VALUES(1,'traj_2.md','done','bob')", + ) + conn.execute( + "INSERT INTO atom_adoption(atom_id,skill,weightscore,ts)" + " VALUES('atom_traj_1_0001','demo',3,'2026-07-30 00:00:00')", + ) + conn.commit() + conn.close() + + +def test_traj_user_map_cached_within_ttl(tmp_path): + db = tmp_path / "r.db" + _seed_db(db) + events._traj_user_cache.clear() + m1 = events._traj_user_map(db) + assert m1["traj_1"] == "alice" + + conn = get_connection(db) + conn.execute( + "INSERT INTO trajectories(watch_dir_id,filename,status,user_key)" + " VALUES(1,'traj_3.md','done','carol')", + ) + conn.commit() + conn.close() + + m2 = events._traj_user_map(db) + assert m2 is m1 # TTL 内同一份共享对象，不重扫 + assert "traj_3" not in m2 # 短窗内允许略陈旧 + + events._traj_user_cache.clear() + m3 = events._traj_user_map(db) + assert m3["traj_3"] == "carol" # 清缓存后立刻可见 + + +def test_visible_traj_info_cached_and_correct(tmp_path): + db = tmp_path / "r.db" + _seed_db(db) + explore._visible_traj_cache.clear() + t1 = explore._visible_traj_info(db) + t2 = explore._visible_traj_info(db) + assert t1 is t2 + assert t1["traj_1"]["user"] == "alice" + assert t1["traj_2"]["user"] == "alice" # 同 watch_dir 的 label + + +def test_adoption_rows_cached_and_fresh_after_clear(tmp_path): + db = tmp_path / "r.db" + _seed_db(db) + console._adoption_rows_cache.clear() + r1 = console._adoption_rows(db) + r2 = console._adoption_rows(db) + assert r1 is r2 + assert r1[0]["skill"] == "demo" + + conn = get_connection(db) + conn.execute( + "INSERT INTO atom_adoption(atom_id,skill,weightscore,ts)" + " VALUES('atom_traj_2_0001','demo2',1,'2026-07-30 00:00:01')", + ) + conn.commit() + conn.close() + assert len(console._adoption_rows(db)) == 1 # 仍走缓存 + console._adoption_rows_cache.clear() + assert len(console._adoption_rows(db)) == 2 + + +def test_atom_adoption_skill_index_exists(tmp_path): + db = tmp_path / "r.db" + _seed_db(db) + conn = get_connection(db) + names = { + r["name"] for r in conn.execute( + "SELECT name FROM sqlite_master" + " WHERE type='index' AND tbl_name='atom_adoption'", + ) + } + conn.close() + assert "idx_atom_adopt_skill" in names + + +def test_skill_lineage_correct_with_shared_traj_cache(tmp_path): + db = tmp_path / "r.db" + _seed_db(db) + explore._visible_traj_cache.clear() + skill_dir = tmp_path / "skills" + (skill_dir / "demo").mkdir(parents=True) + + lin1 = explore.skill_lineage(skill_dir, "demo", db_path=db) + lin2 = explore.skill_lineage(skill_dir, "demo", db_path=db) + assert lin1["by_user"] == [{"user": "alice", "atoms": 1}] + assert lin1["atoms"][0]["atom_id"] == "atom_traj_1_0001" + assert lin2["by_user"] == lin1["by_user"] # 缓存路径结果一致