From f249ee37edfb5ad081418f8172e94a47171175fb Mon Sep 17 00:00:00 2001 From: John Simons Date: Mon, 27 Jul 2026 13:00:47 +1000 Subject: [PATCH] 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. --- .../FailedErrorImportDedupeTests.cs | 50 +++++++++++++ .../FailedErrorImport.cs | 21 ++++++ ...When_deriving_a_failed_error_import_key.cs | 75 +++++++++++++++++++ .../Operations/ErrorIngestionFaultPolicy.cs | 2 +- 4 files changed, 147 insertions(+), 1 deletion(-) create mode 100644 src/ServiceControl.Persistence.Tests.RavenDB/Operations/FailedErrorImportDedupeTests.cs create mode 100644 src/ServiceControl.UnitTests/Operations/When_deriving_a_failed_error_import_key.cs 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