diff --git a/docs/requirements/Component-NotificationOutbox.md b/docs/requirements/Component-NotificationOutbox.md index 0cb0c811..72fdb4c7 100644 --- a/docs/requirements/Component-NotificationOutbox.md +++ b/docs/requirements/Component-NotificationOutbox.md @@ -115,10 +115,11 @@ Delivery has no row-level claim or lease: dispatchers read due rows and persist The dispatcher loop runs on a fixed interval. On each tick the `NotificationOutboxActor`: -1. Polls the `Notifications` table for **due rows** — `Pending` rows, and `Retrying` rows whose `NextAttemptAt` has passed. +1. Polls the `Notifications` table for **due rows** — `Pending` rows, and `Retrying` rows whose `NextAttemptAt` has passed — capped at `DispatchBatchSize` (default 100). The batch is **shared between the two statuses**: due `Retrying` rows take at most half of it and `Pending` rows the rest, each oldest-first, with a share the other status cannot fill back-filled. A single oldest-first claim used to let a retrying backlog (always older than fresh traffic) monopolise every sweep during a partial relay outage, so a new alarm email was not even attempted until the backlog parked. 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`. Deliveries within the sweep run with bounded parallelism (`MaxParallelDeliveries`, default 4), each on its own DI scope, off the actor thread. **That per-delivery scope owns the channel adapter as well as the outbox repository** — see Delivery Adapters below. +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, off the actor thread. **That per-delivery scope owns the channel adapter as well as the outbox repository** — see Delivery Adapters below. Email over the **SMTP** transport is further clamped per sweep to the active (lowest-Id) SMTP row's `MaxConcurrentConnections` when that value is positive, so a relay that limits connections is honoured; the EWS transport (which does not use the field) and SMS keep the full `MaxParallelDeliveries`. A sweep also has a **wall-clock budget** (`DispatchSweepBudget`, default 2 min): once it is spent, claimed notifications that have not yet started an attempt are skipped — left untouched and still due, so the next sweep re-claims them alongside newer rows — while attempts already in flight run to completion. 4. Applies the result: + Every timestamp written for an attempt (`LastAttemptAt`, `DeliveredAt`, `NextAttemptAt` = attempt instant + `RetryDelay`, and the attempt/terminal audit rows' occurrence time) is the **attempt instant**, captured after the adapter returns; the sweep-start instant is used only for the due-claim predicate. Stamping sweep-start skewed `DeliveredAt` for the tail of a slow sweep and scheduled `NextAttemptAt` in the past, collapsing the retry delay. - **success** → `Delivered`, set `DeliveredAt`, snapshot `ResolvedTargets`. - **transient failure** → `Retrying`, increment `RetryCount`, set `NextAttemptAt`, record `LastError`; once retries are exhausted → `Parked`. - **permanent failure** → `Parked`, record `LastError`. diff --git a/docs/requirements/Component-NotificationService.md b/docs/requirements/Component-NotificationService.md index 80ca9178..62ce1ac1 100644 --- a/docs/requirements/Component-NotificationService.md +++ b/docs/requirements/Component-NotificationService.md @@ -51,7 +51,7 @@ The SMTP configuration is defined centrally and used by the central Email delive - **TLS mode**: None, StartTLS, or SSL. - **From address**: The sender email address for all notifications (e.g., `scada-notifications@company.com`). - **Connection timeout**: Maximum time to wait for SMTP connection (default: 30 seconds). -- **Max concurrent connections**: Maximum simultaneous SMTP connections from the central cluster (default: 5). +- **Max concurrent connections**: Maximum simultaneous SMTP connections from the central cluster (default: 5). Enforced by the Notification Outbox dispatcher, which clamps each sweep's Email parallelism to `min(MaxParallelDeliveries, MaxConcurrentConnections)` for the SMTP transport (a non-positive value applies no clamp). - **Retry settings**: Max retry count, fixed time between retries. The Notification Outbox reuses these for delivery retry of transient failures. - **Transport**: The submission mechanism — `Smtp` (default) or `Ews`. Null/empty means `Smtp`, so every pre-existing row keeps its behavior. See EWS Transport below. diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/INotificationOutboxRepository.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/INotificationOutboxRepository.cs index 777b410c..6d1eb1fa 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/INotificationOutboxRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/INotificationOutboxRepository.cs @@ -32,7 +32,10 @@ public interface INotificationOutboxRepository /// Returns notifications ready for a delivery attempt: Pending rows, plus /// Retrying rows whose NextAttemptAt is at or before . /// Terminal rows are excluded. Ordered by CreatedAt ascending, capped at - /// . + /// . The batch is shared between the two statuses — due + /// Retrying rows take at most half, Pending rows the rest, and a share + /// the other status cannot fill is back-filled — so a retrying backlog can never starve + /// fresh traffic (review 2026-10-21 06/PERF-5). /// /// The current time for evaluating due retries. /// Maximum number of rows to return. diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs index 5da38635..6601e9c7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs @@ -146,15 +146,54 @@ VALUES // of notifications — each carrying an nvarchar(max) Body and TypeData — // paid for a full snapshot copy per row plus a DetectChanges scan of the // whole batch on every save. - return await _context.Notifications - .AsNoTracking() - .Where(n => n.Status == NotificationStatus.Pending - || (n.Status == NotificationStatus.Retrying - && n.NextAttemptAt != null - && n.NextAttemptAt <= now)) - .OrderBy(n => n.CreatedAt) - .Take(batchSize) + // + // The batch is SPLIT between due Retrying rows and Pending rows (review + // 2026-10-21 06/PERF-5). A single oldest-first claim let a retrying backlog — + // always older than fresh traffic — monopolise every sweep during a partial + // relay outage, so a new alarm email was not even attempted until the backlog + // parked (hours, at MaxRetries x RetryDelay). Retrying gets floor(batch/2), + // Pending the rest; a share the other status cannot fill is back-filled, so a + // sweep still claims a full batch whenever that many rows are due. Each share is + // oldest-first and the merged batch is returned in CreatedAt order. + if (batchSize <= 0) + { + return Array.Empty(); + } + + var retryingQuota = batchSize / 2; + var pending = await DuePendingQuery() + .Take(batchSize - retryingQuota) .ToListAsync(cancellationToken); + var retrying = await DueRetryingQuery(now) + .Take(batchSize - pending.Count) + .ToListAsync(cancellationToken); + + var remaining = batchSize - pending.Count - retrying.Count; + if (remaining > 0 && pending.Count == batchSize - retryingQuota) + { + // Retrying could not fill its share; back-fill with newer Pending rows. + pending.AddRange(await DuePendingQuery() + .Skip(pending.Count) + .Take(remaining) + .ToListAsync(cancellationToken)); + } + + pending.AddRange(retrying); + return pending.OrderBy(n => n.CreatedAt).ToList(); + + IQueryable DuePendingQuery() => _context.Notifications + .AsNoTracking() + .Where(n => n.Status == NotificationStatus.Pending) + .OrderBy(n => n.CreatedAt) + .ThenBy(n => n.NotificationId); + + IQueryable DueRetryingQuery(DateTimeOffset dueBy) => _context.Notifications + .AsNoTracking() + .Where(n => n.Status == NotificationStatus.Retrying + && n.NextAttemptAt != null + && n.NextAttemptAt <= dueBy) + .OrderBy(n => n.CreatedAt) + .ThenBy(n => n.NotificationId); } /// diff --git a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxActor.cs b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxActor.cs index 193cd001..5d37cdcc 100644 --- a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxActor.cs @@ -11,6 +11,7 @@ using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; using ZB.MOM.WW.ScadaBridge.Commons.Types.Notifications; using ZB.MOM.WW.ScadaBridge.NotificationOutbox.Delivery; using ZB.MOM.WW.ScadaBridge.NotificationOutbox.Messages; +using ZB.MOM.WW.ScadaBridge.NotificationService; namespace ZB.MOM.WW.ScadaBridge.NotificationOutbox; @@ -295,17 +296,35 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers // and rows are distinct per notification, so concurrent deliveries never // contend). The lifetime-scoped audit writer is already safe for concurrent // use. MaxParallelDeliveries = 1 preserves the strictly sequential behaviour. - using var deliveryGate = new SemaphoreSlim( - _options.ResolvedMaxParallelDeliveries, _options.ResolvedMaxParallelDeliveries); + // + // Email over the SMTP transport is additionally clamped to the SMTP row's + // MaxConcurrentConnections (review 2026-10-21 06/IMPL-2): an Email delivery + // takes a permit from the Email gate FIRST and the sweep gate second (a fixed + // acquisition order, so the two gates cannot deadlock), leaving SMS free to use + // the sweep's remaining parallelism. + var maxParallel = _options.ResolvedMaxParallelDeliveries; + using var deliveryGate = new SemaphoreSlim(maxParallel, maxParallel); + using var emailGate = policies.EmailConcurrencyLimit is int emailLimit && emailLimit < maxParallel + ? new SemaphoreSlim(emailLimit, emailLimit) + : null; + var sweep = new SweepBudget(_options.DispatchSweepBudget); var deliveries = new List(due.Count); foreach (var notification in due) { + var typeGate = notification.Type == NotificationType.Email ? emailGate : null; deliveries.Add(DeliverGatedAsync( - notification, now, policies, deliveryGate, cancellationToken)); + notification, policies, deliveryGate, typeGate, sweep, cancellationToken)); } // Each DeliverGatedAsync isolates its own faults, so WhenAll never throws. await Task.WhenAll(deliveries); + + if (sweep.Skipped > 0) + { + _logger.LogInformation( + "Dispatch sweep spent its {Budget} budget; {Skipped} of {Claimed} claimed notification(s) were not attempted and stay due for the next sweep.", + _options.DispatchSweepBudget, sweep.Skipped, due.Count); + } } catch (Exception ex) { @@ -339,19 +358,34 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers /// private async Task DeliverGatedAsync( Notification notification, - DateTimeOffset now, RetryPolicies policies, SemaphoreSlim gate, + SemaphoreSlim? typeGate, + SweepBudget sweep, CancellationToken cancellationToken) { + // Type gate (Email-over-SMTP clamp) first, sweep gate second — the fixed order keeps + // the two gates deadlock-free. + var typeGateHeld = false; try { + if (typeGate is not null) + { + await typeGate.WaitAsync(cancellationToken).ConfigureAwait(false); + typeGateHeld = true; + } + 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. + // Pending and the next active node's sweep re-claims it. No sweep permit to release. + if (typeGateHeld) + { + typeGate!.Release(); + } + return; } @@ -362,6 +396,16 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers return; } + // Per-sweep wall-clock budget (review 2026-10-21 06/PERF-5): a delivery that only + // got its permit after the budget was spent is not attempted. The row is left + // untouched — still Pending, or Retrying with an elapsed NextAttemptAt — so it + // stays due and the next sweep re-claims it alongside any newer traffic. + if (sweep.IsExhausted) + { + sweep.MarkSkipped(); + return; + } + // Each delivery owns its DI scope, repository AND channel adapter so parallel // deliveries never share a DbContext. The audit writer is already safe for // concurrent use. @@ -369,7 +413,7 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers var outboxRepository = scope.ServiceProvider.GetRequiredService(); var adapter = ResolveAdapter(scope.ServiceProvider, notification.Type); await DeliverOneAsync( - notification, now, policies, outboxRepository, adapter, cancellationToken) + notification, policies, outboxRepository, adapter, cancellationToken) .ConfigureAwait(false); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) @@ -386,9 +430,30 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers finally { gate.Release(); + typeGate?.Release(); } } + /// + /// Tracks one sweep's wall-clock budget and how many claimed notifications were skipped + /// because it ran out. Shared by every delivery task of the sweep, hence the interlocked + /// counter. + /// + private sealed class SweepBudget(TimeSpan budget) + { + private readonly System.Diagnostics.Stopwatch _clock = System.Diagnostics.Stopwatch.StartNew(); + private int _skipped; + + /// True once the sweep has run past its budget. + public bool IsExhausted => _clock.Elapsed >= budget; + + /// The number of claimed notifications left unattempted. + public int Skipped => Volatile.Read(ref _skipped); + + /// Records one claimed notification left unattempted. + public void MarkSkipped() => Interlocked.Increment(ref _skipped); + } + /// /// A single channel's transient-failure retry policy: the max attempt count before a /// row parks and the fixed delay between attempts. Resolved once per sweep and selected @@ -399,8 +464,12 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers /// /// The per-type retry policies resolved once per dispatch sweep: /// drives Email (and every non-SMS type), drives SMS. + /// is the active SMTP row's positive + /// MaxConcurrentConnections when that row uses the SMTP transport, else null + /// (no Email-specific clamp). /// - private readonly record struct RetryPolicies(RetryPolicy SmtpPolicy, RetryPolicy SmsPolicy); + private readonly record struct RetryPolicies( + RetryPolicy SmtpPolicy, RetryPolicy SmsPolicy, int? EmailConcurrencyLimit); /// /// Resolves the per-type retry policies for the sweep: the SMTP policy from the first @@ -413,9 +482,9 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers private async Task ResolveRetryPoliciesAsync( INotificationRepository notificationRepository) { - var smtpPolicy = await ResolveSmtpPolicyAsync(notificationRepository); + var (smtpPolicy, emailConcurrencyLimit) = await ResolveSmtpPolicyAsync(notificationRepository); var smsPolicy = await ResolveSmsPolicyAsync(notificationRepository, smtpPolicy); - return new RetryPolicies(smtpPolicy, smsPolicy); + return new RetryPolicies(smtpPolicy, smsPolicy, emailConcurrencyLimit); } /// @@ -423,7 +492,14 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers /// SMTP configuration exists, falls back to the conservative default constants — delivery /// itself will permanently fail in that case, so the policy only acts as a guard. /// - private async Task ResolveSmtpPolicyAsync( + /// + /// Also resolves the Email concurrency clamp from the same row (review 2026-10-21 + /// 06/IMPL-2): MaxConcurrentConnections was bound, validated and shown in the CLI + /// but read by no delivery path. It applies only when the row's transport parses as + /// SMTP and the value is positive — EWS does not use it, and an unparseable transport + /// parks every Email delivery anyway. + /// + private async Task<(RetryPolicy Policy, int? EmailConcurrencyLimit)> ResolveSmtpPolicyAsync( INotificationRepository notificationRepository) { var configurations = await notificationRepository.GetAllSmtpConfigurationsAsync(); @@ -432,7 +508,7 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers var configuration = configurations.OrderBy(c => c.Id).FirstOrDefault(); if (configuration is null) { - return new RetryPolicy(FallbackMaxRetries, FallbackRetryDelay); + return (new RetryPolicy(FallbackMaxRetries, FallbackRetryDelay), null); } if (configurations.Count > 1) @@ -443,8 +519,31 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers configuration.Id); } - return ClampRetryPolicy( - nameof(SmtpConfiguration), configuration.MaxRetries, configuration.RetryDelay); + int? emailConcurrencyLimit = null; + if (configuration.MaxConcurrentConnections > 0 + && TryParseTransport(configuration.Transport, out var transport) + && transport == EmailTransport.Smtp) + { + emailConcurrencyLimit = configuration.MaxConcurrentConnections; + } + + return ( + ClampRetryPolicy(nameof(SmtpConfiguration), configuration.MaxRetries, configuration.RetryDelay), + emailConcurrencyLimit); + } + + private static bool TryParseTransport(string? value, out EmailTransport transport) + { + try + { + transport = EmailTransportParser.Parse(value); + return true; + } + catch (ArgumentException) + { + transport = default; + return false; + } } /// @@ -563,7 +662,6 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers /// private async Task DeliverOneAsync( Notification notification, - DateTimeOffset now, RetryPolicies policies, INotificationOutboxRepository outboxRepository, INotificationDeliveryAdapter? adapter, @@ -581,16 +679,17 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers // attempt that resolved to a terminal park. Emit Attempted then the // terminal Parked row, both carrying the same explanatory error. var missingAdapterError = $"no delivery adapter for type {notification.Type}"; + var parkedAt = DateTimeOffset.UtcNow; notification.Status = NotificationStatus.Parked; notification.LastError = missingAdapterError; - notification.LastAttemptAt = now; + notification.LastAttemptAt = parkedAt; await WarnIfVanishedAsync(outboxRepository, notification, cancellationToken); await EmitAttemptAuditAsync( notification, - now, + parkedAt, durationMs: 0, errorMessage: missingAdapterError); - await EmitTerminalAuditAsync(notification, now, errorMessage: missingAdapterError); + await EmitTerminalAuditAsync(notification, parkedAt, errorMessage: missingAdapterError); return; } @@ -600,21 +699,28 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers // in-flight SMTP send instead of waiting for the SMTP connect/auth/send timeout. var attemptStart = DateTimeOffset.UtcNow; var outcome = await adapter.DeliverAsync(notification, cancellationToken); + + // Stamp the row with the ATTEMPT instant, captured after the adapter returns — not + // the sweep-start instant (review 2026-10-21 06/COR-1). With a slow relay the tail + // of a sweep completes minutes after the sweep began; stamping sweep-start skewed + // DeliveredAt (audit truth and the delivered-last-interval KPI) and scheduled + // NextAttemptAt in the past, collapsing the retry delay to zero. + var attemptAt = DateTimeOffset.UtcNow; var durationMs = (int)Math.Min( - int.MaxValue, Math.Max(0, (DateTimeOffset.UtcNow - attemptStart).TotalMilliseconds)); + int.MaxValue, Math.Max(0, (attemptAt - attemptStart).TotalMilliseconds)); switch (outcome.Result) { case DeliveryResult.Success: notification.Status = NotificationStatus.Delivered; - notification.DeliveredAt = now; - notification.LastAttemptAt = now; + notification.DeliveredAt = attemptAt; + notification.LastAttemptAt = attemptAt; notification.ResolvedTargets = outcome.ResolvedTargets; notification.LastError = null; break; case DeliveryResult.TransientFailure: - notification.LastAttemptAt = now; + notification.LastAttemptAt = attemptAt; notification.RetryCount++; notification.LastError = outcome.Error; if (notification.RetryCount >= maxRetries) @@ -624,13 +730,13 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers else { notification.Status = NotificationStatus.Retrying; - notification.NextAttemptAt = now + retryDelay; + notification.NextAttemptAt = attemptAt + retryDelay; } break; case DeliveryResult.PermanentFailure: notification.Status = NotificationStatus.Parked; - notification.LastAttemptAt = now; + notification.LastAttemptAt = attemptAt; notification.LastError = outcome.Error; break; } @@ -643,7 +749,7 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers // row carries the SMTP failure reason verbatim. await EmitAttemptAuditAsync( notification, - now, + attemptAt, durationMs: durationMs, errorMessage: outcome.Result == DeliveryResult.Success ? null : outcome.Error); @@ -656,7 +762,7 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers { await EmitTerminalAuditAsync( notification, - now, + attemptAt, errorMessage: outcome.Result == DeliveryResult.Success ? null : outcome.Error); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxOptions.cs b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxOptions.cs index 45220c76..20ac0106 100644 --- a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxOptions.cs @@ -19,6 +19,19 @@ public class NotificationOutboxOptions /// public int DispatchBatchSize { get; set; } = 100; + /// + /// Wall-clock budget for one dispatch sweep. Once it is spent, claimed notifications + /// that have not yet started their delivery attempt are skipped — left untouched, so + /// they stay due and the next sweep re-claims them alongside any newer rows — instead + /// of extending the sweep. Attempts already in flight always run to completion. Without + /// it a batch of slow (timing-out) deliveries held the single-flight sweep for + /// DispatchBatchSize / MaxParallelDeliveries x timeout (~12.5 min at the + /// defaults), during which no fresh notification could be claimed + /// (review 2026-10-21 06/PERF-5). + /// Default: 2 minutes. + /// + public TimeSpan DispatchSweepBudget { get; set; } = TimeSpan.FromMinutes(2); + /// /// Age past which a still-Pending/Retrying notification is counted as /// "stuck" on the KPI tile and the per-row badge in the Central UI. @@ -57,6 +70,10 @@ public class NotificationOutboxOptions /// (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 . + /// For Email over the SMTP transport the per-sweep Email parallelism is further clamped + /// to the active SMTP row's MaxConcurrentConnections (when positive), so a relay + /// that limits connections is honoured; the EWS transport and SMS are not clamped + /// (review 2026-10-21 06/IMPL-2). /// Default: 4. /// public int MaxParallelDeliveries { get; set; } = 4; diff --git a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxOptionsValidator.cs index bebeb0b7..4878d449 100644 --- a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/NotificationOutboxOptionsValidator.cs @@ -31,6 +31,11 @@ public sealed class NotificationOutboxOptionsValidator : OptionsValidatorBase TimeSpan.Zero, + $"ScadaBridge:NotificationOutbox:{nameof(NotificationOutboxOptions.DispatchSweepBudget)} " + + $"({options.DispatchSweepBudget}) must be a positive duration; it is the wall-clock budget after " + + "which a sweep stops starting new delivery attempts."); + builder.RequireThat(options.StuckAgeThreshold > TimeSpan.Zero, $"ScadaBridge:NotificationOutbox:{nameof(NotificationOutboxOptions.StuckAgeThreshold)} " + $"({options.StuckAgeThreshold}) must be a positive duration; it is the age past which a " + diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/RepositoryCoverageTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/RepositoryCoverageTests.cs index 40915e3c..e0dd8b40 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/RepositoryCoverageTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/RepositoryCoverageTests.cs @@ -567,6 +567,71 @@ public class NotificationOutboxRepositoryTests : IDisposable Assert.Equal(retryingDue.NotificationId, capped[1].NotificationId); } + /// + /// Verifies that a retrying backlog older than every fresh Pending row cannot monopolise + /// the claim: the batch is split between the two statuses, each oldest-first + /// (review 2026-10-21 06/PERF-5). + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task GetDueAsync_RetryingBacklog_SharesBatchWithPending() + { + var now = new DateTimeOffset(2026, 5, 19, 12, 0, 0, TimeSpan.Zero); + var retrying = Enumerable.Range(0, 4) + .Select(i => MakeNotification(Guid.NewGuid().ToString(), NotificationStatus.Retrying, + createdAt: now.AddHours(-10).AddMinutes(i), nextAttemptAt: now.AddMinutes(-1))) + .ToList(); + var pending = Enumerable.Range(0, 4) + .Select(i => MakeNotification(Guid.NewGuid().ToString(), NotificationStatus.Pending, + createdAt: now.AddMinutes(-10 + i))) + .ToList(); + _context.Notifications.AddRange(retrying); + _context.Notifications.AddRange(pending); + await _context.SaveChangesAsync(); + _context.ChangeTracker.Clear(); + + var due = await _repository.GetDueAsync(now, batchSize: 4); + + Assert.Equal( + new[] + { + retrying[0].NotificationId, retrying[1].NotificationId, + pending[0].NotificationId, pending[1].NotificationId, + }, + due.Select(n => n.NotificationId).ToArray()); + } + + /// + /// Verifies that an unused share of the batch is back-filled from the other status, so a + /// sweep with a short Retrying (or Pending) set still claims a full batch. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task GetDueAsync_ShortRetryingSet_BackfillsFromPending() + { + var now = new DateTimeOffset(2026, 5, 19, 12, 0, 0, TimeSpan.Zero); + var retrying = MakeNotification(Guid.NewGuid().ToString(), NotificationStatus.Retrying, + createdAt: now.AddHours(-10), nextAttemptAt: now.AddMinutes(-1)); + var pending = Enumerable.Range(0, 5) + .Select(i => MakeNotification(Guid.NewGuid().ToString(), NotificationStatus.Pending, + createdAt: now.AddMinutes(-10 + i))) + .ToList(); + _context.Notifications.Add(retrying); + _context.Notifications.AddRange(pending); + await _context.SaveChangesAsync(); + _context.ChangeTracker.Clear(); + + var due = await _repository.GetDueAsync(now, batchSize: 4); + + Assert.Equal( + new[] + { + retrying.NotificationId, + pending[0].NotificationId, pending[1].NotificationId, pending[2].NotificationId, + }, + due.Select(n => n.NotificationId).ToArray()); + } + /// Verifies that get by id returns the row when present and null otherwise. /// A task that represents the asynchronous operation. [Fact] diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorSweepPacingTests.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorSweepPacingTests.cs new file mode 100644 index 00000000..464653c9 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxActorSweepPacingTests.cs @@ -0,0 +1,265 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Notifications; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.NotificationOutbox.Delivery; +using ZB.MOM.WW.ScadaBridge.NotificationOutbox.Messages; +using ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests.TestSupport; + +namespace ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests; + +/// +/// Regression tests for the dispatch sweep's pacing (review 2026-10-21): per-attempt +/// timestamps (06/COR-1), the per-sweep wall-clock budget (06/PERF-5), and the Email +/// concurrency clamp to the SMTP row's MaxConcurrentConnections (06/IMPL-2). +/// +public class NotificationOutboxActorSweepPacingTests : TestKit +{ + private readonly INotificationOutboxRepository _outboxRepository = + OutboxRepositorySubstitute.Healthy(); + + private readonly INotificationRepository _notificationRepository = + Substitute.For(); + + private IActorRef CreateActor( + IEnumerable adapters, NotificationOutboxOptions options) + { + var services = new ServiceCollection(); + services.AddScoped(_ => _outboxRepository); + services.AddScoped(_ => _notificationRepository); + foreach (var adapter in adapters) + { + services.AddScoped(_ => adapter); + } + + return Sys.ActorOf(Props.Create(() => new NotificationOutboxActor( + services.BuildServiceProvider(), + options, + new NoOpCentralAuditWriter(), + NullLogger.Instance))); + } + + private static Notification MakeNotification(NotificationType type = NotificationType.Email) => + new(Guid.NewGuid().ToString(), type, "ops-team", "Subject", "Body", "site-1") + { + CreatedAt = DateTimeOffset.UtcNow, + }; + + private void SetupSmtp(int maxConcurrentConnections = 0, string? transport = null) + { + var config = new SmtpConfiguration("smtp.example.com", "Basic", "noreply@example.com") + { + MaxRetries = 5, + RetryDelay = TimeSpan.FromMinutes(1), + MaxConcurrentConnections = maxConcurrentConnections, + Transport = transport, + }; + _notificationRepository.GetAllSmtpConfigurationsAsync(Arg.Any()) + .Returns(new[] { config }); + } + + private void SetupDueBatchOnce(params Notification[] batch) + { + var claimed = 0; + _outboxRepository.GetDueAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(_ => Interlocked.Increment(ref claimed) == 1 + ? batch + : Array.Empty()); + } + + /// + /// Adapter that delays every delivery, records the concurrency high-water mark and the + /// instant each delivery completed. + /// + private sealed class PacedAdapter( + TimeSpan delay, Func outcome, NotificationType type = NotificationType.Email) + : INotificationDeliveryAdapter + { + private int _current; + private int _highWater; + private int _calls; + + public NotificationType Type => type; + + public int HighWater => Volatile.Read(ref _highWater); + + public int Calls => Volatile.Read(ref _calls); + + public DateTimeOffset LastCompletedAt; + + public async Task DeliverAsync( + Notification notification, CancellationToken cancellationToken = default) + { + Interlocked.Increment(ref _calls); + 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 + } + + await Task.Delay(delay, cancellationToken); + Interlocked.Decrement(ref _current); + LastCompletedAt = DateTimeOffset.UtcNow; + return outcome(); + } + } + + /// + /// 06/COR-1: a delivery that completes well after the sweep began is stamped with the + /// attempt instant, not the sweep-start instant. + /// + [Fact] + public void Success_StampsDeliveredAt_AfterTheAdapterReturns() + { + SetupSmtp(); + var notification = MakeNotification(); + SetupDueBatchOnce(notification); + var adapter = new PacedAdapter(TimeSpan.FromMilliseconds(400), () => DeliveryOutcome.Success("ops@x")); + var actor = CreateActor([adapter], new NotificationOutboxOptions { DispatchInterval = TimeSpan.FromHours(1) }); + var before = DateTimeOffset.UtcNow; + + actor.Tell(InternalMessages.DispatchTick.Instance); + + AwaitAssert(() => _outboxRepository.Received(1).UpdateAsync( + Arg.Is(n => n.Status == NotificationStatus.Delivered), + Arg.Any())); + Assert.True(notification.DeliveredAt >= before + TimeSpan.FromMilliseconds(350), + $"DeliveredAt {notification.DeliveredAt:O} should be after the 400 ms attempt (sweep began {before:O})"); + Assert.Equal(notification.DeliveredAt, notification.LastAttemptAt); + } + + /// + /// 06/COR-1: a transient failure schedules the retry relative to the attempt instant, so a + /// slow attempt cannot collapse the retry delay. + /// + [Fact] + public void TransientFailure_SchedulesNextAttempt_FromTheAttemptInstant() + { + SetupSmtp(); + var notification = MakeNotification(); + SetupDueBatchOnce(notification); + var adapter = new PacedAdapter(TimeSpan.FromMilliseconds(400), () => DeliveryOutcome.Transient("timeout")); + var actor = CreateActor([adapter], new NotificationOutboxOptions { DispatchInterval = TimeSpan.FromHours(1) }); + var before = DateTimeOffset.UtcNow; + + actor.Tell(InternalMessages.DispatchTick.Instance); + + AwaitAssert(() => _outboxRepository.Received(1).UpdateAsync( + Arg.Is(n => n.Status == NotificationStatus.Retrying), + Arg.Any())); + Assert.True(notification.LastAttemptAt >= before + TimeSpan.FromMilliseconds(350)); + Assert.Equal(notification.LastAttemptAt + TimeSpan.FromMinutes(1), notification.NextAttemptAt); + } + + /// + /// 06/PERF-5: once the sweep's wall-clock budget is spent, deliveries still waiting for a + /// permit are skipped (left due for the next sweep) instead of extending the sweep. + /// + [Fact] + public void SweepBudget_Exhausted_SkipsDeliveriesThatHaveNotStarted() + { + SetupSmtp(); + SetupDueBatchOnce(MakeNotification(), MakeNotification(), MakeNotification()); + var adapter = new PacedAdapter(TimeSpan.FromMilliseconds(600), () => DeliveryOutcome.Success("ops@x")); + var actor = CreateActor([adapter], new NotificationOutboxOptions + { + DispatchInterval = TimeSpan.FromHours(1), + MaxParallelDeliveries = 1, + DispatchSweepBudget = TimeSpan.FromMilliseconds(200), + }); + + actor.Tell(InternalMessages.DispatchTick.Instance); + + AwaitAssert(() => _outboxRepository.Received(1).UpdateAsync( + Arg.Any(), Arg.Any())); + // Give the (skipped) rest of the sweep ample time to have run had it not been skipped. + Thread.Sleep(1500); + Assert.Equal(1, adapter.Calls); + _outboxRepository.Received(1).UpdateAsync(Arg.Any(), Arg.Any()); + } + + /// + /// 06/IMPL-2: Email deliveries over the SMTP transport are clamped to the SMTP row's + /// MaxConcurrentConnections even when MaxParallelDeliveries is higher. + /// + [Fact] + public void SmtpTransport_ClampsEmailParallelism_ToMaxConcurrentConnections() + { + SetupSmtp(maxConcurrentConnections: 1); + SetupDueBatchOnce(Enumerable.Range(0, 4).Select(_ => MakeNotification()).ToArray()); + var adapter = new PacedAdapter(TimeSpan.FromMilliseconds(150), () => DeliveryOutcome.Success("ops@x")); + var actor = CreateActor([adapter], new NotificationOutboxOptions + { + DispatchInterval = TimeSpan.FromHours(1), + MaxParallelDeliveries = 4, + }); + + actor.Tell(InternalMessages.DispatchTick.Instance); + + AwaitAssert( + () => _outboxRepository.Received(4).UpdateAsync( + Arg.Is(n => n.Status == NotificationStatus.Delivered), + Arg.Any()), + duration: TimeSpan.FromSeconds(10)); + Assert.Equal(1, adapter.HighWater); + } + + /// + /// 06/IMPL-2: the EWS transport does not use MaxConcurrentConnections, so its deliveries + /// keep the full MaxParallelDeliveries. + /// + [Fact] + public void EwsTransport_IsNotClampedByMaxConcurrentConnections() + { + SetupSmtp(maxConcurrentConnections: 1, transport: "Ews"); + SetupDueBatchOnce(Enumerable.Range(0, 4).Select(_ => MakeNotification()).ToArray()); + var adapter = new PacedAdapter(TimeSpan.FromMilliseconds(400), () => DeliveryOutcome.Success("ops@x")); + var actor = CreateActor([adapter], new NotificationOutboxOptions + { + DispatchInterval = TimeSpan.FromHours(1), + MaxParallelDeliveries = 4, + }); + + actor.Tell(InternalMessages.DispatchTick.Instance); + + AwaitAssert( + () => _outboxRepository.Received(4).UpdateAsync( + Arg.Is(n => n.Status == NotificationStatus.Delivered), + Arg.Any()), + duration: TimeSpan.FromSeconds(10)); + Assert.True(adapter.HighWater >= 2, $"expected concurrent EWS deliveries, got {adapter.HighWater}"); + } + + /// + /// 06/IMPL-2: the SMTP clamp applies to Email only — SMS deliveries in the same sweep keep + /// the full MaxParallelDeliveries. + /// + [Fact] + public void SmtpClamp_DoesNotThrottleSms() + { + SetupSmtp(maxConcurrentConnections: 1); + SetupDueBatchOnce(Enumerable.Range(0, 4).Select(_ => MakeNotification(NotificationType.Sms)).ToArray()); + var sms = new PacedAdapter( + TimeSpan.FromMilliseconds(400), () => DeliveryOutcome.Success("+1555"), NotificationType.Sms); + var actor = CreateActor([sms], new NotificationOutboxOptions + { + DispatchInterval = TimeSpan.FromHours(1), + MaxParallelDeliveries = 4, + }); + + actor.Tell(InternalMessages.DispatchTick.Instance); + + AwaitAssert( + () => _outboxRepository.Received(4).UpdateAsync( + Arg.Is(n => n.Status == NotificationStatus.Delivered), + Arg.Any()), + duration: TimeSpan.FromSeconds(10)); + Assert.True(sms.HighWater >= 2, $"expected concurrent SMS deliveries, got {sms.HighWater}"); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxOptionsValidatorTests.cs index 311e026e..2a467ee8 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxOptionsValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/NotificationOutboxOptionsValidatorTests.cs @@ -98,4 +98,13 @@ public class NotificationOutboxOptionsValidatorTests Assert.True(result.Failed); Assert.Contains("MaxParallelDeliveries", result.FailureMessage); } + + /// Verifies that a zero dispatch sweep budget is rejected (review 2026-10-21 06/PERF-5). + [Fact] + public void ZeroDispatchSweepBudget_IsRejected() + { + var result = Validate(new NotificationOutboxOptions { DispatchSweepBudget = TimeSpan.Zero }); + Assert.True(result.Failed); + Assert.Contains("DispatchSweepBudget", result.FailureMessage); + } }