fix(telemetry): INSERT OR IGNORE + WAL + try/except 三层防御加固
根治遥测写入主键冲突(UNIQUE constraint)和并发写锁(database is locked) 导致的异常冒泡,遥测侧信道错误不再污染 LLM 重试链。 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
+18
-3
@@ -10,6 +10,8 @@ import asyncio
|
|||||||
import sqlite3
|
import sqlite3
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
|
from loguru import logger
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
@@ -45,7 +47,7 @@ class SQLiteTelemetryRecorder:
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
_INSERT_SQL = """
|
_INSERT_SQL = """
|
||||||
INSERT INTO llm_calls (
|
INSERT OR IGNORE INTO llm_calls (
|
||||||
call_id, parent_call_id, session_id,
|
call_id, parent_call_id, session_id,
|
||||||
model_name, provider, messages, response, thinking,
|
model_name, provider, messages, response, thinking,
|
||||||
prompt_tokens, completion_tokens, latency_ms,
|
prompt_tokens, completion_tokens, latency_ms,
|
||||||
@@ -85,9 +87,18 @@ class SQLiteTelemetryRecorder:
|
|||||||
cache_hit: bool,
|
cache_hit: bool,
|
||||||
error: str | None,
|
error: str | None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""同步写入一条 LLM 调用记录到 SQLite。"""
|
"""同步写入一条 LLM 调用记录到 SQLite。
|
||||||
conn = sqlite3.connect(str(self._db_path))
|
|
||||||
|
三层防御加固:
|
||||||
|
1. INSERT OR IGNORE — 主键冲突静默忽略
|
||||||
|
2. WAL + busy_timeout — 并发写锁容忍
|
||||||
|
3. try/except sqlite3.Error — DB 错误不冒泡到调用方
|
||||||
|
"""
|
||||||
try:
|
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)
|
self._ensure_table(conn)
|
||||||
conn.execute(
|
conn.execute(
|
||||||
self._INSERT_SQL,
|
self._INSERT_SQL,
|
||||||
@@ -112,6 +123,10 @@ class SQLiteTelemetryRecorder:
|
|||||||
conn.commit()
|
conn.commit()
|
||||||
finally:
|
finally:
|
||||||
conn.close()
|
conn.close()
|
||||||
|
except sqlite3.Error as exc:
|
||||||
|
logger.warning(
|
||||||
|
"遥测写入失败(已降级),call_id={}: {}", call_id, exc
|
||||||
|
)
|
||||||
|
|
||||||
async def record_llm_call(
|
async def record_llm_call(
|
||||||
self,
|
self,
|
||||||
|
|||||||
@@ -104,3 +104,38 @@ async def test_record_cache_hit(recorder, db_path):
|
|||||||
conn.close()
|
conn.close()
|
||||||
|
|
||||||
assert row["cache_hit"] == 1
|
assert row["cache_hit"] == 1
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_duplicate_call_id_does_not_raise(recorder, db_path):
|
||||||
|
"""重复 call_id 写入应静默忽略(INSERT OR IGNORE),不抛异常。"""
|
||||||
|
kwargs = _make_call_kwargs()
|
||||||
|
await recorder.record_llm_call(**kwargs)
|
||||||
|
await recorder.record_llm_call(**kwargs)
|
||||||
|
conn = sqlite3.connect(str(db_path))
|
||||||
|
rows = conn.execute("SELECT COUNT(*) FROM llm_calls").fetchone()
|
||||||
|
conn.close()
|
||||||
|
assert rows[0] == 1
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_db_error_does_not_propagate(tmp_path):
|
||||||
|
"""SQLite 写入失败时 record_llm_call 应静默降级,不抛异常。"""
|
||||||
|
bad_recorder = SQLiteTelemetryRecorder(db_path=tmp_path / "nonexistent_dir" / "bad.db")
|
||||||
|
kwargs = _make_call_kwargs()
|
||||||
|
await bad_recorder.record_llm_call(**kwargs)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_concurrent_writes_no_lock_error(recorder, db_path):
|
||||||
|
"""16 路并发 record_llm_call 应全部成功,无 database is locked 错误。"""
|
||||||
|
import asyncio
|
||||||
|
tasks = []
|
||||||
|
for _ in range(16):
|
||||||
|
kwargs = _make_call_kwargs()
|
||||||
|
tasks.append(recorder.record_llm_call(**kwargs))
|
||||||
|
await asyncio.gather(*tasks)
|
||||||
|
conn = sqlite3.connect(str(db_path))
|
||||||
|
count = conn.execute("SELECT COUNT(*) FROM llm_calls").fetchone()[0]
|
||||||
|
conn.close()
|
||||||
|
assert count == 16
|
||||||
|
|||||||
Reference in New Issue
Block a user