Feature/outbox wakeup signal

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<IActionResult> 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<BBT.Aether.Polling.IPollingWakeSignal<IInboxProcessor>>()
+                    ?.Signal();
+
                 return Ok();
             }
         }

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.
     /// </summary>
     public string? Schema { get; set; } = "sys_queues";
+
+    /// <summary>
+    /// When true, a unit of work that stored outbox messages publishes a direct pub/sub wake nudge
+    /// (<see cref="OutboxWakeupEvent"/>) after commit so outbox processors poll immediately instead
+    /// of waiting out the idle interval. Default false. Requires an IOutboxWakeupNotifier registration.
+    /// </summary>
+    public bool WakeupSignalEnabled { get; set; }
 }

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;
+
+/// <summary>
+/// Publishes the <see cref="OutboxWakeupEvent"/> nudge. Implementations must be fire-and-forget
+/// safe: a failed notify is swallowed by callers because polling backstops delivery.
+/// </summary>
+public interface IOutboxWakeupNotifier
+{
+    Task NotifyAsync(CancellationToken cancellationToken = default);
+}

framework/src/BBT.Aether.Core/BBT/Aether/Events/OutboxWakeupEvent.cs

@@ -0,0 +1,10 @@
+namespace BBT.Aether.Events;
+
+/// <summary>
+/// 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.
+/// </summary>
+[EventName("aether.outbox.wakeup")]
+public sealed class OutboxWakeupEvent;

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;
+
+/// <summary>
+/// A coalescing wake signal for adaptive polling loops. Producers call <see cref="Signal"/> when
+/// new work becomes available; the polling loop awaits <see cref="WaitAsync"/> 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.
+/// </summary>
+/// <typeparam name="TMarker">Marker type identifying the loop this signal wakes.</typeparam>
+public interface IPollingWakeSignal<TMarker>
+{
+    /// <summary>Wakes the loop. Multiple pending signals coalesce into one.</summary>
+    void Signal();
+
+    /// <summary>
+    /// Waits until <see cref="Signal"/> is called or the timeout elapses.
+    /// Returns true when woken by a signal, false on timeout.
+    /// </summary>
+    Task<bool> WaitAsync(TimeSpan timeout, CancellationToken cancellationToken = default);
+}

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;
+
+/// <summary>
+/// Default <see cref="IPollingWakeSignal{TMarker}"/> over a bounded <see cref="SemaphoreSlim"/>(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.
+/// </summary>
+public sealed class PollingWakeSignal<TMarker> : IPollingWakeSignal<TMarker>
+{
+    private readonly SemaphoreSlim _semaphore = new(0, 1);
+
+    public void Signal()
+    {
+        try
+        {
+            _semaphore.Release();
+        }
+        catch (SemaphoreFullException)
+        {
+            // A wake is already pending — coalesce.
+        }
+    }
+
+    public Task<bool> WaitAsync(TimeSpan timeout, CancellationToken cancellationToken = default)
+        => _semaphore.WaitAsync(timeout, cancellationToken);
+}

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;
+
+/// <summary>
+/// Publishes <see cref="OutboxWakeupEvent"/> straight to the configured pub/sub component,
+/// bypassing the outbox by design (the nudge must not create the work it announces).
+/// </summary>
+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);
+}

framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/EfCoreOutboxStore.cs

@@ -20,7 +20,8 @@ public class EfCoreOutboxStore<TDbContext>(
     IGuidGenerator guidGenerator,
     IClock clock,
     AetherOutboxOptions options,
-    ICurrentSchema? currentSchema) : IOutboxStore
+    ICurrentSchema? currentSchema,
+    OutboxWakeupCoordinator? wakeupCoordinator = null) : IOutboxStore
     where TDbContext : DbContext, IHasEfCoreOutbox
 {
     /// <summary>
@@ -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()

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;
+
+/// <summary>
+/// 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.
+/// </summary>
+public sealed class OutboxWakeupCoordinator(
+    AetherOutboxOptions options,
+    IUnitOfWorkManager? unitOfWorkManager = null,
+    IOutboxWakeupNotifier? wakeupNotifier = null,
+    ILogger<OutboxWakeupCoordinator>? logger = null)
+{
+    private static readonly TimeSpan NotifyTimeout = TimeSpan.FromSeconds(2);
+    private static readonly ConditionalWeakTable<IUnitOfWork, object> WakeupRegistered = new();
+    private static readonly object RegisteredSentinel = new();
+
+    /// <summary>Call once per stored outbox message; registration collapses to one per UoW.</summary>
+    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");
+            }
+        });
+    }
+}

framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/InboxBackgroundService.cs

@@ -10,16 +10,21 @@ namespace BBT.Aether.Events.Processing;
 public sealed class InboxBackgroundService(
     IInboxProcessor processor,
     AetherInboxOptions options,
-    ILogger<InboxBackgroundService> logger) : BackgroundService
+    ILogger<InboxBackgroundService> logger,
+    BBT.Aether.Polling.IPollingWakeSignal<IInboxProcessor>? wakeSignal = null) : BackgroundService
 {
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
         // Spread the first poll of replicas that booted together, so a rolling deployment does not
         // 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);
         }
     }
 }

framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/Processing/OutboxBackgroundService.cs

@@ -10,16 +10,21 @@ namespace BBT.Aether.Events.Processing;
 public sealed class OutboxBackgroundService(
     IOutboxProcessor processor,
     AetherOutboxOptions options,
-    ILogger<OutboxBackgroundService> logger) : BackgroundService
+    ILogger<OutboxBackgroundService> logger,
+    BBT.Aether.Polling.IPollingWakeSignal<IOutboxProcessor>? wakeSignal = null) : BackgroundService
 {
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
         // Spread the first poll of replicas that booted together, so a rolling deployment does not
         // 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);
         }
     }
 }