diff --git a/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/EfCoreOutboxStore.cs b/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/EfCoreOutboxStore.cs index 769c5f1..73d732f 100644 --- a/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/EfCoreOutboxStore.cs +++ b/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/EfCoreOutboxStore.cs @@ -1,4 +1,5 @@ using System; +using System.Diagnostics; using System.Threading; using System.Threading.Tasks; using BBT.Aether.Clock; @@ -61,6 +62,23 @@ public async Task StoreAsync(CloudEventEnvelope envelope, CancellationToken canc if (envelope.Subject != null) outboxMessage.ExtraProperties["Subject"] = envelope.Subject; + // The drop's trace identity, persisted the same way TopicName is. The payload bytes already + // carry a TraceParent for traceable events, but the processor publishes them opaquely — these + // row-level copies are what let Outbox.Process re-join the originating trace without + // deserializing the envelope. Absent (not null) when nothing is ambient, so pre-existing rows + // and non-traced writes keep today's behavior. + if (Activity.Current is { } ambient) + { + outboxMessage.ExtraProperties["TraceParent"] = ambient.Id!; + if (!string.IsNullOrEmpty(ambient.TraceStateString)) + outboxMessage.ExtraProperties["TraceState"] = ambient.TraceStateString; + + // The originating trace's only chance to learn which row the event became: the id is born + // here, and widening IOutboxStore.StoreAsync's return type for one tag is not worth the + // ripple through every implementor. Ambient here is the EventBus.Publish span. + ambient.SetTag("outbox.message_id", outboxMessage.Id.ToString()); + } + await dbContext.OutboxMessages.AddAsync(outboxMessage, cancellationToken); } diff --git a/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/OutboxProcessor.cs b/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/OutboxProcessor.cs index d271900..454e1d3 100644 --- a/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/OutboxProcessor.cs +++ b/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/OutboxProcessor.cs @@ -88,8 +88,29 @@ protected virtual async Task ProcessOutboxMessagesAsync(CancellationToken c { if (cancellationToken.IsCancellationRequested) break; + // Re-join the originating trace when the row carries its drop identity (written by + // EfCoreOutboxStore since the outbox-trace-continuity change): the per-message span + // parents to the stored context and LINKS back to the worker loop — the same shape the + // inbox side's EventTraceScope uses, so publish → outbox drop → outbox publish → inbox + // handle reads as one tree. Rows without the identity (pre-deploy rows, untraced + // writes) keep the worker-loop parent unchanged. + var loopContext = Activity.Current?.Context ?? default; + var parentContext = loopContext; + IEnumerable? links = null; + if (message.ExtraProperties.TryGetValue("TraceParent", out var tpObj) && + ActivityContext.TryParse( + tpObj?.ToString(), + message.ExtraProperties.TryGetValue("TraceState", out var tsObj) ? tsObj?.ToString() : null, + isRemote: true, + out var originContext)) + { + parentContext = originContext; + if (loopContext.TraceId != default) + links = new[] { new ActivityLink(loopContext) }; + } + using var activity = InfrastructureActivitySource.Source.StartActivity( - "Outbox.Process", ActivityKind.Producer, Activity.Current?.Context ?? default); + "Outbox.Process", ActivityKind.Producer, parentContext, links: links); var topicName = message.ExtraProperties.TryGetValue("TopicName", out var topicObj) ? topicObj?.ToString() ?? message.EventName : message.EventName; diff --git a/framework/test/BBT.Aether.Infrastructure.Tests/BBT/Aether/Events/EfCoreOutboxStoreTraceTests.cs b/framework/test/BBT.Aether.Infrastructure.Tests/BBT/Aether/Events/EfCoreOutboxStoreTraceTests.cs new file mode 100644 index 0000000..692e593 --- /dev/null +++ b/framework/test/BBT.Aether.Infrastructure.Tests/BBT/Aether/Events/EfCoreOutboxStoreTraceTests.cs @@ -0,0 +1,136 @@ +using System; +using System.Diagnostics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using BBT.Aether.Clock; +using BBT.Aether.Domain.EntityFrameworkCore; +using BBT.Aether.Domain.EntityFrameworkCore.Modeling; +using BBT.Aether.Events; +using BBT.Aether.Guids; +using BBT.Aether.MultiSchema; +using BBT.Aether.Persistence; +using Microsoft.EntityFrameworkCore; +using NSubstitute; +using Shouldly; +using Xunit; +using OutboxMessage = BBT.Aether.Domain.Events.OutboxMessage; + +namespace BBT.Aether.Infrastructure.Tests.BBT.Aether.Events; + +/// +/// Pins EfCoreOutboxStore.StoreAsync's trace-identity persistence: the drop's ambient trace +/// context (TraceParent/TraceState) is copied onto the stored row's ExtraProperties the same way +/// TopicName already is, and the ambient activity gains an outbox.message_id tag. Together these +/// are what let OutboxProcessor's Outbox.Process span re-join the originating trace without ever +/// deserializing the envelope. +/// +public sealed class EfCoreOutboxStoreTraceTests +{ + private const string TestSourceName = "Test.EfCoreOutboxStoreTrace"; + + public sealed class MessagingDbContext(DbContextOptions options) + : DbContext(options), IHasEfCoreOutbox + { + public DbSet OutboxMessages => Set(); + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.ConfigureOutbox(); + } + } + + [Fact] + public async Task StoreAsync_persists_ambient_trace_context_and_tags_the_ambient_activity() + { + using var listener = CreateListener(); + using var source = new ActivitySource(TestSourceName); + using var ambient = source.StartActivity("EventBus.Publish", ActivityKind.Producer); + ambient.ShouldNotBeNull(); + ambient!.TraceStateString = "congo=t61rcWkgMzE"; + + var sut = CreateSut(out var context); + + await sut.StoreAsync(CreateEnvelope()); + + var stored = context.ChangeTracker.Entries().Select(e => e.Entity).Single(); + + stored.ExtraProperties["TraceParent"].ShouldBe(ambient.Id); + stored.ExtraProperties["TraceState"].ShouldBe("congo=t61rcWkgMzE"); + ambient.GetTagItem("outbox.message_id").ShouldBe(stored.Id.ToString()); + } + + [Fact] + public async Task StoreAsync_omits_trace_state_when_the_ambient_activity_has_none() + { + using var listener = CreateListener(); + using var source = new ActivitySource(TestSourceName); + using var ambient = source.StartActivity("EventBus.Publish", ActivityKind.Producer); + ambient.ShouldNotBeNull(); + + var sut = CreateSut(out var context); + + await sut.StoreAsync(CreateEnvelope()); + + var stored = context.ChangeTracker.Entries().Select(e => e.Entity).Single(); + + stored.ExtraProperties["TraceParent"].ShouldBe(ambient!.Id); + stored.ExtraProperties.ShouldNotContainKey("TraceState"); + } + + [Fact] + public async Task StoreAsync_writes_no_trace_keys_and_does_not_throw_when_nothing_is_ambient() + { + var sut = CreateSut(out var context); + + await Should.NotThrowAsync(async () => await sut.StoreAsync(CreateEnvelope())); + + var stored = context.ChangeTracker.Entries().Select(e => e.Entity).Single(); + + stored.ExtraProperties.ShouldNotContainKey("TraceParent"); + stored.ExtraProperties.ShouldNotContainKey("TraceState"); + } + + private static ActivityListener CreateListener() + { + var listener = new ActivityListener + { + ShouldListenTo = s => s.Name == TestSourceName, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded + }; + ActivitySource.AddActivityListener(listener); + return listener; + } + + private static EfCoreOutboxStore CreateSut(out MessagingDbContext context) + { + context = new MessagingDbContext( + new DbContextOptionsBuilder() + .UseInMemoryDatabase(Guid.NewGuid().ToString("N")) + .Options); + + var provider = Substitute.For>(); + provider.GetDbContextAsync(Arg.Any()).Returns(context); + + var clock = Substitute.For(); + clock.UtcNow.Returns(DateTime.UtcNow); + var guids = Substitute.For(); + guids.Create().Returns(_ => Guid.NewGuid()); + + return new EfCoreOutboxStore( + provider, + new SystemTextJsonEventSerializer(), + guids, + clock, + new AetherOutboxOptions { Schema = "sys_queues" }, + new StaticCurrentSchema("sys_queues")); + } + + private static CloudEventEnvelope CreateEnvelope() => new() + { + Id = Guid.NewGuid().ToString("N"), + Type = "TestEvent", + Topic = "test-event", + Data = new { Value = 42 } + }; +} diff --git a/framework/test/BBT.Aether.Infrastructure.Tests/BBT/Aether/Events/Processing/OutboxProcessorTraceTests.cs b/framework/test/BBT.Aether.Infrastructure.Tests/BBT/Aether/Events/Processing/OutboxProcessorTraceTests.cs new file mode 100644 index 0000000..8a55a2b --- /dev/null +++ b/framework/test/BBT.Aether.Infrastructure.Tests/BBT/Aether/Events/Processing/OutboxProcessorTraceTests.cs @@ -0,0 +1,214 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using BBT.Aether.Clock; +using BBT.Aether.Domain.EntityFrameworkCore; +using BBT.Aether.Domain.EntityFrameworkCore.Modeling; +using BBT.Aether.Events; +using BBT.Aether.MultiSchema; +using BBT.Aether.Persistence; +using BBT.Aether.Telemetry; +using BBT.Aether.Uow; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using Shouldly; +using Xunit; +using OutboxEntity = BBT.Aether.Domain.Events.OutboxMessage; + +namespace BBT.Aether.Events.Processing; + +/// +/// Pins the shape of OutboxProcessor's per-message "Outbox.Process" span: when the leased +/// message's ExtraProperties carry the drop's trace identity (written by EfCoreOutboxStore), the +/// span re-parents into that origin trace and links back to the worker loop — the same shape the +/// inbox side's EventTraceScope uses. Rows without a (parseable) trace identity keep today's +/// behavior: parented to the worker-loop activity, no link. +/// +public sealed class OutboxProcessorTraceTests +{ + private const string OriginSourceName = "Test.Origin"; + private const string LoopSourceName = "Test.WorkerLoop"; + + public sealed class MessagingDbContext(DbContextOptions options) + : DbContext(options), IHasEfCoreOutbox + { + public DbSet OutboxMessages => Set(); + + protected override void OnModelCreating(ModelBuilder modelBuilder) + { + modelBuilder.ConfigureOutbox(); + } + } + + [Fact] + public async Task Message_with_stored_trace_parent_reparents_into_the_origin_trace_and_links_the_worker_loop() + { + using var listener = CreateListener(out var started); + + using var originSource = new ActivitySource(OriginSourceName); + using var origin = originSource.StartActivity("EventBus.Publish", ActivityKind.Producer); + origin.ShouldNotBeNull(); + var traceParent = origin!.Id!; + origin.Stop(); // the drop's publish span has already ended by the time the processor runs + + var message = MakeMessage(traceParent: traceParent); + + using var loopSource = new ActivitySource(LoopSourceName); + using var loop = loopSource.StartActivity("Outbox.Poll", ActivityKind.Internal); + loop.ShouldNotBeNull(); + + await RunProcessorAsync(new[] { message }); + + var activity = started.ShouldHaveSingleItem(); + activity.OperationName.ShouldBe("Outbox.Process"); + activity.TraceId.ShouldBe(origin.TraceId); + activity.ParentSpanId.ShouldBe(origin.SpanId); + activity.Links.ShouldContain(l => l.Context.SpanId == loop!.SpanId); + activity.GetTagItem("event.name").ShouldBe("TestEvent"); + activity.GetTagItem("outbox.message_id").ShouldBe(message.Id.ToString()); + activity.GetTagItem("outbox.retry_count").ShouldBe(0); + } + + [Fact] + public async Task Message_without_trace_parent_keeps_the_worker_loop_as_parent_with_no_link() + { + using var listener = CreateListener(out var started); + + var message = MakeMessage(traceParent: null); + + using var loopSource = new ActivitySource(LoopSourceName); + using var loop = loopSource.StartActivity("Outbox.Poll", ActivityKind.Internal); + loop.ShouldNotBeNull(); + + await Should.NotThrowAsync(async () => await RunProcessorAsync(new[] { message })); + + var activity = started.ShouldHaveSingleItem(); + activity.OperationName.ShouldBe("Outbox.Process"); + activity.TraceId.ShouldBe(loop!.TraceId); + activity.ParentSpanId.ShouldBe(loop.SpanId); + activity.Links.ShouldBeEmpty(); + activity.GetTagItem("event.name").ShouldBe("TestEvent"); + activity.GetTagItem("outbox.message_id").ShouldBe(message.Id.ToString()); + activity.GetTagItem("outbox.retry_count").ShouldBe(0); + } + + [Fact] + public async Task Message_with_garbage_trace_parent_keeps_todays_behavior() + { + using var listener = CreateListener(out var started); + + var message = MakeMessage(traceParent: "not-a-real-traceparent"); + + using var loopSource = new ActivitySource(LoopSourceName); + using var loop = loopSource.StartActivity("Outbox.Poll", ActivityKind.Internal); + loop.ShouldNotBeNull(); + + await Should.NotThrowAsync(async () => await RunProcessorAsync(new[] { message })); + + var activity = started.ShouldHaveSingleItem(); + activity.TraceId.ShouldBe(loop!.TraceId); + activity.ParentSpanId.ShouldBe(loop.SpanId); + activity.Links.ShouldBeEmpty(); + } + + private static OutboxMessage MakeMessage(string? traceParent) + { + var extraProperties = new Dictionary(); + if (traceParent != null) + extraProperties["TraceParent"] = traceParent; + + return new OutboxMessage + { + Id = Guid.NewGuid(), + EventName = "TestEvent", + EventData = [], + Status = OutboxMessageStatus.Pending, + RetryCount = 0, + ExtraProperties = extraProperties + }; + } + + private static ActivityListener CreateListener(out List started) + { + var list = new List(); + started = list; + var listener = new ActivityListener + { + ShouldListenTo = s => s.Name == InfrastructureActivitySource.SourceName + || s.Name == OriginSourceName + || s.Name == LoopSourceName, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded, + ActivityStarted = activity => + { + if (activity.OperationName == "Outbox.Process") list.Add(activity); + } + }; + ActivitySource.AddActivityListener(listener); + return listener; + } + + /// + /// Drives OutboxProcessor.RunAsync through a minimal real DI container (so its own + /// scopeFactory.CreateAsyncScope() call resolves) wired with fakes for every collaborator. + /// The publish call always fails: that routes phase 3 through the read-only + /// "lease expired/not found" branch (FirstOrDefaultAsync over an empty InMemory table returns + /// null and the loop just continues) instead of ExecuteUpdateAsync, which the EF Core InMemory + /// provider does not support — irrelevant to what this test asserts, which is only the shape + /// of the per-message activity started in phase 2. + /// + private static async Task RunProcessorAsync(IReadOnlyList messages) + { + var context = new MessagingDbContext( + new DbContextOptionsBuilder() + .UseInMemoryDatabase(Guid.NewGuid().ToString("N")) + .Options); + + var currentSchema = Substitute.For(); + currentSchema.Change(Arg.Any()).Returns(NullDisposable.Instance); + + var uow = Substitute.For(); + var uowManager = Substitute.For(); + uowManager.Begin(Arg.Any()).Returns(uow); + + var leaseStore = Substitute.For(); + leaseStore.LeaseBatchAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(messages); + + var eventBus = Substitute.For(); + eventBus.PublishEnvelopeAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(Task.FromException(new InvalidOperationException("publish failed (test)"))); + + var dbContextProvider = Substitute.For>(); + dbContextProvider.GetDbContextAsync(Arg.Any()).Returns(context); + + var services = new ServiceCollection(); + services.AddSingleton(currentSchema); + services.AddSingleton(uowManager); + services.AddSingleton(eventBus); + services.AddSingleton(new AetherEventBusOptions { DefaultSource = "urn:vnext:test", PubSubName = "pubsub" }); + services.AddSingleton(leaseStore); + services.AddSingleton(dbContextProvider); + await using var provider = services.BuildServiceProvider(); + + var env = Substitute.For(); + env.ApplicationName.Returns("outbox-processor-trace-tests"); + + var clock = Substitute.For(); + clock.UtcNow.Returns(DateTime.UtcNow); + + var processor = new OutboxProcessor( + provider.GetRequiredService(), + new WorkerIdentity(env), + clock, + NullLogger>.Instance, + new AetherOutboxOptions { Schema = "sys_queues", BatchSize = 10 }); + + await processor.RunAsync(); + } +}