diff --git a/PROGRESS.md b/PROGRESS.md index c5376c8..644b917 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -612,3 +612,34 @@ people 59;外网 `/` `/timeline` 200、`/login` 302、`/api/*` 未登录 401 **待办**:NAS 侧部署 fam-notifier(只能用户手动,密码登录);chat_history 一次性迁移 (`fam-core/scripts/import_chat_history.py`,幂等);frpc.toml 里的 8000 映射可删。 + +## 修复 fam-edge 多线程共用 SQLite 连接(2026-09-13) + +**触发**:用户问「甲骨文剩余硬盘小于 10G 会删东西的逻辑怎么没了?」。查下来逻辑一直在、 +也没被迁云动过(`disk_guard.py` 最后一次改动还是 8/28 新增它那次),但它 95% 的检查在空转: + +``` +DiskGuard「本轮清理完成」 128 次 +DiskGuard「检查异常」 2837 次 ← 22 倍 +``` + +失败原因全是 `cannot start a transaction within a transaction`。表现就是磁盘剩余空间在 +2.7GB 和 16GB 之间来回荡——清理能不能成功全靠运气,赶上 rclone 集中下载 +(实测 5 分钟写入 12.6GB)就掉进危险区。 + +**根因**:`OracleDB.__init__` 建一条 `sqlite3.connect(check_same_thread=False)` 的连接 +给全进程共用,而 VideoQueue / PersonService / DiskGuard 三个后台线程 + gunicorn 的 4 个 +请求线程都在并发读写它。sqlite3 的连接对象本来就不是可并发共享的,事务状态互相踩踏。 +同一个根因在线上刷出三类错误,累计:`database is locked` 71604 次、 +`cannot start a transaction within a transaction` 378 次、`no more rows available` 92 次 +(后者堆栈落在 `self._conn.commit()`,是游标被别的线程重置的典型症状)。 +运动事件推送被 500 打回也是它——NAS 侧失败批次不推进游标,下一轮补推,所以没丢事件。 + +**修复**:`_conn` 改成 `@property`,从 `threading.local()` 取当前线程的连接,没有就新建 +(WAL + busy_timeout=10000)。`close()` 相应改成收掉所有线程开过的连接。因为外部调用方 +(如 `api_gateway` 的 activity 端点)也在直接用 `db._conn.execute(...)`,做成 property +可以让全部现有调用点原样工作,不用逐个改。`_write_lock` 保留,复合写语义不变。 + +**测试**:新增 2 个用例(8 线程 × 25 轮并发读写、close 要收掉所有连接)。在旧代码上 +稳定复现同族错误 `cannot commit transaction - SQL statements in progress`,修复后通过; +fam-edge 全套 159 个测试绿。 diff --git a/README.md b/README.md index d8d291f..7c37a60 100644 --- a/README.md +++ b/README.md @@ -724,6 +724,7 @@ print('NVIDIA NIM 连接成功:', response.choices[0].message.content) - NAS scp 子系统被禁用,同样用 stdin 管道传文件 - **Oracle fam-edge 由 systemd `fam-edge.service` 守护(Restart=always)**:部署代码后必须 `sudo systemctl restart fam-edge`;手动 `setsid` 启动会和守护打架(端口 `Connection in use`) - NAS 远端 kill gunicorn 用 `ps aux | grep "[f]am-core/venv/bin/gunicorn"` 字符类技巧(pkill/pgrep 会匹配 SSH 自身命令行导致断连) +- **sqlite3 连接不能跨线程共享**(2026-09-13 修):`OracleDB` 原来在 `__init__` 里建一条 `check_same_thread=False` 的连接给全进程用,VideoQueue / PersonService / DiskGuard 三个后台线程加 gunicorn 请求线程并发读写它,事务状态互相踩踏,线上累计刷出 `database is locked` 71604 次、`cannot start a transaction within a transaction` 378 次、`no more rows available` 92 次;DiskGuard 的清理被打断 2837 次,磁盘守护形同虚设(剩余空间在 2.7G 和 16G 之间来回荡)。改成 `threading.local()` 每线程一条连接后,WAL 下读不互斥、写由 SQLite 自己排队。**新增后台线程时不要再去共用某一条连接对象** - fam-core 启动模块路径是 `src.fam_core.app:app`(不是 `fam_core.app:app`);`start_core.sh` 会 source 仓库根 `.env` 注入 `DSM_*/ORACLE_SYNC_TOKEN` - Edge 单 worker 处理任务期间 `/health` 可能不响应,属正常 - **运动数据清理**:切换架构/重新提取时清 Oracle `videos/events/people` + `motion_clips/`(保留 `ss_motion_events` 与素材)与 NAS `sync_*` 镜像,重启两端自动重新分割分析 diff --git a/fam-edge/src/fam_edge/oracle_db.py b/fam-edge/src/fam_edge/oracle_db.py index de9b2bd..f635396 100644 --- a/fam-edge/src/fam_edge/oracle_db.py +++ b/fam-edge/src/fam_edge/oracle_db.py @@ -33,13 +33,42 @@ class OracleDB: def __init__(self, db_path: str): os.makedirs(os.path.dirname(db_path), exist_ok=True) self.db_path = db_path - self._conn = sqlite3.connect(db_path, check_same_thread=False) - self._conn.row_factory = sqlite3.Row - self._conn.execute("PRAGMA journal_mode=WAL") - self._conn.execute("PRAGMA busy_timeout=10000") + self._local = threading.local() # 每线程一条连接,见下面的 _conn + self._all_conns = [] # 仅供 close() 收尾 + self._conns_lock = threading.Lock() self._write_lock = threading.Lock() # 复合写(如 DELETE+INSERT+commit)串行化 self._init_schema() + @property + def _conn(self) -> sqlite3.Connection: + """当前线程的连接(2026-09-13 从"全进程共用一条"改成每线程一条)。 + + 原来是 __init__ 里建一条 `check_same_thread=False` 的连接给所有线程共用: + VideoQueue / PersonService / DiskGuard 三个后台线程,加上 gunicorn 的请求 + 线程,并发读写同一个连接对象,事务状态互相踩踏。线上长期刷三类报错, + 全是这一个根因: + + - `cannot start a transaction within a transaction`:一个线程的事务还 + 开着,另一个线程又要开——DiskGuard 的清理被打断 2837 次,磁盘守护基本 + 靠运气生效(表现为剩余空间在 2.7G 和 16G 之间来回荡) + - `no more rows available`:commit 时游标已被别的线程重置,87 次 + - `database is locked`:每小时上百次,NAS 推来的运动事件被 500 打回 + + sqlite3 的连接本来就不是可并发共享的对象。改成各线程各拿一条之后:WAL 下 + 多连接读不互斥,写由 SQLite 自己排队(busy_timeout 兜底等 10 秒),而 + `_write_lock` 继续保证"复合写"在本进程内串行,语义不变。 + """ + conn = getattr(self._local, 'conn', None) + if conn is None: + conn = sqlite3.connect(self.db_path, timeout=10, check_same_thread=False) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=10000") + self._local.conn = conn + with self._conns_lock: + self._all_conns.append(conn) + return conn + # ------------------------------------------------------------------ def _init_schema(self): c = self._conn @@ -1078,4 +1107,12 @@ class OracleDB: self._conn.commit() def close(self): - self._conn.close() + """关掉所有线程开过的连接(不只当前线程这一条)。""" + with self._conns_lock: + conns, self._all_conns = self._all_conns, [] + for c in conns: + try: + c.close() + except sqlite3.Error: + pass + self._local = threading.local() diff --git a/fam-edge/tests/test_oracle_db.py b/fam-edge/tests/test_oracle_db.py index d93c947..9545899 100644 --- a/fam-edge/tests/test_oracle_db.py +++ b/fam-edge/tests/test_oracle_db.py @@ -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 == []