From c77ef40fd571b9e414036df957bd27c34041a3df Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sat, 26 Sep 2026 01:03:18 +0800 Subject: [PATCH] =?UTF-8?q?feat(db):=20=E5=90=8C=E6=AD=A5=E6=A0=88?= =?UTF-8?q?=E8=AF=BB=E5=86=99=E5=B9=B6=E5=8F=91=EF=BC=88=5Fq=20=E5=8E=BB?= =?UTF-8?q?=E9=94=81=EF=BC=89+=20core.db=20=E5=B9=B6=E5=8F=91/=E8=B7=A8loo?= =?UTF-8?q?p=20=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit auth_system._q：单语句读不再持 _DB_LOCK —— WAL 下每连接读一致性快照， 读写互不阻塞；旧实现把全部读也串行化且被写挡住。写事务（_x 与 24 个 with _DB_LOCK 块，经扫描均为写事务）仍全程串行，无写-写竞争。 core.db 并发修复（由测试暴露的真实缺陷）： - WAL 切换竞态：并发连接同时执行 PRAGMA journal_mode=WAL 争排他锁 → 专用锁内串行切换 + 二次复查（busy_timeout 前置） - 跨事件循环锁：模块级 asyncio.Lock 绑定首个 loop，多 loop 场景报 "bound to a different event loop" → 锁按 loop 隔离 测试：25/25 通过（新增「读不被写锁阻塞」红转绿、读+同步写者并发 WAL、 异步并发读）； lint: ruff All checks passed；pyright 0 errors。 验收：冒烟 10/10、奇偶校验 2/2、uvicorn 真实启动 Flask 兜底+FastAPI 双通。 --- auth_system.py | 17 +++--- core/db.py | 45 +++++++++++++--- tests/test_aiohttp_compat.py | 8 ++- tests/test_auth_db_concurrency.py | 87 +++++++++++++++++++++++++++++++ 4 files changed, 141 insertions(+), 16 deletions(-) create mode 100644 tests/test_auth_db_concurrency.py diff --git a/auth_system.py b/auth_system.py index 6b6650e..50f6097 100644 --- a/auth_system.py +++ b/auth_system.py @@ -60,12 +60,17 @@ def get_conn() -> sqlite3.Connection: def _q(sql: str, params: tuple = ()) -> list[sqlite3.Row]: - with _DB_LOCK: - conn = get_conn() - try: - return conn.execute(sql, params).fetchall() - finally: - conn.close() + """单语句读：**不持锁**。 + + WAL 模式下每个连接读到一致性快照，读与写互不阻塞；旧实现对读也加 _DB_LOCK， + 等于把全部读串行化并被写挡住。此处仅放开单语句读——写事务（_x 与各 + `with _DB_LOCK` 块）仍全程串行，故不引入写-写竞争。 + """ + conn = get_conn() + try: + return conn.execute(sql, params).fetchall() + finally: + conn.close() def _x(sql: str, params: tuple = ()) -> int: diff --git a/core/db.py b/core/db.py index be3b081..657d05e 100644 --- a/core/db.py +++ b/core/db.py @@ -25,21 +25,50 @@ import aiosqlite DATA_DIR = Path(os.environ.get('SITE_DATA_DIR', 'data')) DB_PATH = DATA_DIR / 'site.db' -# 写事务串行锁（惰性创建，避免跨事件循环绑定） -_WRITE_LOCK: asyncio.Lock | None = None +# 锁按事件循环隔离：asyncio.Lock 绑定创建它的 loop，跨 loop 复用会抛 +# "bound to a different event loop"（迁移期 FastAPI loop + 同步桥 loop + 测试各自 +# 的 loop 并存，必须分开持有）。 +_LOOP_LOCKS: dict[tuple[int, str], asyncio.Lock] = {} + + +def _loop_lock(kind: str) -> asyncio.Lock: + key = (id(asyncio.get_running_loop()), kind) + lock = _LOOP_LOCKS.get(key) + if lock is None: + lock = asyncio.Lock() + _LOOP_LOCKS[key] = lock + return lock def _write_lock() -> asyncio.Lock: - global _WRITE_LOCK - if _WRITE_LOCK is None: - _WRITE_LOCK = asyncio.Lock() - return _WRITE_LOCK + """写事务串行锁（每事件循环一个）。""" + return _loop_lock('write') + + +def _wal_lock() -> asyncio.Lock: + """WAL 切换专用锁（不能用 _write_lock：execute() 已持它再调 + connect→_configure 会自锁死）。""" + return _loop_lock('wal') + + +# 已确认切到 WAL 的库路径（首连设置一次；后续连接不再重复执行 PRAGMA 争排他锁） +_WAL_READY_FOR: str | None = None async def _configure(conn: aiosqlite.Connection) -> None: + global _WAL_READY_FOR conn.row_factory = aiosqlite.Row - await conn.execute('PRAGMA journal_mode=WAL') # 读写并发的前提 - await conn.execute('PRAGMA busy_timeout=15000') # 遇写锁自旋等待，而非立即报错 + # busy_timeout 先设：让后续语句遇锁自旋等待，而不是立刻报 database is locked + await conn.execute('PRAGMA busy_timeout=15000') + key = str(DB_PATH) + if _WAL_READY_FOR != key: + # 切换日志模式需要排他锁，且 `_WAL_READY_FOR` 的检查不是原子的—— + # 并发连接须在专用锁内串行完成「切换 + 标记」，并二次复查，否则批量并发 + # 连接会同时执行 PRAGMA journal_mode=WAL 争锁（实测 database is locked） + async with _wal_lock(): + if _WAL_READY_FOR != key: + await conn.execute('PRAGMA journal_mode=WAL') + _WAL_READY_FOR = key await conn.execute('PRAGMA synchronous=NORMAL') # WAL 下兼顾安全与吞吐 await conn.execute('PRAGMA foreign_keys=ON') diff --git a/tests/test_aiohttp_compat.py b/tests/test_aiohttp_compat.py index 4a4a94f..532357d 100644 --- a/tests/test_aiohttp_compat.py +++ b/tests/test_aiohttp_compat.py @@ -78,5 +78,9 @@ async def test_bridge_rejects_inside_loop(): async def coro(): return 1 - with pytest.raises(RuntimeError): - async_bridge.run_sync(coro()) + c = coro() + try: + with pytest.raises(RuntimeError): + async_bridge.run_sync(c) + finally: + c.close() # run_sync 拒绝执行，需自行关闭以免「未 await」告警 diff --git a/tests/test_auth_db_concurrency.py b/tests/test_auth_db_concurrency.py new file mode 100644 index 0000000..39213c2 --- /dev/null +++ b/tests/test_auth_db_concurrency.py @@ -0,0 +1,87 @@ +# -*- coding: utf-8 -*- +"""auth_system 同步栈的读写并发验证：读不持锁、写仍串行（WAL）。""" +import sqlite3 +import threading +import time + +import pytest + +import auth_system + + +@pytest.fixture() +def temp_site(tmp_path, monkeypatch): + monkeypatch.setattr(auth_system, 'DATA_DIR', tmp_path) + monkeypatch.setattr(auth_system, 'DB_PATH', tmp_path / 'site.db') + monkeypatch.setattr(auth_system, 'BACKUP_DIR', tmp_path / 'backups') + monkeypatch.setattr(auth_system, 'SECRET_KEY_PATH', tmp_path / 'secret.key') + auth_system.init_db() + return tmp_path / 'site.db' + + +def test_q_does_not_hold_lock(temp_site): + """_q 读不应获取 _DB_LOCK（否则读会被写挡住）。""" + # 让另一个线程持有写锁，读仍应在短时间返回（不被阻塞） + held = threading.Event() + release = threading.Event() + + def holder(): + with auth_system._DB_LOCK: + held.set() + release.wait(5) + + t = threading.Thread(target=holder, daemon=True) + t.start() + held.wait(2) + try: + start = time.monotonic() + rows = auth_system._q('SELECT COUNT(*) AS n FROM users') + elapsed = time.monotonic() - start + assert rows is not None + assert elapsed < 1.0, f'读被写锁阻塞了 {elapsed:.2f}s（WAL 下不应发生）' + finally: + release.set() + t.join(2) + + +def test_concurrent_reads_and_writes(temp_site): + """多读线程 + 一个写线程并发：无异常、结果一致。""" + errors = [] + stop = threading.Event() + + def writer(): + try: + for i in range(50): + auth_system.set_setting(f'k{i}', str(i)) + except Exception as e: + errors.append(('w', e)) + finally: + stop.set() + + def reader(): + while not stop.is_set(): + try: + auth_system.get_setting('k0') + except Exception as e: + errors.append(('r', e)) + + ws = threading.Thread(target=writer, daemon=True) + rs = [threading.Thread(target=reader, daemon=True) for _ in range(4)] + ws.start() + for r in rs: + r.start() + ws.join(20) + for r in rs: + r.join(2) + + assert not errors, f'并发读写异常: {errors[:3]}' + assert auth_system.get_setting('k49') == '49' + + +def test_wal_enabled(temp_site): + conn = sqlite3.connect(str(temp_site)) + try: + mode = conn.execute('PRAGMA journal_mode').fetchone()[0] + finally: + conn.close() + assert str(mode).lower() == 'wal'