AmqpConsumer can block other receives forever

[jwage]

If you run messenger:consumer with multiple like this:

php bin/console messenger:consume transport1 transport2

If transport2 is full of queued messages, AmqpConsumer can yield messages forever and block other transports. We should have some way to configure how often to yield control back to the parent loop either via number of messages or a time limit.

[BullkyCoder]

I haven't found fix for it with basic_consume, what we did is to switch to basic_get and implement round-robin. For our cases where we have always messages on queues polling shouldn't be big performance issue

<?php

declare(strict_types=1);

namespace App\Transport;

use Jwage\PhpAmqpLibMessengerBundle\Transport\AmqpEnvelope;
use Jwage\PhpAmqpLibMessengerBundle\Transport\AmqpReceivedStamp;
use Jwage\PhpAmqpLibMessengerBundle\Transport\AmqpTransport;
use Jwage\PhpAmqpLibMessengerBundle\Transport\Connection;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
use Symfony\Component\Messenger\Transport\Receiver\QueueReceiverInterface;
use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
use Symfony\Component\Messenger\Transport\SetupableTransportInterface;
use Symfony\Component\Messenger\Transport\TransportInterface;

/**
 * Decorator that replaces basic_consume with basic_get + round-robin.
 *
 * Ensures fair consumption across multiple queues — no starvation.
 */
final class PollingAmqpTransport implements TransportInterface, QueueReceiverInterface, MessageCountAwareInterface, SetupableTransportInterface
{
    private int $lastQueueIndex = 0;
    private ?Connection $connection = null;

    public function __construct(
        private readonly AmqpTransport $inner,
        private readonly SerializerInterface $serializer,
        private readonly ?\Psr\Log\LoggerInterface $logger = null,
    ) {
    }

    public function get(): iterable
    {
        return $this->getFromQueues($this->getConnection()->getQueueNames());
    }

    public function getFromQueues(array $queueNames): iterable
    {
        $count = \count($queueNames);
        if ($count === 0) {
            return;
        }

        for ($i = 0; $i < $count; $i++) {
            $index = ($this->lastQueueIndex + $i) % $count;
            $envelope = $this->fetchOneFromQueue($queueNames[$index]);

            if ($envelope !== null) {
                $this->lastQueueIndex = ($index + 1) % $count;
                yield $envelope;
                return;
            }
        }
    }

    public function ack(Envelope $envelope): void
    {
        $this->inner->ack($envelope);
    }

    public function reject(Envelope $envelope): void
    {
        $this->inner->reject($envelope);
    }

    public function send(Envelope $envelope): Envelope
    {
        return $this->inner->send($envelope);
    }

    public function getMessageCount(): int
    {
        return $this->inner->getMessageCount();
    }

    public function setup(): void
    {
        $this->inner->setup();
    }

    private function fetchOneFromQueue(string $queueName): ?Envelope
    {
        $message = $this->getConnection()->channel()->basic_get($queueName);

        if ($message === null) {
            return null;
        }

        $amqpEnvelope = new AmqpEnvelope($message);

        try {
            $envelope = $this->serializer->decode([
                'body' => $amqpEnvelope->getBody(),
                'headers' => $amqpEnvelope->getHeaders(),
            ]);
        } catch (MessageDecodingFailedException $e) {
            // nack without requeue — message is gone from the queue
            // Worker cannot retry because there's no deserializable envelope
            $amqpEnvelope->nack();

            $this->logger?->error('Failed to decode message from queue {queue}: {error}', [
                'queue' => $queueName,
                'error' => $e->getMessage(),
            ]);

            return null;
        }

        if (($messageId = $amqpEnvelope->getMessageId()) !== null) {
            $envelope = $envelope
                ->withoutAll(TransportMessageIdStamp::class)
                ->with(new TransportMessageIdStamp($messageId));
        }

        return $envelope->with(new AmqpReceivedStamp($amqpEnvelope, $queueName));
    }

    private function getConnection(): Connection
    {
        if ($this->connection === null) {
            $this->connection = $this->inner->getConnection();
            $this->inner->setup();
        }

        return $this->connection;
    }
}

/**
 * Rotates receiver order in Worker after each handled message → round-robin across transports.
 *
 * Without this, Worker always checks receivers in declaration order (priority mode).
 * With basic_get (non-blocking), get() returns immediately — rotation has effect
 * because the next iteration starts from a different receiver.
 */
final class RoundRobinWorkerSubscriber implements EventSubscriberInterface
{
    public static function getSubscribedEvents(): array
    {
        return [WorkerRunningEvent::class => 'onWorkerRunning'];
    }

    public function onWorkerRunning(WorkerRunningEvent $event): void
    {
        if ($event->isWorkerIdle()) {
            return;
        }

        $worker = $event->getWorker();
        $ref = new \ReflectionProperty(Worker::class, 'receivers');
        $receivers = $ref->getValue($worker);

        if (\count($receivers) <= 1) {
            return;
        }

        $keys = array_keys($receivers);
        $values = array_values($receivers);
        $firstKey = array_shift($keys);
        $firstValue = array_shift($values);
        $keys[] = $firstKey;
        $values[] = $firstValue;
        $ref->setValue($worker, array_combine($keys, $values));
    }
}