Как обеспечить согласованность между синхронным REST-ответом на создание ресурса и асинхронной обработкой этого ресурса через очередь сообщений?

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

Ресурс нужно сначала надёжно сохранить в основном хранилище (БД) и только после этого (в идеале — в той же транзакции или через надёжный механизм вроде transactional outbox) публиковать сообщение в очередь; клиенту возвращается ответ на основе состояния в БД, а не на основе факта отправки сообщения.

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

если сначала отправить сообщение в очередь и только потом попытаться сохранить запись в БД, при сбое сохранения в очереди уже будет "висеть" сообщение о ресурсе, которого на самом деле нет; если сделать в обратном порядке без гарантий — можно сохранить запись, но не успеть отправить сообщение из-за сбоя (например, падения процесса между двумя операциями), и дальнейшая асинхронная обработка не запустится.

Пример кода

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

def create_order_with_outbox(session, payload: dict):
    with session.begin():
        order = Order(**payload)
        session.add(order)
        session.flush()  # order.id уже доступен внутри транзакции
        outbox_event = OutboxEvent(
            event_type="order_created",
            payload={"order_id": order.id},
        )
        session.add(outbox_event)
    # relay-процесс отдельно читает неотправленные OutboxEvent и публикует их в очередь,
    # помечая отправленными после подтверждения брокером
    return order

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

• Рекомендация. паттерн transactional outbox решает эту проблему: запись о ресурсе и запись о "сообщении для отправки" сохраняются в одной транзакции БД, а отдельный процесс (relay) читает несохранённые исходящие сообщения из таблицы outbox и публикует их в очередь, помечая как отправленные — тем самым публикация становится согласованной с сохранением данных на уровне "по крайней мере один раз". Потребитель сообщения должен быть готов к повторной доставке (идемпотентная обработка), так как outbox-релей обычно не гарантирует ровно один раз.

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

• Риски. публиковать сообщение в очередь до подтверждения коммита транзакции в БД, из-за чего потребитель может начать обработку ресурса, которого в итоге не окажется при откате транзакции; считать, что "просто вызвать send() после commit()" достаточно надёжно — между этими двумя операциями возможен сбой процесса, и тогда сообщение не будет отправлено вовсе без outbox-механизма.