Подтверждение после фиксации эффекта

У каталога и аудита теперь отдельные очереди. Следующий вопрос — когда брокер вправе считать доставку законченной. В этом уроке обработчик отправляет acknowledgement только после сохранения своего эффекта. Результат — ясный переход ответственности от очереди к локальному хранилищу и понимание того, почему потеря соединения всё равно может привести к повтору.

Используем полный consumer.py и store.py итогового проекта. В архиве уже есть защита от дубликатов следующих уроков; здесь рассматриваем порядок обработки и ack. Consumer, SQLite и обрывы соединений автор не запускал. Официальное руководство по acknowledgements и confirms определяет серверные состояния доставки.

Доставка ещё не означает завершённую работу

Когда RabbitMQ передал сообщение consumer, тот может находиться в середине операции. Если удалить сообщение из очереди сразу, а процесс упадёт до записи каталога, работа потеряется. Поэтому auto_ack=False оставляет доставку неподтверждённой до явного решения приложения. У брокера появляется информация о том, что текущий consumer ещё отвечает за неё.

В callback сначала разбираем тело, проверяем договор и вызываем apply_event. Эта функция сохраняет предметный результат в своей SQLite-базе. Только после её успешного возврата отправляем ack. Каталог обновляет проекцию материала, аудит сохраняет конверт события. У каждого получателя собственный момент завершения, даже когда eventId одинаков.

Фрагмент полной функции callback; переменные name, event, channel и method уже определены в consumer.py:

result = apply_event(name, event)
print(name, event['eventId'], result, flush=True)
channel.basic_ack(delivery_tag=method.delivery_tag)

delivery_tag идентифицирует транспортную доставку внутри канала. Он не является eventId и не должен сохраняться как предметный ключ дубликата. При повторе того же события broker выдаёт другую доставку с собственным tag. Подтверждать её нужно через тот канал, к которому tag относится, а не через случайный вновь открытый connection.

Если пользовательская запись не завершилась, переход к ack не выполняется. Позднее временную ошибку направим в отдельную retry-очередь, а постоянную — в dead-очередь. Этот порядок важен и для перенаправления: исходное сообщение подтверждается только после принятия новой ответственной копии выбранным назначением.

Unacked и ограничение незавершённых доставок

Очередь содержит готовые сообщения и доставки, находящиеся у consumer без ack. Когда растёт ready, работа ждёт свободного получателя; когда растёт unacked, получатели уже взяли её, но не завершили. Эти состояния помогают различить отсутствующий процесс и зависшую обработку. Одно общее число «длина очереди» не объясняет причину отставания.

В нашем обработчике задаём prefetch_count=1. Брокер не должен выдавать этому consumer следующий объём неподтверждённой работы сверх установленного предела. Для последовательного BlockingConsumer это помогает удерживать понятное состояние: один текущий факт на обработчика. Это ограничение не является глобальным пределом всей очереди или всех процессов.

При двух процессах каталога каждый имеет собственный prefetch. В сумме может находиться несколько незавершённых доставок. Увеличение числа процессов также требует безопасной конкуренции за предметное хранилище. Наш локальный SQLite-пример сериализует транзакции, но не представляет распределённую базу с независимыми узлами и высокой пропускной способностью.

Prefetch не равен числу рабочих потоков приложения. Если callback сам создаёт параллельные задачи и возвращается до их завершения, управление acknowledgements нужно согласовать с этим жизненным циклом. В этой серии не добавляем скрытые фоновые потоки: эффект фиксируется в последовательном коротком вызове, после которого явно отправляется ack.

Падение до и после записи

Рассмотрим три момента. Процесс падает до начала SQLite-транзакции: предметного эффекта нет, сообщение остаётся неподтверждённым и может вернуться другому consumer. Процесс падает после commit, но до ack: эффект уже сохранён, а RabbitMQ этого не знает. Процесс падает после принятого ack: брокер может убрать доставку, рассчитывая на выполненный эффект.

Наиболее труден второй случай. Повтор будет содержать тот же eventId, но новую транспортную доставку. Если обработчик просто ещё раз выполняет вставку или увеличивает счётчик, эффект повторится. Поэтому manual ack необходим для защиты от потери, но недостаточен для защиты от дубликата. Следующий слой — идемпотентная запись вместе с квитанцией.

Ожидаемая схема состояний:

До commit → доставка unacked, локального эффекта нет
После commit, до ack → эффект есть, доставка ещё может повториться
После принятого ack → брокер завершает эту доставку
Повтор → новый delivery_tag, прежний eventId

Это описание возможных исходов, а не выполненный эксперимент с аварией. Для будущего наблюдения нужна отдельная контролируемая среда и точка остановки. Нельзя утверждать точное число повторов без исполнения: сеть, время обнаружения обрыва и состояние broker влияют на последовательность.

Ack подтверждает выбранный предметный результат

Для каталога результатом является сохранённая локальная проекция. Для аудита — запись факта. Если в будущем consumer отправляет письмо через внешний сервис, успешный HTTP-ответ сам по себе потребует предметного договора повторения. Локальная inbox-транзакция не сможет атомарно включить произвольное внешнее письмо. Нельзя перенести гарантию нашего SQLite-эффекта на любой побочный эффект.

Также нельзя подтверждать сообщение в блоке finally независимо от успеха. Тогда ошибка базы и успешная обработка приводили бы к одному и тому же удалению доставки. Если нужен постоянный отказ, его нужно осознанно сохранить в dead-очереди или другом хранилище, после чего передать ответственность. «Всегда ack, чтобы не мешать очереди» скрывает потерю работы.

Печать журнала перед ack является наблюдением, а не частью атомарной записи. Журнал может появиться, а подтверждение не дойти. Поэтому надпись updated не доказывает отсутствие будущего повтора. Сопоставляйте её с устойчивым eventId и состоянием inbox, а не с предположением, что каждая строка консоли соответствует единственной доставке.

Для будущего запуска читатель использует python consumer.py catalog и отдельный процесс audit. Если один остановлен, его очередь продолжает ждать собственной обработки. Это помогает увидеть независимость обязанностей без необходимости создавать внешние сервисы. Но одноузловое хранение остаётся ограничением учебного окружения и не превращается в отказоустойчивый кластер от количества процессов Python.

Значение multiple у acknowledgement также меняет смысл операции: подтверждение может охватывать несколько более ранних доставок канала. В нашей последовательной реализации используем обычный ack одного delivery_tag. Если перейти к параллельной обработке и массовым подтверждениям, нужно доказать завершение каждой включённой работы. Нельзя подтверждать диапазон только потому, что последняя доставка закончилась раньше остальных.

Выполните мысленное упражнение: broker уже принял сообщение, catalog сохранил эффект, но ack потерялся. Укажите, какое поле должно остановить повторный эффект и почему delivery_tag для этого не подходит. Далее разберём publisher confirms, которые закрывают другой участок пути — между издателем и брокером.

Оглавление серии