symfony · fabpot · Aug 6, 2025 · Jun 10, 2025
diff --git a/src/Symfony/Component/Messenger/Bridge/AmazonSqs/CHANGELOG.md b/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
 ---


diff --git a/src/Symfony/Component/Messenger/Bridge/AmazonSqs/Tests/Transport/AmazonSqsReceiverTest.php b/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 [

diff --git a/src/Symfony/Component/Messenger/Bridge/AmazonSqs/Tests/Transport/AmazonSqsTransportTest.php b/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');

diff --git a/src/Symfony/Component/Messenger/Bridge/AmazonSqs/Tests/Transport/ConnectionTest.php b/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);

diff --git a/src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/AmazonSqsReceiver.php b/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);
        }

diff --git a/src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/AmazonSqsTransport.php b/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);
+    }
 }
diff --git a/src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/AmazonSqsTransportFactory.php b/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

diff --git a/src/Symfony/Component/Messenger/Bridge/AmazonSqs/Transport/Connection.php b/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
     */