fix: avoid SQLite lock on fast position poll by skipping DB writes

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-06-25 15:14:22 +08:00
parent 12a30d4f0c
commit c302e1f3ca
3 changed files with 34 additions and 7 deletions
+25
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import os
import sqlite3
import time
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "futures.db")
@@ -17,3 +18,27 @@ def connect_db(path: str | None = None) -> sqlite3.Connection:
except sqlite3.OperationalError:
pass
return conn
def execute_retry(
conn: sqlite3.Connection,
sql: str,
params: tuple = (),
*,
retries: int = 6,
base_delay: float = 0.05,
) -> sqlite3.Cursor:
"""遇 database is locked 时短暂退避重试。"""
last_exc: Exception | None = None
for attempt in range(retries):
try:
return conn.execute(sql, params)
except sqlite3.OperationalError as exc:
if "locked" not in str(exc).lower():
raise
last_exc = exc
if attempt < retries - 1:
time.sleep(base_delay * (attempt + 1))
if last_exc:
raise last_exc
raise sqlite3.OperationalError("database is locked")