perf(backend): DuckDB 读并发 — cursor 独立连接+信号量限流, DDL 收口 refresh_view

单连接全局互斥下, 同步期任一长查询(全视图扫描)把所有交互读堵到前端 30s 超时,
不同端点轮流失害 (/status 事故即此形态)。cursor() 本就是同一内存库实例的独立连接
(MVCC 快照隔离; duckdb 1.5.3 实测 6 读线程+2 换视图线程共 1800 次零异常),
execute_all/execute_one 改为信号量限并发 (TICK_DUCKDB_READ_CONCURRENCY,
默认 3, 1=完全串行旧行为), 根连接运行期仅作连接工厂。

并发 CREATE OR REPLACE VIEW 同名视图会抛 Catalog write-write conflict (实测),
运行期视图重建一律收口 repo.refresh_view (_ddl_lock 串行, 与读不互斥), 同时消除
kline_sync/daily_pipeline/extend_history/api-kline 共 7 处绕锁直连 db.execute
的线程安全隐患 — 两同步路径同时刷同一视图的静默失败一并修复。

验证: 新增 test_duckdb_read_concurrency (并发重叠/串行回退/视图重建/DDL 零冲突/
坏名字拒绝/env 钳制) 6 条; 全量 pytest 3479 通过; ruff 码级零新增(消 3 条存量);
基准 376ms 长查询持锁时短读 271ms → 0.1ms。

Signed-off-by: shy3130 <shy3130@qq.com>

README.md

@@ -561,6 +561,7 @@ POLARS_MAX_THREADS=2        # 主要收益项: 用约 40% 的构建耗时换 500
 | :-- | :-- | :-- |
 | `TICK_WORKER_RSS_LIMIT_BYTES` | `6442450944` (6GiB) | 回测/优化独立 worker 的 RSS 上限, 超限终止并返回明确错误(而非拖垮整机); `0` = 不限制 |
 | `TICK_HEAVY_MIN_AVAILABLE_BYTES` | `1073741824` (1GiB) | 系统可用内存低于该值时, 新的重任务(回测/因子批量/挖掘)挂起等待内存恢复; `0` = 不检查 |
+| `TICK_DUCKDB_READ_CONCURRENCY` | `3` | DuckDB 并发读上限。读查询走独立连接(MVCC), 同步期的长扫描不再阻塞交互读; 并发数越高峰值内存越大, `1` = 完全串行(旧行为) |
 
 ### 跑起来后的第一次使用
 

backend/app/api/kline.py

@@ -1841,13 +1841,7 @@ async def rebuild_enriched(request: Request):
                 for view_name, glob in [
                     ("kline_enriched", f"{d}/kline_daily_enriched/**/*.parquet"),
                 ]:
-                    try:
-                        repo.db.execute(
-                            f"CREATE OR REPLACE VIEW {view_name} AS "
-                            f"SELECT * FROM read_parquet('{glob}', union_by_name=true)"
-                        )
-                    except Exception:
-                        pass
+                    repo.refresh_view(view_name, glob)
 
                 progress("rebuild_enriched", 100, f"完成,覆盖 {enriched_days} 天")
                 job_store.succeed(job_id, {

backend/app/jobs/daily_pipeline.py

@@ -972,13 +972,7 @@ def _refresh_single_view(repo: KlineRepository, name: str) -> None:
     path = paths.get(name)
     if not path:
         return
-    try:
-        repo.db.execute(
-            f"CREATE OR REPLACE VIEW {name} AS "
-            f"SELECT * FROM read_parquet('{path}', union_by_name=true)"
-        )
-    except Exception as e:  # noqa: BLE001
-        logger.warning("refresh view %s failed: %s", name, e)
+    repo.refresh_view(name, path)
 
 
 def _resolve_minute_symbols(capset: CapabilitySet, repo=None) -> list[str]:
@@ -989,13 +983,7 @@ def _resolve_minute_symbols(capset: CapabilitySet, repo=None) -> list[str]:
 def _refresh_instruments_view(repo: KlineRepository) -> None:
     """单独刷新 instruments 视图。"""
     d = repo.store.data_dir.as_posix()
-    try:
-        repo.db.execute(
-            f"CREATE OR REPLACE VIEW instruments AS "
-            f"SELECT * FROM read_parquet('{d}/instruments/**/*.parquet', union_by_name=true)"
-        )
-    except Exception as e:  # noqa: BLE001
-        logger.warning("refresh instruments view failed: %s", e)
+    repo.refresh_view("instruments", f"{d}/instruments/**/*.parquet")
 
 
 def _push_phase_change_alert(data_dir) -> None:

backend/app/services/extend_history.py

@@ -76,13 +76,7 @@ def _refresh_single_view(repo: KlineRepository, name: str) -> None:
     path = paths.get(name)
     if not path:
         return
-    try:
-        repo.db.execute(
-            f"CREATE OR REPLACE VIEW {name} AS "
-            f"SELECT * FROM read_parquet('{path}', union_by_name=true)"
-        )
-    except Exception as e:
-        logger.warning("refresh view %s failed: %s", name, e)
+    repo.refresh_view(name, path)
 
 
 def compute_offset(value: int, unit: str) -> timedelta:

backend/app/services/kline_sync.py

@@ -373,14 +373,8 @@ def sync_and_persist_daily_batch(
             if "symbol" in df.columns:
                 seen.update(df["symbol"].cast(pl.Utf8).unique().to_list())
             repo.append_daily(df)
-            try:
-                d = repo.store.data_dir.as_posix()
-                repo.db.execute(
-                    f"""CREATE OR REPLACE VIEW kline_daily AS
-                        SELECT * FROM read_parquet('{d}/kline_daily/**/*.parquet', union_by_name=true)"""
-                )
-            except Exception as e:  # noqa: BLE001
-                logger.warning("refresh view failed: %s", e)
+            d = repo.store.data_dir.as_posix()
+            repo.refresh_view("kline_daily", f"{d}/kline_daily/**/*.parquet")
             return _finalize(df.height)
         # 自定义源未配置 daily → 回退 TickFlow
 
@@ -408,14 +402,8 @@ def sync_and_persist_daily_batch(
 
     repo.append_daily(df)
 
-    try:
-        d = repo.store.data_dir.as_posix()
-        repo.db.execute(
-            f"""CREATE OR REPLACE VIEW kline_daily AS
-                SELECT * FROM read_parquet('{d}/kline_daily/**/*.parquet', union_by_name=true)"""
-        )
-    except Exception as e:  # noqa: BLE001
-        logger.warning("refresh view failed: %s", e)
+    d = repo.store.data_dir.as_posix()
+    repo.refresh_view("kline_daily", f"{d}/kline_daily/**/*.parquet")
 
     return _finalize(df.height)
 
@@ -1817,14 +1805,8 @@ def sync_and_persist_minute(
     written = written_box[0]
 
     # 刷新视图
-    try:
-        d = repo.store.data_dir.as_posix()
-        repo.db.execute(
-            f"""CREATE OR REPLACE VIEW kline_minute AS
-                SELECT * FROM read_parquet('{d}/kline_minute/**/*.parquet', union_by_name=true)"""
-        )
-    except Exception as e:  # noqa: BLE001
-        logger.warning("refresh kline_minute view failed: %s", e)
+    d = repo.store.data_dir.as_posix()
+    repo.refresh_view("kline_minute", f"{d}/kline_minute/**/*.parquet")
 
     logger.info("minute K synced: %d rows (%d symbols)", written, len(symbols))
     return written

backend/app/tickflow/repository.py

@@ -14,6 +14,7 @@ from __future__ import annotations
 
 import json
 import logging
+import os
 import sys
 import threading
 import time
@@ -284,12 +285,14 @@ class DataStore:
     def _has_parquet(self, subdir: str) -> bool:
         return any((self.data_dir / subdir).rglob("*.parquet"))
 
-    def _register_unified_views(self) -> None:
+    def _register_unified_views(self, con=None) -> None:
         """Register optional all-asset views when their backing parquet exists.
 
         Physical storage remains split for performance and compatibility. These
-        views are convenience read models for new APIs/features.
+        views are convenience read models for new APIs/features. 运行期调用方
+        传独立 cursor (rebuild_views/refresh_index_views), 初始化走根连接。
         """
+        db = con if con is not None else self.db
         daily_parts: list[str] = []
         enriched_parts: list[str] = []
         minute_parts: list[str] = []
@@ -361,18 +364,40 @@ class DataStore:
             if not parts:
                 continue
             try:
-                self.db.execute(f"CREATE OR REPLACE VIEW {name} AS " + " UNION ALL BY NAME ".join(parts))
+                db.execute(f"CREATE OR REPLACE VIEW {name} AS " + " UNION ALL BY NAME ".join(parts))
             except Exception as e:  # noqa: BLE001
                 logger.debug("unified view %s skipped: %s", name, e)
 
 
+def _read_concurrency_env() -> int:
+    """DuckDB 并发读上限 (默认 3)。
+
+    并发扫描各自持有缓冲, 上限过高会推高峰值内存 (#516 同因); 1 = 恢复
+    完全串行的旧行为。
+    """
+    try:
+        value = int(os.environ.get("TICK_DUCKDB_READ_CONCURRENCY", "3"))
+    except ValueError:
+        return 3
+    return min(max(value, 1), 16)
+
+
 class KlineRepository:
     """日 K / 分钟 K 的读写入口。"""
 
     def __init__(self, store: DataStore) -> None:
         self.store = store
         self.db = store.db
-        self._lock = threading.Lock()
+        # DuckDB 读并发: cursor() 即同一内存库实例的独立连接 (MVCC 快照隔离,
+        # 1.5.3 实测并发读+换视图零异常)。用信号量限制同时进行的扫描数, 取代
+        # 原全局互斥 —— 同步期的长查询 (全视图扫描) 不再把交互读堵到超时。
+        self._read_limiter = threading.Semaphore(_read_concurrency_env())
+        # DDL (视图重建) 互斥: 并发 CREATE OR REPLACE VIEW 同名视图会抛
+        # Catalog write-write conflict (1.5.3 实测), 必须串行; 与读不互斥。
+        self._ddl_lock = threading.RLock()
+        # 根连接运行期只作连接工厂, 不直接执行 SQL (跨线程共用一个
+        # Connection 对象没有契约保证, cursor 各自独立)
+        self._cursor_lock = threading.Lock()
         # 序列化 parquet 分区的读-改-写: 实时轮询线程、手动 refresh、盘后管道
         # 可能并发 merge/flush 同一分区文件, 无锁会互相覆盖丢数据
         self._write_lock = threading.Lock()
@@ -424,24 +449,58 @@ class KlineRepository:
         self._index_inst_glob = str(store.data_dir / "instruments_index" / "**" / "*.parquet")
         self._etf_inst_glob = str(store.data_dir / "instruments_etf" / "**" / "*.parquet")
 
+    def _new_cursor(self):
+        """从根连接复制一个独立连接 (共享同一库实例, 创建开销极低)。"""
+        with self._cursor_lock:
+            return self.db.cursor()
+
     def execute_all(self, sql: str, params: list | None = None) -> list[tuple]:
-        """线程安全的 SELECT → fetchall。DuckDB 单 connection 非线程安全，所有读路径须走此方法。"""
-        with self._lock:
-            cursor = self.db.cursor()
+        """并发安全的 SELECT → fetchall。
+
+        cursor 是同一库实例的独立连接: 读与读并发、读与视图重建并发 (MVCC,
+        运行中的查询继续用旧定义) 均安全; _read_limiter 只限制同时扫描数。
+        """
+        with self._read_limiter:
+            cursor = self._new_cursor()
             try:
                 return cursor.execute(sql, params or []).fetchall()
             finally:
                 cursor.close()
 
     def execute_one(self, sql: str, params: list | None = None) -> tuple | None:
-        """线程安全的 SELECT → fetchone。"""
-        with self._lock:
-            cursor = self.db.cursor()
+        """并发安全的 SELECT → fetchone。"""
+        with self._read_limiter:
+            cursor = self._new_cursor()
             try:
                 return cursor.execute(sql, params or []).fetchone()
             finally:
                 cursor.close()
 
+    def refresh_view(self, name: str, glob: str) -> bool:
+        """重建指向 parquet glob 的视图 (视图懒扫描, 创建本身 O(1))。
+
+        运行期所有视图重建必须走此方法, 不得直连 db.execute:
+        - 并发重建同名视图抛 Catalog write-write conflict (1.5.3 实测),
+          _ddl_lock 内互斥, 也修复了旧代码两处同步路径同时刷视图的静默失败;
+        - 与读不互斥 (MVCC), 换视图期间查询不中断。
+        """
+        if not name.isidentifier():
+            logger.warning("refresh view rejected, invalid name: %r", name)
+            return False
+        with self._ddl_lock:
+            cursor = self._new_cursor()
+            try:
+                cursor.execute(
+                    f"CREATE OR REPLACE VIEW {name} AS "
+                    f"SELECT * FROM read_parquet('{glob}', union_by_name=true)"
+                )
+                return True
+            except Exception as e:  # noqa: BLE001 — 空目录/权限问题降级为警告, 与既有刷新行为一致
+                logger.warning("refresh view %s failed: %s", name, e)
+                return False
+            finally:
+                cursor.close()
+
     # ================================================================
     # Polars 缓存管理
     # ================================================================
@@ -2126,12 +2185,11 @@ class KlineRepository:
         from datetime import timedelta
         try:
             cutoff = reference_date - timedelta(days=min_gap_days)
-            with self._lock:
-                rows = self.db.execute(
-                    "SELECT symbol, max(date) AS mx FROM kline_daily "
-                    "GROUP BY symbol HAVING max(date) < ? ORDER BY mx",
-                    [cutoff],
-                ).fetchall()
+            rows = self.execute_all(
+                "SELECT symbol, max(date) AS mx FROM kline_daily "
+                "GROUP BY symbol HAVING max(date) < ? ORDER BY mx",
+                [cutoff],
+            )
             return [r[0] for r in rows if r and r[0]]
         except Exception:
             return []
@@ -2243,28 +2301,21 @@ class KlineRepository:
     def refresh_index_views(self) -> None:
         """刷新指数相关 DuckDB 视图。"""
         d = self.store.data_dir.as_posix()
-        statements = [
-            f"""CREATE OR REPLACE VIEW kline_index_daily AS
-                SELECT * FROM read_parquet('{d}/kline_index_daily/**/*.parquet', union_by_name=true)""",
-            f"""CREATE OR REPLACE VIEW kline_index_enriched AS
-                SELECT * FROM read_parquet('{d}/kline_index_enriched/**/*.parquet', union_by_name=true)""",
-            f"""CREATE OR REPLACE VIEW kline_etf_daily AS
-                SELECT * FROM read_parquet('{d}/kline_etf_daily/**/*.parquet', union_by_name=true)""",
-            f"""CREATE OR REPLACE VIEW kline_etf_enriched AS
-                SELECT * FROM read_parquet('{d}/kline_etf_enriched/**/*.parquet', union_by_name=true)""",
-            f"""CREATE OR REPLACE VIEW instruments_index AS
-                SELECT * FROM read_parquet('{d}/instruments_index/**/*.parquet', union_by_name=true)""",
-            f"""CREATE OR REPLACE VIEW instruments_etf AS
-                SELECT * FROM read_parquet('{d}/instruments_etf/**/*.parquet', union_by_name=true)""",
-        ]
-        for sql in statements:
+        for name, path in {
+            "kline_index_daily": f"{d}/kline_index_daily/**/*.parquet",
+            "kline_index_enriched": f"{d}/kline_index_enriched/**/*.parquet",
+            "kline_etf_daily": f"{d}/kline_etf_daily/**/*.parquet",
+            "kline_etf_enriched": f"{d}/kline_etf_enriched/**/*.parquet",
+            "instruments_index": f"{d}/instruments_index/**/*.parquet",
+            "instruments_etf": f"{d}/instruments_etf/**/*.parquet",
+        }.items():
+            self.refresh_view(name, path)
+        with self._ddl_lock:
+            cursor = self._new_cursor()
             try:
-                with self._lock:
-                    self.db.execute(sql)
-            except Exception as e:  # noqa: BLE001
-                logger.debug("index/etf view refresh skipped: %s", e)
-        with self._lock:
-            self.store._register_unified_views()
+                self.store._register_unified_views(cursor)
+            finally:
+                cursor.close()
 
     def rebuild_views(self) -> None:
         """重建全部 13 张 parquet 视图并重挂 unified 视图 —— 唯一权威实现。
@@ -2290,16 +2341,13 @@ class KlineRepository:
             "instruments_etf": f"{d}/instruments_etf/**/*.parquet",
         }
         for name, path in views.items():
+            self.refresh_view(name, path)
+        with self._ddl_lock:
+            cursor = self._new_cursor()
             try:
-                with self._lock:
-                    self.db.execute(
-                        f"CREATE OR REPLACE VIEW {name} AS "
-                        f"SELECT * FROM read_parquet('{path}', union_by_name=true)"
-                    )
-            except Exception as e:  # noqa: BLE001
-                logger.warning("rebuild view %s failed: %s", name, e)
-        with self._lock:
-            self.store._register_unified_views()
+                self.store._register_unified_views(cursor)
+            finally:
+                cursor.close()
 
     @staticmethod
     def _atomic_write_parquet(df: pl.DataFrame, out: Path) -> None:

backend/tests/test_duckdb_read_concurrency.py

@@ -0,0 +1,164 @@
+"""DuckDB 读并发与视图重建串行 — repository 连接模型的回归锁。
+
+背景: 原实现单连接全局互斥, 同步期任一长查询(全视图扫描)把所有交互读
+堵在门外直到前端 30s 超时(/status 事故)。改为:
+- execute_all/execute_one 走独立 cursor(=同一内存库实例的独立连接,
+  MVCC 快照隔离), 用信号量限并发数而非全局互斥;
+- 运行期视图重建一律走 refresh_view(_ddl_lock 串行)。
+
+duckdb 1.5.3 实测依据: 并发读 + 换视图零异常(两种视图版本都被观察到);
+并发 CREATE OR REPLACE VIEW 同名视图抛 Catalog write-write conflict,
+因此 DDL 必须串行 —— 旧代码两处同步路径同时刷视图会静默失败。
+"""
+from __future__ import annotations
+
+import threading
+import time
+from pathlib import Path
+
+import polars as pl
+
+from app.tickflow.repository import (
+    DataStore,
+    KlineRepository,
+    _read_concurrency_env,
+)
+
+
+def _write_partition(root: Path, name: str, rows: int) -> Path:
+    d = root / name
+    d.mkdir(parents=True, exist_ok=True)
+    (d / "part.parquet").unlink(missing_ok=True)
+    pl.DataFrame({"x": list(range(rows))}).write_parquet(d / "part.parquet")
+    return d
+
+
+class _SlowCursor:
+    """记录并发数并模拟耗时查询的假 cursor (避免依赖真实查询时长, 不抖动)。"""
+
+    def __init__(self, tracker: dict) -> None:
+        self._tracker = tracker
+
+    def execute(self, sql: str, params: list | None = None) -> _SlowCursor:
+        with self._tracker["lock"]:
+            self._tracker["cur"] += 1
+            self._tracker["max"] = max(self._tracker["max"], self._tracker["cur"])
+        time.sleep(0.12)
+        with self._tracker["lock"]:
+            self._tracker["cur"] -= 1
+        return self
+
+    def fetchall(self) -> list[tuple]:
+        return [(1,)]
+
+    def fetchone(self) -> tuple:
+        return (1,)
+
+    def close(self) -> None:
+        pass
+
+
+def _read_overlap(
+    tmp_path: Path,
+    monkeypatch,
+    env_value: str,
+    n_threads: int,
+) -> int:
+    monkeypatch.setenv("TICK_DUCKDB_READ_CONCURRENCY", env_value)
+    repo = KlineRepository(DataStore(tmp_path))
+    tracker: dict = {"lock": threading.Lock(), "cur": 0, "max": 0}
+    monkeypatch.setattr(repo, "_new_cursor", lambda: _SlowCursor(tracker))
+
+    threads = [
+        threading.Thread(target=repo.execute_all, args=("SELECT 1",))
+        for _ in range(n_threads)
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+    return tracker["max"]
+
+
+def test_reads_overlap_up_to_limiter(tmp_path: Path, monkeypatch) -> None:
+    """并发读 = 信号量容量 (默认 3), 而非旧的全局串行 (1)。"""
+    assert _read_overlap(tmp_path, monkeypatch, "3", 5) == 3
+
+
+def test_reads_serialize_when_concurrency_one(tmp_path: Path, monkeypatch) -> None:
+    """TICK_DUCKDB_READ_CONCURRENCY=1 恢复完全串行的旧行为。"""
+    assert _read_overlap(tmp_path, monkeypatch, "1", 4) == 1
+
+
+def test_read_concurrency_env_clamped(monkeypatch) -> None:
+    monkeypatch.setenv("TICK_DUCKDB_READ_CONCURRENCY", "0")
+    assert _read_concurrency_env() == 1
+    monkeypatch.setenv("TICK_DUCKDB_READ_CONCURRENCY", "99")
+    assert _read_concurrency_env() == 16
+    monkeypatch.setenv("TICK_DUCKDB_READ_CONCURRENCY", "abc")
+    assert _read_concurrency_env() == 3
+    monkeypatch.delenv("TICK_DUCKDB_READ_CONCURRENCY", raising=False)
+    assert _read_concurrency_env() == 3
+
+
+def test_refresh_view_updates_definition(tmp_path: Path) -> None:
+    """视图重建生效: 指向新 glob 后查询立即取新数据。"""
+    repo = KlineRepository(DataStore(tmp_path))
+    a = _write_partition(tmp_path, "tbl_a", 2)
+    b = _write_partition(tmp_path, "tbl_b", 5)
+
+    assert repo.refresh_view("v_switch", f"{a}/**/*.parquet") is True
+    assert repo.execute_one("SELECT count(*) FROM v_switch")[0] == 2
+    assert repo.refresh_view("v_switch", f"{b}/**/*.parquet") is True
+    assert repo.execute_one("SELECT count(*) FROM v_switch")[0] == 5
+
+
+def test_refresh_view_rejects_invalid_name(tmp_path: Path) -> None:
+    """非标识符视图名直接拒绝, 不拼进 DDL。"""
+    repo = KlineRepository(DataStore(tmp_path))
+    a = _write_partition(tmp_path, "tbl_a", 2)
+    assert repo.refresh_view("evil; DROP VIEW x", f"{a}/**/*.parquet") is False
+
+
+def test_concurrent_refresh_view_never_conflicts(tmp_path: Path) -> None:
+    """DDL 串行: 并发重建同名视图全部成功 (无 write-write conflict),
+    且换视图期间并发读者始终读到某个一致版本 (MVCC)。"""
+    repo = KlineRepository(DataStore(tmp_path))
+    a = _write_partition(tmp_path, "tbl_a", 10)
+    b = _write_partition(tmp_path, "tbl_b", 20)
+    assert repo.refresh_view("v_swap", f"{a}/**/*.parquet") is True
+
+    results: list[bool] = []
+    seen_counts: set[int] = set()
+    errors: list[BaseException] = []
+
+    def ddl_worker(src: Path, rounds: int) -> None:
+        try:
+            results.extend(
+                repo.refresh_view("v_swap", f"{src}/**/*.parquet")
+                for _ in range(rounds)
+            )
+        except BaseException as exc:  # noqa: BLE001 — 测试内收集任意失败
+            errors.append(exc)
+
+    def reader(rounds: int) -> None:
+        try:
+            for _ in range(rounds):
+                seen_counts.add(repo.execute_one("SELECT count(*) FROM v_swap")[0])
+        except BaseException as exc:  # noqa: BLE001
+            errors.append(exc)
+
+    threads = [
+        threading.Thread(target=ddl_worker, args=(a, 40)),
+        threading.Thread(target=ddl_worker, args=(b, 40)),
+        threading.Thread(target=reader, args=(120,)),
+        threading.Thread(target=reader, args=(120,)),
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert errors == []
+    assert all(results), "并发视图重建出现 write-write conflict (DDL 未串行)"
+    assert seen_counts == {10, 20}, "读者应看到两个一致版本, 不应有撕裂值"

backend/tests/test_pipeline_capacity.py

@@ -204,6 +204,8 @@ def test_api_jobs_wait_before_computing(
         db=SimpleNamespace(execute=lambda *args: None),
         refresh_cache=lambda: None,
         get_index_symbol_set=lambda: set(),
+        # rebuild_enriched 收尾经 repo.refresh_view 重建视图 (不再直连 db.execute)
+        refresh_view=lambda *args, **kwargs: True,
     )
     app.state.capabilities = SimpleNamespace(has=lambda _: True)
     assert limiter.acquire("normal", timeout=0)