Обработчик получил событие, увеличил баланс и отправил ответ. Ответ потерялся. Отправитель повторил запрос — и баланс увеличился ещё раз. Никакого необычного сбоя: просто одна операция добралась до сервера дважды.

Разберём защиту от такого повтора на Python и SQLite. Без Redis, брокера и веб-фреймворка: только уникальный ключ, транзакция и тесты. Заодно проверим параллельную доставку, падение посередине обработки и неприятный случай, когда под старым идентификатором приходит новое содержимое.

Сначала договоримся, что именно защищаем

Возьмём учебную модель: внешняя система присылает команду начислить пользователю целое количество условных единиц. Это не реализация платёжного сервиса: здесь нет валют, проводок, возвратов и сверки с провайдером. Модель нужна, чтобы не прятать ошибку за бизнес-логикой.

В событии есть четыре поля:

• источник, например provider-a;

• идентификатор события, например evt-1;

• идентификатор счёта;

• сумма в целых единицах.

Наша гарантия узкая: для одной пары «источник, идентификатор события» изменение в локальной базе применяется не более одного раза. Повтор с теми же данными допустим. Повтор с другими данными — ошибка контракта, а не повод молча вернуть успех.

При этом мы предполагаем, что источник уже аутентифицирован, подпись проверена, формат события разобран. Защита от дублей не подтверждает подлинность запроса. Если посторонний может первым прислать любой event_id, уникальный индекс не сделает его событие настоящим.

Почему проверка «мы это уже видели?» недостаточна

Обычный первый вариант выглядит разумно: прочитать таблицу обработанных событий, а если записи нет — выполнить действие и сохранить идентификатор.

Но между чтением и записью есть промежуток. Два обработчика могут пройти его одновременно:

Если уникального ограничения нет, появятся и два начисления, и две записи. Если ограничение есть, вторая вставка завершится ошибкой. Но она не отменит начисление, которое уже закоммичено отдельной транзакцией.

Проблема не в том, что SELECT плохой. Проблема в попытке превратить наблюдение «сейчас записи нет» в эксклюзивное право выполнить действие.

Нужны две вещи одновременно:

1. Уникальное ограничение определяет, кто первым зарегистрировал событие.

2. Регистрация события и изменение баланса фиксируются одной транзакцией.

Тогда проигравший конкурент не делает начисление, а неудачное начисление не оставляет отметку «всё обработано».

Таблица inbox — не архив запросов

В примере будут две таблицы. В accounts лежит баланс, в inbox — уже применённые события. Первичный ключ inbox составной: (source, event_id).

Идентификатор уникален в области, которую обещает отправитель. Разные источники могут прислать одинаковую строку; иногда в ключ нужно включить ещё и клиента.

В inbox сохраняются также счёт и сумма. Без этого обработчик не отличит честный повтор от ситуации «ключ тот же, но теперь начислите другому пользователю».

Сравнивать сырой JSON неудобно: пробелы и порядок полей могут меняться. Здесь сравниваются нормализованные счёт и сумма. В реальном событии нужны все поля, влияющие на результат, включая тип операции и валюту.

Внешний ключ не даёт записать событие для отсутствующего счёта. В SQLite его проверку нужно включать на каждом соединении, поэтому PRAGMA foreign_keys = ON находится в функции подключения, а не только рядом с созданием таблиц.

Один файл, который можно запустить

Ниже полный пример, включая проверки. Сохраните его как webhook_demo.py и выполните python3 webhook_demo.py. Сторонних пакетов нет. Нужна SQLite с поддержкой UPSERT — версия не ниже 3.24.0; фактическую версию библиотеки показывает sqlite3.sqlite_version.

База создаётся во временном каталоге и удаляется после завершения. Скрипт не подключается к существующей базе и ничего не начисляет реальным пользователям.

import sqlite3
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from tempfile import TemporaryDirectory

SCHEMA = """
CREATE TABLE accounts (
    id INTEGER PRIMARY KEY,
    balance INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE inbox (
    source TEXT NOT NULL,
    event_id TEXT NOT NULL,
    account_id INTEGER NOT NULL REFERENCES accounts(id),
    amount INTEGER NOT NULL CHECK(amount > 0),
    PRIMARY KEY (source, event_id)
);
INSERT INTO accounts(id) VALUES (1);
"""


def connect(path):
    db = sqlite3.connect(path, isolation_level=None, timeout=5)
    db.execute("PRAGMA foreign_keys = ON")
    return db


def apply_credit(path, source, event_id, account_id, amount,
                 *, fail_after_insert=False):
    if not isinstance(source, str) or not source:
        raise ValueError("invalid source")
    if not isinstance(event_id, str) or not event_id:
        raise ValueError("invalid event_id")
    if type(account_id) is not int or account_id <= 0:
        raise ValueError("invalid account_id")
    if type(amount) is not int or not 0 < amount <= 10**9:
        raise ValueError("invalid amount")

    db = connect(path)
    try:
        db.execute("BEGIN IMMEDIATE")
        inserted = db.execute("""
            INSERT INTO inbox(source, event_id, account_id, amount)
            VALUES (?, ?, ?, ?)
            ON CONFLICT(source, event_id) DO NOTHING
        """, (source, event_id, account_id, amount)).rowcount
        if inserted == 0:
            old = db.execute("""
                SELECT account_id, amount FROM inbox
                WHERE source = ? AND event_id = ?
            """, (source, event_id)).fetchone()
            if old != (account_id, amount):
                raise ValueError("same key, different payload")
            result = "duplicate"
        else:
            if fail_after_insert:
                raise RuntimeError("simulated failure")
            db.execute("""
                UPDATE accounts SET balance = balance + ?
                WHERE id = ?
            """, (amount, account_id))
            result = "applied"
        db.execute("COMMIT")
        return result
    except BaseException:
        if db.in_transaction:
            db.execute("ROLLBACK")
        raise
    finally:
        db.close()


def main():
    with TemporaryDirectory() as directory:
        path = Path(directory) / "demo.sqlite3"
        db = connect(path)
        db.executescript(SCHEMA)
        db.close()
        assert apply_credit(path, "provider-a", "evt-1", 1, 500) == "applied"
        assert apply_credit(path, "provider-a", "evt-1", 1, 500) == "duplicate"
        try:
            apply_credit(path, "provider-a", "evt-1", 1, 700)
        except ValueError:
            pass
        else:
            raise AssertionError("payload mismatch accepted")
        try:
            apply_credit(path, "provider-a", "evt-2", 1, 500,
                         fail_after_insert=True)
        except RuntimeError:
            pass
        assert apply_credit(path, "provider-a", "evt-2", 1, 500) == "applied"

        def deliver(_):
            return apply_credit(path, "provider-a", "evt-3", 1, 500)

        with ThreadPoolExecutor(max_workers=8) as pool:
            results = list(pool.map(deliver, range(20)))
        assert results.count("applied") == 1
        assert results.count("duplicate") == 19
        db = connect(path)
        assert db.execute("SELECT balance FROM accounts").fetchone() == (1500,)
        assert db.execute("SELECT count(*) FROM inbox").fetchone() == (3,)
        db.close()
        print("OK: balance=1500, inbox=3, applied=1, duplicate=19")


if __name__ == "__main__":
    main()

Что в этом коде действительно важно

Транзакциями управляем явно

isolation_level=None отключает неявное открытие транзакций через старый механизм Python sqlite3. Здесь явно выполняются SQL-команды BEGIN, COMMIT и ROLLBACK. Параметр autocommit не используется, чтобы пример работал и на Python до 3.12, где его ещё не было.

BEGIN IMMEDIATE сразу начинает пишущую транзакцию. SQLite допускает одного писателя одновременно: конкурент ждёт освобождения блокировки или получает ошибку занятости. timeout=5 — ожидание блокировки, не общий дедлайн обработки.

В этом примере первым запросом идёт INSERT, поэтому обычный BEGIN тоже может быть достаточен. IMMEDIATE явно обозначает намерение писать; не переносите этот режим механически в приложение с собственным управлением транзакциями.

Конфликт должен быть ожидаемым и конкретным

ON CONFLICT(source, event_id) DO NOTHING подавляет именно конфликт ключа события. После одиночного INSERT rowcount позволяет понять, вставилась ли строка. Если нет, читаем уже сохранённые бизнес-поля и проверяем совпадение.

Заменять это на широкое INSERT OR IGNORE не стоит. У него другая область подавляемых ошибок: можно случайно проглотить нарушение NOT NULL или CHECK и назвать его повтором. Ошибка суммы — не повторная доставка.

INSERT OR REPLACE тоже не является альтернативой. При конфликте уникальности REPLACE удаляет существующую строку и вставляет новую. Нам же нужна память о первом принятом событии, а не разрешение незаметно переписать её новым содержимым.

Дубликат определяется внутри транзакции

При совпадающем ключе мы не выходим немедленно из функции. Сначала сравниваем поля, затем завершаем транзакцию. Сохранённая строка служит свидетельством того, что локальное изменение уже зафиксировано: отдельно от изменения баланса она в этом коде не коммитится.

Если при повторе нужен прежний ответ API, сохраняйте его в той же транзакции. Из текущего баланса старый ответ восстановить нельзя: могли прийти другие начисления.

Ошибка не оставляет отметку об успехе

Искусственное исключение стоит между INSERT в inbox и UPDATE счёта. Обработчик откатывает транзакцию, поэтому после ошибки нет ни начисления, ни занятого ключа.

Перехват BaseException нужен только для отката: исключение сразу выбрасывается дальше, а не превращается в успешный ответ.

Если процесс исчезнет после COMMIT, но до ответа, повтор увидит inbox. Если до фиксации — восстановление незавершённой транзакции обеспечивает БД. Гарантии при отключении питания зависят также от настроек SQLite и накопителя; пример не отключает журналирование и синхронизацию ради скорости.

Что именно проверили тесты

При запуске получается:

OK: balance=1500, inbox=3, applied=1, duplicate=19

Здесь нет измерения производительности. Двадцать доставок и восемь потоков — параметры небольшого функционального теста, а не утверждение о пропускной способности SQLite.

Проверки покрывают обычный повтор, конфликт содержимого, откат с повторной попыткой и конкурентную доставку. Итоговый баланс и число строк inbox проверяются отдельно.

У каждого вызова своё соединение. Мы не отключаем проверку принадлежности соединения потоку и не пытаемся сделать один общий объект sqlite3 потокобезопасным. База файловая: обычное :memory: с отдельными соединениями создало бы отдельную базу для каждого из них, а не общий стенд.

Тест не имитирует отключение питания, ошибки диска и несколько серверов. Искусственное исключение проверяет откат в приложении, а не восстановление после физического сбоя.

Где эта схема заканчивается

Письмо, HTTP-вызов и БД не становятся одной транзакцией

В UPDATE можно безопасно изменить локальную запись. Но если между INSERT и COMMIT отправить письмо, транзакция SQLite не сможет «отозвать» доставленное письмо после отката.

Если сначала отправить письмо, а потом отметить событие, падение между шагами даст повторную отправку. Если сначала отметить, а потом отправить, падение между шагами даст потерянное письмо.

Обычно здесь появляется transactional outbox: в одной транзакции меняются бизнес-данные, фиксируется inbox и создаётся запись о необходимости отправки. Отдельный воркер читает outbox и доставляет сообщение.

Но сам outbox не обещает ровно одну доставку. Воркер тоже может упасть после отправки, не успев отметить успех. Для эффекта во внешней системе нужна её поддержка идемпотентности, дедупликация на стороне получателя либо допустимость повтора по бизнес-правилам. Транзакция в одной базе не распространяется через сеть по желанию разработчика.

Разные event_id могут означать одну бизнес-операцию

Наш ключ защищает от повторной доставки одного события. Он не запрещает двум разным событиям описывать одно и то же начисление.

Если бизнес-правило звучит «оплаченный заказ активируется один раз», отдельно нужен ключ бизнес-операции, например уникальность активации по заказу. Иначе два разных уведомления об одном заказе пройдут дедупликацию событий вполне законно.

Не стоит и автоматически дедуплицировать по хешу содержимого. Два настоящих начисления могут иметь одинаковые счёт и сумму. Равенство данных ещё не означает равенство намерений.

Удаление истории и блокировки

Удалили запись из inbox — старое событие снова стало новым. Срок хранения должен учитывать период повторной доставки и ручные переотправки. Универсального безопасного срока в днях здесь нет.

Если SQLite не смогла получить блокировку за отведённое время, это не подтверждение ранее выполненной операции. Возвращать в таком случае «уже обработано» нельзя.

На уровне веб-обработчика нужно различать подтверждённый повтор, некорректное содержимое, конфликт ключа и временную недоступность БД. Конкретные HTTP-статусы и правила повторной доставки выбираются по контракту отправителя. Успешный ответ должен означать, что вы действительно приняли ответственность за событие, а не просто поймали исключение.

Что добавить перед реальным запуском

В настоящей интеграции нужны проверка подписи, ограничения размера тела и длины ключей, валидация схемы и мониторинг ошибок БД. Источник события берётся из проверенного контекста интеграции, а не из произвольного поля пользователя.

Целые суммы убирают погрешности float, но не заменяют денежный учёт. В production нужно отдельно ограничить итоговый баланс и исключить переполнение. Для учебного стенда суммы маленькие и заранее известны.

Если у приложения уже есть PostgreSQL, не заводите рядом SQLite ради статьи. Переносится идея: уникальный ключ и локальный эффект в одной транзакции. Блокировки и изоляцию проверяйте для своей СУБД.

Упражнение для учебного проекта

Отправьте evt-1 от provider-b: баланс должен вырасти, ведь источник входит в ключ. Затем попробуйте начислить отсутствующему счёту: inbox не должна пополниться. Наконец, добавьте сохранение результата первого применения, чтобы возвращать его при повторе.

Главное здесь не название паттерна. Запись «событие обработано» должна появляться атомарно с тем изменением, которое она подтверждает. Всё, что выходит за пределы этой транзакции, требует отдельного объяснения — и отдельного теста.