Symfony Messenger под нагрузкой: где очередь ломается первой

Сервис отправки уведомлений на Symfony 6.4 держал очередь на Redis-транспорте. Под пиком в 300–400 сообщений в секунду воркеры начали падать по OOM, а часть уведомлений уходила дважды. Причина оказалась не в Redis, а в том, как Messenger работает с памятью, повторами и подключением к базе внутри долгоживущего процесса. Разберём механику: почему один и тот же воркер, который держал нагрузку неделю, вдруг начинает течь и дублировать доставку.

Как воркер живёт в памяти

Команда messenger:consume — это не короткоживущий скрипт, а процесс, который крутится в бесконечном цикле: забрал сообщение с транспорта, прогнал через шину, отдал ack или отправил в retry. Ключевое отличие от обычного HTTP-запроса: контейнер сервисов, EntityManager, кеши Doctrine и любые ваши синглтоны живут между сообщениями, а не создаются заново.

Отсюда три класса проблем, которые не видны на одном сообщении и вылезают на десятках тысяч:

• Утечка памяти. Doctrine UnitOfWork копит управляемые сущности. Если внутри хендлера вы делаете persist/flush и не чистите, каждое сообщение оставляет объекты в identity map.
• Протухшее соединение с БД. MySQL закрывает соединение по wait_timeout (по умолчанию 8 часов, но у managed-баз бывает 60–120 секунд). Воркер простаивал в тишине, взял сообщение — и получил MySQL server has gone away.
• Устаревший код. Задеплоили новую версию — старый воркер продолжает крутить старый код, пока его не остановят.

Механика памяти на конкретном примере

EntityManager в долгоживущем процессе — главный источник роста RSS. Каждое обработанное сообщение, которое читает или пишет сущности, добавляет их в UnitOfWork. Без явной очистки график памяти воркера — прямая восходящая линия до OOM.

Messenger не решает это сам, но даёт middleware DoctrineClearEntityManagerMiddleware, который чистит EntityManager после обработки каждого сообщения. Он включается автоматически при использовании doctrine_transaction или подключается вручную. Проверить, что он в цепочке, можно так:

php bin/console debug:messenger
# и
php bin/console config:dump-reference framework messenger

Ограничители: почему воркер должен уметь умирать

Правильный воркер под нагрузкой — тот, который регулярно перезапускается. Это не костыль, а штатная модель Messenger. Утечки, протухшие соединения и старый код лечатся тем, что процесс завершается, а супервизор поднимает свежий.

Флаги команды messenger:consume (актуальны для Symfony 6.4 / 7.x):

Рабочая связка под супервизором: --limit=1000 --memory-limit=128M --time-limit=3600. Воркер выйдет по первому сработавшему условию, супервизор его поднимет. Один процесс не накопит ни утечки, ни устаревшего кода.

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

Retry и дубли: почему платёж ушёл дважды

Самая дорогая ошибка. Messenger по умолчанию повторяет неудачные сообщения по стратегии MultiplierRetryStrategy: три попытки с растущей задержкой. Проблема в том, что «неудача» с точки зрения Messenger — это исключение из хендлера. А сообщение к этому моменту могло уже сделать половину работы.

Классический сценарий: хендлер списал деньги через платёжный шлюз, шлюз ответил успехом, но следующей строкой упал flush в базу по протухшему соединению. Messenger видит исключение, ставит сообщение на повтор — и списывает деньги второй раз.

Транспорт вроде Redis или AMQP даёт гарантию at-least-once: сообщение доставится минимум один раз, но может и больше. Ровно-один-раз транспорт не гарантирует. Значит, идемпотентность — забота хендлера, а не транспорта.

Три рабочих подхода к идемпотентности:

1. Ключ идемпотентности. Каждое сообщение несёт уникальный идентификатор операции. Перед выполнением проверяете в БД, не обработан ли уже этот ключ, и пишете результат в той же транзакции.
2. Разделение внешнего вызова и записи. Сначала фиксируете намерение в БД, отдельным сообщением делаете внешний вызов, третьим — подтверждаете. Каждый шаг идемпотентен по своему ключу.
3. Уникальный индекс на стороне БД. Дубликат вставки ловится constraint-нарушением, которое обрабатывается как «уже сделано», а не как ошибка на повтор.

Настройка стратегии повторов в config/packages/messenger.yaml:

framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                retry_strategy:
                    max_retries: 3
                    delay: 1000        # первая задержка, мс
                    multiplier: 2      # умножается на каждой попытке
                    max_delay: 0       # 0 = без потолка
                failure_transport: failed

Если исключение неустранимо в принципе (невалидный формат сообщения), не гоняйте его через все три попытки — бросайте UnrecoverableMessageHandlingException из Symfony\Component\Messenger\Exception. Сообщение уйдёт в failure-транспорт сразу.

Failure-транспорт: куда падают битые сообщения

После исчерпания повторов сообщение попадает в failure_transport. Без него исчерпавшее лимит сообщение теряется. С ним — оседает в отдельной очереди, откуда его видно и можно переотправить руками:

# посмотреть, что накопилось
php bin/console messenger:failed:show

# переотправить конкретное или все
php bin/console messenger:failed:retry --force

На проде на failure-транспорт стоит вешать алерт: если туда что-то капает, значит хендлер систематически падает, а не споткнулся один раз.

Где ещё ломается под нагрузкой

• Один воркер не тянет поток. Если очередь растёт быстрее, чем разгребается, поднимайте несколько процессов messenger:consume на один транспорт. Redis и AMQP отдают сообщения конкурентным потребителям без дублей. Но если хендлер пишет в одну строку БД — получите блокировки, и параллелизм не поможет.
• Prefetch съедает справедливость. AMQP-транспорт по умолчанию забирает пачку сообщений заранее. Медленный воркер, набравший пачку, держит их, пока свободные воркеры простаивают. Параметр prefetch-count настраивается в DSN транспорта — сверьте синтаксис в документации Messenger для вашей версии.
• Порядок не гарантирован. При нескольких воркерах сообщения обрабатываются не в порядке отправки. Если порядок важен (события одной сущности), нужен один потребитель на партицию или транспорт с поддержкой упорядочивания.
• Транзакция шире, чем кажется. Middleware doctrine_transaction оборачивает весь хендлер в транзакцию. Долгий внешний HTTP-вызов внутри держит транзакцию БД открытой всё это время — и держит блокировки.

Что мерить в проде

Три метрики, по которым видно здоровье очереди раньше, чем начнутся инциденты:

• Глубина очереди — количество необработанных сообщений на транспорте. Читается напрямую из Redis (LLEN по ключу очереди) или из AMQP-панели. Растёт монотонно — воркеров не хватает.
• Возраст самого старого сообщения — сколько сообщение ждёт обработки. Важнее глубины: тысяча свежих сообщений не так страшна, как одно, застрявшее на час.
• RSS воркера — если он не сбрасывается при перезапусках, значит --memory-limit не срабатывает или утечка быстрее лимита.

Инвалидация кеша Doctrine и настройка супервизора — отдельные темы; здесь мы разбирали только поведение самого Messenger под потоком. Следующий шаг: настроить failure_transport, повесить алерт на его глубину и прогнать хендлеры на идемпотентность, отправив одно и то же сообщение дважды вручную через messenger:consume с локального дубликата. Если второй прогон изменил состояние — идемпотентность не готова к at-least-once.