队列定义投递，消费者仍须保护副作用

from contextlib import closing
from pathlib import Path
from tempfile import TemporaryDirectory
import sqlite3


def apply(db, message_id, amount, interrupt=False):
    db.execute('BEGIN IMMEDIATE')
    try:
        prior = db.execute('SELECT amount FROM inbox WHERE id=?', (message_id,)).fetchone()
        if prior is not None:
            if prior[0] != amount:
                raise ValueError('same identity, different payload')
            db.commit()
            return False
        db.execute('INSERT INTO inbox VALUES (?, ?)', (message_id, amount))
        db.execute('UPDATE totals SET amount=amount+? WHERE id=1', (amount,))
        if interrupt:
            raise RuntimeError('controlled failure before commit')
        db.commit()
        return True
    except BaseException:
        db.rollback()
        raise


with TemporaryDirectory(prefix='queue-sink-') as directory:
    path = Path(directory) / 'sink.db'
    with closing(sqlite3.connect(path, isolation_level=None)) as db:
        db.execute('CREATE TABLE inbox(id TEXT PRIMARY KEY, amount INTEGER NOT NULL)')
        db.execute('CREATE TABLE totals(id INTEGER PRIMARY KEY, amount INTEGER NOT NULL)')
        db.execute('INSERT INTO totals VALUES (1, 0)')
        try:
            apply(db, 'work-1', 100, interrupt=True)
        except RuntimeError:
            pass
        else:
            raise AssertionError('controlled failure did not occur')
        assert db.execute('SELECT count(*) FROM inbox').fetchone() == (0,)
        assert db.execute('SELECT amount FROM totals').fetchone() == (0,)
        assert apply(db, 'work-1', 100)
        assert apply(db, 'work-2', 50)
    # Close and reopen the real database rather than retaining an in-memory set.
    with closing(sqlite3.connect(path, isolation_level=None)) as db:
        assert not apply(db, 'work-1', 100)
        try:
            apply(db, 'work-1', 999)
        except ValueError:
            pass
        else:
            raise AssertionError('conflicting payload reused the same identity')
        assert db.execute('SELECT count(*) FROM inbox').fetchone() == (2,)
        assert db.execute('SELECT amount FROM totals').fetchone() == (150,)
        assert not db.in_transaction
    print('rollback kept marker and effect together; reopened retry changed nothing; total=150')