Support receive properties of a message as headers

src/Directory.Packages.props

@@ -54,7 +54,7 @@
     <PackageVersion Include="NServiceBus.RabbitMQ" Version="11.2.1" />
     <PackageVersion Include="NServiceBus.SagaAudit" Version="6.0.1" />
     <PackageVersion Include="NServiceBus.Testing" Version="10.1.0" />
-    <PackageVersion Include="NServiceBus.Transport.AzureServiceBus" Version="6.5.0" />
+    <PackageVersion Include="NServiceBus.Transport.AzureServiceBus" Version="6.6.0-alpha.1" />
     <PackageVersion Include="NServiceBus.Transport.AzureStorageQueues" Version="14.1.0" />
     <PackageVersion Include="NServiceBus.Transport.Msmq.Sources" Version="4.0.1" />
     <PackageVersion Include="NServiceBus.Transport.PostgreSql" Version="9.1.0" />

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<Context>()
+                .WithEndpoint<FailingEndpoint>(b => b.When(async ctx =>
+                {
+                    if (ctx.UniqueMessageId == null)
+                    {
+                        return false;
+                    }
+
+                    FailedMessage failedMessage = await this.TryGet<FailedMessage>($"/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<object>($"/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<DefaultServerWithoutAudit>(c =>
+                {
+                    c.GetSettings().Get<TransportDefinition>().TransportTransactionMode =
+                        TransportTransactionMode.ReceiveOnly;
+                    c.EnableFeature<Outbox>();
+
+                    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<MyMessage>
+            {
+                public Task Handle(MyMessage message, IMessageHandlerContext context)
+                {
+                    Console.WriteLine("Message Handled");
+                    if (scenarioContext.AboutToSendRetry)
+                    {
+                        scenarioContext.SecondAttemptFileCreatedAt = context.Extensions.Get<IncomingMessage>().ReceiveProperties["LearningTransport.FileCreatedAt"];
+                        scenarioContext.Retried = true;
+                    }
+                    else
+                    {
+                        scenarioContext.FirstAttemptFileCreatedAt = context.Extensions.Get<IncomingMessage>().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;
+    }
+}

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;

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<string, object>(metadata),
                     failureDetails);
 

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,16 +62,21 @@ static ExceptionDetails GetException(IReadOnlyDictionary<string, string> headers
             return exceptionDetails;
         }
 
-        public FailedMessage.ProcessingAttempt CreateProcessingAttempt(Dictionary<string, string> headers, Dictionary<string, object> metadata, FailureDetails failureDetails)
+        public FailedMessage.ProcessingAttempt CreateProcessingAttempt(Dictionary<string, string> headers, ReceiveProperties receiveProperties, Dictionary<string, object> metadata, FailureDetails failureDetails)
         {
-            return new FailedMessage.ProcessingAttempt
+            var processingAttempt = new FailedMessage.ProcessingAttempt
             {
                 AttemptedAt = failureDetails.TimeOfFailure,
                 FailureDetails = failureDetails,
                 MessageMetadata = metadata,
                 MessageId = headers[Headers.MessageId],
                 Headers = headers
             };
+            foreach (var property in receiveProperties)
+            {
+                headers["ServiceControl.ReceiveProperties." + property.Key] = property.Value;
+            }
+            return processingAttempt;
         }
 
         IFailedMessageEnricher[] failedEnrichers;

src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSender.cs

@@ -3,6 +3,7 @@ namespace ServiceControl.Recoverability
     using System;
     using System.Collections.Generic;
     using System.IO;
+    using System.Linq;
     using System.Threading;
     using System.Threading.Tasks;
     using Microsoft.Extensions.Logging;
@@ -16,6 +17,16 @@ public virtual async Task HandleMessage(MessageContext message, IMessageDispatch
         {
             var outgoingHeaders = new Dictionary<string, string>(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);