Merge pull request #5652 from Particular/john/failedmessages_ingestion_failures

Derive failed error import id from the message, not a random Guid

Author: johnsimons

src/ServiceControl.Persistence.Tests.RavenDB/Operations/FailedErrorImportDedupeTests.cs

@@ -0,0 +1,50 @@
+namespace ServiceControl.Persistence.Tests.RavenDB.Operations
+{
+    using System.Collections.Generic;
+    using System.Threading.Tasks;
+    using NServiceBus;
+    using NUnit.Framework;
+    using Raven.Client.Documents;
+    using ServiceControl.Operations;
+
+    [TestFixture]
+    class FailedErrorImportDedupeTests : RavenPersistenceTestBase
+    {
+        [Test]
+        public async Task Repeated_failure_of_the_same_message_stores_one_document()
+        {
+            var headers = new Dictionary<string, string>
+            {
+                { Headers.MessageId, "message-1" },
+                { Headers.ProcessingEndpoint, "Sales" }
+            };
+
+            await StoreFailure(headers, "the first failure");
+            await StoreFailure(headers, "the second failure");
+
+            DocumentStore.WaitForIndexing();
+
+            using var session = DocumentStore.OpenAsyncSession();
+            var documents = await session.Query<FailedErrorImport>().ToListAsync();
+
+            using (Assert.EnterMultipleScope())
+            {
+                Assert.That(documents, Has.Count.EqualTo(1));
+                Assert.That(documents[0].ExceptionInfo, Is.EqualTo("the second failure"));
+            }
+        }
+
+        Task StoreFailure(IReadOnlyDictionary<string, string> headers, string exceptionInfo) =>
+            ErrorStore.StoreFailedErrorImport(new FailedErrorImport
+            {
+                Id = FailedErrorImport.MakeDocumentId(FailedErrorImport.DeriveKey(headers, "native-1")),
+                Message = new FailedTransportMessage
+                {
+                    Id = "native-1",
+                    Headers = new Dictionary<string, string>(headers),
+                    Body = []
+                },
+                ExceptionInfo = exceptionInfo
+            });
+    }
+}

src/ServiceControl.Persistence/FailedErrorImport.cs

@@ -1,6 +1,8 @@
 namespace ServiceControl.Operations
 {
     using System;
+    using System.Collections.Generic;
+    using ServiceControl.Persistence.Infrastructure;
 
     public class FailedErrorImport
     {
@@ -9,5 +11,24 @@ public class FailedErrorImport
         public string ExceptionInfo { get; set; }
 
         public static string MakeDocumentId(Guid id) => $"FailedErrorImports/{id}";
+
+        public static Guid DeriveKey(IReadOnlyDictionary<string, string> headers, string nativeMessageId)
+        {
+            try
+            {
+                if (Guid.TryParse(headers.UniqueId(), out var uniqueMessageId))
+                {
+                    return uniqueMessageId;
+                }
+            }
+            catch (Exception)
+            {
+                // UniqueId() derives the processing endpoint, which throws when the failed message
+                // carries no endpoint header. Malformed messages are a leading cause of import
+                // failure, so fall back to a key derived from the id the transport always supplies.
+            }
+
+            return DeterministicGuid.MakeId(nativeMessageId);
+        }
     }
 }

src/ServiceControl.UnitTests/Operations/When_deriving_a_failed_error_import_key.cs

@@ -0,0 +1,75 @@
+namespace ServiceControl.UnitTests.Operations;
+
+using System;
+using System.Collections.Generic;
+using NServiceBus;
+using NUnit.Framework;
+using ServiceControl.Operations;
+using ServiceControl.Persistence.Infrastructure;
+
+[TestFixture]
+public class When_deriving_a_failed_error_import_key
+{
+    [Test]
+    public void Uses_the_unique_message_id_when_headers_are_well_formed()
+    {
+        var headers = new Dictionary<string, string>
+        {
+            { Headers.MessageId, "message-1" },
+            { Headers.ProcessingEndpoint, "Sales" }
+        };
+
+        var key = FailedErrorImport.DeriveKey(headers, "native-1");
+
+        Assert.That(key, Is.EqualTo(DeterministicGuid.MakeId("message-1", "Sales")));
+    }
+
+    [Test]
+    public void Prefers_an_existing_retry_unique_message_id()
+    {
+        var uniqueMessageId = Guid.NewGuid();
+        var headers = new Dictionary<string, string>
+        {
+            { "ServiceControl.Retry.UniqueMessageId", uniqueMessageId.ToString() }
+        };
+
+        var key = FailedErrorImport.DeriveKey(headers, "native-1");
+
+        Assert.That(key, Is.EqualTo(uniqueMessageId));
+    }
+
+    [Test]
+    public void Falls_back_to_the_native_id_when_no_processing_endpoint_can_be_derived()
+    {
+        var headers = new Dictionary<string, string>
+        {
+            { Headers.MessageId, "message-1" }
+        };
+
+        var key = FailedErrorImport.DeriveKey(headers, "native-1");
+
+        Assert.That(key, Is.EqualTo(DeterministicGuid.MakeId("native-1")));
+    }
+
+    [Test]
+    public void Falls_back_when_the_retry_unique_message_id_is_not_a_guid()
+    {
+        var headers = new Dictionary<string, string>
+        {
+            { "ServiceControl.Retry.UniqueMessageId", "not-a-guid" }
+        };
+
+        var key = FailedErrorImport.DeriveKey(headers, "native-1");
+
+        Assert.That(key, Is.EqualTo(DeterministicGuid.MakeId("native-1")));
+    }
+
+    [Test]
+    public void Is_stable_across_repeated_failures_of_the_same_malformed_message()
+    {
+        var first = FailedErrorImport.DeriveKey(new Dictionary<string, string>(), "native-1");
+        var second = FailedErrorImport.DeriveKey(new Dictionary<string, string>(), "native-1");
+
+        Assert.That(second, Is.EqualTo(first));
+    }
+}

src/ServiceControl/Operations/ErrorIngestionFaultPolicy.cs

@@ -57,7 +57,7 @@ async Task Handle(ErrorContext errorContext, CancellationToken cancellationToken
                     Body = errorContext.Body.ToArray()
                 },
                 ExceptionInfo = errorContext.Exception.ToFriendlyString(),
-                Id = FailedErrorImport.MakeDocumentId(Guid.NewGuid())
+                Id = FailedErrorImport.MakeDocumentId(FailedErrorImport.DeriveKey(errorContext.Headers, errorContext.MessageId))
             };
 
             try