fix(fam-edge): SQLite 改每线程一条连接,修好 DiskGuard 等一切并发写
OracleDB 原来在 __init__ 里建一条 check_same_thread=False 的连接给全进程共用, 而 VideoQueue / PersonService / DiskGuard 三个后台线程 + gunicorn 的请求线程都在 并发读写它。sqlite3 的连接对象本来就不是可并发共享的,事务状态互相踩踏,线上 累计刷出三类错误(同一个根因): database is locked 71604 次 cannot start a transaction within a transaction 378 次 no more rows available 92 次(堆栈落在 commit()) 最严重的后果是磁盘守护形同虚设:DiskGuard 的清理被打断 2837 次,成功仅 128 次, 剩余空间在 2.7GB 和 16GB 之间来回荡——赶上 rclone 集中下载(实测 5 分钟写入 12.6GB)就掉进危险区。NAS 推来的运动事件被 500 打回也是它(失败批次不推进游标, 下一轮补推,所以没丢事件)。 改法:_conn 改成 @property,从 threading.local() 取当前线程的连接,没有就新建 (WAL + busy_timeout=10000)。做成 property 是因为外部调用方(api_gateway 的 activity 端点)也在直接用 db._conn.execute(...),这样全部现有调用点原样工作。 close() 相应改成收掉所有线程开过的连接。_write_lock 保留,复合写语义不变。 测试:新增 8 线程 × 25 轮并发读写、close 收连接两个用例;在旧代码上稳定复现 同族错误 cannot commit transaction - SQL statements in progress。 线上验证:重启后 DiskGuard 立刻跑通一轮(清 40 个文件,剩余回到 15.3GB), 三类 SQLite 错误在重启后的日志里均为 0。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -463,3 +463,63 @@ def test_delete_video_does_not_touch_ss_motion_events(tmp_path):
|
||||
assert db.get_video_by_motion_event_id(555) is None
|
||||
row = db._conn.execute("SELECT * FROM ss_motion_events WHERE event_id=?", (555,)).fetchone()
|
||||
assert row is not None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 并发(2026-09-13:修"全进程共用一条 sqlite 连接"的回归测试)
|
||||
# ---------------------------------------------------------------------------
|
||||
def test_concurrent_writes_from_many_threads(tmp_path):
|
||||
"""多线程同时读写不能互相踩踏。
|
||||
|
||||
改成每线程一条连接之前,这里会稳定炸出三类错误之一:
|
||||
`cannot start a transaction within a transaction` / `no more rows available`
|
||||
/ `database is locked`——线上 DiskGuard 的清理就是被第一种打断了 2837 次。
|
||||
"""
|
||||
import threading
|
||||
|
||||
db = _db(tmp_path)
|
||||
errors = []
|
||||
rounds = 25
|
||||
|
||||
def writer(tid):
|
||||
try:
|
||||
for i in range(rounds):
|
||||
db.record_activity('t%d' % tid, 'act%d' % i, 'detail')
|
||||
db.set_cursor('cursor_t%d' % tid, str(i))
|
||||
db.record_motion_events([{
|
||||
'event_id': tid * 1000 + i, 'camera_id': 2, 'event_type': 10,
|
||||
'start_time': 1789000000 + i, 'duration': 5,
|
||||
'thumbnail_url': '',
|
||||
}])
|
||||
db.get_recent_activities(5)
|
||||
db.get_motion_heartbeat_age_sec()
|
||||
except Exception as e: # noqa: BLE001 —— 要把原始异常带出来看
|
||||
errors.append(f"线程{tid}: {type(e).__name__}: {e}")
|
||||
|
||||
threads = [threading.Thread(target=writer, args=(t,)) for t in range(8)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join(timeout=60)
|
||||
|
||||
assert not errors, "并发写出错:\n" + "\n".join(errors[:5])
|
||||
assert db._conn.execute("SELECT COUNT(*) FROM ss_motion_events").fetchone()[0] == 8 * rounds
|
||||
for tid in range(8):
|
||||
assert db.get_cursor('cursor_t%d' % tid) == str(rounds - 1)
|
||||
|
||||
|
||||
def test_close_releases_every_thread_connection(tmp_path):
|
||||
"""close() 要收掉所有线程开过的连接,不只当前线程那一条。"""
|
||||
import threading
|
||||
|
||||
db = _db(tmp_path)
|
||||
db.record_activity('main', 'x', '')
|
||||
|
||||
def other():
|
||||
db.record_activity('other', 'y', '')
|
||||
t = threading.Thread(target=other)
|
||||
t.start(); t.join()
|
||||
|
||||
assert len(db._all_conns) == 2 # 主线程 + 子线程各一条
|
||||
db.close()
|
||||
assert db._all_conns == []
|
||||
|
||||
Reference in New Issue
Block a user