SQLite 条件更新实战：配送任务交接时如何拦住并发覆盖

UPDATE task SET owner_id = ? WHERE id = 1;

UPDATE task
SET owner_id = ?, version = version + 1
WHERE id = 1
  AND owner_id = ?
  AND version = ?
  AND state = 'delivering';

import sqlite3
import tempfile
from contextlib import closing
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from threading import Barrier

SCHEMA = """
CREATE TABLE task (
  id INTEGER PRIMARY KEY,
  owner_id INTEGER NOT NULL,
  state TEXT NOT NULL,
  version INTEGER NOT NULL
);
CREATE TABLE handover_event (
  task_id INTEGER NOT NULL,
  old_owner INTEGER NOT NULL,
  new_owner INTEGER NOT NULL,
  version INTEGER NOT NULL,
  UNIQUE(task_id, version)
);
INSERT INTO task VALUES (1, 101, 'delivering', 1);
"""

def connect(db):
    return closing(sqlite3.connect(db, timeout=5, isolation_level=None))

def transfer(db, new_owner, fail=False, gate=None):
    with connect(db) as conn:
        # Both contenders intentionally read the same old version.
        old_owner, version = conn.execute(
            "SELECT owner_id, version FROM task WHERE id=1"
        ).fetchone()
        if gate:
            gate.wait(timeout=5)
        conn.execute("BEGIN IMMEDIATE")
        try:
            changed = conn.execute(
                """UPDATE task SET owner_id=?, version=version+1
                   WHERE id=1 AND owner_id=? AND version=?
                     AND state='delivering'""",
                (new_owner, old_owner, version)
            ).rowcount
            if changed != 1:
                conn.execute("ROLLBACK")
                return "conflict"
            if fail:
                raise RuntimeError("injected before event insert")
            conn.execute(
                "INSERT INTO handover_event VALUES (1, ?, ?, ?)",
                (old_owner, new_owner, version + 1)
            )
            conn.execute("COMMIT")
            return "ok"
        except Exception:
            conn.execute("ROLLBACK")
            raise

def main():
    with tempfile.TemporaryDirectory() as folder:
        db = str(Path(folder) / "demo.db")
        with connect(db) as conn:
            conn.executescript(SCHEMA)
        gate = Barrier(2)
        with ThreadPoolExecutor(max_workers=2) as pool:
            jobs = [pool.submit(transfer, db, owner, gate=gate)
                    for owner in (202, 303)]
            results = sorted(job.result(timeout=10) for job in jobs)
        assert results == ["conflict", "ok"]
        with connect(db) as conn:
            assert conn.execute(
                "SELECT COUNT(*) FROM handover_event"
            ).fetchone()[0] == 1
            stale = conn.execute(
                """UPDATE task SET state='delivered', version=version+1
                   WHERE id=1 AND owner_id=101 AND version=1"""
            ).rowcount
            assert stale == 0
            before = conn.execute("SELECT * FROM task").fetchone()
        try:
            transfer(db, 404, fail=True)
        except RuntimeError:
            pass
        else:
            raise AssertionError("fault injection did not run")
        with connect(db) as conn:
            assert conn.execute("SELECT * FROM task").fetchone() == before
            assert conn.execute(
                "SELECT COUNT(*) FROM handover_event"
            ).fetchone()[0] == 1
        print("competing transfers:", results)
        print("handover events: 1")
        print("stale owner writes:", stale)
        print("injected failure: task and event rolled back")

if __name__ == "__main__":
    main()

python demo.py

competing transfers: ['conflict', 'ok']
handover events: 1
stale owner writes: 0
injected failure: task and event rolled back