src/Symfony/Component/Messenger/Command/FailedMessagesRetryCommand.php - 8.2 - githubmirror/symfony

src/Symfony/Component/Messenger/Command/FailedMessagesRetryCommand.php
330 строк14 KB
Nicolas Grekas
[Messenger] Filter failed messages by class and failure time
01 авг 2026, 10:06
01 авг 2026, 10:06b19cf8f
<?php

/*
 * This file is part of the Symfony package.
 *
 * (c) Fabien Potencier <fabien@symfony.com>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Symfony\Component\Messenger\Command;

use Psr\Log\LoggerInterface;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\SignalableCommandInterface;
use Symfony\Component\Console\Exception\RuntimeException;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
use Symfony\Component\Messenger\Event\WorkerMessageSkipEvent;
use Symfony\Component\Messenger\EventListener\StopWorkerOnMessageLimitListener;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\SentToFailureTransportStamp;
use Symfony\Component\Messenger\Transport\Receiver\ListableReceiverInterface;
use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
use Symfony\Component\Messenger\Transport\Receiver\SingleMessageReceiver;
use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
use Symfony\Component\Messenger\Worker;
use Symfony\Contracts\Service\ServiceProviderInterface;

/**
 * @author Ryan Weaver <ryan@symfonycasts.com>
 */
#[AsCommand(name: 'messenger:failed:retry', description: 'Retry one or more messages from the failure transport')]
class FailedMessagesRetryCommand extends AbstractFailedMessagesCommand implements SignalableCommandInterface
{
    private const DEFAULT_KEEPALIVE_INTERVAL = 5;

    private bool $shouldStop = false;
    private bool $forceExit = false;
    private ?Worker $worker = null;

    public function __construct(
        ?string $globalReceiverName,
        ServiceProviderInterface $failureTransports,
        private MessageBusInterface $messageBus,
        private EventDispatcherInterface $eventDispatcher,
        private ?LoggerInterface $logger = null,
        ?PhpSerializer $phpSerializer = null,
        private ?array $signals = null,
    ) {
        parent::__construct($globalReceiverName, $failureTransports, $phpSerializer);
    }

    protected function configure(): void
    {
        $this
            ->setDefinition([
                new InputArgument('id', InputArgument::IS_ARRAY, 'Specific message id(s) to retry'),
                new InputOption('force', null, InputOption::VALUE_NONE, 'Force action without confirmation'),
                new InputOption('transport', null, InputOption::VALUE_REQUIRED, 'Use a specific failure transport', self::DEFAULT_TRANSPORT_OPTION),
                new InputOption('keepalive', null, InputOption::VALUE_REQUIRED, 'Whether to use the transport\'s keepalive mechanism if implemented', self::DEFAULT_KEEPALIVE_INTERVAL),
                new InputOption('class-filter', null, InputOption::VALUE_REQUIRED, 'Filter by a specific class name'),
                new InputOption('failed-after', null, InputOption::VALUE_REQUIRED, 'Only select messages that failed at or after this date; messages with no known failure time are never selected'),
                new InputOption('failed-before', null, InputOption::VALUE_REQUIRED, 'Only select messages that failed at or before this date; messages with no known failure time are never selected'),
            ])
            ->setHelp(<<<'EOF'
                The <info>%command.name%</info> retries message in the failure transport.

                    <info>php %command.full_name%</info>

                The command will interactively ask if each message should be retried,
                discarded or skipped.

                Some transports support retrying a specific message id, which comes
                from the <info>messenger:failed:show</info> command.

                    <info>php %command.full_name% {id}</info>

                Or pass multiple ids at once to process multiple messages:

                <info>php %command.full_name% {id1} {id2} {id3}</info>

                Instead of ids, messages can be selected by class name, by failure time, or by both:

                    <info>php %command.full_name% --class-filter='App\Message\SendEmail'</info>
                    <info>php %command.full_name% --failed-after='-1 hour'</info>
                    <info>php %command.full_name% --failed-after='2024-05-01 08:00' --failed-before='2024-05-01 09:30'</info>

                The "--failed-after" and "--failed-before" options accept any expression supported by
                DateTimeImmutable and both bounds are inclusive. The failure time comes from the message
                history, so messages that were never redelivered are never selected by these options.

                Filters cannot be combined with message ids. Add "--force" to retry every matching message
                without being asked about each of them.

                EOF
            )
        ;
    }

    protected function initialize(InputInterface $input, OutputInterface $output): void
    {
        if ($input->hasParameterOption('--keepalive')) {
            $this->getApplication()->setAlarmInterval((int) ($input->getOption('keepalive') ?? self::DEFAULT_KEEPALIVE_INTERVAL));
        }
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $ids = $input->getArgument('id');
        [$classFilter, $failedAfter, $failedBefore] = $this->getFilters($input, (bool) $ids);

        $this->eventDispatcher->addSubscriber(new StopWorkerOnMessageLimitListener(1));

        $io = new SymfonyStyle($input, $output);
        $errorIo = $io->getErrorStyle();
        $errorIo->comment('Quit this command with CONTROL-C.');
        if (!$output->isVeryVerbose()) {
            $errorIo->comment('Re-run the command with a -vv option to see logs about consumed messages.');
        }

        $failureTransportName = $input->getOption('transport');
        if (self::DEFAULT_TRANSPORT_OPTION === $failureTransportName) {
            $this->printWarningAvailableFailureTransports($errorIo, $this->getGlobalFailureReceiverName());
        }
        if ('' === $failureTransportName || null === $failureTransportName) {
            $failureTransportName = $this->interactiveChooseFailureTransport($errorIo);
        }
        $failureTransportName = self::DEFAULT_TRANSPORT_OPTION === $failureTransportName ? $this->getGlobalFailureReceiverName() : $failureTransportName;

        $receiver = $this->getReceiver($failureTransportName);
        $this->printPendingMessagesMessage($receiver, $io);

        $io->writeln(\sprintf('To retry all the messages, run <comment>messenger:consume %s</comment>', $failureTransportName));

        $shouldForce = $input->getOption('force');

        if (null !== $classFilter || null !== $failedAfter || null !== $failedBefore) {
            if (!$receiver instanceof ListableReceiverInterface) {
                throw new RuntimeException(\sprintf('The "%s" receiver does not support filtering messages.', $failureTransportName));
            }

            $ids = $this->getMessageIdsByFilter($receiver, $classFilter, $failedAfter, $failedBefore);

            if (!$ids) {
                throw new RuntimeException('No failed messages were found with this filter.');
            }
        }

        if (0 === \count($ids)) {
            if (!$input->isInteractive()) {
                throw new RuntimeException('Message id must be passed when in non-interactive mode.');
            }

            $this->runInteractive($failureTransportName, $io, $errorIo, $shouldForce);

            return 0;
        }

        $this->retrySpecificIds($failureTransportName, $ids, $io, $errorIo, $shouldForce);

        if (!$this->shouldStop) {
            $io->success('All done!');
        }

        return 0;
    }

    public function getSubscribedSignals(): array
    {
        return $this->signals ?? (\extension_loaded('pcntl') ? [\SIGTERM, \SIGINT, \SIGQUIT, \SIGALRM] : []);
    }

    public function handleSignal(int $signal, int|false $previousExitCode = 0): int|false
    {
        if (!$this->worker) {
            return false;
        }

        if (\SIGALRM === $signal) {
            $this->logger?->info('Sending keepalive request.', ['transport_names' => $this->worker->getMetadata()->getTransportNames()]);

            $this->worker->keepalive($this->getApplication()->getAlarmInterval());

            return false;
        }

        $this->logger?->info('Received signal {signal}.', ['signal' => $signal, 'transport_names' => $this->worker->getMetadata()->getTransportNames()]);

        $this->worker->stop();
        $this->shouldStop = true;

        return $this->forceExit ? 0 : false;
    }

    private function runInteractive(string $failureTransportName, SymfonyStyle $io, SymfonyStyle $errorIo, bool $shouldForce): void
    {
        $receiver = $this->failureTransports->get($failureTransportName);
        $count = 0;
        if ($receiver instanceof ListableReceiverInterface) {
            // for listable receivers, find the messages one-by-one
            // this avoids using get(), which for some less-robust
            // transports (like Doctrine), will cause the message
            // to be temporarily "acked", even if the user aborts
            // handling the message
            while (!$this->shouldStop) {
                $envelopes = [];
                $this->phpSerializer?->acceptPhpIncompleteClass();
                try {
                    foreach ($receiver->all(1) as $envelope) {
                        ++$count;
                        $envelopes[] = $envelope;
                    }
                } finally {
                    $this->phpSerializer?->rejectPhpIncompleteClass();
                }

                // break the loop if all messages are consumed
                if (0 === \count($envelopes)) {
                    break;
                }

                $this->retrySpecificEnvelopes($envelopes, $failureTransportName, $io, $errorIo, $shouldForce);
            }
        } else {
            // get() and ask messages one-by-one
            $count = $this->runWorker($failureTransportName, $receiver, $io, $errorIo, $shouldForce);
        }

        // avoid success message if nothing was processed
        if (1 <= $count && !$this->shouldStop) {
            $io->success('All failed messages have been handled or removed!');
        }
    }

    private function runWorker(string $failureTransportName, ReceiverInterface $receiver, SymfonyStyle $io, SymfonyStyle $errorIo, bool $shouldForce): int
    {
        $count = 0;
        $listener = function (WorkerMessageReceivedEvent $messageReceivedEvent) use ($io, $errorIo, $receiver, $shouldForce, &$count) {
            ++$count;
            $envelope = $messageReceivedEvent->getEnvelope();

            $this->displaySingleMessage($envelope, $io, $errorIo);

            $this->forceExit = true;
            try {
                $choice = $shouldForce ? 'retry' : $errorIo->choice('Please select an action', ['retry', 'delete', 'skip'], 'retry');
                $shouldHandle = 'retry' === $choice;
            } finally {
                $this->forceExit = false;
            }

            if ($shouldHandle) {
                return;
            }

            if ('skip' === $choice) {
                $this->eventDispatcher->dispatch(new WorkerMessageSkipEvent($envelope, $envelope->last(SentToFailureTransportStamp::class)->getOriginalReceiverName()));
            }

            $messageReceivedEvent->shouldHandle(false);
            $receiver->reject($envelope);
        };
        $this->eventDispatcher->addListener(WorkerMessageReceivedEvent::class, $listener);

        $this->worker = new Worker(
            [$failureTransportName => $receiver],
            $this->messageBus,
            $this->eventDispatcher,
            $this->logger
        );

        try {
            $this->worker->run();
        } finally {
            $this->worker = null;
            $this->eventDispatcher->removeListener(WorkerMessageReceivedEvent::class, $listener);
        }

        return $count;
    }

    private function retrySpecificIds(string $failureTransportName, array $ids, SymfonyStyle $io, SymfonyStyle $errorIo, bool $shouldForce): void
    {
        $receiver = $this->getReceiver($failureTransportName);

        if (!$receiver instanceof ListableReceiverInterface) {
            throw new RuntimeException(\sprintf('The "%s" receiver does not support retrying messages by id.', $failureTransportName));
        }

        foreach ($ids as $id) {
            $this->phpSerializer?->acceptPhpIncompleteClass();
            try {
                $envelope = $receiver->find($id);
            } finally {
                $this->phpSerializer?->rejectPhpIncompleteClass();
            }
            if (null === $envelope) {
                throw new RuntimeException(\sprintf('The message "%s" was not found.', $id));
            }

            $singleReceiver = new SingleMessageReceiver($receiver, $envelope);
            $this->runWorker($failureTransportName, $singleReceiver, $io, $errorIo, $shouldForce);

            if ($this->shouldStop) {
                break;
            }
        }
    }

    private function retrySpecificEnvelopes(array $envelopes, string $failureTransportName, SymfonyStyle $io, SymfonyStyle $errorIo, bool $shouldForce): void
    {
        $receiver = $this->getReceiver($failureTransportName);

        foreach ($envelopes as $envelope) {
            $singleReceiver = new SingleMessageReceiver($receiver, $envelope);
            $this->runWorker($failureTransportName, $singleReceiver, $io, $errorIo, $shouldForce);

            if ($this->shouldStop) {
                break;
            }
        }
    }
}