[Messenger] prevent infinite redelivery loops and blocked queues

by republishing the redelivered messages as retries with a retry limit and potential delay

Author: Tobion

DependencyInjection/FrameworkExtension.php

@@ -1649,6 +1649,7 @@ private function registerMessengerConfiguration(array $config, ContainerBuilder
         $defaultMiddleware = [
             'before' => [
                 ['id' => 'add_bus_name_stamp_middleware'],
+                ['id' => 'reject_redelivered_message_middleware'],
                 ['id' => 'dispatch_after_current_bus'],
                 ['id' => 'failed_message_processing_middleware'],
             ],

Resources/config/messenger.xml

@@ -48,6 +48,8 @@
             <argument type="service" id="validator" />
         </service>
 
+        <service id="messenger.middleware.reject_redelivered_message_middleware" class="Symfony\Component\Messenger\Middleware\RejectRedeliveredMessageMiddleware" />
+
         <service id="messenger.middleware.failed_message_processing_middleware" class="Symfony\Component\Messenger\Middleware\FailedMessageProcessingMiddleware" />
 
         <service id="messenger.middleware.traceable" class="Symfony\Component\Messenger\Middleware\TraceableMiddleware" abstract="true">

Tests/DependencyInjection/FrameworkExtensionTest.php

@@ -739,6 +739,7 @@ public function testMessengerWithMultipleBuses()
         $this->assertSame([], $container->getDefinition('messenger.bus.commands')->getArgument(0));
         $this->assertEquals([
             ['id' => 'add_bus_name_stamp_middleware', 'arguments' => ['messenger.bus.commands']],
+            ['id' => 'reject_redelivered_message_middleware'],
             ['id' => 'dispatch_after_current_bus'],
             ['id' => 'failed_message_processing_middleware'],
             ['id' => 'send_message'],
@@ -748,6 +749,7 @@ public function testMessengerWithMultipleBuses()
         $this->assertSame([], $container->getDefinition('messenger.bus.events')->getArgument(0));
         $this->assertEquals([
             ['id' => 'add_bus_name_stamp_middleware', 'arguments' => ['messenger.bus.events']],
+            ['id' => 'reject_redelivered_message_middleware'],
             ['id' => 'dispatch_after_current_bus'],
             ['id' => 'failed_message_processing_middleware'],
             ['id' => 'with_factory', 'arguments' => ['foo', true, ['bar' => 'baz']]],

composer.json

@@ -42,7 +42,7 @@
         "symfony/expression-language": "~3.4|~4.0",
         "symfony/http-client": "^4.3",
         "symfony/mailer": "^4.3",
-        "symfony/messenger": "^4.3",
+        "symfony/messenger": "^4.3.6",
         "symfony/mime": "^4.3",
         "symfony/process": "~3.4|~4.0",
         "symfony/security-csrf": "~3.4|~4.0",
@@ -73,7 +73,7 @@
         "symfony/dotenv": "<4.2",
         "symfony/dom-crawler": "<4.3",
         "symfony/form": "<4.3",
-        "symfony/messenger": "<4.3",
+        "symfony/messenger": "<4.3.6",
         "symfony/property-info": "<3.4",
         "symfony/serializer": "<4.2",
         "symfony/stopwatch": "<3.4",