diff --git a/framework/src/BBT.Aether.AspNetCore/BBT/Aether/AspNetCore/Events/EventsController.cs b/framework/src/BBT.Aether.AspNetCore/BBT/Aether/AspNetCore/Events/EventsController.cs index 71875d1..6a94a3f 100644 --- a/framework/src/BBT.Aether.AspNetCore/BBT/Aether/AspNetCore/Events/EventsController.cs +++ b/framework/src/BBT.Aether.AspNetCore/BBT/Aether/AspNetCore/Events/EventsController.cs @@ -7,6 +7,7 @@ using BBT.Aether.MultiSchema; using BBT.Aether.Uow; using Microsoft.AspNetCore.Mvc; +using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; namespace BBT.Aether.AspNetCore.Events; @@ -86,6 +87,12 @@ protected virtual async Task ProcessEventAsync( // Step 7: Commit UoW (flushes inbox) await uow.CommitAsync(cancellationToken); + // Same-process nudge: the inbox poller and this delivery endpoint share the host, + // so a stored row can be processed immediately instead of waiting out the idle interval. + HttpContext.RequestServices + .GetService>() + ?.Signal(); + return Ok(); } } diff --git a/framework/src/BBT.Aether.Core/BBT/Aether/Events/AetherOutboxOptions.cs b/framework/src/BBT.Aether.Core/BBT/Aether/Events/AetherOutboxOptions.cs index 039a887..b4c450f 100644 --- a/framework/src/BBT.Aether.Core/BBT/Aether/Events/AetherOutboxOptions.cs +++ b/framework/src/BBT.Aether.Core/BBT/Aether/Events/AetherOutboxOptions.cs @@ -18,4 +18,11 @@ public class AetherOutboxOptions /// The database schema whose outbox table this processor handles. /// public string? Schema { get; set; } = "sys_queues"; + + /// + /// When true, a unit of work that stored outbox messages publishes a direct pub/sub wake nudge + /// () after commit so outbox processors poll immediately instead + /// of waiting out the idle interval. Default false. Requires an IOutboxWakeupNotifier registration. + /// + public bool WakeupSignalEnabled { get; set; } } diff --git a/framework/src/BBT.Aether.Core/BBT/Aether/Events/IOutboxWakeupNotifier.cs b/framework/src/BBT.Aether.Core/BBT/Aether/Events/IOutboxWakeupNotifier.cs new file mode 100644 index 0000000..bf8c89b --- /dev/null +++ b/framework/src/BBT.Aether.Core/BBT/Aether/Events/IOutboxWakeupNotifier.cs @@ -0,0 +1,13 @@ +using System.Threading; +using System.Threading.Tasks; + +namespace BBT.Aether.Events; + +/// +/// Publishes the nudge. Implementations must be fire-and-forget +/// safe: a failed notify is swallowed by callers because polling backstops delivery. +/// +public interface IOutboxWakeupNotifier +{ + Task NotifyAsync(CancellationToken cancellationToken = default); +} diff --git a/framework/src/BBT.Aether.Core/BBT/Aether/Events/OutboxWakeupEvent.cs b/framework/src/BBT.Aether.Core/BBT/Aether/Events/OutboxWakeupEvent.cs new file mode 100644 index 0000000..6f3722f --- /dev/null +++ b/framework/src/BBT.Aether.Core/BBT/Aether/Events/OutboxWakeupEvent.cs @@ -0,0 +1,10 @@ +namespace BBT.Aether.Events; + +/// +/// Loss-tolerant wake nudge published directly to pub/sub (never through the outbox) after a unit +/// of work that stored at least one outbox message commits. Subscribing outbox processors treat it +/// as "poll now"; the payload is deliberately empty and delivery is best-effort — the adaptive +/// polling interval remains the safety net for lost or early signals. +/// +[EventName("aether.outbox.wakeup")] +public sealed class OutboxWakeupEvent; diff --git a/framework/src/BBT.Aether.Core/BBT/Aether/Polling/IPollingWakeSignal.cs b/framework/src/BBT.Aether.Core/BBT/Aether/Polling/IPollingWakeSignal.cs new file mode 100644 index 0000000..c51650e --- /dev/null +++ b/framework/src/BBT.Aether.Core/BBT/Aether/Polling/IPollingWakeSignal.cs @@ -0,0 +1,24 @@ +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace BBT.Aether.Polling; + +/// +/// A coalescing wake signal for adaptive polling loops. Producers call when +/// new work becomes available; the polling loop awaits with its normal +/// interval as the timeout so a signal cuts the wait short while polling remains the safety net. +/// The marker type parameter distinguishes independent loops (e.g. outbox vs inbox) in DI. +/// +/// Marker type identifying the loop this signal wakes. +public interface IPollingWakeSignal +{ + /// Wakes the loop. Multiple pending signals coalesce into one. + void Signal(); + + /// + /// Waits until is called or the timeout elapses. + /// Returns true when woken by a signal, false on timeout. + /// + Task WaitAsync(TimeSpan timeout, CancellationToken cancellationToken = default); +} diff --git a/framework/src/BBT.Aether.Core/BBT/Aether/Polling/PollingWakeSignal.cs b/framework/src/BBT.Aether.Core/BBT/Aether/Polling/PollingWakeSignal.cs new file mode 100644 index 0000000..ac0532d --- /dev/null +++ b/framework/src/BBT.Aether.Core/BBT/Aether/Polling/PollingWakeSignal.cs @@ -0,0 +1,30 @@ +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace BBT.Aether.Polling; + +/// +/// Default over a bounded (0,1): +/// signals coalesce (a second Signal while one is pending is a no-op), so a burst of producers +/// causes exactly one early wake. +/// +public sealed class PollingWakeSignal : IPollingWakeSignal +{ + private readonly SemaphoreSlim _semaphore = new(0, 1); + + public void Signal() + { + try + { + _semaphore.Release(); + } + catch (SemaphoreFullException) + { + // A wake is already pending — coalesce. + } + } + + public Task WaitAsync(TimeSpan timeout, CancellationToken cancellationToken = default) + => _semaphore.WaitAsync(timeout, cancellationToken); +} diff --git a/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/DaprOutboxWakeupNotifier.cs b/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/DaprOutboxWakeupNotifier.cs new file mode 100644 index 0000000..4098475 --- /dev/null +++ b/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/DaprOutboxWakeupNotifier.cs @@ -0,0 +1,24 @@ +using System.Threading; +using System.Threading.Tasks; +using Dapr.Client; + +namespace BBT.Aether.Events; + +/// +/// Publishes straight to the configured pub/sub component, +/// bypassing the outbox by design (the nudge must not create the work it announces). +/// +public sealed class DaprOutboxWakeupNotifier( + DaprClient daprClient, + ITopicNameStrategy topicNameStrategy, + AetherEventBusOptions eventBusOptions) : IOutboxWakeupNotifier +{ + private readonly string _topic = topicNameStrategy.GetTopicName(typeof(OutboxWakeupEvent)); + + public Task NotifyAsync(CancellationToken cancellationToken = default) + => daprClient.PublishEventAsync( + eventBusOptions.PubSubName, + _topic, + new OutboxWakeupEvent(), + cancellationToken); +} 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 73d732f..e1a14df 100644 --- a/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/EfCoreOutboxStore.cs +++ b/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/EfCoreOutboxStore.cs @@ -20,7 +20,8 @@ public class EfCoreOutboxStore( IGuidGenerator guidGenerator, IClock clock, AetherOutboxOptions options, - ICurrentSchema? currentSchema) : IOutboxStore + ICurrentSchema? currentSchema, + OutboxWakeupCoordinator? wakeupCoordinator = null) : IOutboxStore where TDbContext : DbContext, IHasEfCoreOutbox { /// @@ -37,6 +38,7 @@ public EfCoreOutboxStore( guidGenerator, clock, new AetherOutboxOptions { Schema = null }, + null, null) { } @@ -80,6 +82,8 @@ public async Task StoreAsync(CloudEventEnvelope envelope, CancellationToken canc } await dbContext.OutboxMessages.AddAsync(outboxMessage, cancellationToken); + + wakeupCoordinator?.OnOutboxMessageStored(); } private IDisposable BeginConfiguredSchemaScope() diff --git a/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/OutboxWakeupCoordinator.cs b/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/OutboxWakeupCoordinator.cs new file mode 100644 index 0000000..82a8b55 --- /dev/null +++ b/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/OutboxWakeupCoordinator.cs @@ -0,0 +1,89 @@ +using System; +using System.Diagnostics; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using BBT.Aether.Uow; +using Microsoft.Extensions.Logging; + +namespace BBT.Aether.Events; + +/// +/// Decides when the outbox wakeup nudge fires: once per unit of work that stored at least one +/// outbox message, from the UoW's OnCompleted callback — but WITHOUT extending the commit path: +/// the callback returns immediately and the pub/sub publish runs as an unobserved task with a +/// bounded timeout. A lost or failed nudge is logged and absorbed by the polling safety net. +/// +public sealed class OutboxWakeupCoordinator( + AetherOutboxOptions options, + IUnitOfWorkManager? unitOfWorkManager = null, + IOutboxWakeupNotifier? wakeupNotifier = null, + ILogger? logger = null) +{ + private static readonly TimeSpan NotifyTimeout = TimeSpan.FromSeconds(2); + private static readonly ConditionalWeakTable WakeupRegistered = new(); + private static readonly object RegisteredSentinel = new(); + + /// Call once per stored outbox message; registration collapses to one per UoW. + public void OnOutboxMessageStored() + { + if (wakeupNotifier is null || !options.WakeupSignalEnabled) + return; + + var uow = unitOfWorkManager?.Current; + if (uow is null) + { + // No ambient UoW: the caller flushes on its own SaveChanges, which this coordinator + // cannot observe — the nudge may land BEFORE the row is visible. This branch is an + // early best-effort hint EXCLUDED from the latency guarantee (the row then waits for + // normal polling). Every vnext transition path runs with an ambient UoW, so this is + // never the latency-critical path. + NotifyFireAndForget(); + return; + } + + // Dedupe on the shared transaction root, not the per-call scope object. A `Required` + // participant scope (UnitOfWorkScope with ownsRoot == false) is a distinct object per + // nesting level that forwards OnCompleted to the same CompositeUnitOfWork root — keying + // on `uow` itself would register (and later fire) once per nested scope for a single + // commit. CompositeUnitOfWork itself already implements IUnitOfWork, so it is a valid + // ConditionalWeakTable key; a `uow` that is not a UnitOfWorkScope (e.g. the root itself, + // or a test substitute) dedupes on itself as before. + var dedupeKey = (uow as UnitOfWorkScope)?.SharedRoot ?? uow; + + lock (RegisteredSentinel) + { + if (WakeupRegistered.TryGetValue(dedupeKey, out _)) + return; + WakeupRegistered.Add(dedupeKey, RegisteredSentinel); + } + + // OnCompleted callbacks are awaited inside CommitAsync — return a completed task and let + // the publish run detached so a slow sidecar can never stretch the commit path. + uow.OnCompleted(_ => + { + NotifyFireAndForget(); + return Task.CompletedTask; + }); + } + + private void NotifyFireAndForget() + { + _ = Task.Run(async () => + { + // The nudge is infrastructure, not business flow: sever the ambient Activity captured via + // ExecutionContext so the publish's client span (and the delivery it causes on the worker) + // can never attach to — or propagate the traceparent of — the committing business trace. + Activity.Current = null; + try + { + using var cts = new CancellationTokenSource(NotifyTimeout); + await wakeupNotifier!.NotifyAsync(cts.Token); + } + catch (Exception ex) + { + logger?.LogWarning(ex, "Outbox wakeup nudge failed or timed out; polling will pick the work up"); + } + }); + } +} diff --git a/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/InboxBackgroundService.cs b/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/InboxBackgroundService.cs index 69ad397..df95ffb 100644 --- a/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/InboxBackgroundService.cs +++ b/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/InboxBackgroundService.cs @@ -10,7 +10,8 @@ namespace BBT.Aether.Events.Processing; public sealed class InboxBackgroundService( IInboxProcessor processor, AetherInboxOptions options, - ILogger logger) : BackgroundService + ILogger logger, + BBT.Aether.Polling.IPollingWakeSignal? wakeSignal = null) : BackgroundService { protected override async Task ExecuteAsync(CancellationToken stoppingToken) { @@ -18,8 +19,12 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) // leave the whole fleet polling on the same tick. try { - await Task.Delay(PollingDelay.StartupOffset(options.IdlePollingInterval), stoppingToken) - .ConfigureAwait(false); + // Startup offset: also wake-aware, so a nudge that lands during a rolling restart advances the + // first poll instead of waiting the offset out. + if (wakeSignal is null) + await Task.Delay(PollingDelay.StartupOffset(options.IdlePollingInterval), stoppingToken).ConfigureAwait(false); + else + await wakeSignal.WaitAsync(PollingDelay.StartupOffset(options.IdlePollingInterval), stoppingToken).ConfigureAwait(false); } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { @@ -49,7 +54,11 @@ await Task.Delay(PollingDelay.StartupOffset(options.IdlePollingInterval), stoppi delay = PollingDelay.OnError(delay, options.IdlePollingInterval, options.MaxPollingInterval); } - await Task.Delay(PollingDelay.Jitter(delay), stoppingToken).ConfigureAwait(false); + // A wake signal cuts the interval short; timeout keeps polling as the safety net. + if (wakeSignal is null) + await Task.Delay(PollingDelay.Jitter(delay), stoppingToken).ConfigureAwait(false); + else + await wakeSignal.WaitAsync(PollingDelay.Jitter(delay), stoppingToken).ConfigureAwait(false); } } } diff --git a/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/OutboxBackgroundService.cs b/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/OutboxBackgroundService.cs index 8a92b90..afcfb35 100644 --- a/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/OutboxBackgroundService.cs +++ b/framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/OutboxBackgroundService.cs @@ -10,7 +10,8 @@ namespace BBT.Aether.Events.Processing; public sealed class OutboxBackgroundService( IOutboxProcessor processor, AetherOutboxOptions options, - ILogger logger) : BackgroundService + ILogger logger, + BBT.Aether.Polling.IPollingWakeSignal? wakeSignal = null) : BackgroundService { protected override async Task ExecuteAsync(CancellationToken stoppingToken) { @@ -18,8 +19,12 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) // leave the whole fleet polling on the same tick. try { - await Task.Delay(PollingDelay.StartupOffset(options.IdlePollingInterval), stoppingToken) - .ConfigureAwait(false); + // Startup offset: also wake-aware, so a nudge that lands during a rolling restart advances the + // first poll instead of waiting the offset out. + if (wakeSignal is null) + await Task.Delay(PollingDelay.StartupOffset(options.IdlePollingInterval), stoppingToken).ConfigureAwait(false); + else + await wakeSignal.WaitAsync(PollingDelay.StartupOffset(options.IdlePollingInterval), stoppingToken).ConfigureAwait(false); } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { @@ -49,7 +54,11 @@ await Task.Delay(PollingDelay.StartupOffset(options.IdlePollingInterval), stoppi delay = PollingDelay.OnError(delay, options.IdlePollingInterval, options.MaxPollingInterval); } - await Task.Delay(PollingDelay.Jitter(delay), stoppingToken).ConfigureAwait(false); + // A wake signal cuts the interval short; timeout keeps polling as the safety net. + if (wakeSignal is null) + await Task.Delay(PollingDelay.Jitter(delay), stoppingToken).ConfigureAwait(false); + else + await wakeSignal.WaitAsync(PollingDelay.Jitter(delay), stoppingToken).ConfigureAwait(false); } } } 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 454e1d3..47be160 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,52 +88,57 @@ 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. + // Separate worker execution by design: the publish episode is its own trace. The + // originating transition is causally related, not structurally the parent — a link + // preserves the relation without stretching the origin trace across the worker hop + // (trace_refactor: outbox publish must not be a child of the source transition span). 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, parentContext, links: links); + var links = new List(2); + if (TryParseOrigin(message, out var originContext)) + links.Add(new ActivityLink(originContext)); + if (loopContext != default) + links.Add(new ActivityLink(loopContext)); + + // StartActivity treats `default(ActivityContext)` as "no explicit parent" and + // silently falls back to Activity.Current when one is ambient (a documented + // System.Diagnostics.Activity quirk) — forcing a genuine new root therefore + // requires clearing Activity.Current for the call, then restoring the worker-loop + // ambient once this message's span has ended. + var previousActivity = Activity.Current; + Activity.Current = null; + try + { + using var activity = InfrastructureActivitySource.Source.StartActivity( + "Outbox.Process", ActivityKind.Producer, default(ActivityContext), links: links); - var topicName = message.ExtraProperties.TryGetValue("TopicName", out var topicObj) - ? topicObj?.ToString() ?? message.EventName : message.EventName; - var pubSubName = message.ExtraProperties.TryGetValue("PubSubName", out var pubSubObj) - ? pubSubObj?.ToString() ?? eventBusOptions.PubSubName : eventBusOptions.PubSubName; + var topicName = message.ExtraProperties.TryGetValue("TopicName", out var topicObj) + ? topicObj?.ToString() ?? message.EventName : message.EventName; + var pubSubName = message.ExtraProperties.TryGetValue("PubSubName", out var pubSubObj) + ? pubSubObj?.ToString() ?? eventBusOptions.PubSubName : eventBusOptions.PubSubName; - activity?.SetTag("event.name", message.EventName); - activity?.SetTag("event.topic", topicName); - activity?.SetTag("outbox.message_id", message.Id.ToString()); - activity?.SetTag("outbox.retry_count", message.RetryCount); + activity?.SetTag("event.name", message.EventName); + activity?.SetTag("event.topic", topicName); + activity?.SetTag("outbox.message_id", message.Id.ToString()); + activity?.SetTag("outbox.retry_count", message.RetryCount); - try - { - await eventBus.PublishEnvelopeAsync(message.EventData, topicName, pubSubName, cancellationToken); - outcomes.Add(new OutboxPublishOutcome(message.Id, true, null)); - activity?.SetStatus(ActivityStatusCode.Ok); - logger.LogInformation("Published outbox message {MessageId}", message.Id); + try + { + await eventBus.PublishEnvelopeAsync(message.EventData, topicName, pubSubName, cancellationToken); + outcomes.Add(new OutboxPublishOutcome(message.Id, true, null)); + activity?.SetStatus(ActivityStatusCode.Ok); + logger.LogInformation("Published outbox message {MessageId}", message.Id); + } + catch (Exception ex) + { + logger.LogError(ex, "Failed to publish outbox message {MessageId}", message.Id); + RecordException(activity, ex); + outcomes.Add(new OutboxPublishOutcome(message.Id, false, ex.Message)); + } } - catch (Exception ex) + finally { - logger.LogError(ex, "Failed to publish outbox message {MessageId}", message.Id); - RecordException(activity, ex); - outcomes.Add(new OutboxPublishOutcome(message.Id, false, ex.Message)); + Activity.Current = previousActivity; } } @@ -244,6 +249,27 @@ private DateTime CalculateNextRetryTime(int retryCount) return clock.UtcNow.Add(TimeSpan.FromMilliseconds(delay.TotalMilliseconds)); } + /// + /// Parses the origin trace identity stored on the row (written by EfCoreOutboxStore) so it can + /// be attached as a causation link. Never used as the parent — see the episode-separation note + /// at the call site. + /// + private static bool TryParseOrigin(OutboxMessage message, out ActivityContext originContext) + { + 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 originContext)) + { + return true; + } + + originContext = default; + return false; + } + private static void RecordException(Activity? activity, Exception ex) { if (activity == null) return; diff --git a/framework/src/BBT.Aether.Infrastructure/Microsoft/Extensions/DependencyInjection/AetherOutboxServiceCollectionExtensions.cs b/framework/src/BBT.Aether.Infrastructure/Microsoft/Extensions/DependencyInjection/AetherOutboxServiceCollectionExtensions.cs index 3660605..d960058 100644 --- a/framework/src/BBT.Aether.Infrastructure/Microsoft/Extensions/DependencyInjection/AetherOutboxServiceCollectionExtensions.cs +++ b/framework/src/BBT.Aether.Infrastructure/Microsoft/Extensions/DependencyInjection/AetherOutboxServiceCollectionExtensions.cs @@ -30,6 +30,18 @@ public static IServiceCollection AddAetherOutbox( services.AddSingleton>(); + // Scoped, matching the scoped EfCoreOutboxStore + scoped IUnitOfWorkManager it consumes + // (singleton would be a captive dependency). Per-UoW dedupe survives across scoped instances + // because the registration table is static. + services.TryAddScoped(); + if (options.WakeupSignalEnabled) + { + services.TryAddSingleton(); + } + services.TryAddSingleton< + BBT.Aether.Polling.IPollingWakeSignal, + BBT.Aether.Polling.PollingWakeSignal>(); + if (withHostedService) services.AddHostedService(); @@ -54,6 +66,10 @@ public static IServiceCollection AddAetherInbox( services.AddSingleton>(); + services.TryAddSingleton< + BBT.Aether.Polling.IPollingWakeSignal, + BBT.Aether.Polling.PollingWakeSignal>(); + if (withHostedService) services.AddHostedService(); 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 index 8a55a2b..3d1f067 100644 --- 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 @@ -24,11 +24,12 @@ 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. +/// Pins the shape of OutboxProcessor's per-message "Outbox.Process" span under the episode- +/// separation policy (trace_refactor): the publish is its own trace ROOT — never re-parented onto +/// the originating transition or the worker loop. Both the origin (when the row's TraceParent is +/// present and parseable) and the worker-loop ambient (when one is active) are attached only as +/// causation links, so an origin span's own parent/child duration semantics are never stretched +/// across the worker hop. /// public sealed class OutboxProcessorTraceTests { @@ -47,7 +48,7 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) } [Fact] - public async Task Message_with_stored_trace_parent_reparents_into_the_origin_trace_and_links_the_worker_loop() + public async Task Message_with_stored_trace_parent_is_a_new_root_that_links_the_origin_and_the_worker_loop() { using var listener = CreateListener(out var started); @@ -55,9 +56,11 @@ public async Task Message_with_stored_trace_parent_reparents_into_the_origin_tra using var origin = originSource.StartActivity("EventBus.Publish", ActivityKind.Producer); origin.ShouldNotBeNull(); var traceParent = origin!.Id!; + var traceState = "vendor=origin-state"; + origin.TraceStateString = traceState; origin.Stop(); // the drop's publish span has already ended by the time the processor runs - var message = MakeMessage(traceParent: traceParent); + var message = MakeMessage(traceParent: traceParent, traceState: traceState); using var loopSource = new ActivitySource(LoopSourceName); using var loop = loopSource.StartActivity("Outbox.Poll", ActivityKind.Internal); @@ -67,16 +70,25 @@ public async Task Message_with_stored_trace_parent_reparents_into_the_origin_tra 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); + + // New root: never a child of the origin or the worker loop. + activity.ParentSpanId.ShouldBe(default(ActivitySpanId)); + activity.TraceId.ShouldNotBe(origin.TraceId); + activity.TraceId.ShouldNotBe(loop!.TraceId); + + // Both are attached as links (causation), never as the parent. + activity.Links.ShouldContain(l => l.Context.TraceId == origin.TraceId && l.Context.SpanId == origin.SpanId + && l.Context.TraceState == traceState); + activity.Links.ShouldContain(l => l.Context.SpanId == loop.SpanId); + activity.Links.Count().ShouldBe(2); + 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() + public async Task Message_without_trace_parent_is_still_a_new_root_and_links_only_the_worker_loop() { using var listener = CreateListener(out var started); @@ -90,16 +102,17 @@ public async Task Message_without_trace_parent_keeps_the_worker_loop_as_parent_w var activity = started.ShouldHaveSingleItem(); activity.OperationName.ShouldBe("Outbox.Process"); - activity.TraceId.ShouldBe(loop!.TraceId); - activity.ParentSpanId.ShouldBe(loop.SpanId); - activity.Links.ShouldBeEmpty(); + activity.ParentSpanId.ShouldBe(default(ActivitySpanId)); + activity.TraceId.ShouldNotBe(loop!.TraceId); + activity.Links.ShouldHaveSingleItem(); + 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_with_garbage_trace_parent_keeps_todays_behavior() + public async Task Message_with_garbage_trace_parent_drops_the_origin_link_but_stays_a_root() { using var listener = CreateListener(out var started); @@ -112,16 +125,42 @@ public async Task Message_with_garbage_trace_parent_keeps_todays_behavior() 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(); + activity.ParentSpanId.ShouldBe(default(ActivitySpanId)); + activity.TraceId.ShouldNotBe(loop!.TraceId); + activity.Links.ShouldHaveSingleItem(); + activity.Links.ShouldContain(l => l.Context.SpanId == loop.SpanId); + } + + [Fact] + public async Task Message_with_trace_parent_and_no_ambient_loop_is_a_root_that_links_only_the_origin() + { + 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(); + + var message = MakeMessage(traceParent: traceParent); + + // No ambient worker-loop activity active here — Activity.Current is null. + await RunProcessorAsync(new[] { message }); + + var activity = started.ShouldHaveSingleItem(); + activity.ParentSpanId.ShouldBe(default(ActivitySpanId)); + activity.TraceId.ShouldNotBe(origin.TraceId); + activity.Links.ShouldHaveSingleItem(); + activity.Links.ShouldContain(l => l.Context.TraceId == origin.TraceId && l.Context.SpanId == origin.SpanId); } - private static OutboxMessage MakeMessage(string? traceParent) + private static OutboxMessage MakeMessage(string? traceParent, string? traceState = null) { var extraProperties = new Dictionary(); if (traceParent != null) extraProperties["TraceParent"] = traceParent; + if (traceState != null) + extraProperties["TraceState"] = traceState; return new OutboxMessage { diff --git a/framework/test/BBT.Aether.Infrastructure.Tests/Polling/OutboxWakeupCoordinatorTests.cs b/framework/test/BBT.Aether.Infrastructure.Tests/Polling/OutboxWakeupCoordinatorTests.cs new file mode 100644 index 0000000..4cbb82a --- /dev/null +++ b/framework/test/BBT.Aether.Infrastructure.Tests/Polling/OutboxWakeupCoordinatorTests.cs @@ -0,0 +1,247 @@ +using System; +using System.Diagnostics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using BBT.Aether.Events; +using BBT.Aether.Uow; +using Microsoft.Extensions.DependencyInjection; +using NSubstitute; +using Shouldly; +using Xunit; + +namespace BBT.Aether.Infrastructure.Tests.Polling; + +public sealed class OutboxWakeupCoordinatorTests +{ + private static async Task PollUntilAsync(Func condition, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (condition()) + return true; + await Task.Delay(10); + } + + return condition(); + } + + [Fact] + public async Task OnOutboxMessageStored_WakeupSignalDisabled_DoesNotRegisterOrNotify() + { + var options = new AetherOutboxOptions { WakeupSignalEnabled = false }; + var unitOfWorkManager = Substitute.For(); + var uow = Substitute.For(); + unitOfWorkManager.Current.Returns(uow); + var notifier = Substitute.For(); + + var sut = new OutboxWakeupCoordinator(options, unitOfWorkManager, notifier); + + sut.OnOutboxMessageStored(); + + uow.DidNotReceive().OnCompleted(Arg.Any>()); + await Task.Delay(50); + await notifier.DidNotReceive().NotifyAsync(Arg.Any()); + } + + [Fact] + public void OnOutboxMessageStored_CalledTwiceUnderSameUow_RegistersOnCompletedExactlyOnce() + { + var options = new AetherOutboxOptions { WakeupSignalEnabled = true }; + var unitOfWorkManager = Substitute.For(); + var uow = Substitute.For(); + unitOfWorkManager.Current.Returns(uow); + var notifier = Substitute.For(); + + var sut = new OutboxWakeupCoordinator(options, unitOfWorkManager, notifier); + + sut.OnOutboxMessageStored(); + sut.OnOutboxMessageStored(); + + uow.Received(1).OnCompleted(Arg.Any>()); + } + + [Fact] + public async Task OnCompletedCallback_Invoked_CallsNotifierWithoutAwaitingItInline() + { + var options = new AetherOutboxOptions { WakeupSignalEnabled = true }; + var unitOfWorkManager = Substitute.For(); + var uow = Substitute.For(); + unitOfWorkManager.Current.Returns(uow); + var notifier = Substitute.For(); + notifier.NotifyAsync(Arg.Any()).Returns(Task.CompletedTask); + + Func? capturedCallback = null; + uow.OnCompleted(Arg.Do>(cb => capturedCallback = cb)); + + var sut = new OutboxWakeupCoordinator(options, unitOfWorkManager, notifier); + sut.OnOutboxMessageStored(); + + capturedCallback.ShouldNotBeNull(); + + // The callback itself must return immediately (detached publish), not await the notify. + var callbackTask = capturedCallback!(uow); + callbackTask.IsCompleted.ShouldBeTrue(); + + var notified = await PollUntilAsync( + () => notifier.ReceivedCalls().Count() > 0, + TimeSpan.FromSeconds(1)); + notified.ShouldBeTrue(); + + await notifier.Received(1).NotifyAsync(Arg.Any()); + } + + [Fact] + public async Task OnCompletedCallback_NotifierThrows_CallbackStillCompletesAndDoesNotPropagate() + { + var options = new AetherOutboxOptions { WakeupSignalEnabled = true }; + var unitOfWorkManager = Substitute.For(); + var uow = Substitute.For(); + unitOfWorkManager.Current.Returns(uow); + var notifier = Substitute.For(); + notifier.NotifyAsync(Arg.Any()) + .Returns(Task.FromException(new InvalidOperationException("broker unavailable"))); + + Func? capturedCallback = null; + uow.OnCompleted(Arg.Do>(cb => capturedCallback = cb)); + + var sut = new OutboxWakeupCoordinator(options, unitOfWorkManager, notifier); + sut.OnOutboxMessageStored(); + + capturedCallback.ShouldNotBeNull(); + + // Should not throw even though the underlying notify task faults. + await capturedCallback!(uow); + + var notified = await PollUntilAsync( + () => notifier.ReceivedCalls().Count() > 0, + TimeSpan.FromSeconds(1)); + notified.ShouldBeTrue(); + } + + [Fact] + public async Task OnOutboxMessageStored_NoAmbientUow_NotifiesWithoutRegistration() + { + var options = new AetherOutboxOptions { WakeupSignalEnabled = true }; + var unitOfWorkManager = Substitute.For(); + unitOfWorkManager.Current.Returns((IUnitOfWork?)null); + var notifier = Substitute.For(); + notifier.NotifyAsync(Arg.Any()).Returns(Task.CompletedTask); + + var sut = new OutboxWakeupCoordinator(options, unitOfWorkManager, notifier); + sut.OnOutboxMessageStored(); + + var notified = await PollUntilAsync( + () => notifier.ReceivedCalls().Count() > 0, + TimeSpan.FromSeconds(1)); + notified.ShouldBeTrue(); + + await notifier.Received(1).NotifyAsync(Arg.Any()); + } + + // Item 6 (documented, not tested): rollback needs no test because the coordinator only ever + // registers via IUnitOfWork.OnCompleted, which Aether's UoW implementation fires solely on + // successful commit — a rollback path never invokes the registered callback, so there is no + // coordinator-owned behavior to assert here. + + [Fact] + public async Task OnOutboxMessageStored_TwoNestedScopesSharingOneRoot_FiresExactlyOneNotify() + { + // Reproduces the amplification bug: a `Required` participant scope is a distinct + // UnitOfWorkScope object per nesting level, but both forward OnCompleted to the SAME + // CompositeUnitOfWork root. Two EfCoreOutboxStore.StoreAsync calls at different nesting + // depths within one logical commit must still yield exactly one nudge, not one per scope. + var options = new AetherOutboxOptions { WakeupSignalEnabled = true }; + var notifier = Substitute.For(); + notifier.NotifyAsync(Arg.Any()).Returns(Task.CompletedTask); + + var serviceProvider = new ServiceCollection().BuildServiceProvider(); + var root = new CompositeUnitOfWork(serviceProvider); + root.InitializeCore(new UnitOfWorkOptions()); + + var ambient = new FakeAmbientAccessor(); + var outerScope = new UnitOfWorkScope(root, ambient, ownsRoot: true); + var innerScope = new UnitOfWorkScope(root, ambient, ownsRoot: false); + + var unitOfWorkManager = Substitute.For(); + var sut = new OutboxWakeupCoordinator(options, unitOfWorkManager, notifier); + + // Outer (owning) scope stores a message first... + unitOfWorkManager.Current.Returns(outerScope); + sut.OnOutboxMessageStored(); + + // ...then a nested Required participant, sharing the same root, stores another. + unitOfWorkManager.Current.Returns(innerScope); + sut.OnOutboxMessageStored(); + + await root.CommitAsync(); + + var notified = await PollUntilAsync( + () => notifier.ReceivedCalls().Count() > 0, + TimeSpan.FromSeconds(1)); + notified.ShouldBeTrue(); + + await notifier.Received(1).NotifyAsync(Arg.Any()); + } + + [Fact] + public async Task OnCompletedCallback_Invoked_WithAmbientActivity_NotifierSeesNoAmbientActivity() + { + // Reproduces the trace-leak bug: the OnCompleted callback runs inside the committing + // business transaction's ExecutionContext, which still has an ambient Activity flowing + // through Task.Run. The notify must be severed from it — the nudge is infrastructure, + // not business flow, and must never attach to (or propagate the traceparent of) the + // business trace that triggered it. + var options = new AetherOutboxOptions { WakeupSignalEnabled = true }; + var unitOfWorkManager = Substitute.For(); + var uow = Substitute.For(); + unitOfWorkManager.Current.Returns(uow); + + Activity? activitySeenByNotifier = null; + var notifierInvoked = false; + var notifier = Substitute.For(); + notifier.NotifyAsync(Arg.Any()).Returns(_ => + { + activitySeenByNotifier = Activity.Current; + notifierInvoked = true; + return Task.CompletedTask; + }); + + Func? capturedCallback = null; + uow.OnCompleted(Arg.Do>(cb => capturedCallback = cb)); + + var sut = new OutboxWakeupCoordinator(options, unitOfWorkManager, notifier); + sut.OnOutboxMessageStored(); + capturedCallback.ShouldNotBeNull(); + + using var listener = new ActivityListener + { + ShouldListenTo = _ => true, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllData + }; + ActivitySource.AddActivityListener(listener); + + using var activitySource = new ActivitySource(nameof(OutboxWakeupCoordinatorTests) + ".AmbientActivity"); + using var ambientActivity = activitySource.StartActivity("business-transition"); + ambientActivity.ShouldNotBeNull(); + Activity.Current.ShouldBe(ambientActivity); + + // Fire the OnCompleted callback while the ambient business Activity is current, exactly as + // it happens inside CommitAsync in production. + var callbackTask = capturedCallback!(uow); + callbackTask.IsCompleted.ShouldBeTrue(); + + var notified = await PollUntilAsync(() => notifierInvoked, TimeSpan.FromSeconds(1)); + notified.ShouldBeTrue(); + + activitySeenByNotifier.ShouldBeNull(); + } + + private sealed class FakeAmbientAccessor : IAmbientUnitOfWorkAccessor + { + public IUnitOfWork? Current { get; set; } + + public IUnitOfWork? GetActiveUnitOfWork() => Current; + } +} diff --git a/framework/test/BBT.Aether.Infrastructure.Tests/Polling/PollingWakeSignalTests.cs b/framework/test/BBT.Aether.Infrastructure.Tests/Polling/PollingWakeSignalTests.cs new file mode 100644 index 0000000..c4462f8 --- /dev/null +++ b/framework/test/BBT.Aether.Infrastructure.Tests/Polling/PollingWakeSignalTests.cs @@ -0,0 +1,47 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using BBT.Aether.Polling; +using Shouldly; +using Xunit; + +namespace BBT.Aether.Infrastructure.Tests.Polling; + +public sealed class PollingWakeSignalTests +{ + private interface IMarker; + + [Fact] + public async Task WaitAsync_ReturnsTrue_WhenSignaled() + { + var sut = new PollingWakeSignal(); + sut.Signal(); + (await sut.WaitAsync(TimeSpan.FromSeconds(5))).ShouldBeTrue(); + } + + [Fact] + public async Task WaitAsync_ReturnsFalse_OnTimeout() + { + var sut = new PollingWakeSignal(); + (await sut.WaitAsync(TimeSpan.FromMilliseconds(50))).ShouldBeFalse(); + } + + [Fact] + public async Task Signal_IsCoalesced_NotAccumulated() + { + var sut = new PollingWakeSignal(); + sut.Signal(); + sut.Signal(); // must not throw, must not stack + (await sut.WaitAsync(TimeSpan.FromSeconds(5))).ShouldBeTrue(); + (await sut.WaitAsync(TimeSpan.FromMilliseconds(50))).ShouldBeFalse(); + } + + [Fact] + public async Task WaitAsync_Honors_Cancellation() + { + var sut = new PollingWakeSignal(); + using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(50)); + await Should.ThrowAsync( + () => sut.WaitAsync(TimeSpan.FromSeconds(30), cts.Token)); + } +} diff --git a/framework/test/BBT.Aether.Infrastructure.Tests/Polling/WakeAwarePollingTests.cs b/framework/test/BBT.Aether.Infrastructure.Tests/Polling/WakeAwarePollingTests.cs new file mode 100644 index 0000000..9778c49 --- /dev/null +++ b/framework/test/BBT.Aether.Infrastructure.Tests/Polling/WakeAwarePollingTests.cs @@ -0,0 +1,51 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using BBT.Aether.Events; +using BBT.Aether.Events.Processing; +using BBT.Aether.Polling; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using Shouldly; +using Xunit; + +namespace BBT.Aether.Infrastructure.Tests.Polling; + +public sealed class WakeAwarePollingTests +{ + [Fact] + public async Task OutboxService_PollsImmediately_WhenSignaled() + { + var processed = new SemaphoreSlim(0); + var processor = Substitute.For(); + processor.RunAsync(Arg.Any()) + .Returns(_ => { processed.Release(); return Task.FromResult(0); }); + + var options = new AetherOutboxOptions + { + IdlePollingInterval = TimeSpan.FromSeconds(30), + MaxPollingInterval = TimeSpan.FromSeconds(30), + BusyPollingInterval = TimeSpan.FromMilliseconds(100) + }; + var signal = new PollingWakeSignal(); + var sut = new OutboxBackgroundService( + processor, options, NullLogger.Instance, signal); + + using var cts = new CancellationTokenSource(); + await sut.StartAsync(cts.Token); + try + { + // Startup offset is also wake-aware: signal now, first run must happen fast. + signal.Signal(); + (await processed.WaitAsync(TimeSpan.FromSeconds(5))).ShouldBeTrue(); + // With a 30s idle interval, only a signal can trigger the next run this fast. + signal.Signal(); + (await processed.WaitAsync(TimeSpan.FromSeconds(5))).ShouldBeTrue(); + } + finally + { + cts.Cancel(); + await sut.StopAsync(CancellationToken.None); + } + } +}