API публикации и transactional outbox

Издатель файла показывает обмен, но ещё не связывает событие с действительным изменением материала. Если API сначала сохранит публикацию, а потом процесс упадёт перед отправкой, RabbitMQ о ней не узнает. В этом уроке добавим transactional outbox. Результат — одна локальная транзакция для предметного изменения и устойчивого намерения доставки, затем отдельный relay.

Полные файлы api.py, store.py и relay.py завершают проект «Доставка». API, SQLite, relay и HTTP-команды не запускались. Официальное руководство по надёжности RabbitMQ определяет транспортную неопределённость; outbox является прикладным решением нашего проекта и не выдаётся за встроенную общую транзакцию брокера и базы.

Два независимых действия создают промежуток

Рассмотрим наивный обработчик: UPDATE материала, commit, publish_event. После commit API уже приняло новое состояние. Если процесс останавливается перед publish, событие теряется для получателей. Обратный порядок также плох: сообщение ушло, затем локальная транзакция отказала. Получатель видит факт, который владелец состояния не подтвердил.

Outbox сохраняет конверт в той же SQLite-транзакции, где меняется таблица materials. Внутри этой транзакции сетевого вызова нет. Если commit успешен, есть и материал, и ожидающее доставки событие. Если транзакция откатилась, нет ни нового состояния, ни его факта. Сеть обрабатывается отдельно после устойчивого сохранения намерения.

Таблица outbox имеет event_id, body и dispatched_at. Пока dispatched_at пусто, relay должен пытаться передать событие. Полное тело хранится, поэтому повтор не генерирует новый UUID и не пересобирает payload из уже изменившегося материала. Это существенно: поздняя редакция может существовать к моменту повторной отправки прежнего факта.

Ключ HTTP-команды связывает повторный запрос

Клиент API тоже может потерять ответ после commit. Для него заголовок Idempotency-Key обозначает повтор одной команды. Таблица requests связывает ключ, fingerprint исходной команды и eventId. Повтор с тем же ключом и тем же содержимым возвращает прежний eventId. Тот же ключ с другими данными отклоняется как конфликт договора.

EventId и Idempotency-Key имеют разные роли. Первый путешествует между broker и consumer как идентичность факта. Второй помогает владельцу API распознать повтор входного намерения клиента. Не следует использовать номер сетевой попытки в любом из этих ключей. Повтор должен сохранять связь с уже принятой операцией, а не создавать новую запись.

HTTP-команда из sample-command.json включает articleId, revision, title, url и type. Для первого изменения article-107 ожидается revision 1, затем 2. API проверяет, что новая команда продолжает текущую редакцию. Это защита от конкурирующей старой команды, отдельная от идемпотентного повторения уже принятого requestKey.

Будущая команда читателя для локального API:

curl -X POST http://127.0.0.1:8093/materials \
  -H 'Content-Type: application/json' \
  -H 'Idempotency-Key: article-107:r1:publish' \
  --data-binary @sample-command.json

Она не выполнялась автором. Ожидаемый ответ 202 содержит eventId, статус accepted и адрес проверки delivery. Accepted означает, что локальное изменение и намерение доставки сохранены. Он не обещает, что catalog и audit уже завершили работу. Внешний HTTPServer привязан к loopback и не имеет входа пользователей; рабочая авторизация остаётся отдельной задачей.

Транзакция сохраняет три связанные записи

В accept сначала проверяется known requestKey, затем ожидаемая revision, после чего создаётся событие. Следующий фрагмент полного файла находится внутри BEGIN IMMEDIATE; command и digest уже проверены:

event = make_event(command)
db.execute('INSERT INTO outbox(event_id,body) VALUES(?,?)',
           (event['eventId'], canonical(event)))
db.execute('INSERT INTO requests VALUES(?,?,?)',
           (request_key, digest, event['eventId']))
db.execute(
    'INSERT INTO materials VALUES(?,?,?) '
    'ON CONFLICT(article_id) DO UPDATE SET '
    'revision=excluded.revision,command=excluded.command',
    (command['articleId'], command['revision'], canonical(command))
)
db.commit()

Это исходник, а не выполненная SQL-операция. Параметры передаются отдельно от текста запроса. Если любая запись отказала, функция делает rollback и закрывает соединение. Изменение материала, событие и квитанция входной команды имеют одну локальную границу фиксации. Уникальные ключи помогают сделать связь проверяемой.

Relay передаёт сохранённый конверт

Relay выбирает до ста pending-записей, открывает connection и channel, объявляет topology и включает confirms. Для каждой записи читает сохранённое body и вызывает publish_event. Только после успешного возврата обновляет dispatched_at и делает commit. При исключении неподтверждённая запись остаётся pending для следующей попытки.

Если broker принял событие, а relay упал до SQLite-отметки, запись снова будет отправлена. Это сознательно допустимый дубликат с прежним eventId. Inbox consumer защищает локальный эффект. Если отметить dispatched_at до confirm, противоположный порядок позволил бы потерять работу: relay считал бы её законченной, хотя broker мог ничего не принять.

Название confirmed в GET /events/{eventId} означает отмеченное принятие брокером. Оно не является подтверждением каждого consumer. При повторной HTTP-команде API также не создаёт новый факт ради обновления статуса. Проверка доставки читает состояние существующей outbox-записи; получатели продолжают работу независимо.

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

API commit → материал и outbox существуют
Broker недоступен → outbox pending
Relay confirm → dispatched_at заполнен
Падение после confirm до отметки → возможный повтор того же eventId
Consumer inbox → повтор не создаёт второй локальный эффект

Все строки обозначают договор для будущего наблюдения. Автор не измерял задержку relay и не выполнял потерю ответа. Число попыток зависит от доступности broker и состояния процессов. Фиксированные фразы журнала в исходнике не являются доказательством, что эти переходы уже происходили.

Границы локальной реализации

Relay рассчитан на один экземпляр для source-базы. Несколько relay могут прочитать одинаковые pending-записи и увеличить число дубликатов. Защита consumer снижает риск повторного эффекта, но не заменяет claim/lease-механизм рабочего планировщика. При масштабировании понадобится согласованный захват работы и обработка просроченных claims.

Outbox не очищается автоматически и не является полноценным вечным журналом для новых подписчиков. Срок хранения, архивирование и восстановление требуют отдельного договора. Сохранённые события содержат заголовки и ссылки, поэтому права на source-базу тоже важны. Проблему секретов нельзя решить только защищённым URI RabbitMQ.

Наконец, гарантия API относится к принятому локальному факту. Она не покрывает потерю всех файлов SQLite и не устраняет ранее описанный предел classic TTL/DLX. Поэтому итоговая система не получает универсальное exactly-once или безусловное at-least-once от слова outbox. Она получает понятные границы ответственности и восстановимый повтор там, где сохранено состояние.

Запросы API выполняются последовательно в локальном HTTPServer, но база дополнительно использует BEGIN IMMEDIATE для согласования записи. Это не делает сервис многопользовательским production-API; зато условие ожидаемой revision проверяется в той же транзакции, где сохраняется новая. Если вынести проверку перед транзакцией, две команды могли бы увидеть один прежний номер и обе считать себя следующими. Предметный конфликт относится к владельцу материала, а не перекладывается на consumer.

Для упражнения остановите relay мысленно: API должно продолжать сохранять pending-события в пределах возможностей базы. Затем представьте его падение после confirm и объясните неизменность UUID. В завершающем уроке передадим проект на сопровождение, связав queue, outbox, inbox и предметную revision в одну диагностическую картину.

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