From 065c8ae1b9850f7cc544b4ceebdeff4792e014fe Mon Sep 17 00:00:00 2001 From: iomgaa Date: Thu, 16 Jul 2026 08:47:23 -0400 Subject: [PATCH] fix: telemetry single persistent connection + lock (kill concurrent write lock) --- adapters/telemetry.py | 63 ++++++++++++++++++++---------------- app/harness/log.py | 2 ++ tests/unit/test_telemetry.py | 47 +++++++++++++++++++++++++++ 3 files changed, 85 insertions(+), 27 deletions(-) diff --git a/adapters/telemetry.py b/adapters/telemetry.py index ddc4f7e..aea6d91 100644 --- a/adapters/telemetry.py +++ b/adapters/telemetry.py @@ -8,13 +8,11 @@ from __future__ import annotations import asyncio import sqlite3 -from typing import TYPE_CHECKING +import threading +from pathlib import Path from loguru import logger -if TYPE_CHECKING: - from pathlib import Path - class SQLiteTelemetryRecorder: """基于 SQLite 的 LLM 调用遥测记录器。 @@ -57,16 +55,30 @@ class SQLiteTelemetryRecorder: """ def __init__(self, db_path: Path) -> None: - self._db_path = db_path - self._table_ready = False + """建单持久连接 + 进程内 Lock（对齐 app/harness/log.py:HarnessLog 并发写模式）。 - def _ensure_table(self, conn: sqlite3.Connection) -> None: - """懒初始化：首次写入时创建 llm_calls 表。""" - if self._table_ready: - return - conn.execute(self._CREATE_TABLE_SQL) - conn.commit() - self._table_ready = True + 把并发控制拉到进程内（threading.Lock 串行化写），消除"每次新连接并发写同一 + db、靠 SQLite busy_timeout 跨连接协调"在高频下撑爆 timeout → database is locked + 的根因。check_same_thread=False：record_llm_call 经 asyncio.to_thread 在线程池 + 不同线程调用，共享连接跨线程访问需此 flag，串行性由 self._lock 保证。 + + 遥测哲学（P5）：连接初始化失败降级不冒泡（self._conn=None，写入直接丢弃 warning）， + 绝不因遥测故障拖垮 LLM 调用 / 训练。 + """ + self._db_path = db_path + self._lock = threading.Lock() + self._conn: sqlite3.Connection | None = None + try: + Path(db_path).parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(db_path), check_same_thread=False) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=5000") + conn.execute(self._CREATE_TABLE_SQL) + conn.commit() + self._conn = conn + except sqlite3.Error as exc: + logger.warning("遥测连接初始化失败（已降级，后续写入丢弃）: {}", exc) + self._conn = None def _write( self, @@ -87,20 +99,19 @@ class SQLiteTelemetryRecorder: cache_hit: bool, error: str | None, ) -> None: - """同步写入一条 LLM 调用记录到 SQLite。 + """同步写入一条 LLM 调用记录（单持久连接 + Lock 串行化，对齐 HarnessLog）。 - 三层防御加固： - 1. INSERT OR IGNORE — 主键冲突静默忽略 - 2. WAL + busy_timeout — 并发写锁容忍 - 3. try/except sqlite3.Error — DB 错误不冒泡到调用方 + 三层防御： + 1. INSERT OR IGNORE — call_id 主键冲突静默忽略（幂等） + 2. 进程内 threading.Lock 串行化写 — 消除并发锁竞争（非依赖 SQLite busy_timeout） + 3. try/except sqlite3.Error — DB 错误降级不冒泡，遥测失败绝不拖垮 LLM 调用 """ + if self._conn is None: + logger.warning("遥测连接不可用（已降级），丢弃 call_id={}", call_id) + return try: - conn = sqlite3.connect(str(self._db_path), timeout=10.0) - try: - conn.execute("PRAGMA journal_mode=WAL") - conn.execute("PRAGMA busy_timeout=5000") - self._ensure_table(conn) - conn.execute( + with self._lock: + self._conn.execute( self._INSERT_SQL, ( call_id, @@ -120,9 +131,7 @@ class SQLiteTelemetryRecorder: error, ), ) - conn.commit() - finally: - conn.close() + self._conn.commit() except sqlite3.Error as exc: logger.warning("遥测写入失败（已降级），call_id={}: {}", call_id, exc) diff --git a/app/harness/log.py b/app/harness/log.py index 4cab15f..1fbd79d 100644 --- a/app/harness/log.py +++ b/app/harness/log.py @@ -69,6 +69,8 @@ class HarnessLog: self._run_id = run_id self._register_run = register_run Path(db_path).parent.mkdir(parents=True, exist_ok=True) + # 单持久连接 + 进程内 Lock 串行化写：把并发控制拉到进程内，消除多连接争 + # SQLite 写锁。同款模式复用于 adapters/telemetry.py:SQLiteTelemetryRecorder。 self._conn = sqlite3.connect(db_path, check_same_thread=False) self._lock = threading.Lock() self._conn.row_factory = sqlite3.Row diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index 3514541..edcb405 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -146,3 +146,50 @@ async def test_concurrent_writes_no_lock_error(recorder, db_path): count = conn.execute("SELECT COUNT(*) FROM llm_calls").fetchone()[0] conn.close() assert count == 16 + + +def test_high_concurrency_writes_zero_loss(recorder, db_path): + """64 路线程并发直压 _write 应零丢失——复现生产 concurrency 下 database is locked 丢失。 + + 直接压同步 _write（不经 to_thread 排队），最大化并发连接数以逼出锁竞争； + 单连接 + threading.Lock 串行化模式下应全部落库、零丢失（对齐 HarnessLog）。 + """ + import concurrent.futures + + n = 64 + kwargs_list = [_make_call_kwargs() for _ in range(n)] + with concurrent.futures.ThreadPoolExecutor(max_workers=n) as executor: + list(executor.map(lambda kw: recorder._write(**kw), kwargs_list)) + + conn = sqlite3.connect(str(db_path)) + count = conn.execute("SELECT COUNT(*) FROM llm_calls").fetchone()[0] + conn.close() + assert count == n, f"并发写丢失 {n - count} 条（database is locked 降级丢弃）: 落库 {count}/{n}" + + +def test_uses_single_persistent_connection(db_path, monkeypatch): + """对齐 HarnessLog：单持久连接（构造时建一次），写入复用而非每次新建。 + + 每次写新建连接是并发锁竞争根源（多连接争 SQLite 写锁，撑爆 busy_timeout）； + 单连接 + 进程内 Lock 串行化把并发控制拉到进程内，消除 SQLite 层锁竞争。 + """ + connect_calls = {"n": 0} + real_connect = sqlite3.connect + + def _counting_connect(*args, **kwargs): + connect_calls["n"] += 1 + return real_connect(*args, **kwargs) + + monkeypatch.setattr(sqlite3, "connect", _counting_connect) + + recorder = SQLiteTelemetryRecorder(db_path=db_path) + after_init = connect_calls["n"] + for _ in range(10): + recorder._write(**_make_call_kwargs()) + after_writes = connect_calls["n"] + + assert after_init >= 1, "构造时应建立持久连接（对齐 HarnessLog）" + assert after_writes == after_init, ( + f"写入期间新建了 {after_writes - after_init} 个连接（应复用单持久连接，" + "每次新连接是并发锁竞争根源）" + )