"""SQLite 并发 helper(云端控制面单机阶段)。 只解决一件事:ThreadingHTTPServer 多线程 + 单个 SQLite 文件时的写冲突。 不碰任何业务逻辑、不定义新 schema、不改 server.py 的接线: 本模块只提供连接与读写 helper,协调员后续把 server 的 _connect 换成这里的 connect 即可(表名/字段与 server.py 的 init_db、 memory.py 的 summaries 完全一致,因为本模块根本不建表)。 使用约定(必须遵守,否则 helper 帮不了你): - 每个线程用自己的连接:绝不在多线程间共用一个 sqlite3 连接 (sqlite3 默认 check_same_thread=True,共用会直接报错)。 典型用法是每请求/每线程调用一次 connect(),用完 close()。 - 写操作走 execute_with_retry + transaction:一次语句失败重试是 安全的(单条 INSERT/UPDATE 被锁时 SQLite 已回滚该语句); 不要把多条非幂等语句包在一个重试里——重试只针对单条语句。 单机 SQLite 的上限(诚实边界,见模块末尾注释): - WAL + busy_timeout + 有限重试只能缓解单机多线程写冲突, 不能变成多机/高并发数据库:写仍是单写者串行,读多写少尚可, 写密集或跨机器时必须换 Postgres(下一阶段,不在本轮假装解决)。 """ import sqlite3 import time from contextlib import contextmanager # 与 server.py 旧 _connect 的 timeout=30s 保持一致的连接级等待; # PRAGMA busy_timeout 是语句级等待,两者不冲突,后者以毫秒为单位 # 在每个连接上单独设置(不随库文件持久化,见 connect 的注释)。 DEFAULT_BUSY_TIMEOUT_MS = 5000 DEFAULT_CONNECT_TIMEOUT_S = 30 DEFAULT_MAX_RETRIES = 3 # 指数退避:0.05s、0.1s、0.2s;重试耗尽即抛原始错误,绝不无限循环。 DEFAULT_BACKOFF_SECONDS = (0.05, 0.1, 0.2) _LOCKED_MARKER = "database is locked" def connect(db_path, busy_timeout_ms=DEFAULT_BUSY_TIMEOUT_MS, timeout=DEFAULT_CONNECT_TIMEOUT_S): """返回一个已设好 WAL + busy_timeout + Row 的新连接。 - PRAGMA journal_mode=WAL:读不阻塞写、写不阻塞读,是多线程 server 的前提;该模式随库文件持久化(:memory: 库除外, 内存库的 journal_mode 恒为 memory,属 SQLite 本身行为)。 - PRAGMA busy_timeout:写冲突时语句级等待窗口,单位毫秒; 它是每个连接的属性,不随库文件持久化,所以每次 connect 都设。 - row_factory=sqlite3.Row:与 server.py 各 handler 在 _db() 里 事后设置的 row_factory 一致,行可用 row["id"] 按列名访问。 - 不设 check_same_thread=False:跨线程共用连接是错误用法, 保持默认 True 让误用尽早报错,而不是静默出竞态。 """ conn = sqlite3.connect(str(db_path), timeout=timeout) try: conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=%d" % int(busy_timeout_ms)) conn.row_factory = sqlite3.Row except Exception: conn.close() raise return conn def is_locked_error(exc): """只有 OperationalError 且文案含 database is locked 才算写锁冲突。 其它 OperationalError(如 no such table、语法错误)重试无用, 必须立即抛出,不能被重试掩盖成“锁问题”。 """ return ( isinstance(exc, sqlite3.OperationalError) and _LOCKED_MARKER in str(exc).lower() ) def execute_with_retry(conn, sql, params=(), max_retries=DEFAULT_MAX_RETRIES, backoff_seconds=DEFAULT_BACKOFF_SECONDS): """执行单条写语句,遇写锁冲突时有限重试,返回 cursor。 重试次数 = max_retries(默认 3 次,即最多尝试 1 + 3 = 4 次), 每次重试前按 backoff_seconds 依次 sleep(默认 0.05/0.1/0.2s, 退避表短于重试次数时取最后一项,不越界)。 重试耗尽后把最后一次的原始 OperationalError 原样抛出,不吞掉、 不包装成别的异常类型,调用方看到的仍是 sqlite3.OperationalError。 非锁错误一次都不重试。 """ attempts = 0 while True: try: return conn.execute(sql, params) except sqlite3.OperationalError as exc: if not is_locked_error(exc) or attempts >= max_retries: raise delay = backoff_seconds[ min(attempts, len(backoff_seconds) - 1) ] time.sleep(delay) attempts += 1 def read_all(conn, sql, params=()): """读操作 helper:不重试,只要求连接本身已走 WAL + busy_timeout。 读在 WAL 下不阻塞写,一般不需要重试;如果调用方传的是旧式 连接(非本模块 connect 建的),读结果仍正确,只是没有 WAL/ busy_timeout 保障——这属于调用方误用,本函数不静默替它建连接。 """ return conn.execute(sql, params).fetchall() @contextmanager def transaction(conn): """单连接事务上下文:成功 commit,异常 rollback 后原样抛出。 与 execute_with_retry 配合的典型写法:: conn = connect(db_path) try: with transaction(conn): execute_with_retry(conn, "INSERT INTO tasks(...) ...", (...)) finally: conn.close() 注意:重试发生在语句级(transaction 内部),事务本身不重试; 若事务内某条语句最终抛错,整个事务回滚,不会留下半写状态。 """ try: yield conn conn.commit() except Exception: try: conn.rollback() except Exception: pass raise