GeezerVison

Description

Delayed delivery in RabbitMQHandler (push() with a delay and later()) had three problems.

1. A single delay queue was shared by every priority. queue_{queue}_delay was declared once with x-dead-letter-routing-key fixed to the default priority routing key ({queue}.default). A delayed job with a non-default priority was dead-lettered into the default bucket after its TTL, so priority ordering was lost. In addition, a second process declaring the same argument set differently received PRECONDITION_FAILED (406) and lost its channel.
2. later() rejected the original message before the replacement was published. basic_nack() was called before publishDelayedMessage(). If the publish failed, the original message was already gone — a silently lost job.
3. publishDelayedMessage() relied on the exchange already existing and re-bound the delay queue on every publish. later() can be called by a worker that never went through push(), so the exchange (and its bindings) might not exist at that point. In addition, push() only reset $this->priority/$this->delay on the success path, so a failed push could leak state into the next operation.

Changes

• src/Handlers/RabbitMQHandler.php
  • Delay queues are now created per logical queue and priority: queue_{queue}_{priority}_delay (getDelayQueueName($queue, $priority)). The x-dead-letter-routing-key therefore always targets the priority bucket the job belongs to.
  • The legacy queue_{queue}_delay name is kept only for cleanup (getLegacyDelayQueueName()).
  • later() publishes the delayed copy first and only then calls basic_nack() on the original message. If publishing throws, the original message stays unacknowledged and is not lost.
  • publishDelayedMessage() now ensures the exchange exists (declareExchange()), declares and binds the per-priority delay queue only once (guarded by $this->declaredQueues).
  • declareExchange() binds the priority queues together with the exchange declaration instead of re-binding them on every publish.
  • clearQueue() purges the delay queue of every priority (before the priority queues) and also purges the legacy delay queue, so jobs scheduled before this change cannot leak back after a clear.
  • push() resets priority and delay in a finally block so the state can never leak into the next push.

Testing

Added tests/RabbitMQHandlerUnitTest.php, which exercises the handler logic without a live broker (channel and config are mocked):

• testGetDelayQueueNameIncludesPriority()
• testPublishDelayedMessageDeclaresPerPriorityQueueAndDeadLetterRouting() — asserts that x-dead-letter-routing-key matches the priority bucket.
• testPublishDelayedMessageBindsDelayQueueOnlyOnceAcrossPublications()
• testLaterPublishesDelayedMessageBeforeRejectingOriginalMessage() — asserts the publish/basic_nack ordering.
• testLaterDoesNotRejectOriginalMessageWhenPublishingDelayedFails()
• testClearQueuePurgesDelayQueuesForEachPriority() — including the legacy delay queue.
• testPushResetsPriorityAndDelayInFinallyBlockOnFailure()

The existing tests/RabbitMQHandlerTest.php and tests/RabbitMQDelayTest.php suites still pass against the integration setup.

Verified locally on PHP 8.5 with PHPUnit 11.5:

phpunit ....... OK (237 tests, 682 assertions, 27 skipped)
phpstan ....... [OK] No errors
psalm ......... No errors found
php-cs-fixer .. 0 of 76 files would be fixed