perf(panelcache): single-flight 消除并发同 key 缓存踩踏

PanelCache 把 compute_fn 放锁外避免不同 key 串行化, 但代价是同 key
冷启动踩踏: 优化器/批量回测的 max_workers 个线程几乎同时 miss 空缓存,
全部在锁外并行调 compute_fn, 同一面板被并行加载 N 份 (N=并发线程数),
而非承诺的"只加载一次"。全市场 parquet 冷扫 ×N 的 IO/内存峰值。

改为 single-flight: 同 key 只推举一个 leader 执行 compute, 其余线程
登记为跟随者等 leader 结果复用。要点:
- compute 仍在锁外 → 不同 key 依旧并发, 保留原设计优点, 只消除同 key 重复。
- 失败不缓存: leader 抛错时摘除 inflight 让后续重试, 异常透传所有跟随者。

新增 2 个并发回归测试: 8 线程同 key 踩踏断言 compute 仅 1 次;
失败路径断言异常透传全部等待者且不缓存失败可重试。

backend/app/backtest/engine.py

@@ -121,6 +121,17 @@ class _CacheEntry:
         self.ts = ts
 
 
+class _InFlight:
+    """同 key 正在计算的占位: leader 算完通过 done 唤醒所有跟随者复用结果。"""
+
+    __slots__ = ("done", "df", "error")
+
+    def __init__(self) -> None:
+        self.done = threading.Event()
+        self.df: pl.DataFrame | None = None
+        self.error: BaseException | None = None
+
+
 class PanelCache:
     """LRU + TTL 数据面板缓存。"""
 
@@ -132,6 +143,9 @@ class PanelCache:
         # 无锁的 move_to_end/del/popitem check-then-act 会抛 "OrderedDict mutated"。
         # 用实例锁守护所有 OrderedDict 变更; compute_fn (重扫盘) 放锁外避免串行化。
         self._lock = threading.Lock()
+        # single-flight: 同 key 只让一个线程 compute, 其余等其结果复用。
+        # 否则优化器等场景下 max_workers 个线程冷启动同时 miss, 会并行加载 N 份同一面板。
+        self._inflight: dict[str, _InFlight] = {}
 
     def get_or_compute(
         self,
@@ -146,19 +160,43 @@ class PanelCache:
         now = time.monotonic()
 
         with self._lock:
-            if key in self._cache:
-                entry = self._cache[key]
+            entry = self._cache.get(key)
+            if entry is not None:
                 if now - entry.ts < self._ttl:
                     self._cache.move_to_end(key)
                     return entry.df
-                del self._cache[key]
+                del self._cache[key]  # 过期, 丢弃后重算
+            # single-flight: 同 key 若已有线程在算, 登记为跟随者; 否则本线程当 leader。
+            flight = self._inflight.get(key)
+            leader = flight is None
+            if leader:
+                flight = _InFlight()
+                self._inflight[key] = flight
 
-        # 计算在锁外 (可能重扫 parquet, 耗时); 并发相同 key 至多重复算一次, 不会崩
-        df = compute_fn(symbols, start, end, columns, asset_type)
+        if not leader:
+            # 跟随者: 等 leader 算完直接复用, 不重复 compute (消除缓存踩踏)。
+            flight.done.wait()
+            if flight.error is not None:
+                raise flight.error
+            return flight.df
+
+        # leader: compute 放锁外 (不同 key 仍可并发, 保留原设计优点)。
+        try:
+            df = compute_fn(symbols, start, end, columns, asset_type)
+        except BaseException as e:
+            # 失败不缓存: 摘除 inflight 让后续线程重试, 并把异常透传给已在等的跟随者。
+            with self._lock:
+                self._inflight.pop(key, None)
+            flight.error = e
+            flight.done.set()
+            raise
         with self._lock:
             self._cache[key] = _CacheEntry(df=df, ts=now)
             if len(self._cache) > self._max_size:
                 self._cache.popitem(last=False)
+            self._inflight.pop(key, None)
+        flight.df = df
+        flight.done.set()
         return df
 
     def invalidate(self) -> None:

backend/tests/test_backtest_etf.py

@@ -1,3 +1,4 @@
+import time
 import types
 from datetime import date
 
@@ -77,6 +78,78 @@ def test_engine_stock_uses_daily_enriched_dir(monkeypatch, tmp_path):
     assert "kline_daily_enriched" in captured["path"]
 
 
+def test_panel_cache_single_flight_computes_once():
+    """N 个线程并发同 key 冷启动: compute_fn 只应被调用一次, 其余复用结果 (无缓存踩踏)。"""
+    import threading
+
+    cache = PanelCache()
+    calls = []
+    barrier = threading.Barrier(8)
+    df = pl.DataFrame({"symbol": ["510300"]})
+
+    def slow_compute(symbols, start, end, columns, asset_type):
+        calls.append(1)
+        time.sleep(0.05)  # 拉长窗口, 逼出并发 miss
+        return df
+
+    args = (["510300"], date(2026, 1, 1), date(2026, 1, 2), None)
+    results = []
+    rlock = threading.Lock()
+
+    def worker():
+        barrier.wait()  # 所有线程同时起跑, 制造冷启动踩踏
+        r = cache.get_or_compute(*args, slow_compute, "stock")
+        with rlock:
+            results.append(r)
+
+    threads = [threading.Thread(target=worker) for _ in range(8)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert sum(calls) == 1, f"面板被重复加载 {sum(calls)} 次, single-flight 失效"
+    assert len(results) == 8 and all(r is df for r in results)
+
+
+def test_panel_cache_single_flight_error_propagates_and_retries():
+    """leader compute 抛错: 不缓存失败, 异常透传给所有等待者, 后续调用可重试成功。"""
+    import threading
+
+    cache = PanelCache()
+    barrier = threading.Barrier(4)
+    boom = RuntimeError("scan failed")
+
+    def failing_compute(symbols, start, end, columns, asset_type):
+        time.sleep(0.03)
+        raise boom
+
+    args = (["510300"], date(2026, 1, 1), date(2026, 1, 2), None)
+    errors = []
+    elock = threading.Lock()
+
+    def worker():
+        barrier.wait()
+        try:
+            cache.get_or_compute(*args, failing_compute, "stock")
+        except RuntimeError as e:
+            with elock:
+                errors.append(e)
+
+    threads = [threading.Thread(target=worker) for _ in range(4)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert len(errors) == 4 and all(e is boom for e in errors), "失败未透传给全部跟随者"
+
+    # 失败未被缓存 —— 重试应重新 compute 并成功
+    df = pl.DataFrame({"symbol": ["510300"]})
+    got = cache.get_or_compute(*args, lambda *a: df, "stock")
+    assert got is df
+
+
 def test_job_key_includes_asset_type_and_is_consistent():
     """stream 与 cancel 必须用同一 job_key: asset_type 进 key 且相同入参产出相同 key。"""
     from app.api.backtest import _make_job_key