diff --git a/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_error_forwarding_is_enabled.cs b/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_error_forwarding_is_enabled.cs new file mode 100644 index 0000000000..54d37c6b44 --- /dev/null +++ b/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_error_forwarding_is_enabled.cs @@ -0,0 +1,79 @@ +namespace ServiceControl.AcceptanceTests.Recoverability.MessageFailures +{ + using System; + using System.Threading.Tasks; + using AcceptanceTesting; + using AcceptanceTesting.EndpointTemplates; + using NServiceBus; + using NServiceBus.AcceptanceTesting; + using NUnit.Framework; + + class When_error_forwarding_is_enabled : AcceptanceTest + { + [Test] + public async Task Should_forward_the_failed_message_to_the_error_log_queue() + { + SetSettings = settings => + { + settings.ForwardErrorMessages = true; + settings.ErrorLogQueue = NServiceBus.AcceptanceTesting.Customization.Conventions.EndpointNamingConvention(typeof(Spy)); + }; + + var context = await Define() + .WithEndpoint(b => b + .When(session => session.SendLocal(new ForwardedMessage())) + .DoNotFailOnErrorMessages()) + // The ingestor probes the forwarding address on startup with an empty message the + // spy cannot deserialize, so the spy has to tolerate its own failures. + .WithEndpoint(b => b.DoNotFailOnErrorMessages()) + .Done(c => c.ForwardedMessageId != null) + .Run(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(context.ForwardedMessageId, Is.EqualTo(context.FailedMessageId)); + Assert.That(context.ForwardedWithFailureHeaders, Is.True); + } + } + + public class Failing : EndpointConfigurationBuilder + { + public Failing() => EndpointSetup(c => c.NoRetries()); + + [Handler] + public class ForwardedMessageHandler(MyContext testContext) : IHandleMessages + { + public Task Handle(ForwardedMessage message, IMessageHandlerContext context) + { + testContext.FailedMessageId = context.MessageId; + throw new Exception("Simulated exception"); + } + } + } + + public class Spy : EndpointConfigurationBuilder + { + public Spy() => EndpointSetup(c => c.NoRetries()); + + [Handler] + public class ForwardedMessageHandler(MyContext testContext) : IHandleMessages + { + public Task Handle(ForwardedMessage message, IMessageHandlerContext context) + { + testContext.ForwardedWithFailureHeaders = context.MessageHeaders.ContainsKey("NServiceBus.ExceptionInfo.Message"); + testContext.ForwardedMessageId = context.MessageId; + return Task.CompletedTask; + } + } + } + + public class MyContext : ScenarioContext + { + public string FailedMessageId { get; set; } + public string ForwardedMessageId { get; set; } + public bool ForwardedWithFailureHeaders { get; set; } + } + } + + public class ForwardedMessage : ICommand; +} diff --git a/src/ServiceControl.Audit/Auditing/AuditIngestion.cs b/src/ServiceControl.Audit/Auditing/AuditIngestion.cs index 69e312965c..e500c3aee7 100644 --- a/src/ServiceControl.Audit/Auditing/AuditIngestion.cs +++ b/src/ServiceControl.Audit/Auditing/AuditIngestion.cs @@ -142,8 +142,9 @@ async Task SetUpAndStartInfrastructure(CancellationToken cancellationToken) ); messageReceiver = transportInfrastructure.Receivers[inputEndpoint]; + messageDispatcher = transportInfrastructure.Dispatcher; - await auditIngestor.VerifyCanReachForwardingAddress(cancellationToken); + await auditIngestor.VerifyCanReachForwardingAddress(messageDispatcher, cancellationToken); await messageReceiver.StartReceive(cancellationToken); logger.LogInformation(LogMessages.StartedInfrastructure); @@ -172,10 +173,7 @@ async Task StopAndTeardownInfrastructure(CancellationToken cancellationToken) logger.LogInformation("Stopping infrastructure"); try { - if (messageReceiver != null) - { - await messageReceiver.StopReceive(cancellationToken); - } + await StopReceiving(cancellationToken); } finally { @@ -184,6 +182,7 @@ async Task StopAndTeardownInfrastructure(CancellationToken cancellationToken) messageReceiver = null; transportInfrastructure = null; + receiveStopped = false; logger.LogInformation(LogMessages.StoppedInfrastructure); } @@ -214,6 +213,33 @@ async Task EnsureStopped(CancellationToken cancellationToken) } } + async Task EnsureReceivingStopped(CancellationToken cancellationToken) + { + await startStopSemaphore.WaitAsync(cancellationToken); + + try + { + await StopReceiving(cancellationToken); + } + finally + { + startStopSemaphore.Release(); + } + } + + // Stops the receiver on its own, leaving the infrastructure up. Idempotent because a + // shutdown stops receiving before draining and then tears down, so this runs twice. + async Task StopReceiving(CancellationToken cancellationToken) + { + if (messageReceiver == null || receiveStopped) + { + return; + } + + await messageReceiver.StopReceive(cancellationToken); + receiveStopped = true; + } + async Task OnMessage(MessageContext messageContext, CancellationToken cancellationToken) { using var messageIngestionMetrics = metrics.BeginIngestion(messageContext); @@ -258,7 +284,7 @@ protected override async Task ExecuteAsync(CancellationToken cancellationToken = contexts.Add(context); } - await auditIngestor.Ingest(contexts, cancellationToken); + await auditIngestor.Ingest(contexts, messageDispatcher, cancellationToken); batchMetrics.Complete(contexts.Count); } @@ -300,28 +326,28 @@ public override async Task StopAsync(CancellationToken cancellationToken = defau { try { - await watchdog.Stop(cancellationToken); + // Order matters. Receiving stops under the shutdown token rather than a cancelled + // one, so messages already being processed finish and their receives commit instead + // of being abandoned and redelivered after having been forwarded. Nothing new enters + // the channel after this, and the infrastructure stays up until the channel drains. + await EnsureReceivingStopped(cancellationToken); channel.Writer.Complete(); await base.StopAsync(cancellationToken); } finally { - if (transportInfrastructure != null) - { - try - { - await transportInfrastructure.Shutdown(cancellationToken); - } - catch (OperationCanceledException e) when (cancellationToken.IsCancellationRequested) - { - logger.LogInformation(e, "Shutdown cancelled"); - } - } + // Tears the infrastructure down, now that nothing is left to dispatch. + await watchdog.Stop(cancellationToken); } } TransportInfrastructure transportInfrastructure; IMessageReceiver messageReceiver; + bool receiveStopped; + + // Left in place when the infrastructure is torn down. A shutdown drains before tearing down, + // so this is still usable there. + IMessageDispatcher messageDispatcher; readonly int MaxBatchSize; readonly SemaphoreSlim startStopSemaphore = new(1); diff --git a/src/ServiceControl.Audit/Auditing/AuditIngestor.cs b/src/ServiceControl.Audit/Auditing/AuditIngestor.cs index 784fd15a21..38b01309ad 100644 --- a/src/ServiceControl.Audit/Auditing/AuditIngestor.cs +++ b/src/ServiceControl.Audit/Auditing/AuditIngestor.cs @@ -24,13 +24,11 @@ public AuditIngestor( EndpointInstanceMonitoring endpointInstanceMonitoring, IEnumerable auditEnrichers, // allows extending message enrichers with custom enrichers registered in the DI container IMessageSession messageSession, - Lazy messageDispatcher, ITransportCustomization transportCustomization, ILogger logger ) { this.settings = settings; - this.messageDispatcher = messageDispatcher; this.logger = logger; var enrichers = new IEnrichImportedAuditMessages[] { new MessageTypeEnricher(), new EnrichWithTrackingIds(), new ProcessingStatisticsEnricher(), new DetectNewEndpointsFromAuditImportsEnricher(endpointInstanceMonitoring), new DetectSuccessfulRetriesEnricher(), new SagaRelationshipsEnricher() }.Concat(auditEnrichers).ToArray(); @@ -40,20 +38,19 @@ ILogger logger unitOfWorkFactory, enrichers, messageSession, - messageDispatcher, logger ); } - public async Task Ingest(List contexts, CancellationToken cancellationToken = default) + public async Task Ingest(List contexts, IMessageDispatcher dispatcher, CancellationToken cancellationToken = default) { - var stored = await auditPersister.Persist(contexts, cancellationToken); + var stored = await auditPersister.Persist(contexts, dispatcher, cancellationToken); try { if (settings.ForwardAuditMessages) { - await Forward(stored, logQueueAddress, cancellationToken); + await Forward(stored, logQueueAddress, dispatcher, cancellationToken); } foreach (var context in contexts) @@ -74,7 +71,7 @@ public async Task Ingest(List contexts, CancellationToken cancel } } - Task Forward(IReadOnlyCollection messageContexts, string forwardingAddress, CancellationToken cancellationToken) + Task Forward(IReadOnlyCollection messageContexts, string forwardingAddress, IMessageDispatcher dispatcher, CancellationToken cancellationToken) { var transportOperations = new List(messageContexts.Count); MessageContext anyContext = null; @@ -99,13 +96,13 @@ Task Forward(IReadOnlyCollection messageContexts, string forward } return anyContext != null - ? messageDispatcher.Value.Dispatch( + ? dispatcher.Dispatch( new TransportOperations([.. transportOperations]), anyContext.TransportTransaction, cancellationToken) : Task.CompletedTask; } - public async Task VerifyCanReachForwardingAddress(CancellationToken cancellationToken = default) + public async Task VerifyCanReachForwardingAddress(IMessageDispatcher dispatcher, CancellationToken cancellationToken = default) { if (!settings.ForwardAuditMessages) { @@ -122,7 +119,7 @@ public async Task VerifyCanReachForwardingAddress(CancellationToken cancellation ) ); - await messageDispatcher.Value.Dispatch(transportOperations, new TransportTransaction(), cancellationToken); + await dispatcher.Dispatch(transportOperations, new TransportTransaction(), cancellationToken); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { @@ -136,7 +133,6 @@ public async Task VerifyCanReachForwardingAddress(CancellationToken cancellation readonly AuditPersister auditPersister; readonly Settings settings; - readonly Lazy messageDispatcher; readonly string logQueueAddress; readonly ILogger logger; diff --git a/src/ServiceControl.Audit/Auditing/AuditPersister.cs b/src/ServiceControl.Audit/Auditing/AuditPersister.cs index fdd1c64dd4..832b2b1e06 100644 --- a/src/ServiceControl.Audit/Auditing/AuditPersister.cs +++ b/src/ServiceControl.Audit/Auditing/AuditPersister.cs @@ -19,10 +19,9 @@ class AuditPersister(IAuditIngestionUnitOfWorkFactory unitOfWorkFactory, IEnrichImportedAuditMessages[] enrichers, IMessageSession messageSession, - Lazy messageDispatcher, ILogger logger) { - public async Task> Persist(IReadOnlyList contexts, CancellationToken cancellationToken = default) + public async Task> Persist(IReadOnlyList contexts, IMessageDispatcher dispatcher, CancellationToken cancellationToken = default) { var storedContexts = new List(contexts.Count); IAuditIngestionUnitOfWork unitOfWork = null; @@ -33,7 +32,7 @@ public async Task> Persist(IReadOnlyList(contexts.Count); foreach (var context in contexts) { - inserts.Add(ProcessMessage(context, cancellationToken)); + inserts.Add(ProcessMessage(context, dispatcher, cancellationToken)); } await Task.WhenAll(inserts); @@ -93,7 +92,7 @@ public async Task> Persist(IReadOnlyList messageDispatcher, Settings settings, ILogger logger) { this.settings = settings; this.failedAuditStore = failedAuditStore; this.auditIngestor = auditIngestor; + this.messageDispatcher = messageDispatcher; this.logger = logger; } public async Task Run(CancellationToken cancellationToken = default) { - await auditIngestor.VerifyCanReachForwardingAddress(cancellationToken); + await auditIngestor.VerifyCanReachForwardingAddress(messageDispatcher.Value, cancellationToken); var succeeded = 0; var failed = 0; @@ -47,7 +49,7 @@ await failedAuditStore.ProcessFailedMessages( var taskCompletionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); messageContext.SetTaskCompletionSource(taskCompletionSource); - await auditIngestor.Ingest([messageContext], cancellationToken); + await auditIngestor.Ingest([messageContext], messageDispatcher.Value, cancellationToken); await taskCompletionSource.Task; @@ -78,6 +80,7 @@ await failedAuditStore.ProcessFailedMessages( readonly IFailedAuditStorage failedAuditStore; readonly AuditIngestor auditIngestor; + readonly Lazy messageDispatcher; readonly Settings settings; readonly ILogger logger; diff --git a/src/ServiceControl/Operations/ErrorIngestion.cs b/src/ServiceControl/Operations/ErrorIngestion.cs index a483a94991..cc89696c80 100644 --- a/src/ServiceControl/Operations/ErrorIngestion.cs +++ b/src/ServiceControl/Operations/ErrorIngestion.cs @@ -97,7 +97,7 @@ protected override async Task ExecuteAsync(CancellationToken cancellationToken = batchSizeMeter.Mark(contexts.Count); using (batchDurationMeter.Measure()) { - await ingestor.Ingest(contexts, cancellationToken); + await ingestor.Ingest(contexts, messageDispatcher, cancellationToken); } } catch (OperationCanceledException e) when (cancellationToken.IsCancellationRequested) @@ -138,23 +138,17 @@ public override async Task StopAsync(CancellationToken cancellationToken = defau { try { - await watchdog.Stop(cancellationToken); + // Order matters. Receiving stops first so nothing new enters the channel, but the + // infrastructure is left running while the channel drains, because those batches + // still forward through its dispatcher and Shutdown disposes it. + await EnsureReceivingStopped(cancellationToken); channel.Writer.Complete(); await base.StopAsync(cancellationToken); } finally { - if (transportInfrastructure != null) - { - try - { - await transportInfrastructure.Shutdown(cancellationToken); - } - catch (OperationCanceledException e) when (cancellationToken.IsCancellationRequested) - { - logger.LogInformation(e, "Shutdown cancelled"); - } - } + // Tears the infrastructure down, now that nothing is left to dispatch. + await watchdog.Stop(cancellationToken); } } @@ -226,10 +220,11 @@ async Task SetUpAndStartInfrastructure(CancellationToken cancellationToken) ); messageReceiver = transportInfrastructure.Receivers[errorQueue]; + messageDispatcher = transportInfrastructure.Dispatcher; if (settings.ForwardErrorMessages) { - await ingestor.VerifyCanReachForwardingAddress(cancellationToken); + await ingestor.VerifyCanReachForwardingAddress(messageDispatcher, cancellationToken); } await messageReceiver.StartReceive(cancellationToken); @@ -258,10 +253,7 @@ async Task StopAndTeardownInfrastructure(CancellationToken cancellationToken) logger.LogInformation("Stopping infrastructure"); try { - if (messageReceiver != null) - { - await messageReceiver.StopReceive(cancellationToken); - } + await StopReceiving(cancellationToken); } finally { @@ -270,6 +262,7 @@ async Task StopAndTeardownInfrastructure(CancellationToken cancellationToken) messageReceiver = null; transportInfrastructure = null; + receiveStopped = false; logger.LogInformation(LogMessages.StoppedInfrastructure); } @@ -327,11 +320,43 @@ async Task EnsureStopped(CancellationToken cancellationToken) } } + async Task EnsureReceivingStopped(CancellationToken cancellationToken) + { + await startStopSemaphore.WaitAsync(cancellationToken); + + try + { + await StopReceiving(cancellationToken); + } + finally + { + startStopSemaphore.Release(); + } + } + + // Stops the receiver on its own, leaving the infrastructure up. Idempotent because a + // shutdown stops receiving before draining and then tears down, so this runs twice. + async Task StopReceiving(CancellationToken cancellationToken) + { + if (messageReceiver == null || receiveStopped) + { + return; + } + + await messageReceiver.StopReceive(cancellationToken); + receiveStopped = true; + } + SemaphoreSlim startStopSemaphore = new(1); string errorQueue; ErrorIngestionFaultPolicy errorHandlingPolicy; TransportInfrastructure transportInfrastructure; IMessageReceiver messageReceiver; + bool receiveStopped; + + // Left in place when the infrastructure is torn down. A shutdown drains before tearing down, + // so this is still usable there + IMessageDispatcher messageDispatcher; readonly Settings settings; readonly ITransportCustomization transportCustomization; diff --git a/src/ServiceControl/Operations/ErrorIngestor.cs b/src/ServiceControl/Operations/ErrorIngestor.cs index 9e5040a601..9a05264c37 100644 --- a/src/ServiceControl/Operations/ErrorIngestor.cs +++ b/src/ServiceControl/Operations/ErrorIngestor.cs @@ -26,13 +26,11 @@ public ErrorIngestor(Metrics metrics, IEnumerable failedMessageEnrichers, IDomainEvents domainEvents, IIngestionUnitOfWorkFactory unitOfWorkFactory, - Lazy messageDispatcher, ITransportCustomization transportCustomization, Settings settings, ILogger logger) { this.unitOfWorkFactory = unitOfWorkFactory; - this.messageDispatcher = messageDispatcher; this.settings = settings; this.logger = logger; bulkInsertDurationMeter = metrics.GetMeter("Error ingestion - bulk insert duration", FrequencyInMilliseconds); @@ -51,7 +49,7 @@ public ErrorIngestor(Metrics metrics, logQueueAddress = new UnicastAddressTag(transportCustomization.ToTransportQualifiedQueueName(this.settings.ErrorLogQueue)); } - public async Task Ingest(List contexts, CancellationToken cancellationToken = default) + public async Task Ingest(List contexts, IMessageDispatcher dispatcher, CancellationToken cancellationToken = default) { var failedMessages = new List(contexts.Count); var retriedMessages = new List(contexts.Count); @@ -89,7 +87,7 @@ public async Task Ingest(List contexts, CancellationToken cancel { logger.LogDebug("Forwarding {FailedMessageCount} messages", storedFailed.Count); - await Forward(storedFailed, cancellationToken); + await Forward(storedFailed, dispatcher, cancellationToken); logger.LogDebug("Forwarded messages"); } @@ -149,7 +147,7 @@ async Task> PersistFailedMessages(List messageContexts, CancellationToken cancellationToken) + Task Forward(IReadOnlyCollection messageContexts, IMessageDispatcher dispatcher, CancellationToken cancellationToken) { var transportOperations = new TransportOperation[messageContexts.Count]; //We could allocate based on the actual number of ProcessedMessages but this should be OK var index = 0; @@ -170,13 +168,13 @@ Task Forward(IReadOnlyCollection messageContexts, CancellationTo } return anyContext != null - ? messageDispatcher.Value.Dispatch( + ? dispatcher.Dispatch( new TransportOperations(transportOperations), anyContext.TransportTransaction, cancellationToken) : Task.CompletedTask; } - public async Task VerifyCanReachForwardingAddress(CancellationToken cancellationToken = default) + public async Task VerifyCanReachForwardingAddress(IMessageDispatcher dispatcher, CancellationToken cancellationToken = default) { try { @@ -188,7 +186,7 @@ public async Task VerifyCanReachForwardingAddress(CancellationToken cancellation ) ); - await messageDispatcher.Value.Dispatch(transportOperations, new TransportTransaction(), cancellationToken); + await dispatcher.Dispatch(transportOperations, new TransportTransaction(), cancellationToken); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { @@ -204,7 +202,6 @@ public async Task VerifyCanReachForwardingAddress(CancellationToken cancellation readonly Meter bulkInsertDurationMeter; readonly Settings settings; readonly ErrorProcessor errorProcessor; - readonly Lazy messageDispatcher; readonly RetryConfirmationProcessor retryConfirmationProcessor; readonly UnicastAddressTag logQueueAddress; diff --git a/src/ServiceControl/Operations/ImportFailedErrors.cs b/src/ServiceControl/Operations/ImportFailedErrors.cs index a3eebda858..0dc6254c9e 100644 --- a/src/ServiceControl/Operations/ImportFailedErrors.cs +++ b/src/ServiceControl/Operations/ImportFailedErrors.cs @@ -1,5 +1,6 @@ namespace ServiceControl.Operations { + using System; using System.Threading; using System.Threading.Tasks; using NServiceBus.Extensibility; @@ -10,13 +11,14 @@ public class ImportFailedErrors( IFailedErrorImportDataStore store, ErrorIngestor errorIngestor, + Lazy messageDispatcher, Settings settings) { public async Task Run(CancellationToken cancellationToken = default) { if (settings.ForwardErrorMessages) { - await errorIngestor.VerifyCanReachForwardingAddress(cancellationToken); + await errorIngestor.VerifyCanReachForwardingAddress(messageDispatcher.Value, cancellationToken); } await store.ProcessFailedErrorImports(async (transportMessage, token) => @@ -32,7 +34,7 @@ await store.ProcessFailedErrorImports(async (transportMessage, token) => var taskCompletionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); messageContext.SetTaskCompletionSource(taskCompletionSource); - await errorIngestor.Ingest([messageContext], token); + await errorIngestor.Ingest([messageContext], messageDispatcher.Value, token); await taskCompletionSource.Task; }, cancellationToken); }