Идемпотентность фоновых задач в Python: что ревьюить

Воркер Celery падает по OOM в момент, когда задача уже отправила письмо, но ещё не успела записать в базу флаг notified = True. Брокер (RabbitMQ или Redis) видит, что подтверждения (ack) не было, и отдаёт сообщение другому воркеру. Задача выполняется второй раз целиком. Клиент получает два письма, а если внутри был перевод денег — списание дублируется.

Это не редкий сбой, а нормальный режим работы очередей: почти все брокеры дают гарантию at-least-once, то есть «доставим хотя бы один раз, но можем и больше». Значит, задача обязана быть идемпотентной — повторный запуск с теми же входными данными не должен менять результат. На ревью фоновых задач это первое, что нужно проверять. Разбираем, где именно ломается идемпотентность в Python-коде и как переписывать проблемные места.

Почему повтор — это норма, а не баг

Причин для повторного запуска больше, чем кажется:

• воркер упал после выполнения работы, но до отправки ack брокеру;
• сработал visibility_timeout в Redis — задача выполнялась дольше таймаута, и брокер счёл её потерянной;
• сеть между воркером и брокером моргнула, ack не дошёл;
• кто-то вручную сделал retry или перезапустил зависшую задачу через Flower;
• деплой убил воркер в середине задачи через SIGTERM.

Настройка task_acks_late=True в Celery, которую часто советуют для надёжности, делает ситуацию чаще: ack отправляется после выполнения, поэтому любой сбой в процессе гарантированно приведёт к повтору. Это правильная настройка — но именно она превращает идемпотентность из «хорошо бы» в обязательное требование.

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

Что ревьюить в задаче: три опасных участка

1. Побочные эффекты без ключа идемпотентности

Самое частое замечание. Задача делает внешний вызов — платёж, письмо, вебхук — без ключа, который позволил бы принимающей стороне отбросить дубль.

# Было: повтор задачи спишет деньги дважды
@app.task(acks_late=True)
def charge_order(order_id):
    order = Order.objects.get(id=order_id)
    payment_gateway.charge(
        amount=order.total,
        card_token=order.card_token,
    )
    order.status = "paid"
    order.save()

Формулировка замечания на ревью: «Вызов payment_gateway.charge не передаёт idempotency key. При повторе задачи платёж пройдёт второй раз». У большинства платёжных API (Stripe, YooKassa и других) есть заголовок или параметр идемпотентности — сверьтесь с документацией конкретного шлюза, имя параметра различается.

# Стало: ключ детерминированный, привязан к заказу
@app.task(acks_late=True)
def charge_order(order_id):
    order = Order.objects.get(id=order_id)
    if order.status == "paid":
        return  # уже оплачено, выходим сразу
    # ключ одинаков при любом повторе этой задачи для этого заказа
    idem_key = f"charge-order-{order_id}"
    payment_gateway.charge(
        amount=order.total,
        card_token=order.card_token,
        idempotency_key=idem_key,  # имя параметра — см. доки шлюза
    )
    order.status = "paid"
    order.save(update_fields=["status"])

2. Ключ идемпотентности, зависящий от времени или случайности

Классическая ошибка — сгенерировать ключ через uuid4() или datetime.now() внутри задачи. При повторе значение будет другим, и защита не сработает.

# Плохо: при повторе ключ другой, дубль пройдёт
idem_key = str(uuid.uuid4())

# Плохо: между двумя запусками время сдвинулось
idem_key = f"notify-{int(time.time())}"

# Хорошо: ключ выводится из неизменных входных данных
idem_key = f"notify-user-{user_id}-order-{order_id}"

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

3. Неатомарная связка «внешний эффект + запись в базу»

Даже с ключом идемпотентности остаётся окно: письмо ушло, а флаг в базе не записался из-за падения. Проблему решают не «ещё одной проверкой», а сдвигом порядка операций и транзакцией.

# Django 5.x: сначала фиксируем намерение в БД,
# потом делаем внешний вызов после коммита транзакции
from django.db import transaction

@app.task(acks_late=True)
def send_receipt(order_id):
    with transaction.atomic():
        receipt, created = Receipt.objects.get_or_create(
            order_id=order_id,
            defaults={"status": "pending"},
        )
        if not created and receipt.status == "sent":
            return  # уже отправляли

    # внешний вызов вне транзакции: держать соединение с БД
    # открытым на время сетевого запроса — плохая идея
    email_service.send(order_id=order_id,
                       idempotency_key=f"receipt-{order_id}")

    Receipt.objects.filter(order_id=order_id).update(status="sent")

Здесь get_or_create с уникальным ограничением на order_id выступает как замок: два параллельных воркера не создадут две записи, второй получит created=False. Уникальный индекс в БД — обязателен, иначе гонка проскочит.

Дедупликация на входе: таблица обработанных сообщений

Когда внешний API не поддерживает ключ идемпотентности, защиту строят на своей стороне — через таблицу уже обработанных задач.

# Отдельная таблица processed_tasks с уникальным ключом
@app.task(acks_late=True)
def process_webhook(event_id, payload):
    try:
        with transaction.atomic():
            ProcessedTask.objects.create(task_key=f"webhook-{event_id}")
    except IntegrityError:
        # запись уже есть — задачу с этим event_id уже обработали
        return

    # сюда попадаем ровно один раз на event_id
    handle_event(payload)

Тонкость, которую ревью пропускает чаще всего: если handle_event упадёт после вставки в ProcessedTask, повтор увидит запись и молча выйдет, не доделав работу. Для критичных операций запись факта обработки и саму работу нужно держать в одной транзакции — либо использовать шаблон «сначала pending, в конце done», как в примере с чеком выше.

Сравнение подходов

Где всё это ломается

• Redis-лок с TTL меньше времени задачи. Лок протух, пока задача работала, второй воркер зашёл — и оба выполняются параллельно. TTL лока должен быть с запасом больше максимального времени выполнения, а лучше продлеваться.
• Retry с изменёнными аргументами. Если при self.retry() в задачу подставляются новые данные, ключ идемпотентности едет. Ключ должен считаться от исходных аргументов, не от текущих.
• Частичный откат. Задача сделала три из пяти внешних вызовов и упала. Повтор с проверкой «status == done» не поможет — статуса «done» ещё нет. Разбивайте на отдельные идемпотентные шаги или делайте каждый вызов самопроверяющимся.
• Таблица дедупликации без очистки. processed_tasks растёт бесконечно, индекс распухает, вставки замедляются. Нужна фоновая чистка записей старше разумного окна повторов.

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

Пройдитесь по своим фоновым задачам с одним вопросом к каждой: что произойдёт при втором запуске с теми же аргументами. Задачи, которые трогают деньги, письма и внешние вебхуки, проверяйте в первую очередь. Включите task_acks_late=True вместе с task_reject_on_worker_lost=True — но только после того, как задачи станут идемпотентными, иначе получите гарантированные дубли вместо случайных. В проде смотрите на счётчик повторных выполнений и на дубликаты в таблицах результатов: если растут — где-то идемпотентность не держит.