Symfony Messenger: retry с jitter и failure transport

Transport и очередь failed

framework:
  messenger:
    failure_transport: failed

    transports:
      async:
        dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
        retry_strategy:
          service: App\Messenger\Retry\JitterRetryStrategy

      failed:
        dsn: 'doctrine://default?queue_name=failed'

    routing:
      'App\Message\GenerateReport': async

failed хранится отдельно от рабочей очереди. Сообщение попадает туда после исчерпания попыток и не крутится бесконечно.

RetryStrategy с jitter

<?php

namespace App\Messenger\Retry;

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;
use Symfony\Component\Messenger\Retry\RetryStrategyInterface;
use Symfony\Component\Messenger\Stamp\RedeliveryStamp;

final class JitterRetryStrategy implements RetryStrategyInterface
{
    private const MAX_RETRIES = 4;
    private const INITIAL_DELAY = 1_000;
    private const MAX_DELAY = 60 * 60 * 1_000;

    public function isRetryable(
        Envelope $envelope,
        ?\Throwable $throwable = null,
    ): bool {
        if ($throwable instanceof UnrecoverableMessageHandlingException) {
            return false;
        }

        return $this->retryCount($envelope) < self::MAX_RETRIES;
    }

    public function getWaitingTime(
        Envelope $envelope,
        ?\Throwable $throwable = null,
    ): int {
        $base = min(
            self::INITIAL_DELAY * (2 ** $this->retryCount($envelope)),
            self::MAX_DELAY,
        );

        $spread = (int) round($base * 0.20);

        return max(0, $base + random_int(-$spread, $spread));
    }

    private function retryCount(Envelope $envelope): int
    {
        $stamp = $envelope->last(RedeliveryStamp::class);

        return $stamp instanceof RedeliveryStamp
            ? $stamp->getRetryCount()
            : 0;
    }
}

Messenger ожидает задержку в миллисекундах. При базовых значениях попытки идут примерно через 1, 2, 4 и 8 секунд с разбросом ±20%. Ограничение в час защищает от переполнения integer при большом retry count.

Какие ошибки не повторять

use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;

if (!$report->canBeGenerated()) {
    throw new UnrecoverableMessageHandlingException(
        'Report input is permanently invalid.',
    );
}

Ошибка валидации, отсутствующий обязательный объект или запрещённый переход не станут исправнее через десять секунд. Timeout, 429 и временный 5xx можно повторять. Эту границу лучше выражать типом exception, а не поиском текста в message.

Команды эксплуатации

php bin/console messenger:consume async --time-limit=3600 --memory-limit=256M -vv
php bin/console messenger:failed:show
php bin/console messenger:failed:retry
php bin/console messenger:failed:remove

После deploy воркер перезапускается: долгоживущий процесс не подхватит новый PHP-код сам. Для production messenger:consume держит Supervisor или systemd, а --time-limit обеспечивает регулярный чистый restart.

Retry не должен дублировать эффект

Стратегия повторов бесполезна, если handler дважды списывает деньги или отправляет два письма. В сообщение добавляют operation ID, а в БД — уникальный индекс на него. Запись результата и отметка обработки выполняются в одной транзакции.

Минимальная проверка: первая попытка меняет состояние и падает перед ack, вторая получает то же сообщение и завершает работу без второго побочного эффекта.