perf(notification-outbox): bounded parallel delivery within a dispatch sweep (default 4)

2026-07-09 09:41:48 -04:00
parent acd8a6b8b5
commit 6567216e14
4 changed files with 216 additions and 26 deletions
@@ -567,4 +567,121 @@ public class NotificationOutboxActorDispatchTests : TestKit
                Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>()),
            duration: TimeSpan.FromSeconds(2));
    }
+
+    // ── Task 25: bounded parallel delivery ───────────────────────────────────
+
+    /// <summary>
+    /// Adapter that records the concurrent-invocation high-water mark. Each
+    /// <see cref="DeliverAsync"/> increments a live counter, notes the peak, then blocks
+    /// on a shared gate until <paramref name="releaseWhen"/> deliveries are concurrently
+    /// in flight — proving real overlap. A safety timeout releases the gate so a
+    /// parallelism-1 run (which can never reach <c>releaseWhen</c> concurrently) still
+    /// drains instead of deadlocking.
+    /// </summary>
+    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<DeliveryOutcome> 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<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(_ => Interlocked.Increment(ref claimed) == 1
+                ? batch
+                : Array.Empty<Notification>());
+    }
+
+    [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<Notification>(n => n.Status == NotificationStatus.Delivered),
+                    Arg.Any<CancellationToken>());
+                // 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<Notification>(n => n.Status == NotificationStatus.Delivered),
+                Arg.Any<CancellationToken>()),
+            duration: TimeSpan.FromSeconds(10));
+
+        // Strictly sequential: never more than one delivery in flight at once.
+        Assert.Equal(1, adapter.HighWater);
+    }
 }