v7: Carry trace context in message headers across processes

.github/workflows/dotnet.yml

@@ -59,7 +59,9 @@ jobs:
     - name: Setup .NET
       uses: actions/setup-dotnet@v6
       with:
-        dotnet-version: 10.0.x
+        # Pinned while Microsoft's download origin returns 409 for SDK 10.0.403 / runtime 10.0.14 (October 2026).
+        # Go back to 10.0.x once that release downloads.
+        dotnet-version: 10.0.402
     - name: Setup GitVersion
       uses: gittools/actions/gitversion/setup@v4
       with:

Obvs.RabbitMQ.Tests/Hosting/TestRabbitMqTracing.cs

@@ -0,0 +1,104 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Obvs.Hosting;
+using Obvs.Serialization.Json;
+using Xunit;
+
+namespace Obvs.RabbitMQ.Tests.Hosting
+{
+    // Two hosts connected only through RabbitMQ stand in for two processes
+    public class TestRabbitMqTracing
+    {
+        private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(10);
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldContinueTheSendersTraceAcrossRabbitMq()
+        {
+            const string serviceName = "obvs-tracing-commands";
+            using var activities = new ActivityRecorder();
+            using var receiver = await StartHostAsync(serviceName, service => service.AsServer()
+                .HandleCommand<PlaceOrder>((command, context, ct) => Task.CompletedTask));
+            using var sender = await StartHostAsync(serviceName, service => service.AsClient());
+
+            await sender.Services.GetRequiredService<IServiceBusClient>().SendAsync(new PlaceOrder { OrderId = "o-1" });
+
+            var send = await WaitForAsync(() => activities.For(serviceName).SingleOrDefault(activity => activity.DisplayName == $"send {serviceName}"));
+            var process = await WaitForAsync(() => activities.For(serviceName).SingleOrDefault(activity => activity.DisplayName == $"process {serviceName}"));
+            Assert.Equal(send.TraceId, process.TraceId);
+            Assert.Equal(send.SpanId, process.ParentSpanId);
+        }
+
+        [Fact, Trait("Category", "Explicit")]
+        public async Task ShouldCarryHeadersTheAppSetsOverRabbitMq()
+        {
+            const string serviceName = "obvs-tracing-headers";
+            var received = new TaskCompletionSource<IReadOnlyDictionary<string, string>>(TaskCreationOptions.RunContinuationsAsynchronously);
+            using var receiver = await StartHostAsync(serviceName, service => service.AsServer()
+                .HandleCommand<PlaceOrder>((command, context, ct) =>
+                {
+                    received.TrySetResult(ObvsMessageHeaders.Get(command));
+                    return Task.CompletedTask;
+                }));
+            using var sender = await StartHostAsync(serviceName, service => service.AsClient());
+            var placeOrder = new PlaceOrder { OrderId = "o-1" };
+            ObvsMessageHeaders.Set(placeOrder, new Dictionary<string, string> { ["x-tenant"] = "acme" });
+
+            await sender.Services.GetRequiredService<IServiceBusClient>().SendAsync(placeOrder);
+
+            var headers = await received.Task.WaitAsync(Timeout, TestContext.Current.CancellationToken);
+            Assert.Equal("acme", headers["x-tenant"]);
+        }
+
+        private static async Task<IHost> StartHostAsync(string serviceName, Action<ObvsServiceBuilder> configure)
+        {
+            var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true });
+            builder.Services.AddObvs(obvs => obvs.AddService<IOrderMessage>(serviceName, service => configure(service
+                .UseRabbitMq(options => options.Uri = new Uri(TestBroker.Uri))
+                .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>))))));
+            var host = builder.Build();
+            await host.StartAsync(TestContext.Current.CancellationToken);
+            return host;
+        }
+
+        private static async Task<T> WaitForAsync<T>(Func<T> find) where T : class
+        {
+            var deadline = DateTime.UtcNow + Timeout;
+            T found;
+            while ((found = find()) == null && DateTime.UtcNow < deadline)
+            {
+                await Task.Delay(20, TestContext.Current.CancellationToken);
+            }
+            return found ?? throw new TimeoutException($"The expected {typeof(T).Name} wasn't recorded in time.");
+        }
+
+        private sealed class ActivityRecorder : IDisposable
+        {
+            private readonly ConcurrentQueue<Activity> _stopped = new ConcurrentQueue<Activity>();
+            private readonly ActivityListener _listener;
+
+            public ActivityRecorder()
+            {
+                _listener = new ActivityListener
+                {
+                    ShouldListenTo = source => source.Name == ObvsTelemetry.ActivitySourceName,
+                    Sample = (ref ActivityCreationOptions<ActivityContext> _) => ActivitySamplingResult.AllDataAndRecorded,
+                    ActivityStopped = activity => _stopped.Enqueue(activity),
+                };
+                ActivitySource.AddActivityListener(_listener);
+            }
+
+            public IReadOnlyList<Activity> For(string serviceName)
+            {
+                return _stopped.Where(activity => (activity.GetTagItem("messaging.destination.name") as string) == serviceName).ToList();
+            }
+
+            public void Dispose() => _listener.Dispose();
+        }
+    }
+}

Obvs.RabbitMQ/MessagePublisher.cs

@@ -1,4 +1,6 @@
 using System;
+using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Obvs.Serialization;
@@ -31,12 +33,19 @@ public MessagePublisher(Lazy<IConnection> connection, IMessageSerializer seriali
         public async Task PublishAsync(TMessage message)
         {
             var body = _serializer.Serialize(message);
-            var properties = new BasicProperties {ContentType = "bytes"};
+            var properties = new BasicProperties {ContentType = "bytes", Headers = GetHeaders(message)};
             var routingKey = GetRoutingKey(message);
             var channel = await _channel.Value.ConfigureAwait(false);
             await channel.BasicPublishAsync(_exchange, routingKey, false, properties, body).ConfigureAwait(false);
         }
 
+        // The message's Obvs headers (such as trace context) as AMQP headers
+        private static IDictionary<string, object> GetHeaders(TMessage message)
+        {
+            var headers = ObvsMessageHeaders.Get(message);
+            return headers.Count == 0 ? null : headers.ToDictionary(header => header.Key, header => (object)header.Value);
+        }
+
         private string GetRoutingKey(TMessage message)
         {
             return string.Format("{0}.{1}", _routingKeyPrefix, message.GetType().Name);

Obvs.RabbitMQ/MessageSource.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
+using System.Text;
 using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
@@ -122,7 +123,29 @@ private TMessage Deserialize(BasicDeliverEventArgs deliverEventArgs)
         {
             var deserializer = GetDeserializer(deliverEventArgs);
             // The body's buffer is only valid until the delivery handler returns, so copy it
-            return deserializer.Deserialize(deliverEventArgs.Body.ToArray());
+            var message = deserializer.Deserialize(deliverEventArgs.Body.ToArray());
+            SetHeaders(message, deliverEventArgs.BasicProperties.Headers);
+            return message;
+        }
+
+        // AMQP string headers arrive as UTF-8 bytes
+        private static void SetHeaders(TMessage message, IDictionary<string, object> amqpHeaders)
+        {
+            if (amqpHeaders == null || amqpHeaders.Count == 0)
+            {
+                return;
+            }
+
+            var headers = new Dictionary<string, string>();
+            foreach (var header in amqpHeaders)
+            {
+                var value = header.Value is byte[] bytes ? Encoding.UTF8.GetString(bytes) : header.Value?.ToString();
+                if (value != null)
+                {
+                    headers[header.Key] = value;
+                }
+            }
+            ObvsMessageHeaders.Set(message, headers);
         }
 
         private IMessageDeserializer<TMessage> GetDeserializer(BasicDeliverEventArgs deliverEventArgs)

Obvs.Tests/Hosting/Telemetry/TestTracePropagation.cs

@@ -0,0 +1,142 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Obvs.Hosting;
+using Obvs.Testing;
+using Obvs.Tests.Hosting.Handlers;
+using Xunit;
+
+namespace Obvs.Tests.Hosting.Telemetry
+{
+    public class TestMessageHeaders
+    {
+        [Fact]
+        public void ShouldReturnTheHeadersSetOnAMessage()
+        {
+            var message = new PlaceOrder("o-1");
+
+            ObvsMessageHeaders.Set(message, new Dictionary<string, string> { ["x-tenant"] = "acme" });
+
+            Assert.Equal("acme", ObvsMessageHeaders.Get(message)["x-tenant"]);
+        }
+
+        [Fact]
+        public void ShouldReturnNoHeadersForAMessageWithoutAny()
+        {
+            Assert.Empty(ObvsMessageHeaders.Get(new PlaceOrder("o-1")));
+        }
+
+        [Fact]
+        public void ShouldKeepHeadersSeparateForEqualMessages()
+        {
+            // Records with equal values are equal, but headers belong to the instance
+            var first = new PlaceOrder("o-1");
+            var second = new PlaceOrder("o-1");
+
+            ObvsMessageHeaders.Set(first, new Dictionary<string, string> { ["x-tenant"] = "acme" });
+
+            Assert.Empty(ObvsMessageHeaders.Get(second));
+        }
+    }
+
+    // Two hosts sharing an in-memory broker stand in for two processes
+    public class TestTracePropagation
+    {
+        private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5);
+
+        private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+        private readonly string _service = "orders-" + Guid.NewGuid().ToString("N");
+
+        [Fact]
+        public async Task ShouldPutTheSendSpansTraceContextInTheMessagesHeaders()
+        {
+            using var activities = new ActivityRecorder();
+            await using var harness = await ObvsTestHarness.StartAsync(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>(_service, service => service.UseInMemory())), cancellationToken: Ct);
+            var command = new PlaceOrder("o-1");
+
+            await harness.Client.SendAsync(command);
+
+            var send = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"send {_service}"));
+            Assert.Equal($"00-{send.TraceId}-{send.SpanId}-01", ObvsMessageHeaders.Get(command)[ObvsMessageHeaders.TraceParent]);
+        }
+
+        [Fact]
+        public async Task ShouldKeepHeadersTheAppSetWhenAddingTraceContext()
+        {
+            using var activities = new ActivityRecorder();
+            await using var harness = await ObvsTestHarness.StartAsync(services => services.AddObvs(obvs => obvs
+                .AddService<IOrderMessage>(_service, service => service.UseInMemory())), cancellationToken: Ct);
+            var command = new PlaceOrder("o-1");
+            ObvsMessageHeaders.Set(command, new Dictionary<string, string> { ["x-tenant"] = "acme" });
+
+            await harness.Client.SendAsync(command);
+
+            var headers = ObvsMessageHeaders.Get(command);
+            Assert.Equal("acme", headers["x-tenant"]);
+            Assert.True(headers.ContainsKey(ObvsMessageHeaders.TraceParent));
+        }
+
+        [Fact]
+        public async Task ShouldContinueTheSendersTraceWhenHandling()
+        {
+            using var activities = new ActivityRecorder();
+            var broker = new InMemoryBroker();
+            using var sender = await StartHostAsync(service => service.UseInMemory(broker).AsClient());
+            using var receiver = await StartHostAsync(service => service.UseInMemory(broker).AsServer()
+                .HandleCommand<PlaceOrder>((command, context, ct) => Task.CompletedTask));
+
+            await sender.Services.GetRequiredService<IServiceBusClient>().SendAsync(new PlaceOrder("o-1"));
+
+            var send = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"send {_service}"));
+            var process = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"process {_service}"));
+            Assert.Equal(send.TraceId, process.TraceId);
+            Assert.Equal(send.SpanId, process.ParentSpanId);
+        }
+
+        [Fact]
+        public async Task ShouldContinueTheTraceThroughARequestAndItsReply()
+        {
+            using var activities = new ActivityRecorder();
+            var broker = new InMemoryBroker();
+            using var client = await StartHostAsync(service => service.UseInMemory(broker).AsClient());
+            using var server = await StartHostAsync(service => service.UseInMemory(broker).AsServer()
+                .HandleRequest<GetQuote, Quote>((request, context, ct) => Task.FromResult(new Quote(request.Sku, 9.99m))));
+
+            await client.Services.GetRequiredService<IServiceBusClient>().RequestAsync<GetQuote, Quote>(new GetQuote("sku-1"), Timeout, Ct);
+
+            var request = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"request {_service}"));
+            var process = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"process {_service}"));
+            var reply = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"reply {_service}"));
+            Assert.Equal(request.SpanId, process.ParentSpanId);
+            Assert.Equal(process.SpanId, reply.ParentSpanId);
+            Assert.Equal(request.TraceId, reply.TraceId);
+        }
+
+        private async Task<IHost> StartHostAsync(Action<ObvsServiceBuilder> configure)
+        {
+            var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true });
+            builder.Services.AddObvs(obvs => obvs.AddService<IOrderMessage>(_service, configure));
+            var host = builder.Build();
+            await host.StartAsync(Ct);
+            return host;
+        }
+
+        private static async Task<T> WaitForAsync<T>(Func<T> find) where T : class
+        {
+            var deadline = DateTime.UtcNow + Timeout;
+            T found;
+            while ((found = find()) == null && DateTime.UtcNow < deadline)
+            {
+                await Task.Delay(20, Ct);
+            }
+            return found ?? throw new TimeoutException($"The expected {typeof(T).Name} wasn't recorded in time.");
+        }
+    }
+}

Obvs/Hosting/HandlerRunner.cs

@@ -67,7 +67,7 @@ async Task Handle(IMessage message, CancellationToken cancellationToken)
                 var started = Stopwatch.GetTimestamp();
                 Exception failure = null;
                 // The span is current while the handler runs, so anything it sends becomes a child span
-                using (var activity = ObvsDiagnostics.StartProcess(serviceName, message.GetType()))
+                using (var activity = ObvsDiagnostics.StartProcess(serviceName, message.GetType(), ObvsDiagnostics.ExtractParent(message)))
                 {
                     var scope = scopes.CreateAsyncScope();
                     try

Obvs/Hosting/ObvsDiagnostics.cs

@@ -39,9 +39,10 @@ public static Activity StartProducer(string operation, string serviceName, Type
             return activity;
         }
 
-        public static Activity StartProcess(string serviceName, Type messageType)
+        // A child of the sender's span when the message carries trace context, so the trace continues
+        public static Activity StartProcess(string serviceName, Type messageType, ActivityContext parent)
         {
-            var activity = Source.StartActivity($"process {serviceName}", ActivityKind.Consumer);
+            var activity = Source.StartActivity($"process {serviceName}", ActivityKind.Consumer, parent);
             if (activity != null)
             {
                 activity.SetTag("messaging.system", System);
@@ -53,6 +54,41 @@ public static Activity StartProcess(string serviceName, Type messageType)
             return activity;
         }
 
+        // Adds the span's trace context to the message's headers, keeping any the app set. Falls back to the
+        // ambient span (an incoming web request, say) when nobody listens to Obvs's own spans.
+        public static void InjectTraceContext(Activity activity, object message)
+        {
+            activity = activity ?? Activity.Current;
+            if (activity == null)
+            {
+                return;
+            }
+
+            var headers = new Dictionary<string, string>();
+            foreach (var header in ObvsMessageHeaders.Get(message))
+            {
+                headers[header.Key] = header.Value;
+            }
+            DistributedContextPropagator.Current.Inject(activity, headers,
+                (carrier, name, value) => ((Dictionary<string, string>)carrier)[name] = value);
+            ObvsMessageHeaders.Set(message, headers);
+        }
+
+        // The sender's span from the message's headers, if it has trace context
+        public static ActivityContext ExtractParent(object message)
+        {
+            var headers = ObvsMessageHeaders.Get(message);
+            DistributedContextPropagator.Current.ExtractTraceIdAndState(headers,
+                (object carrier, string name, out string value, out IEnumerable<string> values) =>
+                {
+                    values = null;
+                    ((IReadOnlyDictionary<string, string>)carrier).TryGetValue(name, out value);
+                },
+                out var traceParent, out var traceState);
+
+            return ActivityContext.TryParse(traceParent, traceState, out var parent) ? parent : default(ActivityContext);
+        }
+
         public static void RecordFailure(Activity activity, Exception exception)
         {
             if (activity == null)