Exactly-once в Python: почему воркер обрабатывает сообщение дважды

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

Разберём, почему очередь физически не может гарантировать exactly-once доставку, и как в Python-обработчике добиться того, чтобы повторная обработка сообщения не приводила к повторному эффекту. Примеры на Python 3.11, брокер — RabbitMQ 3.13 через pika, хранилище идемпотентности — PostgreSQL 16.

Почему exactly-once на уровне доставки недостижим

Между «обработал» и «подтвердил обработку» всегда есть зазор. Что бы вы ни делали, остаётся два порядка операций, и оба ломаются:

• ack до обработки. Подтвердили сообщение, потом упали в бизнес-логике. Брокер считает сообщение доставленным, повторной попытки не будет. Это at-most-once: сообщение может потеряться.
• ack после обработки. Обработали, упали до ack. Брокер переотдаёт сообщение. Это at-least-once: сообщение может обработаться дважды.

Третьего варианта на уровне транспорта нет. Атомарно «выполнить сайд-эффект и подтвердить сообщение» невозможно, если сайд-эффект и подтверждение живут в разных системах — в вашей БД и в брокере.

Exactly-once — это не про доставку. Это про эффект. Сообщение можно доставить сколько угодно раз, важно, чтобы результат применился ровно один раз.

Практический вывод: настраивайте транспорт на at-least-once (ack после успешной обработки) и делайте сам обработчик идемпотентным. Тогда повторы безопасны.

Что даёт exactly-once сам брокер

Kafka с включённой транзакционностью (enable.idempotence у продюсера и транзакции) обеспечивает exactly-once только внутри Kafka — от продюсера до консьюмера и обратно в топик Kafka. Как только вы пишете в PostgreSQL или дёргаете платёжный API, гарантия заканчивается на границе Kafka. RabbitMQ такого не обещает вовсе. Так что идемпотентность обработчика нужна в любом случае.

Идемпотентный ключ: основа всего

Каждому сообщению нужен стабильный идентификатор, который не меняется при повторной доставке. Не генерируйте его в обработчике — берите из сообщения. Варианты, откуда взять ключ:

• явный message_id, проставленный продюсером при отправке;
• бизнес-ключ операции — например, order_id + тип операции, если по смыслу операция должна выполниться ровно раз на заказ;
• хеш от полезной нагрузки — хрупкий вариант, ломается при любом изменении сериализации, годится только как крайняя мера.

Дальше — таблица обработанных сообщений и проверка перед выполнением сайд-эффекта.

Проверка и вставка в одной транзакции

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

import psycopg2
from psycopg2.errors import UniqueViolation

# PostgreSQL 16, psycopg2 2.9
# Таблица создана заранее:
# CREATE TABLE processed_messages (
#     message_id TEXT PRIMARY KEY,
#     processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
# );

def handle_message(conn, message_id: str, payload: dict) -> bool:
    """Возвращает True, если сообщение обработано сейчас,
    False — если уже было обработано ранее."""
    with conn:  # транзакция: коммит при выходе, откат при исключении
        with conn.cursor() as cur:
            try:
                cur.execute(
                    "INSERT INTO processed_messages (message_id) VALUES (%s)",
                    (message_id,),
                )
            except UniqueViolation:
                # ключ уже есть — сообщение обработано, повтор игнорируем
                return False
            # Сайд-эффект в ту же БД и ту же транзакцию:
            apply_business_effect(cur, payload)
    return True

Ключевая деталь: INSERT ключа и apply_business_effect — в одной транзакции. Если бизнес-логика упадёт, откатится и вставка ключа, и сообщение честно переобработается. Если всё прошло — оба изменения зафиксируются атомарно.

Проблема двух хранилищ: dual write

Схема выше работает, пока сайд-эффект пишется в ту же БД, где лежит таблица ключей. Как только эффект уходит во внешнюю систему — платёжный шлюз, другой сервис, письмо — атомарность рвётся. Вы не можете закоммитить транзакцию PostgreSQL и вызов HTTP API одновременно.

Три рабочих подхода

Для платежей самый надёжный вариант — комбинация: свой ключ идемпотентности в таблице плюс передача Idempotency-Key провайдеру. Даже если ваша транзакция и вызов API рассинхронизируются, провайдер не спишет дважды по одному ключу.

Outbox в двух словах

# Схематично: эффект и намерение отправить пишутся атомарно
with conn:
    with conn.cursor() as cur:
        cur.execute(
            "INSERT INTO processed_messages (message_id) VALUES (%s)",
            (message_id,),
        )
        # бизнес-изменение
        cur.execute("UPDATE orders SET status = %s WHERE id = %s",
                    ("paid", payload["order_id"]))
        # намерение отправить событие наружу — в ту же транзакцию
        cur.execute(
            "INSERT INTO outbox (event_type, body) VALUES (%s, %s)",
            ("payment.confirmed", json.dumps(payload)),
        )
# Отдельный процесс читает outbox и публикует в брокер,
# помечая строки отправленными. Повторная отправка — at-least-once,
# получатель обязан быть идемпотентным.

Настройка транспорта под at-least-once

Чтобы идемпотентность имела смысл, брокер должен переотдавать неподтверждённые сообщения. В pika для этого:

1. Отключите авто-ack: basic_consume(..., auto_ack=False). Иначе сообщение считается доставленным сразу после получения — теряется при падении.
2. Вызывайте ch.basic_ack(delivery_tag) только после успешной обработки.
3. При обрабатываемой ошибке — ch.basic_nack(delivery_tag, requeue=True), чтобы сообщение вернулось в очередь.
4. Ограничьте basic_qos(prefetch_count=1) на время отладки, чтобы воркер не хватал пачку сообщений, которые потеряет при падении.

Проверка, что настроено верно: убейте воркер (kill -9) в момент обработки. Сообщение должно вернуться в очередь и обработаться повторно, а таблица processed_messages — не дать применить эффект дважды.

Что ломается на практике

• Растущая таблица ключей. processed_messages пухнет бесконечно. Нужен ретеншен: удаляйте записи старше окна, за которое брокер точно не переотдаст сообщение (обычно часы-сутки). Партиционирование по дате упрощает очистку.
• Хеш вместо стабильного ключа. Если ключ — хеш тела, а продюсер поменял порядок полей в JSON, хеш изменится и повтор пройдёт как новое сообщение. Берите явный message_id.
• ack после долгой обработки. Если обработка дольше, чем таймаут доставки RabbitMQ (consumer_timeout, по умолчанию 30 минут в 3.13), брокер закроет канал и переотдаст сообщение ещё до вашего ack. Дробите длинные задачи или поднимайте таймаут осознанно.
• Идемпотентность без транзакции. Проверка ключа отдельным запросом, а вставка — потом: гонка между воркерами. Только атомарная вставка с уникальным ограничением.

Что делать дальше

Начните с малого: добавьте message_id в продюсер и таблицу processed_messages с уникальным ключом. Это закрывает 80% повторов при at-least-once. Затем аудит внешних вызовов: для каждого сайд-эффекта наружу решите, идемпотентен ли он на стороне провайдера, и если нет — заводите outbox. Метрика, за которой стоит следить в проде: доля сообщений, отсечённых по дублю ключа (те самые UniqueViolation). Резкий рост означает либо частые падения воркеров, либо неверно настроенный consumer_timeout.