Как реализовать идемпотентный consumer для очереди с гарантией at-least-once?

Короткий ответ

Идемпотентный consumer хранит признак уже обработанных сообщений (по уникальному ключу) и, получив дубликат, пропускает повторную обработку, возвращая тот же эффект, что и при первой обработке.

Как это работает подробнее

При получении сообщения consumer сначала проверяет, есть ли запись об обработке message_id; если есть — подтверждает получение без повторного эффекта; если нет — выполняет обработку и атомарно фиксирует факт обработки вместе с результатом.

Пример кода

example.pyКопировать

def handle(message):
    with db.transaction():
        inserted = db.execute(
            "INSERT INTO processed_messages (message_id) VALUES (%s) ON CONFLICT DO NOTHING RETURNING message_id",
            [message.id],
        )
        if not inserted:
            return  # уже обработано
        apply_business_effect(message)

Что использовать на практике

• Практика. Ключ идемпотентности должен быть стабильным и уникальным для бизнес-операции (order_id + event_type), а не техническим delivery tag брокера, который меняется при каждой redelivery.

Подводные камни

• Риск. Хранить «обработано» только в памяти процесса (при перезапуске теряется); проверять и фиксировать факт обработки в двух разных нетранзакционных шагах, оставляя окно гонки при параллельной обработке дублей.