多商户外卖系统二开版本评审：配置修订号冲突校验

import json
import sqlite3
import tempfile
from pathlib import Path


def change(connection, request_id, key, expected_revision, rate_bps):
    if not isinstance(request_id, str) or not request_id.strip():
        raise ValueError('INVALID_REQUEST')
    if not isinstance(key, str) or not key.strip():
        raise ValueError('INVALID_KEY')
    if type(expected_revision) is not int or not 0 <= expected_revision < 9223372036854775807:
        raise ValueError('INVALID_REVISION')
    if type(rate_bps) is not int or not 0 <= rate_bps <= 10000:
        raise ValueError('INVALID_RATE')
    connection.execute('BEGIN IMMEDIATE')
    try:
        prior = connection.execute(
            'SELECT region_key, expected_revision, rate_bps, result_revision FROM requests WHERE request_id=?',
            (request_id,),
        ).fetchone()
        if prior:
            if prior[:3] != (key, expected_revision, rate_bps):
                raise ValueError('REQUEST_CONFLICT')
            connection.commit()
            return prior[3]
        connection.execute(
            'INSERT OR IGNORE INTO rules(region_key, revision, rate_bps) VALUES (?, 0, 0)', (key,),
        )
        updated = connection.execute(
            'UPDATE rules SET revision=revision+1, rate_bps=? WHERE region_key=? AND revision=?',
            (rate_bps, key, expected_revision),
        )
        if updated.rowcount != 1:
            raise ValueError('REVISION_CONFLICT')
        result_revision = expected_revision + 1
        connection.execute(
            'INSERT INTO requests VALUES (?, ?, ?, ?, ?)',
            (request_id, key, expected_revision, rate_bps, result_revision),
        )
        connection.commit()
        return result_revision
    except Exception:
        connection.rollback()
        raise


def reject(call, expected):
    try:
        call()
    except ValueError as error:
        assert str(error) == expected
    else:
        raise AssertionError('request unexpectedly accepted')


with tempfile.TemporaryDirectory() as directory:
    database = Path(directory) / 'rules.sqlite'
    first = sqlite3.connect(database, timeout=5)
    second = sqlite3.connect(database, timeout=5)
    try:
        first.executescript('''
            CREATE TABLE rules (
                region_key TEXT PRIMARY KEY NOT NULL,
                revision INTEGER NOT NULL CHECK(revision >= 0),
                rate_bps INTEGER NOT NULL CHECK(rate_bps BETWEEN 0 AND 10000)
            );
            CREATE TABLE requests (
                request_id TEXT PRIMARY KEY NOT NULL,
                region_key TEXT NOT NULL REFERENCES rules(region_key),
                expected_revision INTEGER NOT NULL,
                rate_bps INTEGER NOT NULL,
                result_revision INTEGER NOT NULL
            );
        ''')
        assert change(first, 'R1', 'A', 0, 500) == 1
        reject(lambda: change(second, 'R2', 'A', 0, 900), 'REVISION_CONFLICT')
        assert second.execute('SELECT revision,rate_bps FROM rules WHERE region_key=?', ('A',)).fetchone() == (1, 500)
        assert change(second, 'R3', 'A', 1, 600) == 2
        assert change(first, 'R1', 'A', 0, 500) == 1
        reject(lambda: change(first, 'R1', 'A', 0, 700), 'REQUEST_CONFLICT')
        assert first.execute('SELECT COUNT(*) FROM requests').fetchone()[0] == 2
        reject(lambda: change(first, 'R4', 'A', 2, 10001), 'INVALID_RATE')
        reject(lambda: change(first, 'R4', 'A', 2, 0.5), 'INVALID_RATE')
        reject(lambda: change(first, 'R4', 'A', True, 500), 'INVALID_REVISION')
        reject(lambda: change(first, 'R4', '', 0, 500), 'INVALID_KEY')
        reject(lambda: change(first, '', 'A', 2, 500), 'INVALID_REQUEST')
        reject(lambda: change(second, 'R5', 'B', 1, 300), 'REVISION_CONFLICT')
        assert first.execute('SELECT COUNT(*) FROM rules WHERE region_key=?', ('B',)).fetchone()[0] == 0
        assert change(second, 'R6', 'B', 0, 0) == 1
        assert change(first, 'R7', 'B', 1, 10000) == 2
        output = dict(first.execute('SELECT region_key,revision FROM rules'))
        print(json.dumps({'assertions': 16, 'revisions': output, 'connections': 2, 'staleWrite': 'REVISION_CONFLICT', 'retryResult': 1}))
    finally:
        second.close()
        first.close()