Improve the poison message deletion handling to reduce the chance of duplicates (#2765)

* Add test to verify behavior

* Improve the poison message handling

---------

Co-authored-by: danielmarbach <[email protected]>

Author: danielmarbach

src/NServiceBus.Transport.SQS.Tests/InputQueuePumpTests.cs

@@ -106,6 +106,65 @@ await SetupInitializedPump(onMessage: (ctx, ct) =>
             Assert.That(mockSqsClient.DeleteMessageRequestsSent.Single().receiptHandle, Is.EqualTo(expectedReceiptHandle));
         }
 
+        static IEnumerable<object[]> PoisonMessageExceptions() =>
+        [
+            [new Exception()],
+            [new ReceiptHandleIsInvalidException("Ooops")]
+        ];
+
+        [Theory]
+        [TestCaseSource(nameof(PoisonMessageExceptions))]
+        public async Task Poison_messages_failed_to_be_deleted_are_deleted_on_next_receive_without_processing(Exception exception)
+        {
+            var nativeMessageId = Guid.NewGuid().ToString();
+            var messageId = Guid.NewGuid().ToString();
+            var expectedFirstReceiptHandle = "receipt-handle-1";
+            var expectedSecondReceiptHandle = "receipt-handle-2";
+
+            var processed = false;
+            await SetupInitializedPump(onMessage: (ctx, ct) =>
+            {
+                processed = true;
+                return Task.FromResult(0);
+            });
+
+            var deleteRequest = mockSqsClient.DeleteMessageRequestResponse;
+            mockSqsClient.DeleteMessageRequestResponse = request => throw exception;
+
+            var message = new Message
+            {
+                ReceiptHandle = expectedFirstReceiptHandle,
+                MessageId = nativeMessageId,
+                MessageAttributes = new Dictionary<string, MessageAttributeValue>
+                {
+                    {Headers.MessageId, new MessageAttributeValue {StringValue = messageId}},
+                },
+                Body = null //poison message
+            };
+
+            // First receive attempt
+            await pump.ProcessMessage(message, CancellationToken.None).ConfigureAwait(false);
+
+            // Restore the previous behavior
+            mockSqsClient.DeleteMessageRequestResponse = deleteRequest;
+            // Second receive attempt of the same message that has already been moved to the error queue
+            // but not deleted from the input queue. It will be received with a new receipt handle.
+            message.ReceiptHandle = expectedSecondReceiptHandle;
+            await pump.ProcessMessage(message, CancellationToken.None).ConfigureAwait(false);
+
+            Assert.Multiple(() =>
+            {
+                Assert.That(processed, Is.False);
+                Assert.That(mockSqsClient.RequestsSent, Has.Count.EqualTo(1), "The message should not be attempted to be sent multiple times to the error queue.");
+                Assert.That(mockSqsClient.DeleteMessageRequestsSent, Has.Count.EqualTo(2));
+            });
+            Assert.Multiple(() =>
+            {
+                Assert.That(mockSqsClient.DeleteMessageRequestsSent[0].receiptHandle, Is.EqualTo(expectedFirstReceiptHandle));
+                Assert.That(mockSqsClient.DeleteMessageRequestsSent[1].receiptHandle, Is.EqualTo(expectedSecondReceiptHandle));
+            });
+        }
+
         [Test]
         public async Task Expired_messages_are_deleted_without_processing()
         {

src/NServiceBus.Transport.SQS/InputQueuePump.cs

@@ -590,12 +590,12 @@ await sqsClient
             }
         }
 
-        async Task MovePoisonMessageToErrorQueue(Message message, CancellationToken messageProcessingCancellationToken)
+        async Task MovePoisonMessageToErrorQueue(Message message, CancellationToken cancellationToken)
         {
             string errorQueueUrl = null;
             try
             {
-                errorQueueUrl = await queueCache.GetQueueUrl(errorQueueAddress, messageProcessingCancellationToken)
+                errorQueueUrl = await queueCache.GetQueueUrl(errorQueueAddress, cancellationToken)
                     .ConfigureAwait(false);
                 // Ok to use LINQ here since this is not really a hot path
                 var messageAttributeValues = message.MessageAttributes
@@ -606,12 +606,12 @@ await sqsClient.SendMessageAsync(new SendMessageRequest
                     QueueUrl = errorQueueUrl,
                     MessageBody = message.Body,
                     MessageAttributes = messageAttributeValues
-                }, messageProcessingCancellationToken).ConfigureAwait(false);
+                }, cancellationToken).ConfigureAwait(false);
                 // The Attributes on message are read-only attributes provided by SQS
                 // and can't be re-sent. Unfortunately all the SQS metadata
                 // such as SentTimestamp is reset with this send.
             }
-            catch (Exception ex) when (!ex.IsCausedBy(messageProcessingCancellationToken))
+            catch (Exception ex) when (!ex.IsCausedBy(cancellationToken))
             {
                 Logger.Error($"Error moving poison message to error queue at url {errorQueueUrl}. Moving back to input queue.", ex);
                 try
@@ -621,11 +621,11 @@ await sqsClient.ChangeMessageVisibilityAsync(new ChangeMessageVisibilityRequest
                         QueueUrl = inputQueueUrl,
                         ReceiptHandle = message.ReceiptHandle,
                         VisibilityTimeout = 0
-                    }, messageProcessingCancellationToken).ConfigureAwait(false);
+                    }, cancellationToken).ConfigureAwait(false);
                 }
-                catch (Exception changeMessageVisibilityEx) when (!changeMessageVisibilityEx.IsCausedBy(messageProcessingCancellationToken))
+                catch (Exception changeMessageVisibilityEx) when (!changeMessageVisibilityEx.IsCausedBy(cancellationToken))
                 {
-                    Logger.Warn($"Error returning poison message back to input queue at url {inputQueueUrl}. Poison message will become available at the input queue again after the visibility timeout expires.", changeMessageVisibilityEx);
+                    Logger.Warn($"Error returning poison message with native ID '{message.MessageId}' back to input queue {inputQueueUrl}. Poison message will become available at the input queue again after the visibility timeout expires.", changeMessageVisibilityEx);
                 }
 
                 return;
@@ -637,14 +637,23 @@ await sqsClient.DeleteMessageAsync(new DeleteMessageRequest
                 {
                     QueueUrl = inputQueueUrl,
                     ReceiptHandle = message.ReceiptHandle
-                }, messageProcessingCancellationToken).ConfigureAwait(false);
+                }, CancellationToken.None) // We don't want the delete to be cancellable to avoid unnecessary duplicates of poison messages
+                    .ConfigureAwait(false);
             }
-            catch (Exception ex) when (!ex.IsCausedBy(messageProcessingCancellationToken))
+            catch (ReceiptHandleIsInvalidException ex)
             {
-                Logger.Warn($"Error removing poison message from input queue {inputQueueUrl}. This may cause duplicate poison messages in the error queue for this endpoint.", ex);
+                Logger.Error($"Error removing poison message with native ID '{message.MessageId}' from input queue {inputQueueUrl} because the visibility timeout expired. Poison message will become available at the input queue again and attempted to be removed on a best-effort basis. This may still cause duplicate poison messages in the error queue for this endpoint", ex);
+
+                messagesToBeDeleted.AddOrUpdate(message.MessageId, true);
+            }
+            catch (Exception ex)
+            {
+                Logger.Warn($"Error removing poison message with native ID '{message.MessageId}' from input queue {inputQueueUrl}. Poison message will become available at the input queue again and attempted to be removed on a best-effort basis. This may still cause duplicate poison messages in the error queue for this endpoint", ex);
+
+                messagesToBeDeleted.AddOrUpdate(message.MessageId, true);
             }
 
-            // If there is a message body in S3, simply leave it there
+            // If there is a message body in S3, simply leave it there to be aged out by the S3 lifecycle policy when the TTL expires
         }
 
         List<Task> pumpTasks;