Skip to content

fix: keep RabbitMQ delay queues per priority and publish before rejecting #107

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
gr8man wants to merge 3 commits into
base: develop
Choose a base branch
from
Open

fix: keep RabbitMQ delay queues per priority and publish before rejecting #107

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5521de9
fix: keep RabbitMQ delays per priority and publish before rejecting
gr8man Nov 10, 2026
0059d76
test: add unit tests for RabbitMQ priority-based delay queue naming a…
gr8man Nov 10, 2026
ddf8ef7
fix: ensure exchange declared and bind once in delayed publish
gr8man Nov 10, 2026
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
77 changes: 49 additions & 28 deletions src/Handlers/RabbitMQHandler.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,15 +144,13 @@ public function push(string $queue, string $job, array $data, ?PayloadMetadata $
$this->publishMessage($queue, $message, $routingKey);
} else {
// Use TTL + dead letter pattern for actual delays
$this->publishDelayedMessage($queue, $queueJob, $routingKey, $realDelay);
$this->publishDelayedMessage($queue, $queueJob, $queueJob->priority, $realDelay);
}
} else {
$message = $this->createMessage($queueJob);
$this->publishMessage($queue, $message, $routingKey);
}

$this->priority = $this->delay = null;

// Emit job pushed event
QueueEventManager::jobPushed(
handler: $this->name(),
Expand All @@ -171,6 +169,9 @@ public function push(string $queue, string $job, array $data, ?PayloadMetadata $
);

return QueuePushResult::failure($e->getMessage());
} finally {
// Never leak priority or delay into the next push operation.
$this->priority = $this->delay = null;
}
}

Expand Down Expand Up @@ -209,15 +210,15 @@ public function later(QueueJob $queueJob, int $seconds): bool
$queueJob->status = Status::PENDING->value;
$queueJob->available_at = Time::now()->addSeconds($seconds);

// Reject the original message without requeue
// Publish the delayed copy first. If publishing fails the original
// message must stay unacknowledged so the job is not lost.
$this->publishDelayedMessage($queueJob->queue, $queueJob, $queueJob->priority, $seconds);

// Reject the original message only after the replacement exists.
if (isset($queueJob->amqpDeliveryTag)) {
$this->channel->basic_nack($queueJob->amqpDeliveryTag, false, false);
}

$routingKey = $this->getRoutingKey($queueJob->queue, $queueJob->priority);

$this->publishDelayedMessage($queueJob->queue, $queueJob, $routingKey, $seconds);

return true;
} catch (Throwable $e) {
log_message('error', 'RabbitMQ later error: ' . $e->getMessage());
Expand Down Expand Up @@ -342,17 +343,18 @@ private function declareExchange(string $queue): void
false,
);

$this->declaredExchanges[$exchangeName] = true;
}
// Bind queues to the exchange only once, together with the
// declaration. Re-binding on every publish is unnecessary traffic.
$priorities = $this->config->queuePriorities[$queue] ?? ['default'];

// Bind queues to exchanges
$priorities = $this->config->queuePriorities[$queue] ?? ['default'];
foreach ($priorities as $priority) {
$queueName = $this->getQueueName($queue, $priority);
$routingKey = $this->getRoutingKey($queue, $priority);

foreach ($priorities as $priority) {
$queueName = $this->getQueueName($queue, $priority);
$routingKey = $this->getRoutingKey($queue, $priority);
$this->channel->queue_bind($queueName, $exchangeName, $routingKey);
}

$this->channel->queue_bind($queueName, $exchangeName, $routingKey);
$this->declaredExchanges[$exchangeName] = true;
}
}

Expand Down Expand Up @@ -398,13 +400,22 @@ private function messageToQueueJob(AMQPMessage $message): QueueJob

/**
* Publish message with delay using per-message TTL + dead letter pattern.
*
* Delay queues are separated per priority so the dead-letter routing key
* always points at the priority bucket the job belongs to.
*/
private function publishDelayedMessage(string $queue, QueueJob $queueJob, string $routingKey, int $delaySeconds): void
private function publishDelayedMessage(string $queue, QueueJob $queueJob, string $priority, int $delaySeconds): void
{
$delayQueueName = $this->getDelayQueueName($queue);
// Make sure the exchange exists. later() may be called by a worker that
// never went through push(), so we cannot rely on push() declaring it.
$this->declareExchange($queue);

$delayQueueName = $this->getDelayQueueName($queue, $priority);
$exchangeName = $this->getExchangeName($queue); // Use the main exchange
$routingKey = $this->getRoutingKey($queue, $priority);

// Declare single delay queue (without queue-level TTL)
// Declare and bind delay queue for this priority (without queue-level TTL).
// Binding only happens on the first publish for this delay queue.
if (! isset($this->declaredQueues[$delayQueueName])) {
$this->channel->queue_declare(
$delayQueueName,
Expand All @@ -419,12 +430,11 @@ private function publishDelayedMessage(string $queue, QueueJob $queueJob, string
]),
);

$this->channel->queue_bind($delayQueueName, $exchangeName, $delayQueueName);

$this->declaredQueues[$delayQueueName] = true;
}

// Bind delay queue to main exchange with delay routing key
$this->channel->queue_bind($delayQueueName, $exchangeName, $delayQueueName);

// Create message with per-message expiration (milliseconds string)
$delayedMessage = $this->createMessage($queueJob, [
'expiration' => (string) ($delaySeconds * 1000),
Expand Down Expand Up @@ -466,15 +476,18 @@ private function publishWithOptionalConfirm(AMQPMessage $message, string $exchan
*/
private function clearQueue(string $queue): void
{
// Purge delayed jobs first so they cannot expire into a queue
// that has already been purged.
$this->purgeQueueIfExists($this->getDelayQueueName($queue));

$priorities = $this->config->queuePriorities[$queue] ?? ['default'];

foreach ($priorities as $priority) {
// Purge delayed jobs first so they cannot expire into a queue
// that has already been purged.
$this->purgeQueueIfExists($this->getDelayQueueName($queue, $priority));
$this->purgeQueueIfExists($this->getQueueName($queue, $priority));
}

// Purge the pre-priority delay queue so jobs scheduled before the
// per-priority delay queues were introduced cannot leak after a clear.
$this->purgeQueueIfExists($this->getLegacyDelayQueueName($queue));
}

/**
Expand Down Expand Up @@ -519,9 +532,17 @@ private function getExchangeName(string $queue): string
}

/**
* Get delay queue name (single queue per logical queue).
* Get delay queue name (one queue per logical queue and priority).
*/
private function getDelayQueueName(string $queue, string $priority): string
{
return "queue_{$queue}_{$priority}_delay";
}

/**
* Get the legacy delay queue name (before per-priority delay queues).
*/
private function getDelayQueueName(string $queue): string
private function getLegacyDelayQueueName(string $queue): string
{
return "queue_{$queue}_delay";
}
Expand Down
Loading
Loading