Skip to content

[Messenger] Rabbitmq delayed quorum queues #60298

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 6 commits into
base: 7.2
Choose a base branch
from
Open

[Messenger] Rabbitmq delayed quorum queues #60298

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
93235c8
issues/57867 delayed queue for days to allow quorum queues with high …
Apr 28, 2025
846d52c
issues/57867 improve commend message
Apr 28, 2025
181a108
issues/57867 improve commend message
Apr 29, 2025
8fa00f6
issues/57867 add tests
Apr 29, 2025
a0c0660
issues/57867 fix coding standard issues
Apr 29, 2025
8482fe9
issues/57867 fix coding standard issues
Apr 29, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 35 additions & 4 deletions src/Symfony/Component/Messenger/Bridge/Amqp/Tests/Transport/ConnectionTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -507,6 +507,22 @@ public function testItDelaysTheMessage()
$connection->publish('{}', ['x-some-headers' => 'foo'], 5000);
}

public function testItDelaysTheMessageWithDailyDelayQueues()
{
$delayExchange = $this->createMock(\AMQPExchange::class);
$date = (new \DateTimeImmutable())->format('Y-m-d');
$delayExchange->expects($this->once())
->method('publish')
->with('{}', "delay_messages__5000_delay_$date", \AMQP_NOPARAM, [
'headers' => ['x-some-headers' => 'foo'],
'delivery_mode' => 2,
'timestamp' => time(),
]);
$connection = $this->createDelayOrRetryConnection($delayExchange, self::DEFAULT_EXCHANGE_NAME, "delay_messages__5000_delay_$date", true);

$connection->publish('{}', ['x-some-headers' => 'foo'], 5000);
}

public function testItRetriesTheMessage()
{
$delayExchange = $this->createMock(\AMQPExchange::class);
Expand All @@ -520,6 +536,20 @@ public function testItRetriesTheMessage()
$connection->publish('{}', [], 5000, $amqpStamp);
}

public function testItRetriesTheMessageWithDailyDelayQueues()
{
$delayExchange = $this->createMock(\AMQPExchange::class);
$date = (new \DateTimeImmutable())->format('Y-m-d');
$delayExchange->expects($this->once())
->method('publish')
->with('{}', "delay_messages__5000_retry_$date", \AMQP_NOPARAM);
$connection = $this->createDelayOrRetryConnection($delayExchange, '', "delay_messages__5000_retry_$date", true);

$amqpEnvelope = $this->createMock(\AMQPEnvelope::class);
$amqpStamp = AmqpStamp::createFromAmqpEnvelope($amqpEnvelope, null, '');
$connection->publish('{}', [], 5000, $amqpStamp);
}

public function testItDelaysTheMessageWithADifferentRoutingKeyAndTTLs()
{
$amqpConnection = $this->createMock(\AMQPConnection::class);
Expand Down Expand Up @@ -849,7 +879,7 @@ public function testItWillRetryMaxThreeTimesWhenAMQPConnectionExceptionIsThrown(
$connection->publish('body');
}

private function createDelayOrRetryConnection(\AMQPExchange $delayExchange, string $deadLetterExchangeName, string $delayQueueName): Connection
private function createDelayOrRetryConnection(\AMQPExchange $delayExchange, string $deadLetterExchangeName, string $delayQueueName, bool $dailyDelayQueues = false): Connection
{
$amqpConnection = $this->createMock(\AMQPConnection::class);
$amqpChannel = $this->createMock(\AMQPChannel::class);
Expand All @@ -861,19 +891,20 @@ private function createDelayOrRetryConnection(\AMQPExchange $delayExchange, stri
$delayQueue = $this->createMock(\AMQPQueue::class);
$factory->method('createQueue')->willReturn($this->createMock(\AMQPQueue::class), $delayQueue);
$factory->method('createExchange')->willReturn($this->createMock(\AMQPExchange::class), $delayExchange);

$baseExpire = $dailyDelayQueues ? 86400 * 1000 : 0;
$delayQueue->expects($this->once())->method('setName')->with($delayQueueName);
$delayQueue->expects($this->once())->method('setArguments')->with([
'x-message-ttl' => 5000,
'x-expires' => 5000 + 10000,
'x-expires' => 5000 + 10000 + $baseExpire,
'x-dead-letter-exchange' => $deadLetterExchangeName,
'x-dead-letter-routing-key' => '',
]);

$delayQueue->expects($this->once())->method('declareQueue');
$delayQueue->expects($this->once())->method('bind')->with('delays', $delayQueueName);
$options = $dailyDelayQueues ? ['delay' => ['daily_delay_queues' => true]] : [];

return Connection::fromDsn('amqp://localhost', [], $factory);
return Connection::fromDsn('amqp://localhost', $options, $factory);
}
}

Expand Down
15 changes: 13 additions & 2 deletions src/Symfony/Component/Messenger/Bridge/Amqp/Transport/Connection.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ class Connection
'flags',
'arguments',
];
private const BASE_EXPIRATION = 24 * 60 * 60 * 1000;

private AmqpFactory $amqpFactory;
private mixed $autoSetupExchange;
Expand Down Expand Up @@ -143,6 +144,11 @@ public function __construct(
* * queue_name_pattern: Pattern to use to create the queues (Default: "delay_%exchange_name%_%routing_key%_%delay%")
* * exchange_name: Name of the exchange to be used for the delayed/retried messages (Default: "delays")
* * arguments: array of extra delay queue arguments (for example: ['x-queue-type' => 'classic', 'x-message-deduplication' => true,])
* * daily_delay_queues: When true, delay queues will be created with names including the current date
* (e.g., '%queue_name_pattern%_%current_date%'). These queues are automatically deleted by RabbitMQ after they
* expire (x-expires argument), the x-expires argument is set to 24 hours (24 * 60 * 60 * 1000) plus delay. This is useful for quorum queues.
* because quorum queues do not redeclare expire time.
* (Default: false)
* * auto_setup: Enable or not the auto-setup of queues and exchanges (Default: true)
*
* * Connection tuning options (see http://www.rabbitmq.com/amqp-0-9-1-reference.html#connection.tune for details):
Expand Down Expand Up @@ -389,11 +395,15 @@ private function createDelayQueue(int $delay, ?string $routingKey, bool $isRetry
$queue = $this->amqpFactory->createQueue($this->channel());
$queue->setName($this->getRoutingKeyForDelay($delay, $routingKey, $isRetryAttempt));
$queue->setFlags(\AMQP_DURABLE);
$queueExpirationBase = ($this->connectionOptions['delay']['daily_delay_queues'] ?? false) ?
self::BASE_EXPIRATION : 0;
$queue->setArguments(array_merge([
'x-message-ttl' => $delay,
// delete the delay queue 10 seconds after the message expires
// publishing another message redeclares the queue which renews the lease
'x-expires' => $delay + 10000,
// For quorum queues, redeclaration is not allowed, so using daily_delay_queues=true is recommended to manage cleanup.
// It will create a new queue for each day, with x-expires set to 24 hours (24 * 60 * 60 * 1000) plus delay.
'x-expires' => $queueExpirationBase + $delay + 10000,
// message should be broadcast to all consumers during delay, but to only one queue during retry
// empty name is default direct exchange
'x-dead-letter-exchange' => $isRetryAttempt ? '' : $this->exchangeOptions['name'],
Expand All @@ -408,12 +418,13 @@ private function createDelayQueue(int $delay, ?string $routingKey, bool $isRetry
private function getRoutingKeyForDelay(int $delay, ?string $finalRoutingKey, bool $isRetryAttempt): string
{
$action = $isRetryAttempt ? '_retry' : '_delay';
$date = ($this->connectionOptions['delay']['daily_delay_queues'] ?? false) ? '_'.(new \DateTimeImmutable())->format('Y-m-d') : '';

return str_replace(
['%delay%', '%exchange_name%', '%routing_key%'],
[$delay, $this->exchangeOptions['name'], $finalRoutingKey ?? ''],
$this->connectionOptions['delay']['queue_name_pattern']
).$action;
).$action.$date;
}

/**
Expand Down