From faf69f48aead29f4660d7fa96af8820f4f523cf5 Mon Sep 17 00:00:00 2001 From: SzymonPobiega Date: Mon, 5 Oct 2026 15:31:27 +0200 Subject: [PATCH 1/2] Support transport properties - Store receive properties - Set dispatch properties --- src/Directory.Packages.props | 2 +- .../ASBSTransportCustomization.cs | 1 - src/ServiceControl/Operations/ErrorProcessor.cs | 1 + .../Operations/FailedMessageFactory.cs | 10 ++++++++-- .../Retrying/Infrastructure/ReturnToSender.cs | 13 ++++++++++++- 5 files changed, 22 insertions(+), 5 deletions(-) diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props index 1a712572da..099ff4c023 100644 --- a/src/Directory.Packages.props +++ b/src/Directory.Packages.props @@ -54,7 +54,7 @@ - + diff --git a/src/ServiceControl.Transports.ASBS/ASBSTransportCustomization.cs b/src/ServiceControl.Transports.ASBS/ASBSTransportCustomization.cs index 778dec09d5..45ddca6829 100644 --- a/src/ServiceControl.Transports.ASBS/ASBSTransportCustomization.cs +++ b/src/ServiceControl.Transports.ASBS/ASBSTransportCustomization.cs @@ -51,7 +51,6 @@ protected override AzureServiceBusTransport CreateTransport(TransportSettings tr { transport.HierarchyNamespaceOptions = new HierarchyNamespaceOptions { HierarchyNamespace = connectionSettings.HierarchyNamespace }; } - transport.TransportTransactionMode = transport.GetSupportedTransactionModes().Contains(preferredTransactionMode) ? preferredTransactionMode : TransportTransactionMode.ReceiveOnly; return transport; diff --git a/src/ServiceControl/Operations/ErrorProcessor.cs b/src/ServiceControl/Operations/ErrorProcessor.cs index ab9db44356..e5f17dab92 100644 --- a/src/ServiceControl/Operations/ErrorProcessor.cs +++ b/src/ServiceControl/Operations/ErrorProcessor.cs @@ -114,6 +114,7 @@ async Task ProcessMessage(MessageContext context, IIngestionUnitOfWork unitOfWor var processingAttempt = failedMessageFactory.CreateProcessingAttempt( context.Headers, + context.ReceiveProperties, new Dictionary(metadata), failureDetails); diff --git a/src/ServiceControl/Operations/FailedMessageFactory.cs b/src/ServiceControl/Operations/FailedMessageFactory.cs index b8ec615d28..31974ae0bd 100644 --- a/src/ServiceControl/Operations/FailedMessageFactory.cs +++ b/src/ServiceControl/Operations/FailedMessageFactory.cs @@ -6,6 +6,7 @@ using Infrastructure; using NServiceBus; using NServiceBus.Faults; + using NServiceBus.Transport; using Recoverability; using FailedMessage = MessageFailures.FailedMessage; @@ -61,9 +62,9 @@ static ExceptionDetails GetException(IReadOnlyDictionary headers return exceptionDetails; } - public FailedMessage.ProcessingAttempt CreateProcessingAttempt(Dictionary headers, Dictionary metadata, FailureDetails failureDetails) + public FailedMessage.ProcessingAttempt CreateProcessingAttempt(Dictionary headers, ReceiveProperties receiveProperties, Dictionary metadata, FailureDetails failureDetails) { - return new FailedMessage.ProcessingAttempt + var processingAttempt = new FailedMessage.ProcessingAttempt { AttemptedAt = failureDetails.TimeOfFailure, FailureDetails = failureDetails, @@ -71,6 +72,11 @@ public FailedMessage.ProcessingAttempt CreateProcessingAttempt(Dictionary(message.Headers); + var receiveProperties = message.Headers.Keys.Where(x => x.StartsWith("ServiceControl.ReceiveProperties.")); + DispatchProperties dispatchProperties = null; + foreach (string prop in receiveProperties) + { + dispatchProperties ??= []; + var propValue = outgoingHeaders[prop]; + outgoingHeaders.Remove(prop); + dispatchProperties[prop.Replace("ServiceControl.ReceiveProperties.", "")] = propValue; + } + outgoingHeaders.Remove("ServiceControl.Retry.StagingId"); outgoingHeaders["ServiceControl.Retry.AcknowledgementQueue"] = errorQueueTransportAddress; @@ -50,7 +61,7 @@ public virtual async Task HandleMessage(MessageContext message, IMessageDispatch logger.LogDebug("{MessageId}: Found ServiceControl.RetryTo header. Rerouting to {RetryTo}", messageId, retryTo); } - var transportOp = new TransportOperation(outgoingMessage, new UnicastAddressTag(retryTo)); + var transportOp = new TransportOperation(outgoingMessage, new UnicastAddressTag(retryTo), dispatchProperties); await sender.Dispatch(new TransportOperations(transportOp), message.TransportTransaction, cancellationToken); From 868dd9bb821fbd64546f91465d889c8bd119085a Mon Sep 17 00:00:00 2001 From: SzymonPobiega Date: Fri, 9 Oct 2026 12:02:19 +0200 Subject: [PATCH 2/2] Add a test for property round-trip --- ...sage_with_receive_properties_is_retried.cs | 113 ++++++++++++++++++ 1 file changed, 113 insertions(+) create mode 100644 src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_a_failed_message_with_receive_properties_is_retried.cs diff --git a/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_a_failed_message_with_receive_properties_is_retried.cs b/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_a_failed_message_with_receive_properties_is_retried.cs new file mode 100644 index 0000000000..4e5ba7cf74 --- /dev/null +++ b/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_a_failed_message_with_receive_properties_is_retried.cs @@ -0,0 +1,113 @@ +namespace ServiceControl.AcceptanceTests.RavenDB.Recoverability.MessageFailures +{ + using System; + using System.Linq; + using System.Threading; + using System.Threading.Tasks; + using AcceptanceTesting; + using AcceptanceTesting.EndpointTemplates; + using Infrastructure; + using NServiceBus; + using NServiceBus.AcceptanceTesting; + using NServiceBus.Configuration.AdvancedExtensibility; + using NServiceBus.Features; + using NServiceBus.Settings; + using NServiceBus.Transport; + using NUnit.Framework; + using ServiceControl.MessageFailures; + + class When_a_failed_message_with_receive_properties_is_retried : AcceptanceTest + { + [Test] + public async Task Should_preserve_receive_dispatch_properties() + { + //HINT: LearningTransport defines one receive property "LearningTransport.FileCreatedAt" + // The property should have the same value when initially processed as when retried + var scenarioContext = await Define() + .WithEndpoint(b => b.When(async ctx => + { + if (ctx.UniqueMessageId == null) + { + return false; + } + + FailedMessage failedMessage = await this.TryGet($"/api/errors/{ctx.UniqueMessageId}"); + if (failedMessage == null) + { + return false; + } + + ctx.FailureGroupId = failedMessage.FailureGroups.First().Id; + + return true; + }, async (bus, ctx) => + { + ctx.AboutToSendRetry = true; + await this.Post($"/api/recoverability/groups/{ctx.FailureGroupId}/errors/retry"); + }).DoNotFailOnErrorMessages()) + .Done(ctx => ctx.Retried) + .Run(); + + Assert.That(scenarioContext.SecondAttemptFileCreatedAt, Is.EqualTo(scenarioContext.FirstAttemptFileCreatedAt), "Receive properties are not preserved"); + } + + public class FailingEndpoint : EndpointConfigurationBuilder + { + public FailingEndpoint() => + EndpointSetup(c => + { + c.GetSettings().Get().TransportTransactionMode = + TransportTransactionMode.ReceiveOnly; + c.EnableFeature(); + + c.RegisterStartupTask(new SendMessageAtStart()); + + c.NoRetries(); + }); + + class SendMessageAtStart : FeatureStartupTask + { + protected override Task OnStart(IMessageSession session, CancellationToken cancellationToken = default) + => session.SendLocal(new MyMessage(), cancellationToken); + + protected override Task OnStop(IMessageSession session, CancellationToken cancellationToken = default) + => Task.CompletedTask; + } + + [Handler] + public class MyMessageHandler(Context scenarioContext, IReadOnlySettings settings) + : IHandleMessages + { + public Task Handle(MyMessage message, IMessageHandlerContext context) + { + Console.WriteLine("Message Handled"); + if (scenarioContext.AboutToSendRetry) + { + scenarioContext.SecondAttemptFileCreatedAt = context.Extensions.Get().ReceiveProperties["LearningTransport.FileCreatedAt"]; + scenarioContext.Retried = true; + } + else + { + scenarioContext.FirstAttemptFileCreatedAt = context.Extensions.Get().ReceiveProperties["LearningTransport.FileCreatedAt"]; + scenarioContext.UniqueMessageId = DeterministicGuid.MakeId(context.MessageId, settings.EndpointName()).ToString(); + throw new Exception("Simulated Exception"); + } + + return Task.CompletedTask; + } + } + } + + public class Context : ScenarioContext + { + public string FirstAttemptFileCreatedAt { get; set; } + public string SecondAttemptFileCreatedAt { get; set; } + public string UniqueMessageId { get; set; } + public string FailureGroupId { get; set; } + public bool Retried { get; set; } + public bool AboutToSendRetry { get; set; } + } + + public class MyMessage : ICommand; + } +} \ No newline at end of file