Replace basic_get polling with basic_consume push for AMQP consumers (cron path)

[thebraziliandeveloper]

Description

Magento already uses basic_consume (push-based) when running a consumer in daemon mode from the CLI — Consumer::process() calls Queue::subscribe() directly when no message limit is set, and Queue::subscribe() has always used basic_consume + basic_qos prefetch:

// Consumer::process() — lib/internal/Magento/Framework/MessageQueue/Consumer.php
if (!isset($maxNumberOfMessages)) {
    $queue->subscribe($this->getTransactionCallback($queue)); // ← basic_consume ✓
} else {
    $this->invoker->invoke($queue, $maxNumberOfMessages, ...); // ← basic_get ✗
}

However, when --max-messages is set — which is the path taken by every cron-spawned consumer via ConsumersRunner — execution falls into CallbackInvoker::invoke(), which uses a dequeue() loop. dequeue() calls basic_get() once per message: a synchronous request/response that requires one full network round-trip to RabbitMQ per message.

This means the same consumer that efficiently subscribes when run manually in the CLI degrades to sequential polling the moment cron spawns it with --max-messages=10000. RabbitMQ's own documentation explicitly discourages this pattern:

"BasicGet is not suitable for high throughput consumption … prefer Basic.Consume."
— https://www.rabbitmq.com/docs/consumers#basics

This PR makes the cron path consistent with the CLI daemon path. It adds Queue::subscribeWithLimit(callable $callback, int $maxMessages, int $waitTimeout) — structurally identical to the existing Queue::subscribe(), with the addition of a basic_cancel call once $maxMessages have been processed. CallbackInvoker::invoke() now routes AMQP queues with a message limit through this method instead of the dequeue() loop.

As part of this change, the duplicated AMQP message-to-envelope property mapping that existed separately in subscribe() and the new method has been extracted into a private helper createEnvelopeFromAmqpMessage().

The existing dequeue() polling loop is preserved unchanged for all other queue backends (MySQL, STOMP).

Behaviour matrix:

Scenario	Before	After
AMQP + --max-messages (cron)	basic_get loop — 1 RTT/message	basic_consume + prefetch — batch push
AMQP + no limit (CLI daemon)	basic_consume via subscribe()	basic_consume via subscribe() (unchanged)
MySQL / STOMP + --max-messages	dequeue() loop	dequeue() loop (unchanged)

The consumers_wait_for_messages deployment config is respected:

• 1 (default): channel->wait() blocks indefinitely — same as original
• 0: 1-second idle timeout on channel->wait(), exits promptly when the queue drains — matches original immediate-exit behaviour

The poison pill check runs before each message callback, consistent with the original implementation.

---

Benchmark

Environment: Warden / Docker on Apple M-series, RabbitMQ co-located in the same Docker network (~0.1 ms RTT).

Method: A self-contained PHP script bootstraps Magento to obtain the AMQP connection config, declares a temporary durable queue, publishes N messages, then times each approach independently before deleting the queue. Both rounds use identical message payloads and per-message basic_ack.

Benchmark script (click to expand)

<?php
// Usage: warden env exec php-fpm php dev/tools/amqp-benchmark.php [--messages=5000] [--prefetch=100]
declare(strict_types=1);

use Magento\Framework\Amqp\Config as AmqpConfig;
use Magento\Framework\App\Bootstrap;
use Magento\Framework\App\State;

require __DIR__ . '/../../app/bootstrap.php';

$opts          = getopt('', ['messages::', 'prefetch::']);
$totalMessages = (int) ($opts['messages'] ?? 5000);
$prefetchCount = (int) ($opts['prefetch'] ?? 100);

$bootstrap  = Bootstrap::create(BP, $_SERVER);
$om         = $bootstrap->getObjectManager();
$om->get(State::class)->setAreaCode('global');

/** @var AmqpConfig $amqpConfig */
$amqpConfig = $om->get(AmqpConfig::class);

const QUEUE_NAME = 'amqp_benchmark_tmp';

function getChannel(AmqpConfig $cfg): \PhpAmqpLib\Channel\AMQPChannel
{
    $ch = $cfg->getChannel();
    $ch->queue_declare(QUEUE_NAME, false, true, false, false);
    return $ch;
}

function publish(AmqpConfig $cfg, int $n): void
{
    $ch      = getChannel($cfg);
    $payload = json_encode(['product_ids' => [1], 'attributes' => ['meta_description' => 'bench'], 'store_id' => 0]);
    echo "  Publishing $n messages... ";
    $t = microtime(true);
    for ($i = 0; $i < $n; $i++) {
        $ch->basic_publish(new \PhpAmqpLib\Message\AMQPMessage($payload, ['delivery_mode' => 2]), '', QUEUE_NAME);
    }
    printf("done [%.3fs]\n", microtime(true) - $t);
}

function benchGet(AmqpConfig $cfg, int $total): array
{
    $ch = getChannel($cfg); $done = 0; $t = microtime(true);
    while ($done < $total) {
        $msg = $ch->basic_get(QUEUE_NAME);
        if ($msg === null) { usleep(1000); continue; }
        $ch->basic_ack($msg->delivery_info['delivery_tag']);
        $done++;
    }
    $e = microtime(true) - $t;
    return ['elapsed' => $e, 'msg_per_s' => round($done / $e)];
}

function benchConsume(AmqpConfig $cfg, int $total, int $prefetch): array
{
    $ch = getChannel($cfg); $done = 0; $t = microtime(true);
    $ch->basic_qos(0, $prefetch, false);
    $ch->basic_consume(QUEUE_NAME, '', false, false, false, false,
        function ($msg) use ($ch, &$done, $total) {
            $ch->basic_ack($msg->delivery_info['delivery_tag']);
            if (++$done >= $total) { $ch->basic_cancel($msg->delivery_info['consumer_tag']); }
        });
    while (count($ch->callbacks)) { $ch->wait(null, false, 5.0); }
    $e = microtime(true) - $t;
    return ['elapsed' => $e, 'msg_per_s' => round($done / $e)];
}

printf("\nMessages: %d | Prefetch: %d\n\n", $totalMessages, $prefetchCount);
echo "[1/2] basic_get\n"; getChannel($amqpConfig)->queue_purge(QUEUE_NAME); publish($amqpConfig, $totalMessages);
$rGet = benchGet($amqpConfig, $totalMessages);
printf("  %.3fs — %d msg/s\n\n", $rGet['elapsed'], $rGet['msg_per_s']);

echo "[2/2] basic_consume + prefetch\n"; getChannel($amqpConfig)->queue_purge(QUEUE_NAME); publish($amqpConfig, $totalMessages);
$rCon = benchConsume($amqpConfig, $totalMessages, $prefetchCount);
printf("  %.3fs — %d msg/s\n\n", $rCon['elapsed'], $rCon['msg_per_s']);

printf("Speedup: %.1fx | Time saved: %.3fs\n", $rGet['elapsed'] / $rCon['elapsed'], $rGet['elapsed'] - $rCon['elapsed']);
getChannel($amqpConfig)->queue_delete(QUEUE_NAME);

Results — 5,000 messages, prefetch=100 (3 runs):

Run	basic_get before	basic_consume after	Speedup
1	0.720s (6,945 msg/s)	0.302s (16,535 msg/s)	2.4×
2	0.674s (7,415 msg/s)	0.307s (16,285 msg/s)	2.2×
3	0.884s (5,659 msg/s)	0.336s (14,871 msg/s)	2.6×

Prefetch value comparison (5,000 messages, single run each):

Prefetch	Time	Throughput
10	0.363s	13,791 msg/s
50	0.373s	13,414 msg/s
100 (default)	0.337s	14,859 msg/s
200	0.444s	11,267 msg/s
500	0.299s	16,727 msg/s

These numbers are from a co-located Docker environment where RTT is ~0.1 ms. In a production deployment with a dedicated RabbitMQ host the difference is proportionally larger — at a typical 2 ms RTT:

• basic_get: 5,000 × 2 ms = ~10 seconds of pure queue overhead per cron consumer run
• basic_consume: ~50 round-trips × 2 ms = ~0.1 seconds

---

Manual testing scenarios

1. Reproduce the benchmark

   # Create 5,000 fixture products (run once)
   warden env exec php-fpm php dev/tools/amqp-benchmark-fixtures.php
   
   # Run the benchmark — prints before/after comparison
   warden env exec php-fpm php dev/tools/amqp-benchmark.php --messages=5000 --prefetch=100

2. Real consumer end-to-end — trigger a mass product attribute update from Admin to populate the product_action_attribute.update queue, then:

   php bin/magento queue:consumers:start product_action_attribute.update --max-messages=100 --single-thread

   Verify products are updated and var/log/ is clean.

3. MySQL queue unaffected — configure a consumer on a db connection and confirm the original dequeue() loop is still used (no subscribeWithLimit call).

4. consumers_wait_for_messages=0 — set in env.php, run a consumer against an empty queue, confirm it exits within ~1 second instead of blocking indefinitely.

5. --max-messages=0 edge case — confirm the consumer exits immediately without connecting to RabbitMQ (guarded by if ($maxMessages <= 0) { return; }).

---

Related Pull Requests

Fixed Issues (if relevant)

Questions or comments

This PR is intentionally narrow in scope: only CallbackInvoker and Queue are changed, and only for the AMQP + limited-message path. The dequeue() method is left untouched. Happy to split into smaller commits or add integration test coverage if the maintainers prefer.

Contribution checklist

• Pull request has a meaningful description of its purpose
• All commits are accompanied by meaningful commit messages
• All new or changed code is covered with unit/integration tests (if applicable)
• All automated tests passed successfully (all builds are green)

[rhoerr]

Hi @gowrizrh -- I thought of you given your AMQP experience. Any chance you can review this?

Thanks @thebraziliandeveloper for the PR

[gowrizrh]

This is a nice addition and it looks good overall, but I'll need more time to dive deeper.

@rhoerr It looks like there are additional changes apart from the author's original modifications, is this expected? Should I ignore everything except the changes part of 3ea0cc0

[rhoerr]

This is a nice addition and it looks good overall, but I'll need more time to dive deeper.

@rhoerr It looks like there are additional changes apart from the author's original modifications, is this expected? Should I ignore everything except the changes part of 3ea0cc0

Pardon that -- I changed the PR target and that brought some other work along. I updated the target branch to fix that. This PR is isolated to the single commit now. Thank you

[gowrizrh]

This is a neat change, thanks. I've approved it

[thebraziliandeveloper]

Thanks @gowrizrh

I see some failing checks, should I fix them so the team can merge the PR?

[rhoerr]

Thanks @thebraziliandeveloper , I reran PR checks. Can you look at the AMQP unit test and integration test errors?

[thebraziliandeveloper]

@rhoerr will check this week