---
title: "乱序同步把新数据覆盖了：用带版本条件的 UPSERT"
date: 2026-11-13
slug: sqlite-versioned-upsert
canonical: https://kernelranch.com/blog/2026/11/sqlite-versioned-upsert
markdown_url: https://kernelranch.com/blog/2026/11/sqlite-versioned-upsert.md
tags: ["sqlite", "upsert", "synchronization"]
---同步服务收到商品版本 8，随后又收到此前延迟的版本 7。如果每次都直接覆盖，商品会倒退到旧状态。在主键冲突分支里比较传入版本与当前版本，让数据库在同一条语句里完成判断与写入。

## 1. 按乱序送入四条快照

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

```python
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]}")
```

本地实测输出如下。

```text
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 实测乱序和重复投递对数据库状态的影响。多生产者的版本分配需要在上游确定。