From 5521de9bf50e7753c0210899fbc93fbe4d0a0686 Mon Sep 17 00:00:00 2001 From: Bogdan Date: Tue, 10 Nov 2026 21:42:03 +0100 Subject: [PATCH 1/3] fix: keep RabbitMQ delays per priority and publish before rejecting --- src/Handlers/RabbitMQHandler.php | 40 ++++++++++++++++++-------------- 1 file changed, 22 insertions(+), 18 deletions(-) diff --git a/src/Handlers/RabbitMQHandler.php b/src/Handlers/RabbitMQHandler.php index 0afe85c..25feda0 100644 --- a/src/Handlers/RabbitMQHandler.php +++ b/src/Handlers/RabbitMQHandler.php @@ -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(), @@ -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; } } @@ -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()); @@ -398,13 +399,17 @@ 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); + $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 delay queue for this priority (without queue-level TTL) if (! isset($this->declaredQueues[$delayQueueName])) { $this->channel->queue_declare( $delayQueueName, @@ -466,13 +471,12 @@ 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)); } } @@ -519,11 +523,11 @@ 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 + private function getDelayQueueName(string $queue, string $priority): string { - return "queue_{$queue}_delay"; + return "queue_{$queue}_{$priority}_delay"; } /** From 0059d76ba2b51d77bb14c70f93f2e090a4f7b677 Mon Sep 17 00:00:00 2001 From: Bogdan Date: Tue, 10 Nov 2026 22:03:38 +0100 Subject: [PATCH 2/3] test: add unit tests for RabbitMQ priority-based delay queue naming and later reschedule order --- tests/RabbitMQHandlerUnitTest.php | 252 ++++++++++++++++++++++++++++++ 1 file changed, 252 insertions(+) create mode 100644 tests/RabbitMQHandlerUnitTest.php diff --git a/tests/RabbitMQHandlerUnitTest.php b/tests/RabbitMQHandlerUnitTest.php new file mode 100644 index 0000000..12537f5 --- /dev/null +++ b/tests/RabbitMQHandlerUnitTest.php @@ -0,0 +1,252 @@ + + * + * For the full copyright and license information, please view + * the LICENSE file that was distributed with this source code. + */ + +namespace Tests; + +use CodeIgniter\I18n\Time; +use CodeIgniter\Queue\Entities\QueueJob; +use CodeIgniter\Queue\Enums\Status; +use CodeIgniter\Queue\Handlers\RabbitMQHandler; +use CodeIgniter\Queue\Payloads\Payload; +use CodeIgniter\Test\ReflectionHelper; +use PhpAmqpLib\Channel\AMQPChannel; +use PhpAmqpLib\Connection\AbstractConnection; +use PhpAmqpLib\Wire\AMQPTable; +use ReflectionClass; +use RuntimeException; +use Tests\Support\Config\Queue as QueueConfig; +use Tests\Support\TestCase; + +/** + * Unit tests for RabbitMQHandler logic that do not require a live RabbitMQ server. + * + * @internal + */ +final class RabbitMQHandlerUnitTest extends TestCase +{ + use ReflectionHelper; + + private QueueConfig $config; + + protected function setUp(): void + { + parent::setUp(); + + $this->config = config(QueueConfig::class); + } + + public function testGetDelayQueueNameIncludesPriority(): void + { + $handler = $this->createHandlerWithMocks(); + $invoker = $this->getPrivateMethodInvoker($handler, 'getDelayQueueName'); + + $this->assertSame('queue_emails_high_delay', $invoker('emails', 'high')); + $this->assertSame('queue_emails_low_delay', $invoker('emails', 'low')); + $this->assertSame('queue_tasks_default_delay', $invoker('tasks', 'default')); + } + + public function testPublishDelayedMessageDeclaresPerPriorityQueueAndDeadLetterRouting(): void + { + $declaredQueue = null; + /** @var AMQPTable|null $declaredArgs */ + $declaredArgs = null; + + $channel = $this->createMock(AMQPChannel::class); + $channel->expects($this->once()) + ->method('queue_declare') + ->willReturnCallback(static function ($queue, $passive, $durable, $exclusive, $autoDelete, $nowait, $arguments) use (&$declaredQueue, &$declaredArgs) { + $declaredQueue = $queue; + $declaredArgs = $arguments; + + return [$queue, 0, 0]; + }); + + $channel->expects($this->once()) + ->method('queue_bind') + ->with('queue_emails_high_delay', 'queue_emails_exchange', 'queue_emails_high_delay'); + + $channel->expects($this->once()) + ->method('basic_publish'); + + $handler = $this->createHandlerWithMocks($channel); + $invoker = $this->getPrivateMethodInvoker($handler, 'publishDelayedMessage'); + + $job = new QueueJob([ + 'id' => 1, + 'queue' => 'emails', + 'payload' => new Payload('success', ['foo' => 'bar']), + 'priority' => 'high', + 'status' => Status::PENDING->value, + 'attempts' => 0, + ]); + + $invoker('emails', $job, 'high', 30); + + $this->assertSame('queue_emails_high_delay', $declaredQueue); + $this->assertInstanceOf(AMQPTable::class, $declaredArgs); + $nativeArgs = $declaredArgs->getNativeData(); + $this->assertSame('queue_emails_exchange', $nativeArgs['x-dead-letter-exchange']); + $this->assertSame('emails.high', $nativeArgs['x-dead-letter-routing-key']); + } + + public function testLaterPublishesDelayedMessageBeforeRejectingOriginalMessage(): void + { + Time::setTestNow('2026-11-10 12:00:00'); + + $callOrder = []; + + $channel = $this->createMock(AMQPChannel::class); + $channel->method('queue_declare') + ->willReturnCallback(static function () use (&$callOrder) { + $callOrder[] = 'queue_declare'; + + return ['queue_emails_high_delay', 0, 0]; + }); + $channel->method('queue_bind') + ->willReturnCallback(static function () use (&$callOrder): void { + $callOrder[] = 'queue_bind'; + }); + $channel->method('basic_publish') + ->willReturnCallback(static function () use (&$callOrder): void { + $callOrder[] = 'basic_publish'; + }); + $channel->expects($this->once()) + ->method('basic_nack') + ->with('delivery-tag-123', false, false) + ->willReturnCallback(static function () use (&$callOrder): void { + $callOrder[] = 'basic_nack'; + }); + + $handler = $this->createHandlerWithMocks($channel); + + $job = new QueueJob([ + 'id' => 1, + 'queue' => 'emails', + 'payload' => new Payload('success', ['foo' => 'bar']), + 'priority' => 'high', + 'status' => Status::RESERVED->value, + 'attempts' => 1, + ]); + $job->amqpDeliveryTag = 'delivery-tag-123'; + + $result = $handler->later($job, 45); + + $this->assertTrue($result); + $this->assertSame(Status::PENDING->value, $job->status); + $this->assertSame(Time::now()->addSeconds(45)->timestamp, $job->available_at->timestamp); + + // Verify order: publishing occurred before basic_nack + $publishIndex = array_search('basic_publish', $callOrder, true); + $nackIndex = array_search('basic_nack', $callOrder, true); + $this->assertNotFalse($publishIndex); + $this->assertNotFalse($nackIndex); + $this->assertLessThan($nackIndex, $publishIndex); + } + + public function testLaterDoesNotRejectOriginalMessageWhenPublishingDelayedFails(): void + { + $channel = $this->createMock(AMQPChannel::class); + $channel->method('basic_publish') + ->willThrowException(new RuntimeException('Failed to publish delayed replacement')); + $channel->expects($this->never()) + ->method('basic_nack'); + + $handler = $this->createHandlerWithMocks($channel); + + $job = new QueueJob([ + 'id' => 1, + 'queue' => 'emails', + 'payload' => new Payload('success', ['foo' => 'bar']), + 'priority' => 'high', + 'status' => Status::RESERVED->value, + 'attempts' => 1, + ]); + $job->amqpDeliveryTag = 'delivery-tag-456'; + + $result = $handler->later($job, 60); + + $this->assertFalse($result); + $this->assertLogContains('error', 'RabbitMQ later error: Failed to publish delayed replacement'); + } + + public function testClearQueuePurgesDelayQueuesForEachPriority(): void + { + $purgedQueues = []; + + $disposableChannel = $this->createMock(AMQPChannel::class); + $disposableChannel->method('queue_purge') + ->willReturnCallback(static function (string $queue) use (&$purgedQueues): int { + $purgedQueues[] = $queue; + + return 0; + }); + $disposableChannel->method('is_open') + ->willReturn(true); + + $connection = $this->createMock(AbstractConnection::class); + $connection->method('channel') + ->willReturn($disposableChannel); + + $config = clone $this->config; + $config->queuePriorities['reports'] = ['high', 'low']; + + $handler = $this->createHandlerWithMocks(null, $connection, $config); + $invoker = $this->getPrivateMethodInvoker($handler, 'clearQueue'); + + $invoker('reports'); + + $this->assertContains('queue_reports_high_delay', $purgedQueues); + $this->assertContains('reports_high', $purgedQueues); + $this->assertContains('queue_reports_low_delay', $purgedQueues); + $this->assertContains('reports_low', $purgedQueues); + } + + public function testPushResetsPriorityAndDelayInFinallyBlockOnFailure(): void + { + $channel = $this->createMock(AMQPChannel::class); + $channel->method('basic_publish') + ->willThrowException(new RuntimeException('Network failure on push')); + + $handler = $this->createHandlerWithMocks($channel); + $handler->setPriority('high')->setDelay(30); + + $this->assertSame('high', self::getPrivateProperty($handler, 'priority')); + $this->assertSame(30, self::getPrivateProperty($handler, 'delay')); + + $result = $handler->push('queue', 'success', ['test' => 'data']); + + $this->assertFalse($result->getStatus()); + $this->assertNull(self::getPrivateProperty($handler, 'priority')); + $this->assertNull(self::getPrivateProperty($handler, 'delay')); + } + + private function createHandlerWithMocks(?AMQPChannel $channel = null, ?AbstractConnection $connection = null, ?QueueConfig $config = null): RabbitMQHandler + { + $ref = new ReflectionClass(RabbitMQHandler::class); + $handler = $ref->newInstanceWithoutConstructor(); + + $config ??= $this->config; + $this->setPrivateProperty($handler, 'config', $config); + + $channel ??= $this->createMock(AMQPChannel::class); + $connection ??= $this->createMock(AbstractConnection::class); + + $this->setPrivateProperty($handler, 'channel', $channel); + $this->setPrivateProperty($handler, 'connection', $connection); + $this->setPrivateProperty($handler, 'declaredLogicalQueues', []); + $this->setPrivateProperty($handler, 'declaredQueues', []); + $this->setPrivateProperty($handler, 'declaredExchanges', []); + + return $handler; + } +} From ddf8ef7a013efff03345ba4eda9e121ce580573c Mon Sep 17 00:00:00 2001 From: Bogdan Date: Wed, 11 Nov 2026 00:00:28 +0100 Subject: [PATCH 3/3] fix: ensure exchange declared and bind once in delayed publish --- src/Handlers/RabbitMQHandler.php | 41 +++++++++++++++------- tests/RabbitMQHandlerUnitTest.php | 57 +++++++++++++++++++++++++++++-- 2 files changed, 83 insertions(+), 15 deletions(-) diff --git a/src/Handlers/RabbitMQHandler.php b/src/Handlers/RabbitMQHandler.php index 25feda0..e0f251d 100644 --- a/src/Handlers/RabbitMQHandler.php +++ b/src/Handlers/RabbitMQHandler.php @@ -343,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; } } @@ -405,11 +406,16 @@ private function messageToQueueJob(AMQPMessage $message): QueueJob */ private function publishDelayedMessage(string $queue, QueueJob $queueJob, string $priority, int $delaySeconds): void { + // 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 delay queue for this priority (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, @@ -424,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), @@ -479,6 +484,10 @@ private function clearQueue(string $queue): void $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)); } /** @@ -530,6 +539,14 @@ 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 getLegacyDelayQueueName(string $queue): string + { + return "queue_{$queue}_delay"; + } + /** * Get routing key for priority. */ diff --git a/tests/RabbitMQHandlerUnitTest.php b/tests/RabbitMQHandlerUnitTest.php index 12537f5..336b4e0 100644 --- a/tests/RabbitMQHandlerUnitTest.php +++ b/tests/RabbitMQHandlerUnitTest.php @@ -60,8 +60,16 @@ public function testPublishDelayedMessageDeclaresPerPriorityQueueAndDeadLetterRo $declaredQueue = null; /** @var AMQPTable|null $declaredArgs */ $declaredArgs = null; + $bindings = []; + $exchanges = []; $channel = $this->createMock(AMQPChannel::class); + $channel->expects($this->once()) + ->method('exchange_declare') + ->willReturnCallback(static function (string $exchange) use (&$exchanges): void { + $exchanges[] = $exchange; + }); + $channel->expects($this->once()) ->method('queue_declare') ->willReturnCallback(static function ($queue, $passive, $durable, $exclusive, $autoDelete, $nowait, $arguments) use (&$declaredQueue, &$declaredArgs) { @@ -71,9 +79,10 @@ public function testPublishDelayedMessageDeclaresPerPriorityQueueAndDeadLetterRo return [$queue, 0, 0]; }); - $channel->expects($this->once()) - ->method('queue_bind') - ->with('queue_emails_high_delay', 'queue_emails_exchange', 'queue_emails_high_delay'); + $channel->method('queue_bind') + ->willReturnCallback(static function (string $queueName, string $exchange, string $routingKey) use (&$bindings): void { + $bindings[] = [$queueName, $exchange, $routingKey]; + }); $channel->expects($this->once()) ->method('basic_publish'); @@ -97,6 +106,46 @@ public function testPublishDelayedMessageDeclaresPerPriorityQueueAndDeadLetterRo $nativeArgs = $declaredArgs->getNativeData(); $this->assertSame('queue_emails_exchange', $nativeArgs['x-dead-letter-exchange']); $this->assertSame('emails.high', $nativeArgs['x-dead-letter-routing-key']); + + // The exchange is declared because later() may run on a worker that never pushed. + $this->assertSame(['queue_emails_exchange'], $exchanges); + + $delayBind = ['queue_emails_high_delay', 'queue_emails_exchange', 'queue_emails_high_delay']; + $this->assertCount(1, array_keys($bindings, $delayBind, true), 'The delay queue must be bound exactly once.'); + } + + public function testPublishDelayedMessageBindsDelayQueueOnlyOnceAcrossPublications(): void + { + $delayBindCount = 0; + + $channel = $this->createMock(AMQPChannel::class); + $channel->method('queue_declare') + ->willReturn(['queue_emails_high_delay', 0, 0]); + $channel->method('queue_bind') + ->willReturnCallback(static function (string $queueName) use (&$delayBindCount): void { + if ($queueName === 'queue_emails_high_delay') { + $delayBindCount++; + } + }); + $channel->expects($this->exactly(2)) + ->method('basic_publish'); + + $handler = $this->createHandlerWithMocks($channel); + $invoker = $this->getPrivateMethodInvoker($handler, 'publishDelayedMessage'); + + $job = new QueueJob([ + 'id' => 1, + 'queue' => 'emails', + 'payload' => new Payload('success', ['foo' => 'bar']), + 'priority' => 'high', + 'status' => Status::PENDING->value, + 'attempts' => 0, + ]); + + $invoker('emails', $job, 'high', 30); + $invoker('emails', $job, 'high', 60); + + $this->assertSame(1, $delayBindCount); } public function testLaterPublishesDelayedMessageBeforeRejectingOriginalMessage(): void @@ -209,6 +258,8 @@ public function testClearQueuePurgesDelayQueuesForEachPriority(): void $this->assertContains('reports_high', $purgedQueues); $this->assertContains('queue_reports_low_delay', $purgedQueues); $this->assertContains('reports_low', $purgedQueues); + // Legacy (pre-priority) delay queue must be purged as well. + $this->assertContains('queue_reports_delay', $purgedQueues); } public function testPushResetsPriorityAndDelayInFinallyBlockOnFailure(): void