Prevent DeliveryConstraints from leaking in nested operations (#6356)

authored-by: Tim Bussmann <[email protected]>

src/NServiceBus.AcceptanceTests/Core/Pipeline/When_setting_ttbr_in_outer_publish.cs

@@ -0,0 +1,129 @@
+namespace NServiceBus.AcceptanceTests.Core.Pipeline
+{
+    using System;
+    using System.Threading.Tasks;
+    using AcceptanceTesting;
+    using DeliveryConstraints;
+    using EndpointTemplates;
+    using NServiceBus.Pipeline;
+    using NUnit.Framework;
+    using Performance.TimeToBeReceived;
+
+    public class When_setting_ttbr_in_outer_publish : NServiceBusAcceptanceTest
+    {
+        [Test]
+        public async Task Should_not_apply_ttbr_to_inner_publish()
+        {
+            Requires.NativePubSubSupport();
+
+            var context = await Scenario.Define<Context>()
+                .WithEndpoint<PublisherWithTtbr>(e => e
+                    .When(s => s.Publish(new OuterEvent())))
+                .WithEndpoint<Subscriber>()
+                .Done(c => c.OuterEventReceived && c.InnerEventReceived)
+                .Run();
+
+            Assert.IsTrue(context.OuterEventHasTtbr, "Outer event should have TTBR settings applied");
+            Assert.IsFalse(context.InnerEventHasTtbr, "Inner event should not have TTBR settings applied");
+        }
+
+        class Context : ScenarioContext
+        {
+            public bool InnerEventReceived { get; set; }
+            public bool OuterEventReceived { get; set; }
+            public bool OuterEventHasTtbr { get; set; }
+            public bool InnerEventHasTtbr { get; set; }
+        }
+
+        class PublisherWithTtbr : EndpointConfigurationBuilder
+        {
+            public PublisherWithTtbr()
+            {
+                EndpointSetup<DefaultServer>((c, r) =>
+                {
+                    c.Pipeline.Register(new InnerPublishBehavior(), "publishes an additional event when publishing an OuterEvent");
+                    c.Pipeline.Register(new TtbrObserver((Context)r.ScenarioContext), "Checks outgoing messages for their TTBR setting");
+                });
+            }
+
+            // Behavior needs to be in the OutgoingPhysical stage as the TTBR settings are applied in the OutgoingLogical stage
+            class InnerPublishBehavior : Behavior<IOutgoingPhysicalMessageContext>
+            {
+                public override async Task Invoke(IOutgoingPhysicalMessageContext context, Func<Task> next)
+                {
+                    await next();
+
+                    if (context.Extensions.Get<OutgoingLogicalMessage>().MessageType == typeof(OuterEvent))
+                    {
+                        await context.Publish(new InnerEvent());
+                    }
+                }
+            }
+
+            class TtbrObserver : Behavior<IDispatchContext>
+            {
+                Context testContext;
+
+                public TtbrObserver(Context testContext)
+                {
+                    this.testContext = testContext;
+                }
+
+                public override Task Invoke(IDispatchContext context, Func<Task> next)
+                {
+                    var outgoingMessage = context.Extensions.Get<OutgoingLogicalMessage>();
+                    if (outgoingMessage.MessageType == typeof(OuterEvent))
+                    {
+                        testContext.OuterEventHasTtbr = context.Extensions.TryGetDeliveryConstraint<DiscardIfNotReceivedBefore>(out var _);
+                    }
+
+                    if (outgoingMessage.MessageType == typeof(InnerEvent))
+                    {
+                        testContext.InnerEventHasTtbr = context.Extensions.TryGetDeliveryConstraint<DiscardIfNotReceivedBefore>(out var _);
+                    }
+
+                    return next();
+                }
+            }
+        }
+
+        class Subscriber : EndpointConfigurationBuilder
+        {
+            public Subscriber()
+            {
+                EndpointSetup<DefaultServer>();
+            }
+
+            class EventHandler : IHandleMessages<OuterEvent>, IHandleMessages<InnerEvent>
+            {
+                Context testContext;
+
+                public EventHandler(Context testContext)
+                {
+                    this.testContext = testContext;
+                }
+
+                public Task Handle(OuterEvent message, IMessageHandlerContext context)
+                {
+                    testContext.OuterEventReceived = true;
+                    return Task.FromResult(0);
+                }
+
+                public Task Handle(InnerEvent message, IMessageHandlerContext context)
+                {
+                    testContext.InnerEventReceived = true;
+                    return Task.FromResult(0);
+                }
+            }
+        }
+
+        [TimeToBeReceived("00:30:00")]
+        public class OuterEvent : IEvent
+        {
+        }
+
+        public class InnerEvent : IEvent
+        {
+        }
+    }
+}

src/NServiceBus.AcceptanceTests/DelayedDelivery/When_deferring_outer_send.cs

@@ -0,0 +1,120 @@
+namespace NServiceBus.AcceptanceTests.DelayedDelivery
+{
+    using System;
+    using System.Threading.Tasks;
+    using AcceptanceTesting;
+    using AcceptanceTesting.Customization;
+    using EndpointTemplates;
+    using Features;
+    using NServiceBus.Pipeline;
+    using NUnit.Framework;
+
+    public class When_deferring_outer_send : NServiceBusAcceptanceTest
+    {
+        [Test]
+        public async Task Should_not_defer_inner_send()
+        {
+            var context = await Scenario.Define<Context>()
+                .WithEndpoint<SenderWithNestedSend>(e => e
+                    .When(s =>
+                    {
+                        var sendOptions = new SendOptions();
+                        sendOptions.DelayDeliveryWith(TimeSpan.FromSeconds(2));
+                        return s.Send(new DelayedMessage(), sendOptions);
+                    }))
+                .WithEndpoint<Receiver>()
+                .Done(c => c.ReceivedNonDelayedMessage && c.ReceivedDelayedMessage)
+                .Run();
+
+            Assert.IsTrue(context.DelayedMessageDelayed, "should delay the message sent with 'DelayDeliveryWith'");
+            Assert.IsFalse(context.NonDelayedMessageDelayed, "should not delay the message sent with default options");
+        }
+
+        class Context : ScenarioContext
+        {
+            public bool DelayedMessageDelayed { get; set; }
+            public bool NonDelayedMessageDelayed { get; set; }
+
+            public bool ReceivedNonDelayedMessage { get; set; }
+            public bool ReceivedDelayedMessage { get; set; }
+        }
+
+        class SenderWithNestedSend : EndpointConfigurationBuilder
+        {
+            public SenderWithNestedSend() => EndpointSetup<DefaultServer>((c, r) =>
+            {
+                c.Pipeline.Register(new NestedSendBehavior(), "Sends an additional message when sending a delayed message");
+                c.ConfigureTransport().Routing().RouteToEndpoint(typeof(DelayedMessage).Assembly, Conventions.EndpointNamingConvention(typeof(Receiver)));
+                if (!TestSuiteConstraints.Current.SupportsNativeDeferral)
+                {
+                    c.EnableFeature<TimeoutManager>();
+                }
+            });
+
+            class NestedSendBehavior : Behavior<IOutgoingSendContext>
+            {
+                public override async Task Invoke(IOutgoingSendContext context, Func<Task> next)
+                {
+                    await next();
+                    if (context.Message.MessageType == typeof(DelayedMessage))
+                    {
+                        await context.Send(new NonDelayedMessage()); // use default options
+                    }
+                }
+            }
+        }
+
+        class Receiver : EndpointConfigurationBuilder
+        {
+            public Receiver() => EndpointSetup<DefaultServer>();
+
+            class DelayedMessageHandler : IHandleMessages<DelayedMessage>, IHandleMessages<NonDelayedMessage>
+            {
+                Context testContext;
+
+                public DelayedMessageHandler(Context testContext)
+                {
+                    this.testContext = testContext;
+                }
+
+                public Task Handle(DelayedMessage message, IMessageHandlerContext context)
+                {
+                    testContext.ReceivedDelayedMessage = true;
+                    if (TestSuiteConstraints.Current.SupportsNativeDeferral)
+                    {
+                        testContext.DelayedMessageDelayed = context.MessageHeaders.TryGetValue(Headers.DeliverAt, out var _); // header value not set when routing to timeout manager    
+                    }
+                    else
+                    {
+                        testContext.DelayedMessageDelayed = context.MessageHeaders.TryGetValue("NServiceBus.Timeout.RouteExpiredTimeoutTo", out var _); // header value when routing to timeout manager queue
+                    }
+
+                    return Task.FromResult(0);
+                }
+
+                public Task Handle(NonDelayedMessage message, IMessageHandlerContext context)
+                {
+                    testContext.ReceivedNonDelayedMessage = true;
+                    if (TestSuiteConstraints.Current.SupportsNativeDeferral)
+                    {
+                        testContext.NonDelayedMessageDelayed = context.MessageHeaders.TryGetValue(Headers.DeliverAt, out var _); // header value not set when routing to timeout manager    
+                    }
+                    else
+                    {
+                        testContext.NonDelayedMessageDelayed = context.MessageHeaders.TryGetValue("NServiceBus.Timeout.RouteExpiredTimeoutTo", out var _); // header value when routing to timeout manager queue
+                    }
+
+                    return Task.FromResult(0);
+                }
+            }
+        }
+
+        public class DelayedMessage : IMessage
+        {
+        }
+
+        public class NonDelayedMessage : IMessage
+        {
+        }
+    }
+}

src/NServiceBus.Core.Tests/Pipeline/Outgoing/OutgoingPublishContextTests.cs

@@ -1,7 +1,11 @@
 namespace NServiceBus.Core.Tests.Pipeline.Outgoing
 {
+    using System.Collections.Generic;
+    using DeliveryConstraints;
+    using Extensibility;
     using NServiceBus.Pipeline;
     using NUnit.Framework;
+    using Testing;
 
     [TestFixture]
     public class OutgoingPublishContextTests
@@ -40,5 +44,56 @@ public void ShouldNotMergeOptionsToParentContext()
 
             Assert.IsFalse(valueFound);
         }
+
+        [Test]
+        public void ShouldExposePublishOptionsExtensionsAsOperationProperties()
+        {
+            var message = new OutgoingLogicalMessage(typeof(object), new object());
+            var parentContext = new FakeRootContext(); // exact parent context doesn't matter
+            var options = new ContextBag();
+            options.Set("some key", "some value");
+
+            var publishContext = new OutgoingPublishContext(message, "message-id", new Dictionary<string, string>(), options, parentContext);
+
+            var operationProperties = publishContext.GetOperationProperties();
+            Assert.AreEqual("some value", operationProperties.Get<string>("some key"));
+        }
+
+        [Test]
+        public void ShouldNotLeakParentsOperationProperties()
+        {
+            var outerOptions = new ContextBag();
+            outerOptions.Set("outer key", "outer value");
+            outerOptions.Set("shared key", "outer shared value");
+            var parentContext = new OutgoingPublishContext(new OutgoingLogicalMessage(typeof(object), new object()), "message-id", new Dictionary<string, string>(), outerOptions, new FakeRootContext());
+
+            var innerOptions = new ContextBag();
+            innerOptions.Set("inner key", "inner value");
+            innerOptions.Set("shared key", "inner shared value");
+            var innerContext = new OutgoingPublishContext(new OutgoingLogicalMessage(typeof(object), new object()), "message-id", new Dictionary<string, string>(), innerOptions, parentContext);
+
+            var innerOperationProperties = innerContext.GetOperationProperties();
+            Assert.AreEqual("inner value", innerOperationProperties.Get<string>("inner key"));
+            Assert.AreEqual("inner shared value", innerOperationProperties.Get<string>("shared key"));
+            Assert.IsFalse(innerOperationProperties.TryGet("outer key", out string _));
+
+            var outerOperationProperties = parentContext.GetOperationProperties();
+            Assert.AreEqual("outer value", outerOperationProperties.Get<string>("outer key"));
+            Assert.AreEqual("outer shared value", outerOperationProperties.Get<string>("shared key"));
+            Assert.IsFalse(outerOperationProperties.TryGet("inner key", out string _));
+        }
+
+        [Test]
+        public void ShouldNotLeakParentsDeliveryConstraints()
+        {
+            var message = new OutgoingLogicalMessage(typeof(object), new object());
+            var publishOptions = new ContextBag();
+            var parentContext = new FakeRootContext();
+            parentContext.Extensions.AddDeliveryConstraint(new NonDurableDelivery());
+
+            var publishContext = new OutgoingPublishContext(message, "message-id", new Dictionary<string, string>(), publishOptions, parentContext);
+
+            Assert.IsFalse(publishContext.TryGetDeliveryConstraint(out NonDurableDelivery _));
+        }
     }
 }

src/NServiceBus.Core.Tests/Pipeline/Outgoing/OutgoingReplyContextTests.cs

@@ -1,5 +1,8 @@
 namespace NServiceBus.Core.Tests.Pipeline.Outgoing
 {
+    using System.Collections.Generic;
+    using DeliveryConstraints;
+    using Extensibility;
     using NServiceBus.Pipeline;
     using NUnit.Framework;
 
@@ -41,5 +44,56 @@ public void ShouldNotMergeOptionsToParentContext()
 
             Assert.IsFalse(valueFound);
         }
+
+        [Test]
+        public void ShouldExposeSendOptionsExtensionsAsOperationProperties()
+        {
+            var message = new OutgoingLogicalMessage(typeof(object), new object());
+            var parentContext = new FakeRootContext(); // exact parent context doesn't matter
+            var options = new ContextBag();
+            options.Set("some key", "some value");
+
+            var context = new OutgoingReplyContext(message, "message-id", new Dictionary<string, string>(), options, parentContext);
+
+            var operationProperties = context.GetOperationProperties();
+            Assert.AreEqual("some value", operationProperties.Get<string>("some key"));
+        }
+
+        [Test]
+        public void ShouldNotLeakParentsOperationProperties()
+        {
+            var outerOptions = new ContextBag();
+            outerOptions.Set("outer key", "outer value");
+            outerOptions.Set("shared key", "outer shared value");
+            var parentContext = new OutgoingReplyContext(new OutgoingLogicalMessage(typeof(object), new object()), "message-id", new Dictionary<string, string>(), outerOptions, new FakeRootContext());
+
+            var innerOptions = new ContextBag();
+            innerOptions.Set("inner key", "inner value");
+            innerOptions.Set("shared key", "inner shared value");
+            var innerContext = new OutgoingReplyContext(new OutgoingLogicalMessage(typeof(object), new object()), "message-id", new Dictionary<string, string>(), innerOptions, parentContext);
+
+            var innerOperationProperties = innerContext.GetOperationProperties();
+            Assert.AreEqual("inner value", innerOperationProperties.Get<string>("inner key"));
+            Assert.AreEqual("inner shared value", innerOperationProperties.Get<string>("shared key"));
+            Assert.IsFalse(innerOperationProperties.TryGet("outer key", out string _));
+
+            var outerOperationProperties = parentContext.GetOperationProperties();
+            Assert.AreEqual("outer value", outerOperationProperties.Get<string>("outer key"));
+            Assert.AreEqual("outer shared value", outerOperationProperties.Get<string>("shared key"));
+            Assert.IsFalse(outerOperationProperties.TryGet("inner key", out string _));
+        }
+
+        [Test]
+        public void ShouldNotLeakParentsDeliveryConstraints()
+        {
+            var message = new OutgoingLogicalMessage(typeof(object), new object());
+            var options = new ContextBag();
+            var parentContext = new FakeRootContext();
+            parentContext.Extensions.AddDeliveryConstraint(new NonDurableDelivery());
+
+            var context = new OutgoingReplyContext(message, "message-id", new Dictionary<string, string>(), options, parentContext);
+
+            Assert.IsFalse(context.TryGetDeliveryConstraint(out NonDurableDelivery _));
+        }
     }
 }