接口先查询计数器，再更新它。连接已经设置 timeout=5，更新却几乎立即返回 database is locked。把超时调到 30 秒仍然失败，而同一条 SQL 在新连接上可以成功。此时要检查失败连接是否持有旧的读快照。
本文针对 Python 3.12+ 标准库 sqlite3、本机文件数据库、WAL 模式和独立连接。示例不使用共享缓存，也不依赖线程调度；通过两条连接的确定执行顺序复现事务交错。
1. 先看错误码，再判断能不能等待
database is locked 只是异常文本，不能据此认定所有失败都能靠等待解决。捕获 sqlite3.OperationalError 时，记录 sqlite_errorcode、sqlite_errorname、事务阶段、connection.in_transaction 和耗时；不要记录原始 SQL 参数或业务数据。
| 观察结果 | 含义与处理方向 |
|---|---|
BEGIN IMMEDIATE 遇到 SQLITE_BUSY |
其他连接正在写入；检查长事务，设置有上限的等待预算 |
读事务升级写入遇到 SQLITE_BUSY_SNAPSHOT，扩展码 517 |
快照已经过时；退出旧事务，重新读取并计算 |
SQLITE_LOCKED、I/O 或约束错误 |
分别检查游标、共享缓存、存储和数据规则，不能统称为可重试锁竞争 |
SQLite 官方给出的典型顺序是：A 开启读事务并查询，B 修改数据库并提交，A 再尝试写入。A 仍在旧快照上，升级会失败；等待不能把这份快照变新。必须结束旧事务，才能重新开始。隔离说明、扩展错误码给出了这一语义。
2. 一个能直接运行的复现与回归脚本
将下面代码保存为 sqlite_snapshot_check.py，用 Python 3.12+ 执行 python sqlite_snapshot_check.py -v。只用标准库，不需要安装包。脚本在当前目录创建临时数据库，退出时关闭连接并清理文件，不会连接业务库。
import sqlite3
import tempfile
import unittest
from pathlib import Path
def open_connection(path: Path, timeout: float = 5.0) -> sqlite3.Connection:
"""采用显式 SQL 事务，避免混用驱动自动开启的事务。"""
return sqlite3.connect(path, timeout=timeout, autocommit=True)
def increase_counter(connection: sqlite3.Connection) -> None:
"""短写事务在读取之前取得写入资格，失败时不留下未完成事务。"""
connection.execute("BEGIN IMMEDIATE")
try:
current = connection.execute(
"SELECT value FROM counter WHERE id = ?", (1,)
).fetchone()[0]
connection.execute(
"UPDATE counter SET value = ? WHERE id = ?", (current + 1, 1)
)
connection.execute("COMMIT")
except sqlite3.Error:
if connection.in_transaction:
connection.execute("ROLLBACK")
raise
class SnapshotTests(unittest.TestCase):
"""用确定的连接交错验证快照失效、恢复和写锁边界。"""
def setUp(self) -> None:
self.directory = tempfile.TemporaryDirectory(
prefix="sqlite-snapshot-", dir=Path.cwd()
)
self.addCleanup(self.directory.cleanup)
path = Path(self.directory.name) / "sample.db"
self.first = open_connection(path)
self.addCleanup(self.first.close)
mode = self.first.execute("PRAGMA journal_mode=WAL").fetchone()[0]
self.assertEqual(mode, "wal")
self.first.execute(
"CREATE TABLE counter ("
"id INTEGER PRIMARY KEY, "
"value INTEGER NOT NULL CHECK(value BETWEEN 0 AND 100))"
)
self.first.execute("INSERT INTO counter VALUES (?, ?)", (1, 0))
self.second = open_connection(path, timeout=0)
self.addCleanup(self.second.close)
def read_value(self, connection: sqlite3.Connection) -> int:
"""读取测试中的唯一计数器。"""
return connection.execute(
"SELECT value FROM counter WHERE id = ?", (1,)
).fetchone()[0]
def create_old_snapshot(self) -> None:
"""第二条连接提交后，第一条连接仍持有值为零的快照。"""
self.first.execute("BEGIN")
self.assertEqual(self.read_value(self.first), 0)
self.second.execute("UPDATE counter SET value = value + 1 WHERE id = 1")
self.assertEqual(self.read_value(self.first), 0)
def test_old_snapshot_cannot_become_writer(self) -> None:
self.create_old_snapshot()
with self.assertRaises(sqlite3.OperationalError) as caught:
self.first.execute("UPDATE counter SET value = value + 1 WHERE id = 1")
self.assertEqual(caught.exception.sqlite_errorcode, sqlite3.SQLITE_BUSY_SNAPSHOT)
self.assertEqual(caught.exception.sqlite_errorname, "SQLITE_BUSY_SNAPSHOT")
self.assertTrue(self.first.in_transaction)
self.first.execute("ROLLBACK")
def test_retry_in_old_transaction_still_fails(self) -> None:
self.create_old_snapshot()
for _ in range(2):
with self.assertRaises(sqlite3.OperationalError) as caught:
self.first.execute("UPDATE counter SET value = value + 1 WHERE id = 1")
self.assertEqual(caught.exception.sqlite_errorcode, sqlite3.SQLITE_BUSY_SNAPSHOT)
self.first.execute("ROLLBACK")
def test_restart_reads_new_value(self) -> None:
self.create_old_snapshot()
self.first.execute("ROLLBACK")
increase_counter(self.first)
self.assertFalse(self.first.in_transaction)
self.assertEqual(self.read_value(self.second), 2)
def test_immediate_moves_contention_to_begin(self) -> None:
self.first.execute("BEGIN IMMEDIATE")
self.assertEqual(self.read_value(self.first), 0)
with self.assertRaises(sqlite3.OperationalError) as caught:
self.second.execute("BEGIN IMMEDIATE")
self.assertEqual(caught.exception.sqlite_errorcode, sqlite3.SQLITE_BUSY)
self.assertFalse(self.second.in_transaction)
self.first.execute("ROLLBACK")
increase_counter(self.second)
self.assertEqual(self.read_value(self.first), 1)
def test_wal_reader_does_not_block_short_writer(self) -> None:
self.first.execute("BEGIN")
self.assertEqual(self.read_value(self.first), 0)
increase_counter(self.second)
self.assertEqual(self.read_value(self.first), 0)
self.first.execute("ROLLBACK")
self.assertEqual(self.read_value(self.first), 1)
def test_constraint_failure_rolls_back(self) -> None:
self.first.execute("UPDATE counter SET value = 100 WHERE id = 1")
with self.assertRaises(sqlite3.IntegrityError):
increase_counter(self.first)
self.assertFalse(self.first.in_transaction)
self.assertEqual(self.read_value(self.second), 100)
if __name__ == "__main__":
unittest.main()
预期是六项测试通过。第一条连接的超时仍为 5 秒，但旧快照场景会返回 SQLITE_BUSY_SNAPSHOT。脚本断言错误码和事务状态，不用脆弱的“必须小于几十毫秒”计时断言。
第二条连接使用 timeout=0，是为了确定地测试写入资格被占用时的失败路径，并非生产配置建议。测试没有模拟两个线程同时运行，但确实让两条独立连接的事务重叠，覆盖了造成旧快照问题的操作顺序。
3. 修复要落在事务边界
已确定会写入的短事务，可以像 increase_counter 一样，先 BEGIN IMMEDIATE，再读取和计算。成功取得写入资格之后，其他写入者不能插进这段读改写流程；竞争提前发生在事务入口。BEGIN IMMEDIATE 本身仍可能失败，必须由调用边界处理。事务文档说明了这两种事务的区别。
如果只是把数据库内的计数器加一，进一步简化成单条参数化语句通常更好：UPDATE counter SET value = value + 1 WHERE id = ?。应用无需先取旧值；独立单语句事务也避开了示例中的读快照升级。若还有关联记录必须同时写入，仍要把这些语句放入同一短事务。
已经收到 SQLITE_BUSY_SNAPSHOT 时，正确恢复顺序是：回滚整个旧事务，开启新事务，重新读取业务数据，重新验证条件，再执行写入。不能只重试最后一条 UPDATE，也不能沿用第一次 SELECT 算出的新值。脚本第三项测试验证了最终值为 2，保留了另一个连接先前提交的增量。
不要给所有 SELECT 换成 BEGIN IMMEDIATE：纯读请求会无谓争抢唯一写入资格。也不要在写事务内等待 HTTP 请求、人工确认或长时间计算。必须先读取大量数据才能计算时，应考虑事务外计算加版本条件更新，并在版本冲突时重新读取，而不是长时间占住写入位置。
4. Python 的事务控制不要混用
这里显式传入 autocommit=True，再使用 SQL 的 BEGIN、COMMIT、ROLLBACK。该参数从 Python 3.12 开始提供；这种模式下，连接对象的 commit()、rollback() 不承担手工 SQL 事务的提交和回滚，所以示例全程执行 SQL。
如果项目使用 autocommit=False，驱动会维护事务生命周期，直接照搬 BEGIN IMMEDIATE 可能触发“事务中再次开启事务”。先明确 ORM 或驱动当前的控制模式，统一入口后再调整；不要在公共函数里偷偷修改调用者连接的 autocommit，因为更改该设置可能影响已有事务。Python sqlite3 事务控制说明可用于核对行为。
increase_counter 假定传入一条没有活动事务、且不被其他请求共用的连接，计数器行已由初始化建立。它只演示本地数据库事务，不接受外部记录 ID。接入接口时还需要服务端校验参数、授权和租户边界，并在进入这段函数前确定事务所有权。
5. 上线检查与失败处理
先从失败日志区分错误码，再检查事务开始到提交之间包含哪些工作。把连接超时、上游请求截止时间和应用重试预算一起计算，避免每次等待 5 秒再重试多次，最终拖垮请求队列。
对可重试的竞争设置总截止时间和少量尝试次数；每次重做完整事务。约束错误、只读文件、磁盘故障不能进入同一个重试分支。如果事务已经夹带发券、付款或通知等外部动作，先设计幂等和提交后投递机制，不能直接重跑整个业务函数。
灰度时观察错误码分布、事务耗时、写入等待和接口失败率。若改用 IMMEDIATE 后写入口排队上升，先缩短事务、减少不必要的先读后写，再判断是否需要收敛写入并发或迁移数据库。回退应用改动也要保留正确的回滚和错误分类，不能重新吞掉锁异常。
不要把删除 -wal、-shm 文件作为解锁办法，也不必为这个问题切换日志模式。WAL 允许读写并行，但仍只有一个写入者；本文修复的是事务组织方式。
本文在 Python 3.14.6、SQLite 3.50.4 上验证上述六项测试。未验证真实 ORM、连接池、多进程、网络文件系统或生产吞吐；这些边界应在实际部署环境补测。Ruff 在验证环境不可用，未执行其格式化与静态检查。
这类故障的关键证据是“先读、其他连接提交、旧事务再写”的顺序和扩展错误码。让等待发生在短写事务入口，或用单条数据库更新消除应用侧读改写；已有旧快照则先退出，再基于新数据重新计算。
Discussion
评论