diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Operations/FailedErrorImportDedupeTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Operations/FailedErrorImportDedupeTests.cs new file mode 100644 index 0000000000..910f8fbcba --- /dev/null +++ b/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 + { + { 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().ToListAsync(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(documents, Has.Count.EqualTo(1)); + Assert.That(documents[0].ExceptionInfo, Is.EqualTo("the second failure")); + } + } + + Task StoreFailure(IReadOnlyDictionary headers, string exceptionInfo) => + ErrorStore.StoreFailedErrorImport(new FailedErrorImport + { + Id = FailedErrorImport.MakeDocumentId(FailedErrorImport.DeriveKey(headers, "native-1")), + Message = new FailedTransportMessage + { + Id = "native-1", + Headers = new Dictionary(headers), + Body = [] + }, + ExceptionInfo = exceptionInfo + }); + } +} diff --git a/src/ServiceControl.Persistence/FailedErrorImport.cs b/src/ServiceControl.Persistence/FailedErrorImport.cs index 751f6f3418..034ab53ee8 100644 --- a/src/ServiceControl.Persistence/FailedErrorImport.cs +++ b/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 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); + } } } \ No newline at end of file diff --git a/src/ServiceControl.UnitTests/Operations/When_deriving_a_failed_error_import_key.cs b/src/ServiceControl.UnitTests/Operations/When_deriving_a_failed_error_import_key.cs new file mode 100644 index 0000000000..9161c81ce0 --- /dev/null +++ b/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 + { + { 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 + { + { "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 + { + { 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 + { + { "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(), "native-1"); + var second = FailedErrorImport.DeriveKey(new Dictionary(), "native-1"); + + Assert.That(second, Is.EqualTo(first)); + } +} diff --git a/src/ServiceControl/Operations/ErrorIngestionFaultPolicy.cs b/src/ServiceControl/Operations/ErrorIngestionFaultPolicy.cs index d60dae5c63..dfa6b38be5 100644 --- a/src/ServiceControl/Operations/ErrorIngestionFaultPolicy.cs +++ b/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