Messenger і outbox: як не втратити подію між транзакцією і брокером

Усе нижче перевірено на Symfony 7.4 LTS і 8.0, symfony/doctrine-messenger, doctrine/dbal 4 і PHP 8.4. Приклади з PostgreSQL; де поведінка MySQL відрізняється, це зазначено окремо.

Проблема подвійного запису

Типовий обробник виглядає так:

$this->entityManager->wrapInTransaction(function () use ($order): void {
    $order->markPaid();
    $this->entityManager->persist($order);
});

$this->bus->dispatch(new OrderPaid($order->getId()));

Тут два записи в дві різні системи: рядок у базі і повідомлення в брокері. Атомарної операції «зроби обидва або жодного» між ними немає, і будь-який збій між рядками дає розходження. Процес помер після COMMIT, але до dispatch() — замовлення оплачене, лист не пішов, склад не зарезервував товар. Перенесете dispatch() всередину транзакції з AMQP-транспортом — отримаєте дзеркальну проблему: повідомлення вже в черзі, споживач хапає його, читає SELECT по order_id і не знаходить рядка, бо транзакція ще не закомічена. Або комітом і не закінчиться — унікальний індекс упаде, транзакція відкотиться, а подія про неіснуючу оплату вже полетіла далі.

Двофазний коміт між Postgres і RabbitMQ теоретично існує, практично його ніхто не тримає в проді: він вимагає координатора, блокує ресурси на час невизначеності й перетворює падіння одного вузла на зависання всієї гілки. Outbox — це відмова від розподіленої транзакції на користь однієї локальної: пишемо подію в ту саму базу, тією самою транзакцією, а доставку в брокер робить окремий процес, який читає цю таблицю. Ціна — доставка стає асинхронною і at-least-once.

Doctrine transport як outbox

Головне, чого часто не помічають: Doctrine transport у Messenger — це вже готовий outbox, окрема бібліотека не потрібна. Його send() виконує звичайний INSERT INTO messenger_messages через Doctrine\DBAL\Connection. Якщо на цьому з'єднанні відкрита транзакція, INSERT стає її частиною.

# config/packages/messenger.yaml
framework:
    messenger:
        failure_transport: failed
        transports:
            outbox:
                dsn: 'doctrine://default?queue_name=outbox&auto_setup=false'
                retry_strategy:
                    max_retries: 5
                    delay: 1000
                    multiplier: 3
                    max_delay: 3600000
            failed: 'doctrine://default?queue_name=failed&auto_setup=false'
        routing:
            App\Message\OrderPaid: outbox

Тепер dispatch() заходить усередину транзакції:

$this->entityManager->wrapInTransaction(function () use ($order): void {
    $order->markPaid();
    $this->entityManager->persist($order);
    $this->entityManager->flush();

    $this->bus->dispatch(new OrderPaid($order->getId()->toRfc4122()));
});

Чотири умови, без яких це не працює:

Те саме з'єднання. doctrine://default має вказувати на той самий Doctrine connection, у якому живуть ваші сутності. doctrine://analytics або окремий пул — і атомарності немає, ви просто переклали проблему на іншу базу.

flush() до dispatch(). Транспорт серіалізує повідомлення одразу під час dispatch(), а не на коміті. Якщо в конструктор OrderPaid іде згенерований базою ID, його на момент виклику ще не існує. Передавайте UUID, згенерований у застосунку (Uuid::v7()), або робіть flush() явно перед dispatch().

auto_setup=false у проді. За замовчуванням транспорт перевіряє й створює таблицю на першому зверненні — це зайвий DDL-запит під навантаженням і DDL від додаткового користувача. Створіть схему один раз: php bin/console messenger:setup-transports.

Воркер — окремий процес. php bin/console messenger:consume outbox --time-limit=3600 --memory-limit=128M. Він робить SELECT ... FOR UPDATE SKIP LOCKED, проставляє delivered_at, віддає повідомлення шині й видаляє рядок після успіху. SKIP LOCKED означає, що кілька воркерів не заважають одне одному — але й що порядок доставки не гарантований. Якщо порядок критичний, тримайте один воркер на queue_name і партиціонуйте потік за ключем через окремі значення queue_name.

На PostgreSQL транспорт за замовчуванням використовує LISTEN/NOTIFY (опція use_notify), і це вдало лягає на outbox: NOTIFY транзакційний, підписник отримує сигнал тільки після COMMIT. На MySQL натомість звичайний полінг із паузою --sleep (типово 1 секунда), тож затримка доставки — до секунди плюс час обробки.

Якщо повідомлення все-таки має потрапити в RabbitMQ чи Kafka (наприклад, його чекає інший сервіс), робіть outbox першим кроком, а брокер — другим. Обробник, прив'язаний до транспорту outbox, просто пересилає повідомлення далі:

#[AsMessageHandler(fromTransport: 'outbox')]
final readonly class ForwardOrderPaidToBroker
{
    public function __construct(private MessageBusInterface $bus) {}

    public function __invoke(OrderPaid $message): void
    {
        $this->bus->dispatch($message, [new TransportNamesStamp(['amqp'])]);
    }
}

TransportNamesStamp перекриває routing, тож циклу не буде, а fromTransport гарантує, що цей обробник не спрацює на повідомленнях, які прийшли з amqp. Рядок в messenger_messages видаляється лише після того, як basic.publish пройшов, тож між базою і брокером втрати вже немає — є дублікати.

Ретраї й failed transport

retry_strategy з конфіга вище дає затримки 1с, 3с, 9с, 27с, 81с — множення на multiplier, обрізане max_delay. Для Doctrine transport затримка не потребує окремої черги: воркер просто пише в available_at час у майбутньому, і SELECT його не бачить, поки той не настане.

Двома винятками ви керуєте цим ззовні:

// не ретраїти взагалі: помилка детермінована, повтор дасть той самий результат
throw new UnrecoverableMessageHandlingException('Невідомий тип платежу.');

// ретраїти, навіть якщо стратегія сказала б ні, і саме через 30 секунд
throw new RecoverableMessageHandlingException('Провайдер віддав 503.', retryDelay: 30_000);

Валідаційна помилка в payload, відсутній обробник, зіпсована серіалізація — це Unrecoverable. Таймаут HTTP, deadlock detected, 503 від провайдера — звичайні ретраї. Різницю варто провести явно, інакше ви п'ять разів поспіль намагаєтесь обробити повідомлення, яке не обробиться ніколи, і забиваєте цим воркер.

Коли ретраї вичерпані, повідомлення переїжджає у failure_transport — у нашому випадку в ту саму таблицю з queue_name=failed. Це навмисно: інциденти зручніше розбирати в базі, а не в UI брокера.

php bin/console messenger:failed:show
php bin/console messenger:failed:show 4821 -vv   # повний стектрейс з ErrorDetailsStamp
php bin/console messenger:failed:retry 4821 4822
php bin/console messenger:failed:remove 4821

Дві поради з практики. По-перше, ставте --failure-limit на messenger:consume: якщо за одну сесію воркера в failed переїхало більше N повідомлень, процес зупиняється, супервізор його перезапускає, а ваш моніторинг бачить рестарти. Без цього зламаний зовнішній сервіс тихо перемелює всю чергу у failed за хвилину. По-друге, моніторте не лише розмір failed, а й вік найстарішого рядка з queue_name = 'outbox': зростання now() - created_at — це єдиний надійний сигнал, що воркер помер або відстає.

Ще одна опція, про яку згадують уже під час інциденту: redeliver_timeout (типово 3600 секунд). Якщо воркер узяв повідомлення, проставив delivered_at і помер, рядок висітиме забраним годину, після чого повернеться в обіг. Для швидких обробників це занадто довго — ставте секунд шістдесят із запасом на найдовшу легітимну обробку.

Ідемпотентність споживача

Outbox дає at-least-once, і це не дрібний нюанс, а контракт. Воркер може впасти між успішним побічним ефектом і DELETE рядка — тоді повідомлення буде доставлено вдруге. redeliver_timeout може повернути в обіг повідомлення, яке насправді ще обробляється повільним воркером. messenger:failed:retry руками — те саме. Обробник, який не витримує повторного виклику, зламається; питання лише коли.

Дедуплікація має бути в тій самій транзакції, що й побічний ефект:

public function __invoke(OrderPaid $message): void
{
    $this->connection->transactional(function (Connection $connection) use ($message): void {
        $inserted = $connection->executeStatement(
            'INSERT INTO processed_message (message_id, consumer, processed_at)
             VALUES (?, ?, ?) ON CONFLICT DO NOTHING',
            [$message->eventId, self::class, new \DateTimeImmutable()],
            [ParameterType::STRING, ParameterType::STRING, Types::DATETIME_IMMUTABLE],
        );

        if ($inserted === 0) {
            return;
        }

        $this->invoices->issueFor($message->orderId);
    });
}

Три деталі, які тут не випадкові.

message_id — це UUID, згенерований продюсером і покладений у payload, а не ID рядка в messenger_messages і не TransportMessageIdStamp. Транспортний ID змінюється при перекладанні повідомлення у failed і назад, тож дедуплікація по ньому пропустить дубль саме в тому сценарії, заради якого писалась.

ON CONFLICT DO NOTHING, а не try/catch (UniqueConstraintViolationException). На PostgreSQL порушення унікального індексу переводить транзакцію в стан 25P02, і всі наступні запити в ній падають — перехопити виняток і продовжити роботу в тій самій транзакції не вийде без savepoint. На MySQL еквівалент — INSERT IGNORE, семантика та сама.

Ключ складений: (message_id, consumer). Одну подію зазвичай слухає кілька обробників, і кожен має свій прапорець «оброблено».

Таблиці processed_message потрібен ретеншен — рядки старші за максимальний час життя повідомлення (з урахуванням усіх ретраїв і ручних retry) видаляються плановим завданням. Інакше вона росте вічно.

Там, де побічний ефект і так природно ідемпотентний — UPDATE ... SET status = 'paid' WHERE id = ?, PUT у зовнішнє API з ключем ідемпотентності, upsert — окрема таблиця не потрібна. Пишіть її тільки для операцій, повтор яких видно назовні: списання коштів, відправлення листа, виклик API без ключа ідемпотентності.

Коли це зайве

Outbox — не безкоштовний. Кожне повідомлення — це INSERT, UPDATE delivered_at, DELETE, плюс SELECT ... FOR UPDATE SKIP LOCKED від кожного воркера кожну секунду. На PostgreSQL це ще й потік мертвих версій рядків у гарячій таблиці, тобто постійна робота для autovacuum. Тому перевірте, чи вам це взагалі потрібно.

Побічний ефект живе в тій самій базі. Якщо обробник тільки пише в ті самі таблиці, розподіленої транзакції немає, і немає що узгоджувати — окремий forwarder не потрібен, Doctrine transport сам і є вашою чергою.

Подія необов'язкова. Прогрів кешу, метрика, перерахунок рекомендацій — втрата однієї події нічого не ламає. Тут дешевший прямий AMQP або взагалі синхронний виклик.

Ви віддаєте результат користувачу в тому ж запиті. Синхронна шина (sync://) чесніша за асинхронну обіцянку.

Потік справді великий. Коли outbox стає найгарячішою таблицею в базі, дивіться в бік CDC: Debezium читає WAL або binlog і будує потік змін без жодного коду в застосунку, зберігаючи ту саму гарантію атомарності — бо джерелом подій стає сам факт коміту.

Споживач принципово не ідемпотентний і зробити його таким не можна. Outbox гарантує «щонайменше один раз». Якщо повторний виклик неприпустимий, а дедуплікувати нема за чим, проблема не в транспорті, і outbox її не вирішить.

І останнє, про що варто домовитися з командою заздалегідь: outbox прибирає втрату подій, але не прибирає затримку і не дає глобального порядку. Споживач побачить подію через частку секунди після коміту, а при кількох воркерах — можливо, не в тому порядку, у якому вона писалась. Якщо бізнес-логіка на це не розрахована, її треба міняти разом із транспортом, а не після першого інциденту.