Introduce incoming message support for envelope unwrapping

src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Metrics/When_envelope_handler_fails.cs

@@ -0,0 +1,73 @@
+namespace NServiceBus.AcceptanceTests.Core.OpenTelemetry.Metrics;
+
+using System;
+using System.Buffers;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using NServiceBus;
+using NServiceBus.AcceptanceTesting;
+using NServiceBus.AcceptanceTests.Core.OpenTelemetry;
+using NServiceBus.AcceptanceTests.EndpointTemplates;
+using NServiceBus.Extensibility;
+using NServiceBus.Features;
+using NUnit.Framework;
+using Conventions = AcceptanceTesting.Customization.Conventions;
+
+public class When_envelope_handler_fails : OpenTelemetryAcceptanceTest
+{
+    [Test]
+    public async Task Should_report_failed_unwrapping_metric()
+    {
+        using var metricsListener = TestingMetricListener.SetupNServiceBusMetricsListener();
+
+        _ = await Scenario.Define<Context>()
+            .WithEndpoint<EndpointWithMetrics>(b => b.CustomConfig(x =>
+                {
+                    x.MakeInstanceUniquelyAddressable("discriminator");
+                    x.EnableFeature<TestEnvelopeFeature>();
+                })
+                .When(session => session.SendLocal(new OutgoingMessage())))
+            .Run();
+
+        metricsListener.AssertMetric("nservicebus.envelope.unwrapped", 1);
+
+        metricsListener.AssertTags("nservicebus.envelope.unwrapped",
+            new Dictionary<string, object>
+            {
+                ["nservicebus.queue"] = Conventions.EndpointNamingConvention(typeof(EndpointWithMetrics)),
+                ["nservicebus.discriminator"] = "discriminator",
+                ["nservicebus.envelope.unwrapper_type"] = typeof(ThrowingHandler).FullName,
+                ["error.type"] = typeof(InvalidOperationException).FullName
+            });
+    }
+
+    class ThrowingHandler : IEnvelopeHandler
+    {
+        public Dictionary<string, string> UnwrapEnvelope(string nativeMessageId, IDictionary<string, string> incomingHeaders, ReadOnlySpan<byte> incomingBody, ContextBag extensions, IBufferWriter<byte> bodyWriter)
+            => throw new InvalidOperationException("Some exception");
+    }
+
+    class TestEnvelopeFeature : Feature
+    {
+        protected override void Setup(FeatureConfigurationContext context) => context.AddEnvelopeHandler<ThrowingHandler>();
+    }
+
+    class Context : ScenarioContext;
+
+    class EndpointWithMetrics : EndpointConfigurationBuilder
+    {
+        public EndpointWithMetrics() => EndpointSetup<DefaultServer>();
+
+        public class MessageHandler(Context testContext) : IHandleMessages<OutgoingMessage>
+        {
+            public Task Handle(OutgoingMessage message, IMessageHandlerContext context)
+            {
+                testContext.MarkAsCompleted();
+                return Task.CompletedTask;
+            }
+        }
+    }
+
+    public class OutgoingMessage : IMessage;
+}

src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Metrics/When_envelope_handler_succeeds.cs

@@ -0,0 +1,76 @@
+namespace NServiceBus.AcceptanceTests.Core.OpenTelemetry.Metrics;
+
+using System;
+using System.Buffers;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using NServiceBus;
+using NServiceBus.AcceptanceTesting;
+using NServiceBus.AcceptanceTests.Core.OpenTelemetry;
+using NServiceBus.AcceptanceTests.EndpointTemplates;
+using NServiceBus.Extensibility;
+using NServiceBus.Features;
+using NUnit.Framework;
+using Conventions = AcceptanceTesting.Customization.Conventions;
+
+public class When_envelope_handler_succeeds : OpenTelemetryAcceptanceTest
+{
+    [Test]
+    public async Task Should_report_successful_unwrapping_metric()
+    {
+        using var metricsListener = TestingMetricListener.SetupNServiceBusMetricsListener();
+
+        _ = await Scenario.Define<Context>()
+            .WithEndpoint<EndpointWithMetrics>(b => b.CustomConfig(x =>
+                {
+                    x.MakeInstanceUniquelyAddressable("discriminator");
+                    x.EnableFeature<TestEnvelopeFeature>();
+                })
+                .When(session => session.SendLocal(new OutgoingMessage())))
+            .Run();
+
+        // The metric should be explicitly emitted with a value of 0 to indicate no errors occurred
+        Assert.That(metricsListener.ReportedMeters["nservicebus.envelope.unwrapped"], Is.EqualTo(0));
+
+        metricsListener.AssertTags("nservicebus.envelope.unwrapped",
+            new Dictionary<string, object>
+            {
+                ["nservicebus.queue"] = Conventions.EndpointNamingConvention(typeof(EndpointWithMetrics)),
+                ["nservicebus.discriminator"] = "discriminator",
+                ["nservicebus.envelope.unwrapper_type"] = typeof(SuccessfulCloudEventHandler).FullName
+            });
+    }
+
+    class SuccessfulCloudEventHandler : IEnvelopeHandler
+    {
+        public Dictionary<string, string> UnwrapEnvelope(string nativeMessageId, IDictionary<string, string> incomingHeaders, ReadOnlySpan<byte> incomingBody, ContextBag extensions, IBufferWriter<byte> bodyWriter)
+        {
+            bodyWriter.Write(incomingBody);
+            return incomingHeaders.ToDictionary();
+        }
+    }
+
+    class TestEnvelopeFeature : Feature
+    {
+        protected override void Setup(FeatureConfigurationContext context) => context.AddEnvelopeHandler<SuccessfulCloudEventHandler>();
+    }
+
+    class Context : ScenarioContext;
+
+    class EndpointWithMetrics : EndpointConfigurationBuilder
+    {
+        public EndpointWithMetrics() => EndpointSetup<DefaultServer>();
+
+        public class MessageHandler(Context testContext) : IHandleMessages<OutgoingMessage>
+        {
+            public Task Handle(OutgoingMessage message, IMessageHandlerContext context)
+            {
+                testContext.MarkAsCompleted();
+                return Task.CompletedTask;
+            }
+        }
+    }
+
+    public class OutgoingMessage : IMessage;
+}

src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/When_incoming_message_handled.cs

@@ -81,7 +81,6 @@ static async Task<TestingMetricListener> WhenMessagesHandled(Func<IMessage> mess
                                 }
                             }
                         }))
-                .Done(c => c.TotalHandledMessages == 5)
                 .Run();
             return metricsListener;
         }
@@ -120,7 +119,7 @@ class MyMessageHandler(Context testContext) : IHandleMessages<MyMessage>
     {
         public Task Handle(MyMessage message, IMessageHandlerContext context)
         {
-            Interlocked.Increment(ref testContext.TotalHandledMessages);
+            testContext.MarkAsCompleted(Interlocked.Increment(ref testContext.TotalHandledMessages) == 5);
             return Task.CompletedTask;
         }
     }
@@ -129,7 +128,8 @@ class MyExceptionalHandler(Context testContext) : IHandleMessages<MyExceptionalM
     {
         public Task Handle(MyExceptionalMessage message, IMessageHandlerContext context)
         {
-            Interlocked.Increment(ref testContext.TotalHandledMessages);
+            var count = Interlocked.Increment(ref testContext.TotalHandledMessages);
+            testContext.MarkAsCompleted(count == 5);
             throw new Exception();
         }
     }

src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/When_incoming_message_handled_successfully.cs

@@ -25,7 +25,6 @@ public async Task Should_record_handling_time()
                         await session.SendLocal(new MyMessage());
                     }
                 }))
-            .Done(c => c.TotalHandledMessages == 5)
             .Run();
 
         string handlingTime = "nservicebus.messaging.handler_time";
@@ -52,7 +51,8 @@ class MyMessageHandler(Context testContext) : IHandleMessages<MyMessage>
     {
         public Task Handle(MyMessage message, IMessageHandlerContext context)
         {
-            Interlocked.Increment(ref testContext.TotalHandledMessages);
+            var count = Interlocked.Increment(ref testContext.TotalHandledMessages);
+            testContext.MarkAsCompleted(count == 5);
             return Task.CompletedTask;
         }
     }

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

@@ -297,6 +297,14 @@ namespace NServiceBus
     {
         public static NServiceBus.IStartableEndpointWithExternallyManagedContainer Create(NServiceBus.EndpointConfiguration configuration, Microsoft.Extensions.DependencyInjection.IServiceCollection serviceCollection) { }
     }
+    public static class EnvelopeConfigExtensions
+    {
+        extension(NServiceBus.Features.FeatureConfigurationContext context)
+        {
+            public void AddEnvelopeHandler<THandler>()
+                where THandler :  class, NServiceBus.IEnvelopeHandler { }
+        }
+    }
     public static class ErrorQueueSettings
     {
         public const string SettingsKey = "errorQueue";
@@ -456,6 +464,10 @@ namespace NServiceBus
     {
         System.Threading.Tasks.Task Stop(System.Threading.CancellationToken cancellationToken = default);
     }
+    public interface IEnvelopeHandler
+    {
+        System.Collections.Generic.Dictionary<string, string>? UnwrapEnvelope(string nativeMessageId, System.Collections.Generic.IDictionary<string, string> incomingHeaders, System.ReadOnlySpan<byte> incomingBody, NServiceBus.Extensibility.ContextBag extensions, System.Buffers.IBufferWriter<byte> bodyWriter);
+    }
     public interface IHandleMessages { }
     public interface IHandleMessages<T> : NServiceBus.IHandleMessages
     {

src/NServiceBus.Core.Tests/ApprovalFiles/MeterTests.Verify_MeterAPI.approved.txt

@@ -6,12 +6,14 @@
     "error.type",
     "execution.result",
     "nservicebus.discriminator",
+    "nservicebus.envelope.unwrapper_type",
     "nservicebus.message_handler_type",
     "nservicebus.message_handler_types",
     "nservicebus.message_type",
     "nservicebus.queue"
   ],
   "Metrics": [
+    "nservicebus.envelope.unwrapped => Counter",
     "nservicebus.messaging.critical_time => Histogram, Unit: s",
     "nservicebus.messaging.failures => Counter",
     "nservicebus.messaging.fetches => Counter",