Download the PHP package gian-tiaga/spiral-outbox without Composer

Spiral Outbox

gian-tiaga/spiral-outbox — transactional outbox для Spiral. Пакет сохраняет интеграционное событие в бизнес-транзакции, а после commit создаёт и независимо выполняет одну или несколько доставок через Spiral Queue.

Установка

Приложение подключает Spiral\RoadRunnerBridge\Bootloader\QueueBootloader и GianTiaga\SpiralOutbox\Bootloader\OutboxBootloader после bootloader миграций Cycle. Пакет добавляет свой каталог миграций, команды outbox:relay и outbox:status и перехватчик Worker.

Целевой алгоритм

Что хранит Outbox

Событие неизменяемо и общего статуса не имеет. Каждая доставка хранит свой класс Job, очередь, список задержек, таймаут, статус и число неудачных попыток. Успех одной доставки не закрывает соседние доставки события.

Запись события

OutboxEventStoreContract::add() вызывается внутри уже открытой бизнес-транзакции. Хранилище просит у сериализатора строку полезной нагрузки, добавляет одну строку outbox_events с routed_at = null и возвращает её идентификатор. Карта маршрутов и очередь в этой транзакции не читаются. Без транзакции запись запрещена; ошибка записи откатывает доменные изменения вместе с событием.

Сериализация события

Полезную нагрузку собирает не хранилище, а реализация OutboxEventSerializerContract: событие превращается в строку, а строка колонки payload вместе с классом события — обратно в объект. Реализация по умолчанию — JSON на cuyz/valinor, который собирает событие по типам его конструктора. Приложение подменяет её своей привязкой контракта в контейнере: ни хранилище, ни загрузчик о формате не знают.

Обычный проход relay: этап 1 — создать доставки

Relay выбирает события с routed_at = null через FOR UPDATE SKIP LOCKED. Для каждого OutboxRoute он создаёт строку outbox_deliveries и ставит событию routed_at. Доставки и отметка маршрутизации фиксируются одним commit.

Если маршрут отсутствует или неверен, частичные доставки не создаются, а routed_at остаётся null. Новый маршрут не применяется к уже маршрутизированной истории. Unique по event_id, job_class и queue_name защищает от одинаковой доставки.

Обычный проход relay: этап 2 — отправить доставки

В новой транзакции relay выбирает готовые pending, блокирует их через FOR UPDATE SKIP LOCKED, ставит dispatched, dispatched_at и delivery_timeout_at, затем делает commit. Только после commit он отправляет Job в очередь доставки.

Полное имя класса Job служит типом задачи Spiral Queue. Отдельного имени Job и записей в queue.registry.handlers нет. Полезная нагрузка содержит outboxEventId, outboxDeliveryId, внутренний outboxDispatchId, уникальный для каждой отправки, и признак последней попытки outboxLastAttempt. Идентификатор события relay уже читает из доставки, поэтому отдельного запроса он не стоит.

Маршрут проверяет публичный invoke(): обязательные скалярные параметры должны приходить из этой полезной нагрузки. Объектные зависимости из контейнера и необязательные параметры разрешены.

Значение статусов доставки

Работа Worker и Job

Перехватчик пакета находит доставку по outboxDeliveryId, требует статус dispatched, совпадение внутреннего outboxDispatchId с текущей отправкой, сверяет полный класс Job и загружает событие. Поздний результат старой Job не меняет новую отправку.

Успех переводит только эту доставку в completed. RetryableOutboxException оформляет повторяемую ошибку. Любое другое исключение сразу переводит только эту доставку в failed. Все переходы условно проверяют текущий статус.

Отметка успеха ищет строку по паре «доставка и отправка». Если за время работы попытку перехватили — тайм-аут истёк, и relay выдал доставку заново, — строки не найдётся, и пакет пишет предупреждение с обоими идентификаторами. Иначе успешно доработавшая Job не оставляла бы следа ни в базе, ни в журнале, и вторую копию пришлось бы искать сверкой времени руками.

Расписание повторов

Число повторов равно длине retryDelaysSeconds; отдельного maxAttempts нет. Список [60, 120] даёт первую попытку и два повтора. Пустой список [] означает финальный отказ после первой ошибки.

При повторе доставка возвращается в pending, получает новое available_at, а dispatched_at и delivery_timeout_at очищаются. Другие доставки события продолжают работу.

Отправка сообщает Job, последняя ли это попытка: bool $outboxLastAttempt равен true, когда следующей паузы у доставки уже нет и её неудача станет окончательной. Признак считается из снимка пауз и числа попыток самой доставки, поэтому своего счётчика попыток Job не заводит и повторами по-прежнему владеет только Outbox. Job принимает признак необязательно — он нужен только тем обработчикам, которым на последней неудаче нужно сделать что-то отдельное.

Потерянная Job и таймаут результата

deliveryTimeoutSeconds покрывает ожидание Worker, выполнение и запас. Пока delivery_timeout_at не наступил, вторую Job relay не создаёт. После таймаута попытка считается потерянной и проходит обычную ветку повтора или финального отказа.

Таймаут не останавливает старую Job. Обработчик обязан безопасно переносить повторный запуск; гарантия пакета остаётся at-least-once.

Job, которая идёт порциями или страницами, может оборваться сама. OutboxAttemptContract отвечает на один вопрос: её ли попытка ещё текущая. Обе половины ответа приходят Job в полезной нагрузке, поэтому спрашивать можно между порциями, ничего больше не зная.

Смерть Worker посреди работы

Нехватка памяти убивает процесс фатальной ошибкой, а она не исключение: catch вокруг Job её не видит. Пакет регистрирует обработчик завершения процесса, и доставка, которую Job не отпустила своим ходом, сразу получает статус failed с причиной worker_died и запись уровня ERROR. Ждать конца тайм-аута для этого не нужно.

Обычные исключения до обработчика не доходят: доработавшая Job снимает присмотр сама, и одна неудача не записывается дважды. Текст фатальной ошибки в журнал не идёт — стек печатает фрагменты тел внешних ответов; запись собирается из идентификаторов и числового кода ошибки.

Ошибка отправки в RabbitMQ

Если очередь отвергла Job, relay оформляет повторяемую ошибку только этой доставки и продолжает остальные строки пачки. Он не ждёт delivery_timeout_at, потому что отказ известен.

Кто делает повторы

Повторами владеет только Outbox. RabbitMQ хранит и отдаёт текущую Job, Worker выполняет её и сообщает результат. RetryPolicyInterceptor, retry-атрибуты Job и возврат упавшей Job в RabbitMQ не используются. Для AMQP pipeline задаётся requeue_on_fail: false.

Очереди и Worker

Имя очереди — строковое значение enum приложения. То же значение используется в OutboxRoute, конфигурации Spiral Queue и RoadRunner. Один relay обслуживает все очереди, а каждую физическую очередь читает отдельная группа Worker.

Постоянный relay

Без параметров команда выполняет один проход. Постоянный процесс запускается отдельно от HTTP-приложения и после миграций:

При пустом проходе он ждёт секунду. --sleep=1 принимает целое значение от 1 до 3600. SIGTERM и SIGINT запрещают новый проход и завершают процесс после текущего. Для цикла нужно расширение PHP pcntl. Фатальная ошибка базы или конфигурации завершает процесс, а отказ отправки одной доставки остаётся ошибкой этой доставки.

Состояние Outbox

Команда outbox:status печатает таблицей текущее состояние обмена по базе окружения:

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

Граница будущего микросервиса

Будущий микросервис не читает таблицы приложения. Он получает стабильное integration event через свою RabbitMQ-очередь и владеет своими inbox и Outbox. PHP-класс локальной Job не является межсервисным контрактом.

Контракты реализации

Событие

Событие реализует IntegrationEventContract. Поле события — string, int, float, bool, null, backed enum, DateTimeImmutable, вложенный объект по тем же правилам, list<T> и array<string, T> с указанным T. Видимость свойства не ограничена: непубличные свойства тоже попадают в полезную нагрузку, — но одноимённый параметр конструктора обязателен для свойства любой видимости, потому что именно им событие собирается обратно. Статическое свойство полем события не является.

Нетипизированный array, iterable, mixed, объект без класса и свойство без одноимённого параметра конструктора ловит правило статического анализа gian-tiaga/phpstan-strict-rules: параметр integrationEventInterface в phpstan.neon называет ему интерфейс события. В рантайме остаётся отказ значения, которое нельзя закодировать, — NAN, INF, строка с невалидным UTF-8, замыкание или ресурс: пакет отвергает его до вставки строки. Секреты и персональные данные в событие не кладут.

Маршрут

Для класса события задаётся непустой список объектов OutboxRoute:

job — конкретный наследник JobHandler с публичным invoke() и обязательным параметром string $outboxDeliveryId; кроме него из полезной нагрузки разрешены только string $outboxEventId, string $outboxDispatchId и bool $outboxLastAttempt; queue — непустая строка; retryDelaysSeconds — список положительных целых секунд; deliveryTimeoutSeconds — положительное целое число. Настройки копируются в доставку и потом не меняются вместе с конфигурацией.

Таблицы и миграции

Имена таблиц задаются до первой миграции. После неё переименование выполняется отдельной миграцией приложения. Пакет добавляет свой каталог в migration.vendorDirectories.

outbox_events хранит id, event_type, payload, created_at, routed_at. outbox_deliveries хранит id, event_id, job_class, queue_name, снимок повторов и таймаута, статус, попытки, dispatch_id, времена переходов и безопасную причину отказа.

Логирование

Записи связываются полями outboxEventId и outboxDeliveryId. В контекст входят класс события, класс Job, очередь, статус, попытки и времена. Полезная нагрузка, исходные сообщения внешних исключений, секреты и персональные данные в журнал не попадают.

WARNING получают назначенный повтор и отметка успеха, не нашедшая своей доставки. ERROR — окончательный отказ доставки и смерть Worker, не доработавшего Job.

Проверки пакета

Тестам нужен настоящий PostgreSQL: поведение транзакций и SKIP LOCKED живёт в самой СУБД и подделкой не проверяется. Адрес базы приходит переменными окружения — своей базы у пакета нет, он создаёт в чужой только собственные таблицы и удаляет их после прогона: