Outbox

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

Схема текстом

1. Обработчик-издатель: отслеживаемое сообщение → Слой публикации: INSERT записи силами сервиса, на своём соединении
2. Слой публикации: INSERT записи силами сервиса, на своём соединении → Любой брокер RustStream: сообщение и id записи: затем отправляет
3. При старте: каждая запись без отметки → Любой брокер RustStream: сообщение и id записи: со своим id
4. Любой брокер RustStream: сообщение и id записи → Слой подписки: запись получена по id, пока она не обработана: доставляет
5. Слой подписки: запись получена по id, пока она не обработана → Уже обработана: ack, без обработчика
6. Слой подписки: запись получена по id, пока она не обработана → Обработчик-получатель: не обработана
7. Обработчик-получатель → ack или drop: запись отмечена, processed_at = now, или удалена: подтверждает
8. Обработчик-получатель → retry: запись ждёт старта: просит снова
9. Запись фиксируется отдельно, не в транзакции доставки: публикация доставляется хотя бы один раз

Outbox - это плагин рантайма RustStream, а не часть таблицы очереди: два middleware и шаг старта, которые оборачивают издателей и обработчики любого брокера, включая inbox. Реестр, построенный через outbox! или Outbox::track, называет отслеживаемые сообщения и тип записи, которую сохраняет каждое из них. Слой публикации сохраняет запись каждого отслеживаемого сообщения через собственную реализацию Publish типа записи, а затем отправляет сообщение с id записи в заголовке x-ruststream-outbox-id. Если запись не сохранилась, публикация возвращает ошибку, поэтому без записи ничего не отправляется.

Слой подписки читает заголовок на стороне получателя. Он получает запись по её id, WHERE id = ? AND processed_at IS NULL, если таблица отмечает свои записи. Уже обработанная запись подтверждает доставку без запуска обработчика: так копия, отправленная дважды, обрабатывается один раз. Если получить запись не удалось, доставка уходит на переобработку без обработчика. Затем исход обработчика завершает запись: подтверждение и drop отмечают её, устанавливая processed_at по часам базы данных или удаляя строку; retry и retry_after оставляют её, и запись ждёт следующего старта. Отметка, вернувшая ошибку, попадает в журнал, сохраняет исход обработчика и тоже оставляет запись до следующего старта.

При старте повторная публикация выбирает каждую необработанную запись каждого имени и снова публикует её через издателя брокера, с её id, когда подписки уже открыты. Запись, которую повторная публикация не может прочитать или отправить, прерывает старт. Два экземпляра, которые стартуют одновременно, могут оба отправить одну и ту же запись, поэтому получатель, который не должен обрабатывать сообщение дважды, проверяет это сам.

Собственное соединение записи

Отслеживаемая публикация берёт собственное соединение для вставки записи, и запись фиксируется на нём. Она никогда не входит в транзакцию доставки: запись, сохранённая во время работы транзакционного обработчика, фиксируется при любом исходе обработчика. Поэтому сообщение и его запись не фиксируются вместе с бизнес-данными сервиса. Outbox гарантирует, что публикация доставляется хотя бы один раз: запись сохраняется до отправки, и каждая отправка, которая вернула ошибку или так и не была обработана, оставляет запись, которую следующий старт отправит снова. Обычный транзакционный outbox, где бизнес-строка и строка события фиксируются вместе, а ретранслятор опрашивает таблицу, - это таблица очереди inbox, которую читает обработчик, публикующий в другой брокер.

Стоимость

Реестр - это список в системе типов, один узел на зарегистрированное имя, поэтому сообщение с незарегистрированным именем стоит одно сравнение имён на узел и ничего больше: ни выделения памяти, ни обращения к базе данных, ни динамического вызова. Отслеживаемая публикация берёт соединение для вставки и выделяет память под текст заголовка с id. Отслеживаемая доставка берёт соединение, чтобы получить запись, возвращает его до запуска обработчика, чтобы обработчик мог публиковать собственные отслеживаемые сообщения, и берёт снова для отметки. Пул, заданный после построения реестра внутри #[ruststream::app], читается через OnceLock: одна атомарная загрузка на отслеживаемое сообщение. Тестовая сборка оставляет outbox выключенным, если не задано RUSTSTREAM_SQLX_OUTBOX=on, а рабочая сборка отслеживает всегда: переключатель вырезается из неё при компиляции.

Это паттерн транзакционный outbox с добавленной стороной доставки: запись завершает отметка получателя, а не отправка.