Паттерн Outbox

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

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

Ошибка в одной из этих операций может привести к не консистентным данным.

Проблема

Когда микросервис отправляет уведомление после обновления базы данных, эти две операции должны выполняться атомарно, чтобы обеспечить целостность и надежность данных.

• Если обновление базы данных прошло успешно, но сообщение не удалось отправить, сервис B не узнает об изменении, и система может войти в несогласованное состояние.

• Если обновление базы данных не удалось, но сообщение было отправлено, данные могут быть повреждены, что может повлиять на надежность системы.

Область применения

Обычно паттерн Outbox используется в следующих случаях:

• Вы создаете приложение, в котором обновление базы данных инициирует уведомление о событии.
• Вы хотите обеспечить атомарность операций, которые затрагивают два сервиса.
• Вы хотите реализовать паттерн event sourcing.

С чем этот подход не поможет и о чем нужно подумать отдельно:

• Дубликаты сообщений: сервис может отправлять дубликаты сообщений или событий, поэтому рекомендуется сделать потребляющий сервис идемпотентным, отслеживая обработанные сообщения на его стороне.
• Порядок сообщений: отправляйте сообщения или события в том же порядке, в котором сервис обновляет базу данных. Это критично для паттерна event sourcing, где вы можете использовать хранилище событий для восстановления данных на определенный момент времени. Если порядок неверен, это может все испортить. При этом, eventual consistency и откат транзакций могут усугубить проблему, если порядок уведомлений не сохраняется.
• Откат транзакций: если транзакция откатывается, сообщение не должно отправляться.
• Транзакционность только в рамках одного сервиса: паттерн Outbox не решает проблему транзакционной целостности между несколькими сервисами. Для этого используйте паттерн saga orchestration.

Архитектура

1. Клиент отправляет запрос на обновление данных в Order service.
2. Order service обновляет базу данных.
3. Order service отправляет сообщение в очередь сообщений.

Если запись в базе данных Order service не удалась, а сообщение уже отправлено, то Payment сервис получит сообщение и обновит свою базу данных, что приведет к несогласованному состоянию: в одной базе данных заказ будет отсутствовать, а в другой - присутствовать.

Реализация c использованием сервисов Yandex Cloud

Для этого мы будем использовать сервисы Yandex Cloud:

• Yandex Database для хранения данных.
• Yandex Message Queue для отправки сообщений.
• Yandex Function для обработки сообщений.

Если в сервисе Order происходит ошибка после коммита транзакции, сообщение не будет отправлено, и Payment сервис не получит уведомление.

Однако, может упасть и транзакция, но сообщение при этом все равно будет отправлено. В этом случае, Payment сервис получит уведомление и возьмёт с пользователя оплату за заказ, который не был создан.

Чтобы решить эту проблему, мы рассмотрим 2 подхода:

• таблица Outbox в базе данных
• CDC (Change Data Capture) поток

Таблица Outbox в базе данных

Таблица Outbox хранит все события из сервиса Orders с timestamp и id события.

Когда в таблицу orders добавляется запись, также в той же транзакции запись добавляется и в таблицу outbox. Другой сервис (например, функция, вызываемая cron-триггером) читает из таблицы outbox и отправляет сообщение в очередь сообщений Yandex Message Queue. Далее сервис Payment получает сообщение из очереди и обрабатывает его. YMQ гарантирует, что сообщение будет доставлено хотя бы один раз и не потеряется.

Стоит учитывать, что при использовании стандартных очередей Yandex Message Queue, одно и то же сообщение может быть доставлено более одного раза. Поэтому нужно убедиться, что сервис обработки сообщений идемпотентен (то есть обработка одного и того же сообщения не приведет к нежелательным последствиям). Если требуется, чтобы сообщение было доставлено ровно один раз, с сохранением порядка сообщений, можно использовать очереди YMQ FIFO.

Если обновление таблицы orders, либо outbox не удалось, транзакция откатывается, таким образом, не возникает несогласованности данных между сервисами.

В следующей диаграмме архитектура транзакционного outbox реализована с использованием базы данных YDB. Когда сервис обработки событий читает таблицу outbox, он видит только те строки, которые являются частью закоммиченой (успешной) транзакции. Затем помещает сообщение для события в очередь YMQ, которая читается сервисом Payment для дальнейшей обработки. Такая архитектура решает проблему двойной записи и сохраняет порядок сообщений и событий с использованием timestamp.

1. Изменение данных в таблице orders.
2. Добавление записи в таблицу outbox в той же транзакции.
3. Сервис обработки событий читает таблицу outbox;
4. И отправляет сообщение в очередь сообщений YMQ.
5. Сервис Payment получает сообщение из очереди и обрабатывает его.

Пример кода на TypeScript реализующий этот паттерн.

CDC (Change Data Capture) поток

YDB поддерживает CDC (Change Data Capture) поток, который публикует изменения в таблице в виде событий в Yandex Data Stream. Вы можете создать триггер, который будет отправлять сообщение в очередь сообщений Yandex Message Queue или подписаться в сервисе консьюмере на поток напрямую.

В YDB вы можете создавать CDC на отдельные таблицы.

ALTER TABLE `orders` ADD CHANGEFEED `updates` WITH (
    FORMAT = 'JSON',
    MODE = 'UPDATES'
);

После этого, все изменения в таблице orders будут публиковаться в поток updates.

Поток изменений предоставляет следующие гарантии:

• Записи об изменении шардированы между партициями топика по первичному ключу.
• Каждое изменение доставляется ровно один раз (exactly-once семантика).
• Изменения по одному и тому же первичному ключу доставляются в том же порядке, в котором они происходили в таблице в одну и ту же партицию топика.

Но следует учитывать и существующие ограничения:

• Количество партиций топика фиксируется на момент создания потока изменений и остается неизменным (топики, в отличие от таблиц, не являются эластичными).
• В потоке изменений поддерживаются записи о следующих видах операций: обновление и удаление. Таким образом, добавление строки является частным случаем обновления, и в потоке изменений запись о добавлении строки будет выглядеть аналогично записи об обновлении.

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

Пример кода на TypeScript реализующий этот паттерн.