[Messenger] Allow SQS to handle it's own retry/DLQ

Allow SQS to handle retries rather then handling this by Symfony.
This allows applications to use the retry strategy from SQS rather then Symfony.
The default is for the message to be deleted from SQS at which point Symfony
will handle the retry by deleting and then adding back in to the queue.
If `delete_on_rejection` is set to `false` instead it will change the message
visibility of the message on SQS and thus SQS to handle the retry mechanism

Author: maxbaldanza

src/Symfony/Component/Messenger/Bridge/AmazonSqs/CHANGELOG.md

@@ -1,6 +1,11 @@
 CHANGELOG
 =========
 
+7.4
+---
+
+* Allow SQS to handle it's own retry/DLQ
+
 7.3
 ---
 

src/Symfony/Component/Messenger/Bridge/AmazonSqs/Tests/Transport/AmazonSqsReceiverTest.php

@@ -50,7 +50,7 @@ public function testItRejectTheMessageIfThereIsAMessageDecodingFailedException()
         $sqsEnvelop = $this->createSqsEnvelope();
         $connection = $this->createMock(Connection::class);
         $connection->method('get')->willReturn($sqsEnvelop);
-        $connection->expects($this->once())->method('delete');
+        $connection->expects($this->once())->method('reject');
 
         $receiver = new AmazonSqsReceiver($connection, $serializer);
         iterator_to_array($receiver->get());
@@ -67,6 +67,17 @@ public function testKeepalive()
         $receiver->keepalive(new Envelope(new DummyMessage('foo'), [new AmazonSqsReceivedStamp('123')]), 10);
     }
 
+    public function testReject()
+    {
+        $serializer = $this->createSerializer();
+
+        $connection = $this->createMock(Connection::class);
+        $connection->expects($this->once())->method('reject')->with('123');
+
+        $receiver = new AmazonSqsReceiver($connection, $serializer);
+        $receiver->reject(new Envelope(new DummyMessage('foo'), [new AmazonSqsReceivedStamp('123')]));
+    }
+
     private function createSqsEnvelope()
     {
         return [

src/Symfony/Component/Messenger/Bridge/AmazonSqs/Tests/Transport/AmazonSqsTransportTest.php

@@ -22,6 +22,7 @@
 use Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\Connection;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\TransportException;
+use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
 use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
 use Symfony\Component\Messenger\Transport\Sender\SenderInterface;
@@ -116,6 +117,22 @@ public function testItCanSendAMessageViaTheSender()
         $this->assertSame($envelope, $this->transport->send($envelope));
     }
 
+    public function testItSendsAMessageViaTheSenderWithRedeliveryStamp()
+    {
+        $envelope = new Envelope(new \stdClass(), [new RedeliveryStamp(1)]);
+        $this->sender->expects($this->once())->method('send')->with($envelope)->willReturn($envelope);
+        $this->assertSame($envelope, $this->transport->send($envelope));
+    }
+
+    public function testItDoesNotSendRedeliveredMessageWhenNotHandlingRetries()
+    {
+        $transport = new AmazonSqsTransport($this->connection, null, $this->receiver, $this->sender, false);
+
+        $envelope = new Envelope(new \stdClass(), [new RedeliveryStamp(1)]);
+        $this->sender->expects($this->never())->method('send')->with($envelope)->willReturn($envelope);
+        $this->assertSame($envelope, $transport->send($envelope));
+    }
+
     public function testItCanSetUpTheConnection()
     {
         $this->connection->expects($this->once())->method('setup');

src/Symfony/Component/Messenger/Bridge/AmazonSqs/Tests/Transport/ConnectionTest.php

@@ -375,6 +375,35 @@ public function testKeepalive()
         $connection->keepalive($id);
     }
 
+    public function testDeleteOnReject()
+    {
+        $expectedParams = [
+            'QueueUrl' => $queueUrl = 'https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue',
+            'ReceiptHandle' => $id = 'abc',
+        ];
+
+        $client = $this->createMock(SqsClient::class);
+        $client->expects($this->once())->method('deleteMessage')->with($expectedParams);
+
+        $connection = new Connection([], $client, $queueUrl);
+        $connection->reject($id);
+    }
+
+    public function testDoNotDeleteOnRejection()
+    {
+        $expectedParams = [
+            'QueueUrl' => $queueUrl = 'https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue',
+            'ReceiptHandle' => $id = 'abc',
+            'VisibilityTimeout' => $visibilityTimeout = 10,
+        ];
+
+        $client = $this->createMock(SqsClient::class);
+        $client->expects($this->once())->method('changeMessageVisibility')->with($expectedParams);
+
+        $connection = new Connection(['delete_on_rejection' => false, 'visibility_timeout' => $visibilityTimeout], $client, $queueUrl);
+        $connection->reject($id);
+    }
+
     public function testKeepaliveWithTooSmallTtl()
     {
         $client = $this->createMock(SqsClient::class);

src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/AmazonSqsReceiver.php

@@ -52,7 +52,7 @@ public function get(): iterable
                 'headers' => $sqsEnvelope['headers'],
             ]);
         } catch (MessageDecodingFailedException $exception) {
-            $this->connection->delete($sqsEnvelope['id']);
+            $this->connection->reject($sqsEnvelope['id']);
 
             throw $exception;
         }
@@ -72,7 +72,7 @@ public function ack(Envelope $envelope): void
     public function reject(Envelope $envelope): void
     {
         try {
-            $this->connection->delete($this->findSqsReceivedStamp($envelope)->getId());
+            $this->connection->reject($this->findSqsReceivedStamp($envelope)->getId());
         } catch (HttpException $e) {
             throw new TransportException($e->getMessage(), 0, $e);
         }

src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/AmazonSqsTransport.php

@@ -14,6 +14,7 @@
 use AsyncAws\Core\Exception\Http\HttpException;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\TransportException;
+use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
 use Symfony\Component\Messenger\Transport\CloseableTransportInterface;
 use Symfony\Component\Messenger\Transport\Receiver\KeepaliveReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
@@ -37,6 +38,7 @@ public function __construct(
         ?SerializerInterface $serializer = null,
         private (ReceiverInterface&MessageCountAwareInterface)|null $receiver = null,
         private ?SenderInterface $sender = null,
+        private bool $handleRetries = true,
     ) {
         $this->serializer = $serializer ?? new PhpSerializer();
     }
@@ -71,6 +73,10 @@ public function getMessageCount(): int
 
     public function send(Envelope $envelope): Envelope
     {
+        if (false === $this->handleRetries && $this->isRedelivered($envelope)) {
+            return $envelope;
+        }
+
         return $this->getSender()->send($envelope);
     }
 
@@ -106,4 +112,9 @@ private function getSender(): SenderInterface
     {
         return $this->sender ??= new AmazonSqsSender($this->connection, $this->serializer);
     }
+
+    private function isRedelivered(Envelope $envelope): bool
+    {
+        return null !== $envelope->last(RedeliveryStamp::class);
+    }
 }

src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/AmazonSqsTransportFactory.php

@@ -32,7 +32,7 @@ public function createTransport(#[\SensitiveParameter] string $dsn, array $optio
     {
         unset($options['transport_name']);
 
-        return new AmazonSqsTransport(Connection::fromDsn($dsn, $options, null, $this->logger), $serializer);
+        return new AmazonSqsTransport(Connection::fromDsn($dsn, $options, null, $this->logger), $serializer, null, null, !($options['delete_on_rejection'] ?? false));
     }
 
     public function supports(#[\SensitiveParameter] string $dsn, array $options): bool

src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/Connection.php

@@ -39,6 +39,7 @@ class Connection
         'wait_time' => 20,
         'poll_timeout' => 0.1,
         'visibility_timeout' => null,
+        'delete_on_rejection' => true,
         'auto_setup' => true,
         'access_key' => null,
         'secret_key' => null,
@@ -101,6 +102,7 @@ public function __destruct()
      * * wait_time: long polling duration in seconds (Default: 20)
      * * poll_timeout: amount of seconds the transport should wait for new message
      * * visibility_timeout: amount of seconds the message won't be visible
+     * * delete_on_rejection: Whether to delete message on rejection or allow SQS to handle retries. (Default: true).
      * * sslmode: Can be "disable" to use http for a custom endpoint
      * * auto_setup: Whether the queue should be created automatically during send / get (Default: true)
      * * debug: Log all HTTP requests and responses as LoggerInterface::DEBUG (Default: false)
@@ -134,6 +136,7 @@ public static function fromDsn(#[\SensitiveParameter] string $dsn, array $option
             'wait_time' => (int) $options['wait_time'],
             'poll_timeout' => $options['poll_timeout'],
             'visibility_timeout' => null !== $options['visibility_timeout'] ? (int) $options['visibility_timeout'] : null,
+            'delete_on_rejection' => filter_var($options['delete_on_rejection'], \FILTER_VALIDATE_BOOL),
             'auto_setup' => filter_var($options['auto_setup'], \FILTER_VALIDATE_BOOL),
             'queue_name' => (string) $options['queue_name'],
             'queue_attributes' => $options['queue_attributes'],
@@ -312,6 +315,19 @@ public function delete(string $id): void
         ]);
     }
 
+    public function reject(string $id): void
+    {
+        if ($this->configuration['delete_on_rejection']) {
+            $this->delete($id);
+        } else {
+            $this->client->changeMessageVisibility([
+                'QueueUrl' => $this->getQueueUrl(),
+                'ReceiptHandle' => $id,
+                'VisibilityTimeout' => $this->configuration['visibility_timeout'] ?? 30,
+            ]);
+        }
+    }
+
     /**
      * @param int|null $seconds the minimum duration the message should be kept alive
      */