diff --git a/docs/requirements/Component-NotificationOutbox.md b/docs/requirements/Component-NotificationOutbox.md index a5918759..6e5293eb 100644 --- a/docs/requirements/Component-NotificationOutbox.md +++ b/docs/requirements/Component-NotificationOutbox.md @@ -20,7 +20,7 @@ Central cluster. The `NotificationOutboxActor` is a **singleton on the active ce - Compute delivery KPIs from the `Notifications` table for the Health Monitoring dashboard and the Central UI. - Purge terminal rows daily after a configurable retention window. -SMTP and HTTP delivery is blocking I/O. Delivery work runs on a **dedicated blocking-I/O dispatcher**, the same pattern used by Script Execution Actors, so delivery never blocks the actor's dispatcher loop. +SMTP and HTTP delivery is blocking I/O. A dispatch sweep runs **off the actor thread** on the thread pool (the sweep task pipes its completion back to the actor), so delivery never blocks the actor's message loop. Within a sweep, deliveries run with **bounded per-sweep parallelism** — a `SemaphoreSlim` caps the number of concurrent deliveries at `MaxParallelDeliveries` (default 4; set to 1 for strictly sequential delivery). Each concurrent delivery uses its own DI scope/repository, so there is no shared-`DbContext` contention. ## End-to-End Flow @@ -117,7 +117,7 @@ The dispatcher loop runs on a fixed interval. On each tick the `NotificationOutb 1. Polls the `Notifications` table for **due rows** — `Pending` rows, and `Retrying` rows whose `NextAttemptAt` has passed. 2. Resolves the target notification list to its recipients/targets at central, at delivery time. -3. Hands the notification to the delivery adapter registered for its `Type`, running on the dedicated blocking-I/O dispatcher. +3. Hands the notification to the delivery adapter registered for its `Type`. Deliveries within the sweep run with bounded parallelism (`MaxParallelDeliveries`, default 4), each on its own DI scope/repository, off the actor thread. 4. Applies the result: - **success** → `Delivered`, set `DeliveredAt`, snapshot `ResolvedTargets`. - **transient failure** → `Retrying`, increment `RetryCount`, set `NextAttemptAt`, record `LastError`; once retries are exhausted → `Parked`. diff --git a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxActor.cs b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxActor.cs index 22d2da81..a55fff05 100644 --- a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxActor.cs @@ -318,33 +318,26 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers var (maxRetries, retryDelay) = await ResolveRetryPolicyAsync(notificationRepository); + // Deliver the claimed batch with bounded parallelism. Delivery is I/O-bound + // (multi-second SMTP/Twilio round trips), so a strictly sequential sweep caps + // throughput at ~1/attempt-latency — an alarm-storm backlog behind one slow + // host would drain far too slowly (arch-review 04). A SemaphoreSlim caps the + // number of in-flight deliveries; each delivery runs on its OWN DI scope/ + // repository (the EF DbContext is not thread-safe and rows are distinct per + // notification, so concurrent deliveries never contend). The cached adapters + // and the lifetime-scoped audit writer are already safe for concurrent use. + // MaxParallelDeliveries = 1 preserves the strictly sequential behaviour. + using var deliveryGate = new SemaphoreSlim( + _options.ResolvedMaxParallelDeliveries, _options.ResolvedMaxParallelDeliveries); + var deliveries = new List(due.Count); foreach (var notification in due) { - // Between deliveries, observe shutdown so we don't kick off a fresh - // SMTP send when the actor is already tearing down. - if (cancellationToken.IsCancellationRequested) - { - return; - } - - try - { - await DeliverOneAsync( - notification, now, maxRetries, retryDelay, outboxRepository, adapters, cancellationToken); - } - catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) - { - // In-flight delivery interrupted by shutdown. Row remains in its - // pre-attempt state; next active sweep retries. - return; - } - catch (Exception ex) - { - // Isolate per-notification failures so the remainder of the batch still runs. - _logger.LogError( - ex, "Dispatch failed for notification {NotificationId}.", notification.NotificationId); - } + deliveries.Add(DeliverGatedAsync( + notification, now, maxRetries, retryDelay, adapters, deliveryGate, cancellationToken)); } + + // Each DeliverGatedAsync isolates its own faults, so WhenAll never throws. + await Task.WhenAll(deliveries); } catch (Exception ex) { @@ -354,6 +347,67 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers } } + /// + /// Delivers one notification under the sweep's concurrency gate. Acquires a permit, + /// runs on a fresh per-notification DI scope (so parallel + /// deliveries never share an EF DbContext), and always releases the permit. + /// Faults are isolated here — shutdown cancellation is swallowed (the row stays in its + /// pre-attempt state for the next active sweep) and any other exception is logged — so + /// one bad delivery never aborts the rest of the batch and Task.WhenAll never faults. + /// + private async Task DeliverGatedAsync( + Notification notification, + DateTimeOffset now, + int maxRetries, + TimeSpan retryDelay, + IReadOnlyDictionary adapters, + SemaphoreSlim gate, + CancellationToken cancellationToken) + { + try + { + await gate.WaitAsync(cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // Shutdown cancelled before this delivery acquired a permit; the row stays + // Pending and the next active node's sweep re-claims it. No permit to release. + return; + } + + try + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + // Each delivery owns its DI scope + repository so parallel UpdateAsync calls + // never share a DbContext. The adapters map + audit writer are already safe + // for concurrent use. + using var scope = _serviceProvider.CreateScope(); + var outboxRepository = scope.ServiceProvider.GetRequiredService(); + await DeliverOneAsync( + notification, now, maxRetries, retryDelay, outboxRepository, adapters, cancellationToken) + .ConfigureAwait(false); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // In-flight delivery interrupted by shutdown. Row remains in its pre-attempt + // state; next active sweep retries. + } + catch (Exception ex) + { + // Isolate per-notification failures so the remainder of the batch still runs. + _logger.LogError( + ex, "Dispatch failed for notification {NotificationId}.", notification.NotificationId); + } + finally + { + gate.Release(); + } + } + /// /// Resolves the retry policy from the first SMTP configuration row. When no SMTP /// configuration exists, falls back to a conservative default — delivery itself will diff --git a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxOptions.cs b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxOptions.cs index 7c10473d..45220c76 100644 --- a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxOptions.cs @@ -48,4 +48,23 @@ public class NotificationOutboxOptions /// Default: 1 minute. /// public TimeSpan DeliveredKpiWindow { get; set; } = TimeSpan.FromMinutes(1); + + /// + /// Maximum number of notifications delivered concurrently within a single dispatch + /// sweep. Delivery is I/O-bound (SMTP / Twilio round trips of seconds each), so a + /// purely sequential sweep caps throughput at ~1/attempt-latency regardless of batch + /// size — an alarm storm queued behind one slow SMTP host would drain far too slowly + /// (arch-review 04). Each concurrent delivery runs on its own DI scope/repository, so + /// there is no shared-DbContext contention. Set to 1 to preserve the strictly + /// sequential behaviour. Clamped to a minimum of 1 via . + /// Default: 4. + /// + public int MaxParallelDeliveries { get; set; } = 4; + + /// + /// Resolves the effective delivery parallelism, clamped to at least 1 so a + /// misconfigured 0/negative value cannot stall the dispatcher (a zero-count + /// semaphore would deadlock the sweep). + /// + public int ResolvedMaxParallelDeliveries => MaxParallelDeliveries < 1 ? 1 : MaxParallelDeliveries; } diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorDispatchTests.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorDispatchTests.cs index ce9f1086..fd3289ad 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorDispatchTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorDispatchTests.cs @@ -567,4 +567,121 @@ public class NotificationOutboxActorDispatchTests : TestKit Arg.Any(), Arg.Any(), Arg.Any()), duration: TimeSpan.FromSeconds(2)); } + + // ── Task 25: bounded parallel delivery ─────────────────────────────────── + + /// + /// Adapter that records the concurrent-invocation high-water mark. Each + /// increments a live counter, notes the peak, then blocks + /// on a shared gate until deliveries are concurrently + /// in flight — proving real overlap. A safety timeout releases the gate so a + /// parallelism-1 run (which can never reach releaseWhen concurrently) still + /// drains instead of deadlocking. + /// + private sealed class ConcurrencyHighWaterAdapter : INotificationDeliveryAdapter + { + private readonly int _releaseWhen; + private readonly TaskCompletionSource _gate = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _current; + private int _highWater; + private int _entered; + + public ConcurrencyHighWaterAdapter(int releaseWhen) => _releaseWhen = releaseWhen; + + public NotificationType Type => NotificationType.Email; + + public int HighWater => Volatile.Read(ref _highWater); + + public async Task DeliverAsync( + Notification notification, CancellationToken cancellationToken = default) + { + var current = Interlocked.Increment(ref _current); + int hw; + while ((hw = Volatile.Read(ref _highWater)) < current + && Interlocked.CompareExchange(ref _highWater, current, hw) != hw) + { + // retry until our peak is recorded + } + + if (Interlocked.Increment(ref _entered) >= _releaseWhen) + { + _gate.TrySetResult(); + } + + // Overlap window: wait until enough deliveries are concurrently in flight, + // or a short safety timeout (so the strictly-sequential config still drains). + await Task.WhenAny(_gate.Task, Task.Delay(TimeSpan.FromMilliseconds(400), cancellationToken)) + .ConfigureAwait(false); + + Interlocked.Decrement(ref _current); + return DeliveryOutcome.Success("ops@example.com"); + } + } + + private void SetupDueBatchOnce(int count) + { + var batch = Enumerable.Range(0, count).Select(_ => MakeNotification()).ToArray(); + var claimed = 0; + _outboxRepository.GetDueAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(_ => Interlocked.Increment(ref claimed) == 1 + ? batch + : Array.Empty()); + } + + [Fact] + public void Dispatch_WithMaxParallelFour_DeliversConcurrently_AndAllReachTerminal() + { + SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1)); + SetupDueBatchOnce(8); + var adapter = new ConcurrencyHighWaterAdapter(releaseWhen: 2); + var actor = CreateActor( + [adapter], + new NotificationOutboxOptions + { + DispatchInterval = TimeSpan.FromHours(1), + MaxParallelDeliveries = 4, + }); + + actor.Tell(InternalMessages.DispatchTick.Instance); + + AwaitAssert( + () => + { + // All 8 notifications reached a terminal Delivered update. + _outboxRepository.Received(8).UpdateAsync( + Arg.Is(n => n.Status == NotificationStatus.Delivered), + Arg.Any()); + // At least two deliveries overlapped — the sweep is no longer sequential. + Assert.True(adapter.HighWater >= 2, + $"expected concurrent deliveries (high-water >= 2), got {adapter.HighWater}"); + }, + duration: TimeSpan.FromSeconds(10)); + } + + [Fact] + public void Dispatch_WithMaxParallelOne_DeliversSequentially_AndAllReachTerminal() + { + SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1)); + SetupDueBatchOnce(8); + var adapter = new ConcurrencyHighWaterAdapter(releaseWhen: 2); + var actor = CreateActor( + [adapter], + new NotificationOutboxOptions + { + DispatchInterval = TimeSpan.FromHours(1), + MaxParallelDeliveries = 1, + }); + + actor.Tell(InternalMessages.DispatchTick.Instance); + + AwaitAssert( + () => _outboxRepository.Received(8).UpdateAsync( + Arg.Is(n => n.Status == NotificationStatus.Delivered), + Arg.Any()), + duration: TimeSpan.FromSeconds(10)); + + // Strictly sequential: never more than one delivery in flight at once. + Assert.Equal(1, adapter.HighWater); + } }