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

Failed error imports were keyed by Guid.NewGuid(), so each failed import
attempt produced a distinct document (and log file), and nothing tied a
document back to its FailedImports/Error/{id}.txt log.

Add FailedErrorImport.DeriveKey(headers, nativeMessageId), which uses the
message's UniqueId() when it can be derived and falls back to a
deterministic id built from the native transport id when it cannot.
UniqueId() throws when a message carries no processing endpoint header,
and malformed or header-less messages are a leading cause of import
failure, so the fallback is required rather than optional.

ErrorIngestionFaultPolicy now keys the stored failure by this derived id.
Repeated failures of the same message collapse onto one document, the
latest attempt's details win, and the log file name is recoverable from
the document key. This is a shared change: the derivation lives in
ServiceControl.Persistence and is reused by the upcoming EF persister so
the two cannot drift.

Existing RavenDB data is unaffected and needs no migration. The read path
is index-based and deletes by document id, so legacy random-id documents
are still found, replayed, and removed. The store session uses no
optimistic concurrency, so writing a derived id that already exists is a
plain upsert.

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