Preserve transport receive properties through message dispatch

src/NServiceBus.Core.Tests/ApprovalFiles/APIApprovals.ApproveNServiceBus.approved.txt

@@ -2419,6 +2419,7 @@ namespace NServiceBus.Transport
     public class ErrorContext
     {
         public ErrorContext(System.Exception exception, System.Collections.Generic.Dictionary<string, string> headers, string nativeMessageId, System.ReadOnlyMemory<byte> body, NServiceBus.Transport.TransportTransaction transportTransaction, int immediateProcessingFailures, string receiveAddress, NServiceBus.Extensibility.ContextBag context) { }
+        public ErrorContext(System.Exception exception, System.Collections.Generic.Dictionary<string, string> headers, string nativeMessageId, System.ReadOnlyMemory<byte> body, NServiceBus.Transport.ReceiveProperties receiveProperties, NServiceBus.Transport.TransportTransaction transportTransaction, int immediateProcessingFailures, string receiveAddress, NServiceBus.Extensibility.ContextBag context) { }
         public int DelayedDeliveriesPerformed { get; }
         public System.Exception Exception { get; }
         public NServiceBus.Extensibility.ContextBag Extensions { get; }
@@ -2481,10 +2482,12 @@ namespace NServiceBus.Transport
     public class IncomingMessage
     {
         public IncomingMessage(string nativeMessageId, System.Collections.Generic.Dictionary<string, string> headers, System.ReadOnlyMemory<byte> body) { }
+        public IncomingMessage(string nativeMessageId, System.Collections.Generic.Dictionary<string, string> headers, System.ReadOnlyMemory<byte> body, NServiceBus.Transport.ReceiveProperties receiveProperties) { }
         public System.ReadOnlyMemory<byte> Body { get; }
         public System.Collections.Generic.Dictionary<string, string> Headers { get; }
         public string MessageId { get; }
         public string NativeMessageId { get; }
+        public NServiceBus.Transport.ReceiveProperties ReceiveProperties { get; }
     }
     public static class IncomingMessageExtensions
     {
@@ -2494,11 +2497,13 @@ namespace NServiceBus.Transport
     public class MessageContext : NServiceBus.Extensibility.IExtendable
     {
         public MessageContext(string nativeMessageId, System.Collections.Generic.Dictionary<string, string> headers, System.ReadOnlyMemory<byte> body, NServiceBus.Transport.TransportTransaction transportTransaction, string receiveAddress, NServiceBus.Extensibility.ContextBag context) { }
+        public MessageContext(string nativeMessageId, System.Collections.Generic.Dictionary<string, string> headers, System.ReadOnlyMemory<byte> body, NServiceBus.Transport.ReceiveProperties receiveProperties, NServiceBus.Transport.TransportTransaction transportTransaction, string receiveAddress, NServiceBus.Extensibility.ContextBag context) { }
         public System.ReadOnlyMemory<byte> Body { get; }
         public NServiceBus.Extensibility.ContextBag Extensions { get; }
         public System.Collections.Generic.Dictionary<string, string> Headers { get; }
         public string NativeMessageId { get; }
         public string ReceiveAddress { get; }
+        public NServiceBus.Transport.ReceiveProperties ReceiveProperties { get; }
         public NServiceBus.Transport.TransportTransaction TransportTransaction { get; }
     }
     public class MulticastTransportOperation : NServiceBus.Transport.IOutgoingTransportOperation
@@ -2541,6 +2546,19 @@ namespace NServiceBus.Transport
         public System.Collections.Generic.IReadOnlyCollection<string> SendingAddresses { get; }
         public void BindSending(string transportAddress) { }
     }
+    public sealed class ReceiveProperties : System.Collections.Generic.IEnumerable<System.Collections.Generic.KeyValuePair<string, string>>, System.Collections.Generic.IReadOnlyCollection<System.Collections.Generic.KeyValuePair<string, string>>, System.Collections.Generic.IReadOnlyDictionary<string, string>, System.Collections.IEnumerable
+    {
+        public ReceiveProperties() { }
+        public ReceiveProperties(System.Collections.Generic.Dictionary<string, string> dictionary) { }
+        public int Count { get; }
+        public string this[string key] { get; }
+        public System.Collections.Generic.IEnumerable<string> Keys { get; }
+        public System.Collections.Generic.IEnumerable<string> Values { get; }
+        public static NServiceBus.Transport.ReceiveProperties Empty { get; }
+        public bool ContainsKey(string key) { }
+        public System.Collections.Generic.IEnumerator<System.Collections.Generic.KeyValuePair<string, string>> GetEnumerator() { }
+        public bool TryGetValue(string key, [System.Diagnostics.CodeAnalysis.MaybeNullWhen(false)] out string value) { }
+    }
     public class ReceiveSettings
     {
         public ReceiveSettings(string id, NServiceBus.Transport.QueueAddress receiveAddress, bool usePublishSubscribe, bool purgeOnStartup, string errorQueue) { }

src/NServiceBus.Core.Tests/Recoverability/ErrorContextTests.cs

@@ -1,9 +1,10 @@
 namespace NServiceBus.Core.Tests.Recoverability;
 
 using System;
-using Extensibility;
+using System.Collections.Generic;
+using NServiceBus.Extensibility;
+using NServiceBus.Transport;
 using NUnit.Framework;
-using Transport;
 
 [TestFixture]
 public class ErrorContextTests
@@ -17,4 +18,48 @@ public void Can_pass_additional_information_via_context_bag()
 
         Assert.That(context.Extensions.Get<string>("MyKey"), Is.EqualTo("MyValue"));
     }
+
+    [Test]
+    public void Should_carry_receive_properties_on_incoming_message()
+    {
+        var context = new ContextBag();
+        var receiveProperties = new ReceiveProperties(new Dictionary<string, string> { ["Native.Key"] = "Value" });
+
+        var errorContext = new ErrorContext(
+            exception: new InvalidOperationException("Test"),
+            headers: new Dictionary<string, string> { [Headers.MessageId] = "id" },
+            nativeMessageId: "native-id",
+            body: ReadOnlyMemory<byte>.Empty,
+            receiveProperties: receiveProperties,
+            transportTransaction: new TransportTransaction(),
+            immediateProcessingFailures: 1,
+            receiveAddress: "queue",
+            context: context
+        );
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(errorContext.Message.ReceiveProperties, Is.SameAs(receiveProperties));
+            Assert.That(errorContext.Message.ReceiveProperties["Native.Key"], Is.EqualTo("Value"));
+        }
+    }
+
+    [Test]
+    public void Should_default_receive_properties_to_empty_when_not_provided()
+    {
+        var context = new ContextBag();
+
+        var errorContext = new ErrorContext(
+            exception: new InvalidOperationException("Test"),
+            headers: [],
+            nativeMessageId: "native-id",
+            body: ReadOnlyMemory<byte>.Empty,
+            transportTransaction: new TransportTransaction(),
+            immediateProcessingFailures: 1,
+            receiveAddress: "queue",
+            context: context
+        );
+
+        Assert.That(errorContext.Message.ReceiveProperties, Is.SameAs(ReceiveProperties.Empty));
+    }
 }

src/NServiceBus.Core.Tests/Routing/RoutingToDispatchConnectorTests.cs

@@ -1,6 +1,7 @@
 namespace NServiceBus.Core.Tests.Routing;
 
 using System;
+using System.Collections.Frozen;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
@@ -60,7 +61,7 @@ await behavior.Invoke(testableRoutingContext, context =>
     public async Task Should_copy_message_state_for_multiple_routing_strategies()
     {
         var behavior = new RoutingToDispatchConnector();
-        IEnumerable<TransportOperation> operations = null;
+        List<TransportOperation> operations = null;
         var testableRoutingContext = new TestableRoutingContext
         {
             RoutingStrategies =
@@ -78,11 +79,11 @@ public async Task Should_copy_message_state_for_multiple_routing_strategies()
         testableRoutingContext.Message = new OutgoingMessage("ID", originalHeaders, Array.Empty<byte>());
         await behavior.Invoke(testableRoutingContext, context =>
         {
-            operations = context.Operations;
+            operations = [.. context.Operations];
             return Task.CompletedTask;
         });
 
-        Assert.That(operations.ToList(), Has.Count.EqualTo(2));
+        Assert.That(operations, Has.Count.EqualTo(2));
 
         TransportOperation destination1Operation = operations.ElementAt(0);
         using (Assert.EnterMultipleScope())
@@ -91,13 +92,20 @@ await behavior.Invoke(testableRoutingContext, context =>
             Assert.That((destination1Operation.AddressTag as UnicastAddressTag)?.Destination, Is.EqualTo("destination1"));
         }
         Dictionary<string, string> destination1Headers = destination1Operation.Message.Headers;
-        Assert.That(destination1Headers, Contains.Item(new KeyValuePair<string, string>("SomeHeaderKey", "SomeHeaderValue")));
-        Assert.That(destination1Headers, Contains.Item(new KeyValuePair<string, string>("HeaderKeyAddedByTheRoutingStrategy1", "HeaderValueAddedByTheRoutingStrategy1")));
-        Assert.That(destination1Headers, Does.Not.Contain(new KeyValuePair<string, string>("HeaderKeyAddedByTheRoutingStrategy2", "HeaderValueAddedByTheRoutingStrategy2")));
-        Assert.That(destination1Headers, Is.Not.SameAs(originalHeaders));
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(destination1Headers, Contains.Item(new KeyValuePair<string, string>("SomeHeaderKey", "SomeHeaderValue")));
+            Assert.That(destination1Headers, Contains.Item(new KeyValuePair<string, string>("HeaderKeyAddedByTheRoutingStrategy1", "HeaderValueAddedByTheRoutingStrategy1")));
+            Assert.That(destination1Headers, Does.Not.Contain(new KeyValuePair<string, string>("HeaderKeyAddedByTheRoutingStrategy2", "HeaderValueAddedByTheRoutingStrategy2")));
+            Assert.That(destination1Headers, Is.Not.SameAs(originalHeaders));
+        }
+
         DispatchProperties destination1DispatchProperties = destination1Operation.Properties;
-        Assert.That(destination1DispatchProperties, Contains.Item(new KeyValuePair<string, string>("SomeKey", "SomeValue")));
-        Assert.That(destination1DispatchProperties, Is.Not.SameAs(originalDispatchProperties));
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(destination1DispatchProperties, Contains.Item(new KeyValuePair<string, string>("SomeKey", "SomeValue")));
+            Assert.That(destination1DispatchProperties, Is.Not.SameAs(originalDispatchProperties));
+        }
 
         TransportOperation destination2Operation = operations.ElementAt(1);
         using (Assert.EnterMultipleScope())
@@ -106,10 +114,14 @@ await behavior.Invoke(testableRoutingContext, context =>
             Assert.That((destination2Operation.AddressTag as UnicastAddressTag)?.Destination, Is.EqualTo("destination2"));
         }
         Dictionary<string, string> destination2Headers = destination2Operation.Message.Headers;
-        Assert.That(destination2Headers, Contains.Item(new KeyValuePair<string, string>("SomeHeaderKey", "SomeHeaderValue")));
-        Assert.That(destination2Headers, Contains.Item(new KeyValuePair<string, string>("HeaderKeyAddedByTheRoutingStrategy2", "HeaderValueAddedByTheRoutingStrategy2")));
-        Assert.That(destination2Headers, Does.Not.Contain(new KeyValuePair<string, string>("HeaderKeyAddedByTheRoutingStrategy1", "HeaderValueAddedByTheRoutingStrategy1")));
-        Assert.That(destination2Headers, Is.Not.SameAs(originalHeaders));
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(destination2Headers, Contains.Item(new KeyValuePair<string, string>("SomeHeaderKey", "SomeHeaderValue")));
+            Assert.That(destination2Headers, Contains.Item(new KeyValuePair<string, string>("HeaderKeyAddedByTheRoutingStrategy2", "HeaderValueAddedByTheRoutingStrategy2")));
+            Assert.That(destination2Headers, Does.Not.Contain(new KeyValuePair<string, string>("HeaderKeyAddedByTheRoutingStrategy1", "HeaderValueAddedByTheRoutingStrategy1")));
+            Assert.That(destination2Headers, Is.Not.SameAs(originalHeaders));
+        }
+
         DispatchProperties destination2DispatchProperties = destination2Operation.Properties;
         Assert.That(destination2DispatchProperties, Is.Not.SameAs(originalDispatchProperties));
         using (Assert.EnterMultipleScope())
@@ -242,4 +254,114 @@ public override AddressTag Apply(Dictionary<string, string> headers)
     }
 
     class MyMessage : IMessage;
+
+    [Test]
+    public async Task Should_merge_receive_properties_when_declared_by_transport()
+    {
+        var behavior = new RoutingToDispatchConnector();
+
+        var receiveProperties = new ReceiveProperties(new Dictionary<string, string>
+        {
+            ["MessageGroupId"] = "group-123",
+            ["MessageDeduplicationId"] = "dedup-456"
+        });
+
+        TransportOperation transportOperation = null;
+
+        var routingContext = new TestableRoutingContext
+        {
+            RoutingStrategies = [new UnicastRoutingStrategy("destination")]
+        };
+        routingContext.Extensions.Set(new IncomingMessage(routingContext.Message.MessageId, [], Array.Empty<byte>(), receiveProperties));
+
+        await behavior.Invoke(routingContext, context =>
+        {
+            transportOperation = context.Operations.Single();
+            return Task.CompletedTask;
+        });
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(transportOperation, Is.Not.Null);
+            Assert.That(transportOperation!.Properties["MessageGroupId"], Is.EqualTo("group-123"));
+            Assert.That(transportOperation!.Properties["MessageDeduplicationId"], Is.EqualTo("dedup-456"));
+        }
+    }
+
+    [Test]
+    public async Task Should_not_override_user_set_dispatch_property()
+    {
+        var behavior = new RoutingToDispatchConnector();
+
+        var receiveProperties = new ReceiveProperties(new Dictionary<string, string>
+        {
+            ["MessageGroupId"] = "incoming-group"
+        });
+
+        var userDispatchProperties = new DispatchProperties
+        {
+            ["MessageGroupId"] = "user-specified-group"
+        };
+
+        TransportOperation transportOperation = null;
+
+        var routingContext = new TestableRoutingContext
+        {
+            RoutingStrategies = [new UnicastRoutingStrategy("destination")]
+        };
+        routingContext.Extensions.Set(new IncomingMessage(routingContext.Message.MessageId, [], Array.Empty<byte>(), receiveProperties));
+        routingContext.Extensions.Set(userDispatchProperties);
+
+        await behavior.Invoke(routingContext, context =>
+        {
+            transportOperation = context.Operations.Single();
+            return Task.CompletedTask;
+        });
+
+        Assert.That(transportOperation!.Properties["MessageGroupId"], Is.EqualTo("user-specified-group"),
+            "User-set DispatchProperties should take precedence over ReceiveProperties");
+    }
+
+    [Test]
+    public async Task Should_preserve_user_dispatch_properties_even_with_receive_properties()
+    {
+        var behavior = new RoutingToDispatchConnector();
+
+        var receiveProperties = new ReceiveProperties(new Dictionary<string, string>
+        {
+            ["MessageGroupId"] = "incoming-group",
+            ["DeduplicationId"] = "incoming-dedup"
+        });
+
+        var userDispatchProperties = new DispatchProperties
+        {
+            ["MessageGroupId"] = "user-group",
+            ["Custom.Property"] = "custom-value"
+        };
+
+        TransportOperation transportOperation = null;
+
+        var routingContext = new TestableRoutingContext
+        {
+            RoutingStrategies = [new UnicastRoutingStrategy("destination")]
+        };
+        routingContext.Extensions.Set(new IncomingMessage(routingContext.Message.MessageId, [], Array.Empty<byte>(), receiveProperties));
+        routingContext.Extensions.Set(userDispatchProperties);
+
+        await behavior.Invoke(routingContext, context =>
+        {
+            transportOperation = context.Operations.Single();
+            return Task.CompletedTask;
+        });
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(transportOperation!.Properties["MessageGroupId"], Is.EqualTo("user-group"),
+                "User-set property wins");
+            Assert.That(transportOperation.Properties["DeduplicationId"], Is.EqualTo("incoming-dedup"),
+                "Receive property merged when not set by user");
+            Assert.That(transportOperation.Properties["Custom.Property"], Is.EqualTo("custom-value"),
+                "User custom property preserved");
+        }
+    }
 }

src/NServiceBus.Core.Tests/Transports/MessageContextTests.cs

@@ -0,0 +1,48 @@
+namespace NServiceBus.Core.Tests.Transports;
+
+using System;
+using System.Collections.Generic;
+using NServiceBus.Extensibility;
+using NServiceBus.Transport;
+using NUnit.Framework;
+
+[TestFixture]
+public class MessageContextTests
+{
+    [Test]
+    public void Should_accept_receive_properties_via_ctor()
+    {
+        var receiveProperties = new ReceiveProperties(new Dictionary<string, string>
+        {
+            ["Native.CustomProperty"] = "CustomValue",
+            ["MessageGroupId"] = "group-123"
+        });
+
+        var context = new MessageContext(
+            nativeMessageId: "native-id",
+            headers: [],
+            body: ReadOnlyMemory<byte>.Empty,
+            receiveProperties: receiveProperties,
+            transportTransaction: new TransportTransaction(),
+            receiveAddress: "queue@machine",
+            context: new ContextBag()
+        );
+
+        Assert.That(context.ReceiveProperties, Is.SameAs(receiveProperties));
+    }
+
+    [Test]
+    public void Should_default_receive_properties_to_empty()
+    {
+        var context = new MessageContext(
+            nativeMessageId: "native-id",
+            headers: [],
+            body: ReadOnlyMemory<byte>.Empty,
+            transportTransaction: new TransportTransaction(),
+            receiveAddress: "queue@machine",
+            context: new ContextBag()
+        );
+
+        Assert.That(context.ReceiveProperties, Is.SameAs(ReceiveProperties.Empty));
+    }
+}