重复请求不应重复扣库存：用SQLite验证幂等键与事务回滚

作者：lgbisha。本文由AI辅助起草、编写示例并在本地执行验证。所有事件和库存均为合成数据，仅使用内存SQLite数据库；不包含真实订单、生产库存或云部署，也不是经过并发压测的库存服务。

一次请求超时后，客户端可能重试。对于修改状态的操作，问题不只是“第二次能否成功”，更是“第一次是否已经成功”。本文把范围缩到一个商品、一条数据库连接，用可观察的状态验证两件事：相同事件不能重复修改库存；业务失败不能留下一个假装成功的事件标记。

一、先区分三种输入

假设初始库存100，事件receive-1表示入库20：

最后一行很重要。只判断“这个键出现过”，会把调用者误用幂等键也当作正常重放。这个演示只有delta一个业务字段，所以直接比较整数；真实接口应定义完整的请求语义、租户范围及规范化策略，不能照搬为只比较金额或数量。

二、让事件标记和库存修改处于同一事务

以下代码在Python 3.12.3、SQLite 3.45.1上执行。示例显式选择LEGACY_TRANSACTION_CONTROL及DEFERRED，不依赖未来版本的默认事务模式。with connection处理事务退出时的提交或回滚，但不负责关闭连接，因此最后仍需close()。

import sqlite3


def new_store():
    connection = sqlite3.connect(
        ":memory:",
        autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL,
        isolation_level="DEFERRED",
    )
    connection.executescript("""
        CREATE TABLE stock (
            item_id INTEGER PRIMARY KEY,
            units INTEGER NOT NULL CHECK (
                typeof(units) = 'integer' AND units >= 0
            )
        );
        CREATE TABLE applied_change (
            event_id TEXT PRIMARY KEY NOT NULL,
            delta INTEGER NOT NULL
        );
        INSERT INTO stock VALUES (1, 100);
    """)
    return connection


def apply_change(connection, event_id, delta):
    if not isinstance(event_id, str) or not event_id.strip():
        raise ValueError("event_id必须是非空字符串")
    if type(delta) is not int or not -1000 <= delta <= 1000 or delta == 0:
        raise ValueError("演示delta必须是-1000到1000之间的非零整数")
    if connection.in_transaction:
        raise RuntimeError("必须传入无活动事务的连接")
    with connection:
        inserted = connection.execute(
            "INSERT INTO applied_change VALUES (?, ?) "
            "ON CONFLICT(event_id) DO NOTHING",
            (event_id, delta),
        )
        if inserted.rowcount == 0:
            previous = connection.execute(
                "SELECT delta FROM applied_change WHERE event_id = ?",
                (event_id,),
            ).fetchone()[0]
            if previous != delta:
                raise ValueError("同一event_id对应了不同delta")
            return False
        updated = connection.execute(
            "UPDATE stock SET units = units + ? "
            "WHERE item_id = 1 AND units + ? >= 0",
            (delta, delta),
        )
        if updated.rowcount != 1:
            raise ValueError("库存不足或演示商品不存在")
    return True

event_id是数据库唯一键。这里先尝试插入事件，并且仅针对该键的冲突执行DO NOTHING，再根据是否插入判断首次请求或重放。其他数据库错误仍然向外传播，不能用一个宽泛的异常捕获把约束错误全部吞成“已经处理”。

库存更新没有在Python里先查余额再写，而是把非负条件放在同一条UPDATE中。若更新不到目标行，抛出异常；此前刚插入的事件也跟着回滚。stock还有整数类型与非负约束，避免示例中的整数溢出变成浮点库存后悄悄通过。

这里没有使用REPLACE覆盖旧事件，也没有拼接SQL字符串。函数拒绝已有活动事务的连接，避免离开with时把调用者此前的未提交工作一起提交或回滚。这是此函数的接口限制，不是通用事务组合方案。

三、失败后，幂等键不能被提前占住

把上一段和下面代码放在同一个Python文件即可运行：

connection = new_store()
try:
    assert apply_change(connection, "receive-1", 20) is True
    assert apply_change(connection, "receive-1", 20) is False
    try:
        apply_change(connection, "ship-1", -121)
    except ValueError:
        pass
    else:
        raise AssertionError("应当拒绝库存不足")

    assert connection.execute("SELECT units FROM stock").fetchone()[0] == 120
    assert connection.execute(
        "SELECT COUNT(*) FROM applied_change WHERE event_id = ?",
        ("ship-1",),
    ).fetchone()[0] == 0
    assert apply_change(connection, "ship-1", -120) is True
    assert apply_change(connection, "ship-1", -120) is False
    assert connection.execute("SELECT units FROM stock").fetchone()[0] == 0
finally:
    connection.close()

第一次ship-1, -121失败时，表中没有这个事件，所以随后可以用同一键提交修正后的-120。这是“失败不占键”的演示策略，并非所有幂等协议都必须如此：有些系统会持久保存失败响应，要求修改请求后换键，应由接口合同明确。

最后一次重放发生在库存已经为0时，仍应返回False，而不是再次检查是否能扣120。因为这次请求对应的状态变更早已发生，重放不应该再次执行。

四、本地验证结果与覆盖范围

完整本地检查脚本执行结果：44项检查通过，0失败。上面的短代码只展示核心路径，不等同于完整44项测试清单；完整检查还读取事件表、事务状态，并覆盖下列分组：

44项是断言覆盖数量，不是44种生产故障，更不代表并发正确性或崩溃恢复已经验证。测试使用一个进程、一条连接、内存数据库，没有模拟断电、锁等待、网络超时或多个写入者。

五、不能从这个小实验推出什么

• :memory:随连接关闭消失，演示了事务内的状态变化，没有验证磁盘持久化与进程重启后的去重。
• 事件和库存都在同一数据库事务中；若中间还要调用外部支付、发短信或发送消息，数据库回滚不会自动撤销外部副作用。
• 没有实现认证、租户隔离、多商品、锁等待重试、事件过期、响应缓存或运维备份。
• 不同数据库、驱动及事务模式需要独立验证。不能把本例直接宣布为分布式“恰好一次”交付。
• delta的范围只是为了让输入规则具体可测，不是行业库存规则。空白字符串会被拒绝，但非空键没有被自动裁剪或大小写折叠。

结论

重试安全需要先定义“同一次操作”的身份，再定义失败后的状态。这个小实验给出了两个可直接观察的证据：相同成功事件不重复更新；库存修改失败时，事件标记也不会独立留下。把证据限制在已运行的范围内，比把一次成功执行扩写成“生产级幂等”更可靠。

参考资料：Python 3.12标准库官方文档sqlite3，重点为事务控制、连接上下文管理器、参数占位符及Connection.in_transaction；SQLite官方文档《UPSERT》《CHECK constraints》。本文代码和44项结果来自上述合成实验，不以官方文档代替本地执行证据。