ADD message deletion when they cannot be decoded

Deleting messages with decoding failures will avoid the consumers
being stuck on the same message forever.
It also converts the decoding exception to a custom one with an
envelope with an empty message for logging purposes.

Author: AymDev

CHANGELOG.md

@@ -6,6 +6,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.2.0] - 2022-02-28
+### Added
+ - Converts the `MessageDecodingFailedException` to a new `SerializerDecodingException` containing an envelope with an empty message for logging purposes.
+
+### Fixed
+ - Deletes messages for consumers when the serializer throws a `MessageDecodingFailedException` to avoid retying them forever.
+
 ## [1.1.1] - 2022-02-15
 ### Fixed
  - **AzureBrokerPropertiesStamp** **DateTime** properties timezones are now set to the current default timezone.
@@ -21,7 +28,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
  - **Symfony Messenger** transport for **Azure Service Bus** *queues* and *topics*
 
-[Unreleased]: https://github.com/AymDev/MessengerAzureBundle/compare/v1.1.1...HEAD
+[Unreleased]: https://github.com/AymDev/MessengerAzureBundle/compare/v1.2.0...HEAD
+[1.2.0]: https://github.com/AymDev/MessengerAzureBundle/releases/tag/v1.2.0
 [1.1.1]: https://github.com/AymDev/MessengerAzureBundle/releases/tag/v1.1.1
 [1.1.0]: https://github.com/AymDev/MessengerAzureBundle/releases/tag/v1.1.0
 [1.0.0]: https://github.com/AymDev/MessengerAzureBundle/releases/tag/v1.0.0

README.md

@@ -83,7 +83,16 @@ The `AymDev\MessengerAzureBundle\Messenger\Stamp\AzureReceivedStamp` stamp holds
 
 ## Serialization
 
+### Creating your serializers
 There is no serializer provided, but here is the expected array structure of an encoded envelope:
 
  - `body`: your plain text message
- - `headers`: optional HTTP headers (either received from *Azure Service Bus* response or to send to the REST API)
+ - `headers`: optional HTTP headers (either received from *Azure Service Bus* response or to send to the REST API)
+
+### Logging decoding errors
+When a serializer throws a `Symfony\Component\Messenger\Exception\MessageDecodingFailedException` while decoding a message,
+it will be converted to a `AymDev\MessengerAzureBundle\Messenger\Exception\SerializerDecodingException` which contains an
+envelope with an empty message but with the same stamps as a successfully decoded message.
+
+You can then [listen](https://symfony.com/doc/current/event_dispatcher.html) to the `console.error` Symfony event and get
+the topic/queue name where then decoding failure happened, the original message, etc.

src/Messenger/Exception/SerializerDecodingException.php

@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+
+namespace AymDev\MessengerAzureBundle\Messenger\Exception;
+
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
+use Throwable;
+
+/**
+ * This exception class contains an envelope for a message that could not be decoded. Its only purpose is to provide the
+ * same metadata as a decoded messages with its stamps.
+ * It is thrown when a consumer serializer throws a MessageDecodingFailedException and can be used for logging using the
+ * console.error Symfony event.
+ */
+final class SerializerDecodingException extends MessageDecodingFailedException
+{
+    /** @var Envelope */
+    private $envelope;
+
+    /**
+     * @param Envelope $envelope an envelope with an empty message
+     */
+    public function __construct(Envelope $envelope, string $message = "", int $code = 0, Throwable $previous = null)
+    {
+        parent::__construct($message, $code, $previous);
+        $this->envelope = $envelope;
+    }
+
+    public function getEnvelope(): Envelope
+    {
+        return $this->envelope;
+    }
+}

src/Messenger/Transport/AzureTransport.php

@@ -4,17 +4,20 @@
 
 namespace AymDev\MessengerAzureBundle\Messenger\Transport;
 
+use AymDev\MessengerAzureBundle\Messenger\Exception\SerializerDecodingException;
 use AymDev\MessengerAzureBundle\Messenger\Stamp\AzureBrokerPropertiesStamp;
 use AymDev\MessengerAzureBundle\Messenger\Stamp\AzureMessageStamp;
 use AymDev\MessengerAzureBundle\Messenger\Stamp\AzureReceivedStamp;
 use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
 use Symfony\Component\Messenger\Exception\TransportException;
 use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
 use Symfony\Component\Messenger\Transport\TransportInterface;
 use Symfony\Contracts\HttpClient\Exception\HttpExceptionInterface;
 use Symfony\Contracts\HttpClient\Exception\TransportExceptionInterface;
 use Symfony\Contracts\HttpClient\HttpClientInterface;
+use Symfony\Contracts\HttpClient\ResponseInterface;
 
 /**
  * Messenger transport for Azure Service Bus
@@ -92,12 +95,34 @@ public function get(): iterable
         }
 
         // Decode message
-        $envelope = $this->serializer->decode([
-            'body' => $body,
-            'headers' => $headers,
-        ]);
+        try {
+            $envelope = $this->serializer->decode([
+                'body' => $body,
+                'headers' => $headers,
+            ]);
+        } catch (MessageDecodingFailedException $e) {
+            // Delete the message using an empty message
+            $envelope = $this->addStamps(new Envelope(new EmptyMessage()), $response);
+            $this->delete($envelope);
+
+            throw new SerializerDecodingException(
+                $envelope,
+                $e->getMessage(),
+                1646061041,
+                $e
+            );
+        }
 
-        // Add stamps
+        return [$this->addStamps($envelope, $response)];
+    }
+
+    /**
+     * Add stamps to an envelope
+     * @throws HttpExceptionInterface
+     * @throws TransportExceptionInterface
+     */
+    private function addStamps(Envelope $envelope, ResponseInterface $response): Envelope
+    {
         $brokerPropertiesStamp = AzureBrokerPropertiesStamp::createFromResponse($response);
         $envelope = $envelope
             ->with(AzureReceivedStamp::createFromResponse($response))
@@ -110,7 +135,7 @@ public function get(): iterable
             $envelope = $envelope->with(new TransportMessageIdStamp($brokerPropertiesStamp->getMessageId()));
         }
 
-        return [$envelope];
+        return $envelope;
     }
 
     /**

src/Messenger/Transport/EmptyMessage.php

@@ -0,0 +1,13 @@
+<?php
+
+declare(strict_types=1);
+
+namespace AymDev\MessengerAzureBundle\Messenger\Transport;
+
+/**
+ * This message class is used when creating an envelope for decoding failures
+ * @internal
+ */
+final class EmptyMessage
+{
+}

tests/Messenger/Exception/SerializerDecodingExceptionTest.php

@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Messenger\Exception;
+
+use AymDev\MessengerAzureBundle\Messenger\Exception\SerializerDecodingException;
+use AymDev\MessengerAzureBundle\Messenger\Transport\EmptyMessage;
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Messenger\Envelope;
+
+final class SerializerDecodingExceptionTest extends TestCase
+{
+    public function testGetMessageProperties(): void
+    {
+        $envelope = new Envelope(new EmptyMessage());
+
+        $exception = new SerializerDecodingException($envelope);
+
+        self::assertSame($envelope, $exception->getEnvelope());
+    }
+}

tests/Messenger/Transport/AzureTransportTest.php

@@ -4,6 +4,7 @@
 
 namespace Tests\AymDev\MessengerAzureBundle\Messenger\Transport;
 
+use AymDev\MessengerAzureBundle\Messenger\Exception\SerializerDecodingException;
 use AymDev\MessengerAzureBundle\Messenger\Stamp\AzureMessageStamp;
 use AymDev\MessengerAzureBundle\Messenger\Transport\AzureTransport;
 use AymDev\MessengerAzureBundle\Messenger\Stamp\AzureBrokerPropertiesStamp;
@@ -12,6 +13,7 @@
 use Symfony\Component\HttpClient\MockHttpClient;
 use Symfony\Component\HttpClient\Response\MockResponse;
 use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
 use Symfony\Component\Messenger\Exception\TransportException;
 use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
@@ -86,6 +88,54 @@ public function testGetWithUnexpectedStatusCode(): void
         $transport->get();
     }
 
+    /**
+     * Messages that can't be decoded must throw a converted exception and delete the message
+     */
+    public function testGetSerializationExceptionIsConvertedAndDeletesMessage(): void
+    {
+        self::expectException(SerializerDecodingException::class);
+        self::expectExceptionCode(1646061041);
+
+        $serializer = self::createMock(SerializerInterface::class);
+        $serializer->expects(self::once())
+            ->method('decode')
+            ->willThrowException(new MessageDecodingFailedException());
+
+        $location = 'https://test-location/deletion';
+
+        $httpReceiver = new MockHttpClient(
+            [
+                new MockResponse('test-body', [
+                        'http_code' => 201,
+                        'response_headers' => [
+                            'Location' => $location
+                        ],
+                ]),
+                function (string $method, string $url) use ($location): ResponseInterface {
+                    self::assertSame('DELETE', $method);
+                    self::assertSame($location, $url);
+
+                    return new MockResponse();
+                },
+            ]
+        );
+
+        $transport = new AzureTransport(
+            $serializer,
+            new MockHttpClient(),
+            $httpReceiver,
+            AzureTransport::RECEIVE_MODE_PEEK_LOCK,
+            'test-entity',
+            'test-subscription'
+        );
+
+        try {
+            $transport->get();
+        } finally {
+            self::assertSame(2, $httpReceiver->getRequestsCount());
+        }
+    }
+
     /**
      * Read messages must be returned in an envelope with specific stamps
      * @dataProvider provideValidGetCases