bug symfony#54842 [Messenger] Don't drop stamps when message validation fails (valtzu)

This PR was squashed before being merged into the 6.4 branch.

Discussion
----------

[Messenger] Don't drop stamps when message validation fails

| Q             | A
| ------------- | ---
| Branch?       | 6.4
| Bug fix?      | yes
| New feature?  | no
| Deprecations? | no
| License       | MIT

`ValidationMiddleware` has the same issue as `DispatchAfterCurrentBusMiddleware` did before the [fix](symfony#51468): if message validation fails, stamps added by previous middlewares in the stack are dropped. What this means in practice is:

1. `bin/console messenger:consume --bus=external` receives a message and `AddBusNameStampMiddleware` adds the `BusNameStamp` so that in case of failure it would be routed to a correct bus
2. The message fails validation – the original envelope without the added `BusNameStamp` is sent to failure queue
3. When running `bin/console messenger:failed:retry`, the message is dispatched on wrong bus (the default one)

This has really bad implications if you handle the message differently depending on which bus it is received.

Some refactoring was done to reduce duplication, similar to `WrappedExceptionsTrait`/`WrappedExceptionsInterface`.

Commits
-------

fed32dc [Messenger] Don't drop stamps when message validation fails

Author: fabpot

src/Symfony/Component/Messenger/Exception/DelayedMessageHandlingException.php

@@ -19,12 +19,12 @@
  *
  * @author Tobias Nyholm <tobias.nyholm@gmail.com>
  */
-class DelayedMessageHandlingException extends RuntimeException implements WrappedExceptionsInterface
+class DelayedMessageHandlingException extends RuntimeException implements WrappedExceptionsInterface, EnvelopeAwareExceptionInterface
 {
+    use EnvelopeAwareExceptionTrait;
     use WrappedExceptionsTrait;
 
     private array $exceptions;
-    private ?Envelope $envelope;
 
     public function __construct(array $exceptions, ?Envelope $envelope = null)
     {
@@ -55,9 +55,4 @@ public function getExceptions(): array
 
         return $this->exceptions;
     }
-
-    public function getEnvelope(): ?Envelope
-    {
-        return $this->envelope;
-    }
 }

src/Symfony/Component/Messenger/Exception/EnvelopeAwareExceptionInterface.php

@@ -0,0 +1,22 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Exception;
+
+use Symfony\Component\Messenger\Envelope;
+
+/**
+ * @internal
+ */
+interface EnvelopeAwareExceptionInterface
+{
+    public function getEnvelope(): ?Envelope;
+}

src/Symfony/Component/Messenger/Exception/EnvelopeAwareExceptionTrait.php

@@ -0,0 +1,27 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Exception;
+
+use Symfony\Component\Messenger\Envelope;
+
+/**
+ * @internal
+ */
+trait EnvelopeAwareExceptionTrait
+{
+    private ?Envelope $envelope = null;
+
+    public function getEnvelope(): ?Envelope
+    {
+        return $this->envelope;
+    }
+}

src/Symfony/Component/Messenger/Exception/HandlerFailedException.php

@@ -13,7 +13,7 @@
 
 use Symfony\Component\Messenger\Envelope;
 
-class HandlerFailedException extends RuntimeException implements WrappedExceptionsInterface
+class HandlerFailedException extends RuntimeException implements WrappedExceptionsInterface, EnvelopeAwareExceptionInterface
 {
     use WrappedExceptionsTrait;
 

src/Symfony/Component/Messenger/Exception/ValidationFailedException.php

@@ -11,20 +11,24 @@
 
 namespace Symfony\Component\Messenger\Exception;
 
+use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Validator\ConstraintViolationListInterface;
 
 /**
  * @author Tobias Nyholm <tobias.nyholm@gmail.com>
  */
-class ValidationFailedException extends RuntimeException
+class ValidationFailedException extends RuntimeException implements EnvelopeAwareExceptionInterface
 {
+    use EnvelopeAwareExceptionTrait;
+
     private ConstraintViolationListInterface $violations;
     private object $violatingMessage;
 
-    public function __construct(object $violatingMessage, ConstraintViolationListInterface $violations)
+    public function __construct(object $violatingMessage, ConstraintViolationListInterface $violations, ?Envelope $envelope = null)
     {
         $this->violatingMessage = $violatingMessage;
         $this->violations = $violations;
+        $this->envelope = $envelope;
 
         parent::__construct(sprintf('Message of type "%s" failed validation.', $this->violatingMessage::class));
     }

src/Symfony/Component/Messenger/Middleware/ValidationMiddleware.php

@@ -39,7 +39,7 @@ public function handle(Envelope $envelope, StackInterface $stack): Envelope
 
         $violations = $this->validator->validate($message, null, $groups);
         if (\count($violations)) {
-            throw new ValidationFailedException($message, $violations);
+            throw new ValidationFailedException($message, $violations, $envelope);
         }
 
         return $stack->next()->handle($envelope, $stack);

src/Symfony/Component/Messenger/Tests/FailureIntegrationTest.php

@@ -24,6 +24,7 @@
 use Symfony\Component\Messenger\EventListener\StopWorkerOnMessageLimitListener;
 use Symfony\Component\Messenger\Exception\DelayedMessageHandlingException;
 use Symfony\Component\Messenger\Exception\HandlerFailedException;
+use Symfony\Component\Messenger\Exception\ValidationFailedException;
 use Symfony\Component\Messenger\Handler\HandlerDescriptor;
 use Symfony\Component\Messenger\Handler\HandlersLocator;
 use Symfony\Component\Messenger\MessageBus;
@@ -32,6 +33,7 @@
 use Symfony\Component\Messenger\Middleware\FailedMessageProcessingMiddleware;
 use Symfony\Component\Messenger\Middleware\HandleMessageMiddleware;
 use Symfony\Component\Messenger\Middleware\SendMessageMiddleware;
+use Symfony\Component\Messenger\Middleware\ValidationMiddleware;
 use Symfony\Component\Messenger\Retry\MultiplierRetryStrategy;
 use Symfony\Component\Messenger\Stamp\BusNameStamp;
 use Symfony\Component\Messenger\Stamp\DispatchAfterCurrentBusStamp;
@@ -42,6 +44,9 @@
 use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
 use Symfony\Component\Messenger\Transport\Sender\SendersLocator;
 use Symfony\Component\Messenger\Worker;
+use Symfony\Component\Validator\ConstraintViolation;
+use Symfony\Component\Validator\ConstraintViolationList;
+use Symfony\Component\Validator\Validator\ValidatorInterface;
 
 class FailureIntegrationTest extends TestCase
 {
@@ -440,6 +445,87 @@ public function testStampsAddedByMiddlewaresDontDisappearWhenDelayedMessageFails
         $this->assertCount(1, $messagesWaiting);
         $this->assertSame('some.bus', $messagesWaiting[0]->last(BusNameStamp::class)?->getBusName());
     }
+
+    public function testStampsAddedByMiddlewaresDontDisappearWhenValidationFails()
+    {
+        $transport1 = new DummyFailureTestSenderAndReceiver();
+
+        $transports = [
+            'transport1' => $transport1,
+        ];
+
+        $locator = $this->createMock(ContainerInterface::class);
+        $locator->expects($this->any())
+            ->method('has')
+            ->willReturn(true);
+        $locator->expects($this->any())
+            ->method('get')
+            ->willReturnCallback(fn ($transportName) => $transports[$transportName]);
+        $senderLocator = new SendersLocator([], $locator);
+
+        $retryStrategyLocator = $this->createMock(ContainerInterface::class);
+        $retryStrategyLocator->expects($this->any())
+            ->method('has')
+            ->willReturn(true);
+        $retryStrategyLocator->expects($this->any())
+            ->method('get')
+            ->willReturn(new MultiplierRetryStrategy(1));
+
+        $violationList = new ConstraintViolationList([new ConstraintViolation('validation failed', null, [], null, null, null)]);
+        $validator = $this->createMock(ValidatorInterface::class);
+        $validator->expects($this->once())->method('validate')->willReturn($violationList);
+
+        $middlewareStack = new \ArrayIterator([
+            new AddBusNameStampMiddleware('some.bus'),
+            new ValidationMiddleware($validator),
+            new SendMessageMiddleware($senderLocator),
+        ]);
+
+        $bus = new MessageBus($middlewareStack);
+
+        $transport1Handler = fn () => $bus->dispatch(new \stdClass(), [new DispatchAfterCurrentBusStamp()]);
+
+        $handlerLocator = new HandlersLocator([
+            DummyMessage::class => [new HandlerDescriptor($transport1Handler)],
+        ]);
+
+        $middlewareStack->append(new HandleMessageMiddleware($handlerLocator));
+
+        $dispatcher = new EventDispatcher();
+
+        $dispatcher->addSubscriber(new SendFailedMessageForRetryListener($locator, $retryStrategyLocator));
+        $dispatcher->addSubscriber(new StopWorkerOnMessageLimitListener(1));
+
+        $runWorker = function (string $transportName) use ($transports, $bus, $dispatcher): ?\Throwable {
+            $throwable = null;
+            $failedListener = function (WorkerMessageFailedEvent $event) use (&$throwable) {
+                $throwable = $event->getThrowable();
+            };
+            $dispatcher->addListener(WorkerMessageFailedEvent::class, $failedListener);
+
+            $worker = new Worker([$transportName => $transports[$transportName]], $bus, $dispatcher);
+
+            $worker->run();
+
+            $dispatcher->removeListener(WorkerMessageFailedEvent::class, $failedListener);
+
+            return $throwable;
+        };
+
+        // Simulate receive from external source
+        $transport1->send(new Envelope(new DummyMessage('API')));
+
+        // Receive the message from "transport1"
+        $throwable = $runWorker('transport1');
+
+        $this->assertInstanceOf(ValidationFailedException::class, $throwable, $throwable->getMessage());
+
+        $messagesWaiting = $transport1->getMessagesWaitingToBeReceived();
+
+        // Stamps should not be dropped on message that's queued for retry
+        $this->assertCount(1, $messagesWaiting);
+        $this->assertSame('some.bus', $messagesWaiting[0]->last(BusNameStamp::class)?->getBusName());
+    }
 }
 
 class DummyFailureTestSenderAndReceiver implements ReceiverInterface, SenderInterface

src/Symfony/Component/Messenger/Worker.php

@@ -22,8 +22,7 @@
 use Symfony\Component\Messenger\Event\WorkerRunningEvent;
 use Symfony\Component\Messenger\Event\WorkerStartedEvent;
 use Symfony\Component\Messenger\Event\WorkerStoppedEvent;
-use Symfony\Component\Messenger\Exception\DelayedMessageHandlingException;
-use Symfony\Component\Messenger\Exception\HandlerFailedException;
+use Symfony\Component\Messenger\Exception\EnvelopeAwareExceptionInterface;
 use Symfony\Component\Messenger\Exception\RejectRedeliveredMessageException;
 use Symfony\Component\Messenger\Exception\RuntimeException;
 use Symfony\Component\Messenger\Stamp\AckStamp;
@@ -189,7 +188,7 @@ private function ack(): bool
                     $receiver->reject($envelope);
                 }
 
-                if ($e instanceof HandlerFailedException || ($e instanceof DelayedMessageHandlingException && null !== $e->getEnvelope())) {
+                if ($e instanceof EnvelopeAwareExceptionInterface && null !== $e->getEnvelope()) {
                     $envelope = $e->getEnvelope();
                 }