Как обрабатывать дубликаты сообщений при асинхронной записи из очереди в базу

Что проверяют

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

Как отвечать

Суть ответа: идемпотентный консьюмер, где дедупликацию гарантирует база, а не проверка в коде.

• ключ идемпотентности: у сообщения есть стабильный ID, который задаёт отправитель один раз, а не генерирует заново при каждом ретрае. Подойдёт и естественный ключ: номер заказа плюс тип события;
• уникальное ограничение: INSERT ... ON CONFLICT (message_id) DO NOTHING в PostgreSQL. Через RowsAffected() из database/sql или CommandTag.RowsAffected() в pgx вы узнаёте, была ли вставка;
• inbox-таблица: ID обработанного сообщения записывается в той же транзакции, что и бизнес-изменение. Либо закоммитилось всё, либо ничего;
• подтверждение после коммита: в Kafka смещение фиксируется только после успешной транзакции (в segmentio/kafka-go это FetchMessage и затем CommitMessages, а не ReadMessage с автокоммитом в группе). Упали между ними, получите сообщение повторно, и его отсечёт ключ;
• порядок: если событие может прийти раньше предыдущего, храните версию и обновляйте через UPDATE ... WHERE version < $1.

Для батчей: pgx.CopyFrom не умеет ON CONFLICT, поэтому либо COPY во временную таблицу и INSERT ... SELECT ... ON CONFLICT, либо вставка массивами через unnest. Таблицу обработанных ID чистите по TTL, удобно партициями по дню.

Где спотыкаются

Делают SELECT EXISTS, а потом INSERT. При двух воркерах оба увидят «записи нет», и без уникального индекса в базе окажутся два экземпляра. Транзакция на уровне READ COMMITTED от этого не спасает.

Держат дедупликацию в Redis через SETNX отдельно от записи в базу. Ключ поставили, запись упала, и сообщение потеряно навсегда.

Путают exactly-once в Kafka с гарантией для базы. Транзакции Kafka покрывают только цепочку «прочитал из топика, записал в топик», внешнюю базу они не охватывают.