乱序同步把新数据覆盖了：用带版本条件的 UPSERT

2026-11-13 #sqlite#upsert#synchronization

同步服务收到商品版本 8，随后又收到此前延迟的版本 7。如果每次都直接覆盖，商品会倒退到旧状态。在主键冲突分支里比较传入版本与当前版本，让数据库在同一条语句里完成判断与写入。

1. 按乱序送入四条快照

保存为 versioned_upsert_lab.py，运行 python versioned_upsert_lab.py。每条消息包含一份完整状态，版本号由上游为同一个商品单调分配。

from contextlib import closing
import sqlite3

UPSERT = """
INSERT INTO products (id, revision, name) VALUES (?, ?, ?)
ON CONFLICT(id) DO UPDATE
SET revision = excluded.revision, name = excluded.name
WHERE excluded.revision > products.revision
"""

with closing(sqlite3.connect(":memory:")) as db, db:
    db.execute("""
        CREATE TABLE products(
            id TEXT PRIMARY KEY NOT NULL,
            revision INTEGER NOT NULL CHECK(revision >= 0),
            name TEXT NOT NULL
        ) STRICT
    """)
    deliveries = [(8, "new"), (7, "old"), (8, "new"), (9, "latest")]
    expected = [(1, 8, "new"), (0, 8, "new"), (0, 8, "new"), (1, 9, "latest")]
    for (revision, name), result in zip(deliveries, expected):
        changed = db.execute(UPSERT, ("sku-42", revision, name)).rowcount
        current = db.execute("SELECT revision, name FROM products").fetchone()
        assert (changed, *current) == result
        print(f"incoming={revision} changed={changed} stored={current[0]}:{current[1]}")

本地实测输出如下。

incoming=8 changed=1 stored=8:new
incoming=7 changed=0 stored=8:new
incoming=8 changed=0 stored=8:new
incoming=9 changed=1 stored=9:latest

2. 把版本比较放在数据库写入路径上

先查旧版本、再由应用执行无条件 UPDATE，会让判断与写入之间出现并发窗口。这里的 WHERE 在处理主键冲突时判断被更新的那一行。rowcount 为 0 代表本次没有改变持久状态。若业务需要区分重放、过时事件和同版本不同内容，应额外记录或核验版本与内容的对应关系。

消息类型	接入前要确认
完整快照	版本 9 能直接替换版本 7，缺失的版本 8 不影响最终状态
增量事件	跳过中间版本会丢失操作，需要按序消费或重新拉取快照
删除事件	保留带版本的删除标记，否则迟到的旧快照可能重新插入已删除记录
多写入方	必须约定版本如何产生与比较，避免同一对象两个不同事件使用相同版本

同一个事务还可以记录供后续发送的事件。已经执行的邮件或外部 API 调用，需要各自的幂等机制。

小结

对完整快照，用主键确定实体，用版本决定能否覆盖。测试序列应包含旧事件、重复事件和一条更高版本，并断言最终数据。

数据来源与检索时间

检索时间 2026-11-13。

来源	URL	对应内容
SQLite UPSERT	https://www.sqlite.org/lang_upsert.html	excluded、冲突目标与更新条件
SQLite STRICT Tables	https://www.sqlite.org/stricttables.html	严格列类型

核验说明

本地 SQLite 3.45.1 实测乱序和重复投递对数据库状态的影响。多生产者的版本分配需要在上游确定。

#sqlite #upsert #synchronization