fix(notification-outbox): per-attempt timestamps, Pending/Retrying split claim + sweep budget, SMTP MaxConcurrentConnections clamp (review 2026-10-21 06/COR-1, 06/PERF-5, 06/IMPL-2)

2026-10-22 09:22:10 -04:00
parent ebf4b96789
commit 811a561885
10 changed files with 547 additions and 37 deletions
@@ -32,7 +32,10 @@ public interface INotificationOutboxRepository
    /// Returns notifications ready for a delivery attempt: <c>Pending</c> rows, plus
    /// <c>Retrying</c> rows whose <c>NextAttemptAt</c> is at or before <paramref name="now"/>.
    /// Terminal rows are excluded. Ordered by <c>CreatedAt</c> ascending, capped at
-    /// <paramref name="batchSize"/>.
+    /// <paramref name="batchSize"/>. The batch is shared between the two statuses — due
+    /// <c>Retrying</c> rows take at most half, <c>Pending</c> 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).
    /// </summary>
    /// <param name="now">The current time for evaluating due retries.</param>
    /// <param name="batchSize">Maximum number of rows to return.</param>
@@ -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<Notification>();
+        }
+
+        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<Notification> DuePendingQuery() => _context.Notifications
+            .AsNoTracking()
+            .Where(n => n.Status == NotificationStatus.Pending)
+            .OrderBy(n => n.CreatedAt)
+            .ThenBy(n => n.NotificationId);
+
+        IQueryable<Notification> 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);
    }

    /// <inheritdoc />
@@ -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<Task>(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
    /// </remarks>
    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<INotificationOutboxRepository>();
            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();
        }
    }

+    /// <summary>
+    /// 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.
+    /// </summary>
+    private sealed class SweepBudget(TimeSpan budget)
+    {
+        private readonly System.Diagnostics.Stopwatch _clock = System.Diagnostics.Stopwatch.StartNew();
+        private int _skipped;
+
+        /// <summary>True once the sweep has run past its budget.</summary>
+        public bool IsExhausted => _clock.Elapsed >= budget;
+
+        /// <summary>The number of claimed notifications left unattempted.</summary>
+        public int Skipped => Volatile.Read(ref _skipped);
+
+        /// <summary>Records one claimed notification left unattempted.</summary>
+        public void MarkSkipped() => Interlocked.Increment(ref _skipped);
+    }
+
    /// <summary>
    /// 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
    /// <summary>
    /// The per-type retry policies resolved once per dispatch sweep: <see cref="SmtpPolicy"/>
    /// drives Email (and every non-SMS type), <see cref="SmsPolicy"/> drives SMS.
+    /// <see cref="EmailConcurrencyLimit"/> is the active SMTP row's positive
+    /// <c>MaxConcurrentConnections</c> when that row uses the SMTP transport, else null
+    /// (no Email-specific clamp).
    /// </summary>
-    private readonly record struct RetryPolicies(RetryPolicy SmtpPolicy, RetryPolicy SmsPolicy);
+    private readonly record struct RetryPolicies(
+        RetryPolicy SmtpPolicy, RetryPolicy SmsPolicy, int? EmailConcurrencyLimit);

    /// <summary>
    /// 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<RetryPolicies> 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);
    }

    /// <summary>
@@ -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.
    /// </summary>
-    private async Task<RetryPolicy> ResolveSmtpPolicyAsync(
+    /// <remarks>
+    /// Also resolves the Email concurrency clamp from the same row (review 2026-10-21
+    /// 06/IMPL-2): <c>MaxConcurrentConnections</c> 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.
+    /// </remarks>
+    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;
+        }
    }

    /// <summary>
@@ -563,7 +662,6 @@ public class NotificationOutboxActor : ReceiveActor, IWithTimers
    /// </remarks>
    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);
        }
    }
@@ -19,6 +19,19 @@ public class NotificationOutboxOptions
    /// </summary>
    public int DispatchBatchSize { get; set; } = 100;

+    /// <summary>
+    /// 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
+    /// <c>DispatchBatchSize / MaxParallelDeliveries x timeout</c> (~12.5 min at the
+    /// defaults), during which no fresh notification could be claimed
+    /// (review 2026-10-21 06/PERF-5).
+    /// Default: 2 minutes.
+    /// </summary>
+    public TimeSpan DispatchSweepBudget { get; set; } = TimeSpan.FromMinutes(2);
+
    /// <summary>
    /// Age past which a still-<c>Pending</c>/<c>Retrying</c> 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 <c>1</c> to preserve the strictly
    /// sequential behaviour. Clamped to a minimum of 1 via <see cref="ResolvedMaxParallelDeliveries"/>.
+    /// For Email over the SMTP transport the per-sweep Email parallelism is further clamped
+    /// to the active SMTP row's <c>MaxConcurrentConnections</c> (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.
    /// </summary>
    public int MaxParallelDeliveries { get; set; } = 4;
@@ -31,6 +31,11 @@ public sealed class NotificationOutboxOptionsValidator : OptionsValidatorBase<No
            $"ScadaBridge:NotificationOutbox:{nameof(NotificationOutboxOptions.DispatchBatchSize)} " +
            $"({options.DispatchBatchSize}) must be positive; it is the max rows claimed per sweep.");

+        builder.RequireThat(options.DispatchSweepBudget > 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 " +