fix(api): /api/data/status 同步期忙时读旧缓存 — 不再 30s 超时

同步进行中 (active job pending/running 或指标预热未完成) 时, /status 的 12 张表聚合 + 存储扫描仍要抢数据槽并全量扫描, 与前端 2s 轮询叠加导致「请求超时(30s)」。 busy 模式改为直接回最近一次成功聚合的过期缓存 (无缓存则该表置 null), 不重算; 空闲路径 TTL 语义不变; 僵死 active_id 由 job 终态判定兜底, 不会永久卡 busy。验证: test_data_status_busy 7 条 (忙时旧缓存/无缓存置空/空闲 TTL 不变/触发条件/ 端点接线); 全量 pytest 通过。 Signed-off-by: shy3130 <shy3130@qq.com>
2026-11-02 10:26:13 +00:00 · 2026-10-28 12:46:45 +08:00
parent bc5d213bea
commit 977248ceb2
2 changed files with 171 additions and 18 deletions
@@ -13,6 +13,7 @@ from fastapi import APIRouter, Request

 from app.enriched_generation import EnrichedPublication
 from app.indicators.pipeline import ENRICHED_COLUMNS
+from app.services.pipeline_jobs import job_store

 logger = logging.getLogger(__name__)

@@ -31,6 +32,7 @@ _LARGE_TABLES = {"minute"}

 _storage_cache: dict[str, Any] | None = None
 _storage_cache_ts: float = 0.0
+_storage_cache_stale: dict[str, Any] | None = None
 _storage_lock = threading.Lock()

 _table_cache: dict[str, dict | None] = {
@@ -48,6 +50,9 @@ _table_cache: dict[str, dict | None] = {
    "financials": None,
 }
 _table_cache_ts: dict[str, float] = {k: 0.0 for k in _table_cache}
+# 最后一次非空统计 (失效事件不清空): 同步任务期间 /status 的兜底展示 —
+# 宁可数字旧一点, 也不能让重聚合在 DuckDB 锁上排队到前端超时
+_table_cache_stale: dict[str, dict | None] = {}
 _table_cache_lock = threading.Lock()

 _last_finished_cache: dict[str, str | None] | None = None
@@ -79,11 +84,22 @@ def invalidate_storage_cache() -> None:
    invalidate_data_cache(None)


-def _get_table_stats(name: str, fetch: Callable[[], dict | None]) -> dict | None:
-    """走 TTL+事件 双重缓存。fetch 在锁外执行避免阻塞别的请求。"""
+def _get_table_stats(
+    name: str, fetch: Callable[[], dict | None], *, busy: bool = False
+) -> dict | None:
+    """走 TTL+事件 双重缓存。fetch 在锁外执行避免阻塞别的请求。
+
+    busy (同步任务运行中 / 指标预热未完成) 时完全不重算, 直接回最后一次非空值
+    (可能为 None): 这些聚合夹着 DuckDB 查询, 与同步管道共用 repository 的
+    DuckDB 单连接全局锁 — 管道持锁跑长查询时重算会在锁上排队, 2s 的 /status
+    轮询堆积后触发前端 30s 超时 (2026-10-28 用户反馈)。任务结束的失效事件
+    会清 TTL 缓存, 结束后第一次调用即恢复正常重算。
+    """
    ttl = _TABLE_TTL_LARGE if name in _LARGE_TABLES else _TABLE_TTL
    now = time.time()
    with _table_cache_lock:
+        if busy:
+            return _table_cache_stale.get(name)
        cached = _table_cache.get(name)
        cached_ts = _table_cache_ts.get(name, 0.0)
        if cached is not None and (now - cached_ts) < ttl:
@@ -94,6 +110,8 @@ def _get_table_stats(name: str, fetch: Callable[[], dict | None]) -> dict | None
    with _table_cache_lock:
        _table_cache[name] = fresh
        _table_cache_ts[name] = now
+        if fresh is not None:
+            _table_cache_stale[name] = fresh
    return fresh


@@ -544,17 +562,22 @@ def _next_cron_run(scheduler, job_id: str) -> str | None:
    return None


-def _get_storage(data_dir: Path) -> dict:
-    """返回缓存的 storage 统计；走独立 TTL，stage 写完不触发重算。"""
-    global _storage_cache, _storage_cache_ts
+def _get_storage(data_dir: Path, *, busy: bool = False) -> dict:
+    """返回缓存的 storage 统计；走独立 TTL，stage 写完不触发重算。
+
+    busy 时同样不重算 (目录 du 在同步写盘期间会显著变慢), 回最后一次值。"""
+    global _storage_cache, _storage_cache_ts, _storage_cache_stale
    now = time.time()
    with _storage_lock:
+        if busy:
+            return _storage_cache_stale
        if _storage_cache is not None and (now - _storage_cache_ts) < _STORAGE_TTL:
            return _storage_cache
    fresh = _compute_storage(data_dir)
    with _storage_lock:
        _storage_cache = fresh
        _storage_cache_ts = now
+        _storage_cache_stale = fresh
    return fresh


@@ -580,28 +603,40 @@ def _last_finished(job_label: str) -> str | None:
    return cache.get(job_label)


+def _data_pipeline_busy(request: Request) -> bool:
+    """同步任务运行中或指标预热未完成 — /status 的重聚合应退化为纯缓存读。"""
+    if not getattr(request.app.state, "indicators_ready", True):
+        return True
+    active_id = job_store.active_id()
+    if not active_id:
+        return False
+    active = job_store.get(active_id) or {}
+    return active.get("status") in ("pending", "running")
+
+
@router.get("/status")
 def status(request: Request) -> dict:
    repo = request.app.state.repo
    scheduler = getattr(request.app.state, "scheduler", None)
    data_dir = repo.store.data_dir
+    busy = _data_pipeline_busy(request)

    return {
-        "daily":       _get_table_stats("daily",       lambda: _safe_aggregate_daily(repo)),
-        "enriched":    _get_table_stats("enriched",    lambda: _safe_aggregate_enriched(repo)),
-    "index_daily":       _get_table_stats("index_daily",       lambda: _safe_aggregate_index_daily(repo)),
-    "index_enriched":    _get_table_stats("index_enriched",    lambda: _safe_aggregate_index_enriched(repo)),
-    "index_instruments": _get_table_stats("index_instruments", lambda: _safe_aggregate_index_instruments(repo)),
-    "etf_daily":         _get_table_stats("etf_daily",         lambda: _safe_aggregate_etf_daily(repo)),
-    "etf_enriched":      _get_table_stats("etf_enriched",      lambda: _safe_aggregate_etf_enriched(repo)),
-    "etf_instruments":   _get_table_stats("etf_instruments",   lambda: _safe_aggregate_etf_instruments(repo)),
-    "minute":      _get_table_stats("minute",      lambda: _safe_aggregate_minute(repo)),
-        "adj_factor":  _get_table_stats("adj_factor",  lambda: _safe_aggregate_adj_factor(repo)),
-        "instruments": _get_table_stats("instruments", lambda: _safe_aggregate_instruments(repo)),
-        "financials":  _get_table_stats("financials",  lambda: _safe_aggregate_financials(repo)),
+        "daily":       _get_table_stats("daily",       lambda: _safe_aggregate_daily(repo), busy=busy),
+        "enriched":    _get_table_stats("enriched",    lambda: _safe_aggregate_enriched(repo), busy=busy),
+    "index_daily":       _get_table_stats("index_daily",       lambda: _safe_aggregate_index_daily(repo), busy=busy),
+    "index_enriched":    _get_table_stats("index_enriched",    lambda: _safe_aggregate_index_enriched(repo), busy=busy),
+    "index_instruments": _get_table_stats("index_instruments", lambda: _safe_aggregate_index_instruments(repo), busy=busy),
+    "etf_daily":         _get_table_stats("etf_daily",         lambda: _safe_aggregate_etf_daily(repo), busy=busy),
+    "etf_enriched":      _get_table_stats("etf_enriched",      lambda: _safe_aggregate_etf_enriched(repo), busy=busy),
+    "etf_instruments":   _get_table_stats("etf_instruments",   lambda: _safe_aggregate_etf_instruments(repo), busy=busy),
+    "minute":      _get_table_stats("minute",      lambda: _safe_aggregate_minute(repo), busy=busy),
+        "adj_factor":  _get_table_stats("adj_factor",  lambda: _safe_aggregate_adj_factor(repo), busy=busy),
+        "instruments": _get_table_stats("instruments", lambda: _safe_aggregate_instruments(repo), busy=busy),
+        "financials":  _get_table_stats("financials",  lambda: _safe_aggregate_financials(repo), busy=busy),

        # 文件层面信息(缓存)
-        "storage": _get_storage(data_dir),
+        "storage": _get_storage(data_dir, busy=busy),

        # 调度
        "next_instruments_run": _next_cron_run(scheduler, "pre_market_instruments"),
@@ -0,0 +1,118 @@
+"""数据状态接口 busy 降级 — 同步任务期间 /status 不再重算聚合。
+
+回归 (2026-10-28 用户反馈): 同步任务运行时前端把 /api/data/status 轮询加速到
+2s, 而聚合里的 DuckDB 查询与管道共用 repository 单连接全局锁 — 管道持锁跑长
+查询时请求在锁上排队堆积, 反复触发前端 30s 超时 (请求超时（30s）/api/data/status)。
+修复: busy (活跃任务/指标预热中) 时聚合退化为纯缓存读 (最后一次非空值), 不进锁。
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from app.api import data as data_api
+from app.services.pipeline_jobs import job_store
+
+
+@pytest.fixture()
+def clean_caches():
+    data_api.invalidate_data_cache(None)
+    data_api._table_cache_stale.clear()
+    data_api._storage_cache_stale = None
+    yield
+    data_api.invalidate_data_cache(None)
+    data_api._table_cache_stale.clear()
+    data_api._storage_cache_stale = None
+
+
+# ---------- _get_table_stats: busy 语义 ----------
+
+def test_busy_serves_stale_without_fetch(clean_caches):
+    calls = []
+
+    def fetch():
+        calls.append(1)
+        return {"rows": 42}
+
+    first = data_api._get_table_stats("daily", fetch)
+    assert first == {"rows": 42} and len(calls) == 1
+    # 失效事件 (stage 写完) 清 TTL 缓存, 但 stale 兜底保留
+    data_api.invalidate_data_cache("daily")
+    got = data_api._get_table_stats("daily", fetch, busy=True)
+    assert got == {"rows": 42}
+    assert len(calls) == 1  # busy 期间绝不重算 → 不进 DuckDB 锁
+
+
+def test_busy_without_stale_returns_none_and_skips_fetch(clean_caches):
+    calls = []
+
+    def fetch():
+        calls.append(1)
+        return {"rows": 1}
+
+    assert data_api._get_table_stats("minute", fetch, busy=True) is None
+    assert calls == []
+
+
+def test_not_busy_still_refetches_after_invalidation(clean_caches):
+    calls = []
+
+    def fetch():
+        calls.append(1)
+        return {"rows": len(calls)}
+
+    data_api._get_table_stats("daily", fetch)
+    data_api.invalidate_data_cache("daily")
+    got = data_api._get_table_stats("daily", fetch)
+    assert got == {"rows": 2} and len(calls) == 2  # 原有 TTL+事件语义不变
+
+
+def test_get_storage_busy_serves_stale(clean_caches, tmp_path):
+    data_api._get_storage(tmp_path)          # 首次计算并缓存
+    data_api.invalidate_data_cache(None)     # 全清 (storage 一并清空)
+    got = data_api._get_storage(tmp_path, busy=True)
+    assert got is not None                   # stale 兜底仍在
+
+
+# ---------- busy 判定 ----------
+
+def _req(indicators_ready=True):
+    return SimpleNamespace(
+        app=SimpleNamespace(state=SimpleNamespace(indicators_ready=indicators_ready))
+    )
+
+
+def test_busy_when_indicators_warming(clean_caches):
+    assert data_api._data_pipeline_busy(_req(indicators_ready=False)) is True
+
+
+def test_busy_follows_active_job(monkeypatch, clean_caches):
+    monkeypatch.setattr(job_store, "active_id", lambda: None)
+    assert data_api._data_pipeline_busy(_req()) is False
+    monkeypatch.setattr(job_store, "active_id", lambda: "abc123")
+    monkeypatch.setattr(job_store, "get", lambda jid: {"status": "running"})
+    assert data_api._data_pipeline_busy(_req()) is True
+    # 终态任务不算 busy (防僵尸 active_id 把 /status 永久钉死在缓存)
+    monkeypatch.setattr(job_store, "get", lambda jid: {"status": "succeeded"})
+    assert data_api._data_pipeline_busy(_req()) is False
+
+
+# ---------- status 端点: busy 接线 ----------
+
+def test_status_endpoint_busy_skips_aggregates(monkeypatch, clean_caches, tmp_path):
+    """indicators_ready=False → busy: 聚合全部跳过 (不触碰 repo), 轻字段照常返回。"""
+    repo = SimpleNamespace(store=SimpleNamespace(data_dir=Path(tmp_path)))
+    req = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(
+        repo=repo, scheduler=None, indicators_ready=False,
+    )))
+    # 任何聚合尝试都算失败 — busy 下不应被调用
+    def _boom(*_a, **_k):
+        raise AssertionError("busy 下不应重算聚合")
+    for fn in ("daily", "enriched", "index_daily", "minute", "financials"):
+        monkeypatch.setattr(data_api, f"_safe_aggregate_{fn}", _boom)
+    res = data_api.status(req)
+    assert res["indicators_ready"] is False
+    assert res["checked_at"]
+    assert res["daily"] is None and res["minute"] is None