fix(asb): de-quadratic session listeners, surface MaxConcurrentCalls, settle the original on a batched defer (GH-3494)

AZURESERVICEBUS-PERF-DEEP-DIVE-PLAN.md

@@ -160,7 +160,19 @@ feed the #3488/GH-3471 release notes directly** — that's the first ledger entr
 
 | Optimization | PR | Scenario (AE-cell) | Metric | Before | After | Release-note one-liner |
 |---|---|---|---|---|---|---|
-| _(PrefetchCount validation (#3488), AO2 session de-quadratic, AO4 settlement concurrency, ... — rows added as measured)_ | | | | | | |
+| AO2 session de-quadratic (accept-loop listener) | this wave | AE5 | concurrent accept loops at `RequireSessions(5)` | 25 | 5 | `RequireSessions(n)` opens n session accept loops instead of n-squared |
+| AO2 session de-quadratic (`ServiceBusSessionProcessor` path, GH-3533) | this wave | AE5 | concurrent sessions at `RequireSessions(8)` | 64 | 8 | Same n-squared on the session-processor path: `MaxConcurrentSessions` was set per processor while ListeningAgent was already building n of them |
+| AO3 `MaximumConcurrentCalls` | this wave | AE1 | inline handler concurrency | 1 (SDK default, unreachable) | configurable | Inline Azure Service Bus listeners can now process messages concurrently without dropping to the raw `ConfigureProcessor` hook |
+| AO8 batched defer settles the original | this wave | AE8 | duplicates per deferral | 2 deliveries | 1 | Deferring on a buffered/durable Azure Service Bus listener no longer leaves the original message locked to be redelivered |
+
+**Not measured on a real namespace.** AO2's 25→5 is structural and unit-tested; AO3 and AO8 are
+correctness/ergonomics fixes verified by tests. The throughput numbers this plan asks for -- and
+the PrefetchCount validation in AE2 -- still need the real Standard/Premium namespace runs
+described in §2 and §8. The emulator is explicitly not publishable.
+
+**Deferred, needs a JasperFx change:** AO4 (settlement concurrency). `RetryBlock` hard-codes a
+parallel count of 1 (`Block<Item>(1, Unbounded, executeAsync)`), so widening ASB's per-message
+`CompleteMessageAsync` concurrency is not a Wolverine-side change. Track upstream before revisiting.
 
 ## 7. Sequencing & exit criteria
 

docs/guide/messaging/transports/azureservicebus/performance.md

@@ -36,15 +36,23 @@ means silent redelivery and a rising delivery count.
 ### Inline endpoints process one message at a time by default
 
 Inline ASB endpoints use a `ServiceBusProcessor`, whose `MaxConcurrentCalls` defaults to **1**.
-Wolverine does not change that default, so an inline listener is single-threaded unless you
-raise it:
+Wolverine leaves that default alone, so an inline listener is single-threaded unless you raise it:
 
 ```cs
 opts.ListenToAzureServiceBusQueue("orders")
     .ProcessInline()
-    .ConfigureProcessor(o => o.MaxConcurrentCalls = 10);
+    .MaximumConcurrentCalls(10);
 ```
 
+`MaximumParallelMessages` has no effect on an inline endpoint — that knob sizes Wolverine's own
+in-process worker queue, which inline listeners bypass. The raw
+`ConfigureProcessor(o => o.MaxConcurrentCalls = 10)` hook still works and takes precedence.
+
+On a **session** listener driven by a `ServiceBusSessionProcessor`, `MaximumConcurrentCalls` maps
+to `MaxConcurrentCallsPerSession` instead, which trades away the per-session FIFO ordering that is
+usually the reason for using sessions at all. Leave it alone there unless you mean it — to process
+more sessions at once, use `RequireSessions(n)`.
+
 ## Lock duration vs. processing window
 
 For Buffered/Durable endpoints, Wolverine does not renew message locks while messages wait in
@@ -86,6 +94,7 @@ outgoing batches are additionally grouped by session id so each batch shares a p
 Sessions give broker-enforced ordering per `SessionId` (mapped automatically from Wolverine's
 `Envelope.GroupId`) with cluster-wide exclusivity — but session processing is inherently more
 expensive than plain consumption: each session must be accepted, locked, drained, and released.
+`RequireSessions(n)` opens exactly `n` concurrent session accept loops — one per listener.
 Keep `RequireSessions(n)` counts modest, and note that strict per-session *processing* order on
 Buffered/Durable endpoints also needs `PartitionProcessingByGroupId(...)` (or inline
 execution), since the local worker queue otherwise executes a session's batch in parallel —

src/Transports/Azure/Wolverine.AzureServiceBus.Tests/configuring_session_processor_options.cs

@@ -77,7 +77,7 @@ public void build_session_processor_options_reasserts_peek_lock_and_disables_aut
     }
 
     [Fact]
-    public void build_session_processor_options_maps_listener_count_to_max_concurrent_sessions()
+    public void build_session_processor_options_accepts_one_session_per_processor()
     {
         var transport = new AzureServiceBusTransport();
         var queue = transport.Queues["incoming"];
@@ -88,12 +88,30 @@ public void build_session_processor_options_maps_listener_count_to_max_concurren
         ((IDelayedEndpointConfiguration)configuration).Apply();
 
         var options = AzureServiceBusTransport.BuildSessionProcessorOptions(queue);
-        options.MaxConcurrentSessions.ShouldBe(8);
+
+        // GH-3494 (AO2): ListeningAgent builds ListenerCount of these listeners, so mapping
+        // RequireSessions(8) onto each processor's MaxConcurrentSessions gave 8 x 8 = 64
+        // concurrent sessions. One per processor keeps the total at the 8 that was asked for.
+        options.MaxConcurrentSessions.ShouldBe(1);
 
         // FIFO ordering per session is preserved
         options.MaxConcurrentCallsPerSession.ShouldBe(1);
     }
 
+    [Fact]
+    public void a_user_can_still_override_max_concurrent_sessions()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+
+        var configuration = new AzureServiceBusQueueListenerConfiguration(queue);
+        configuration.RequireSessions(8).ConfigureSessionProcessor(o => o.MaxConcurrentSessions = 3);
+
+        ((IDelayedEndpointConfiguration)configuration).Apply();
+
+        AzureServiceBusTransport.BuildSessionProcessorOptions(queue).MaxConcurrentSessions.ShouldBe(3);
+    }
+
     [Fact]
     public void build_session_processor_options_composes_multiple_actions()
     {

src/Transports/Azure/Wolverine.AzureServiceBus.Tests/maximum_concurrent_calls_3494.cs

@@ -0,0 +1,98 @@
+using Shouldly;
+using Wolverine.AzureServiceBus.Internal;
+using Wolverine.Configuration;
+using Xunit;
+
+namespace Wolverine.AzureServiceBus.Tests;
+
+/// <summary>
+/// GH-3494 (AO3). Wolverine never set MaxConcurrentCalls, so an inline Azure Service Bus listener
+/// ran on the SDK default of 1 -- one message at a time per endpoint -- reachable only through the
+/// raw ConfigureProcessor hook.
+/// </summary>
+public class maximum_concurrent_calls_3494
+{
+    [Fact]
+    public void defaults_to_null_so_the_sdk_default_still_applies()
+    {
+        var transport = new AzureServiceBusTransport();
+        transport.Queues["incoming"].MaximumConcurrentCalls.ShouldBeNull();
+    }
+
+    [Fact]
+    public void must_be_at_least_one()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+
+        Should.Throw<ArgumentOutOfRangeException>(() => queue.MaximumConcurrentCalls = 0);
+    }
+
+    [Fact]
+    public void queue_listener_configuration_sets_it()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+
+        var configuration = new AzureServiceBusQueueListenerConfiguration(queue);
+        configuration.MaximumConcurrentCalls(12);
+        ((IDelayedEndpointConfiguration)configuration).Apply();
+
+        queue.MaximumConcurrentCalls.ShouldBe(12);
+    }
+
+    [Fact]
+    public void subscription_listener_configuration_sets_it()
+    {
+        var transport = new AzureServiceBusTransport();
+        var subscription = transport.Topics["topic1"].FindOrCreateSubscription("sub1");
+
+        var configuration = new AzureServiceBusSubscriptionListenerConfiguration(subscription);
+        configuration.MaximumConcurrentCalls(4);
+        ((IDelayedEndpointConfiguration)configuration).Apply();
+
+        subscription.MaximumConcurrentCalls.ShouldBe(4);
+    }
+
+    [Fact]
+    public void flows_onto_the_processor_options()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+        queue.MaximumConcurrentCalls = 8;
+
+        AzureServiceBusTransport.BuildProcessorOptions(queue).MaxConcurrentCalls.ShouldBe(8);
+    }
+
+    [Fact]
+    public void configure_processor_still_wins_over_the_endpoint_value()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+        queue.MaximumConcurrentCalls = 8;
+        queue.ConfigureProcessor = o => o.MaxConcurrentCalls = 3;
+
+        AzureServiceBusTransport.BuildProcessorOptions(queue).MaxConcurrentCalls.ShouldBe(3);
+    }
+
+    [Fact]
+    public void session_processor_keeps_per_session_fifo_unless_asked_otherwise()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+        queue.ConfigureSessionProcessor = _ => { };
+
+        AzureServiceBusTransport.BuildSessionProcessorOptions(queue).MaxConcurrentCallsPerSession.ShouldBe(1);
+    }
+
+    [Fact]
+    public void session_processor_honors_the_opt_in()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+        queue.MaximumConcurrentCalls = 5;
+        queue.ConfigureSessionProcessor = _ => { };
+
+        AzureServiceBusTransport.BuildSessionProcessorOptions(queue).MaxConcurrentCallsPerSession.ShouldBe(5);
+    }
+}

src/Transports/Azure/Wolverine.AzureServiceBus.Tests/session_listener_accept_loops_3494.cs

@@ -0,0 +1,44 @@
+using Azure.Messaging.ServiceBus;
+using Microsoft.Extensions.Logging.Abstractions;
+using NSubstitute;
+using Shouldly;
+using Wolverine.AzureServiceBus.Internal;
+using Wolverine.Transports;
+using Wolverine.Transports.Sending;
+using Xunit;
+
+namespace Wolverine.AzureServiceBus.Tests;
+
+/// <summary>
+/// GH-3494 (AO2). ListeningAgent already builds Endpoint.ListenerCount listeners, so the session
+/// listener spawning ListenerCount accept loops of its own made RequireSessions(n) open n-squared
+/// concurrent AcceptNextSessionAsync loops -- 25 of them for RequireSessions(5).
+/// </summary>
+public class session_listener_accept_loops_3494
+{
+    private static AzureServiceBusSessionListener buildListener(int listenerCount)
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+        queue.Options.RequiresSession = true;
+        queue.ListenerCount = listenerCount;
+
+        // The accept loop itself immediately fails against a transport with no live connection and
+        // backs off inside its own catch, which is all this test needs -- it only asserts on how
+        // many loops were started.
+        return new AzureServiceBusSessionListener(transport, queue, Substitute.For<IReceiver>(),
+            Substitute.For<IEnvelopeMapper<ServiceBusReceivedMessage, ServiceBusMessage>>(), NullLogger.Instance,
+            Substitute.For<ISender>());
+    }
+
+    [Theory]
+    [InlineData(0)]
+    [InlineData(1)]
+    [InlineData(5)]
+    [InlineData(10)]
+    public async Task always_runs_exactly_one_accept_loop_per_listener(int listenerCount)
+    {
+        await using var listener = buildListener(listenerCount);
+        listener.AcceptLoopCount.ShouldBe(1);
+    }
+}

src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusQueueListenerConfiguration.cs

@@ -170,6 +170,22 @@ public AzureServiceBusQueueListenerConfiguration PrefetchCount(int prefetchCount
         return this;
     }
 
+    /// <summary>
+    ///     How many messages an <c>Inline</c> listener for this queue processes concurrently. This
+    ///     is the Azure Service Bus SDK's <c>MaxConcurrentCalls</c>, which Wolverine left at the SDK
+    ///     default of 1 -- so an inline Azure Service Bus listener consumed strictly one message at
+    ///     a time per endpoint. On a session listener driven by a <c>ServiceBusSessionProcessor</c>
+    ///     this sets <c>MaxConcurrentCallsPerSession</c> instead, which trades away the per-session
+    ///     FIFO ordering, so leave it alone on session endpoints unless you mean it. Has no effect
+    ///     on the default Buffered/Durable batch receive loop. See GH-3494.
+    /// </summary>
+    /// <param name="concurrency">Concurrent handler invocations. Must be at least 1</param>
+    public AzureServiceBusQueueListenerConfiguration MaximumConcurrentCalls(int concurrency)
+    {
+        add(e => e.MaximumConcurrentCalls = concurrency);
+        return this;
+    }
+
     /// <summary>
     /// Completely disable all SQS dead letter queueing for just this queue
     /// </summary>

src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusSubscriptionListenerConfiguration.cs

@@ -193,6 +193,21 @@ public AzureServiceBusSubscriptionListenerConfiguration PrefetchCount(int prefet
         return this;
     }
 
+    /// <summary>
+    ///     How many messages an <c>Inline</c> listener for this subscription processes concurrently.
+    ///     This is the Azure Service Bus SDK's <c>MaxConcurrentCalls</c>, which Wolverine left at
+    ///     the SDK default of 1 -- so an inline listener consumed strictly one message at a time per
+    ///     endpoint. On a session listener driven by a <c>ServiceBusSessionProcessor</c> this sets
+    ///     <c>MaxConcurrentCallsPerSession</c> instead, which trades away the per-session FIFO
+    ///     ordering. Has no effect on the default Buffered/Durable batch receive loop. See GH-3494.
+    /// </summary>
+    /// <param name="concurrency">Concurrent handler invocations. Must be at least 1</param>
+    public AzureServiceBusSubscriptionListenerConfiguration MaximumConcurrentCalls(int concurrency)
+    {
+        add(e => e.MaximumConcurrentCalls = concurrency);
+        return this;
+    }
+
     /// <summary>
     /// Force this subscription listener to require session identifiers. Use this for FIFO semantics
     /// </summary>

src/Transports/Azure/Wolverine.AzureServiceBus/AzureServiceBusTransport.Listening.cs

@@ -190,6 +190,14 @@ internal static ServiceBusProcessorOptions BuildProcessorOptions(AzureServiceBus
             PrefetchCount = endpoint.PrefetchCount
         };
 
+        // GH-3494 (AO3): Wolverine never set MaxConcurrentCalls, so an inline listener ran on the
+        // SDK default of 1 -- single threaded per endpoint, reachable only through the raw
+        // ConfigureProcessor hook. Applied before ConfigureProcessor so that hook still wins.
+        if (endpoint.MaximumConcurrentCalls.HasValue)
+        {
+            options.MaxConcurrentCalls = endpoint.MaximumConcurrentCalls.Value;
+        }
+
         endpoint.ConfigureProcessor?.Invoke(options);
 
         // Reserved by Wolverine: the inline listener relies on the peek-lock model to complete,
@@ -229,14 +237,25 @@ internal static ServiceBusSessionProcessorOptions BuildSessionProcessorOptions(A
         {
             PrefetchCount = endpoint.PrefetchCount,
 
-            // Map the existing "parallel sessions" knob (ListenerCount, set via RequireSessions(count))
-            // onto the processor's concurrency. A user may override this in ConfigureSessionProcessor.
-            MaxConcurrentSessions = endpoint.ListenerCount > 0 ? endpoint.ListenerCount : 1,
+            // GH-3494 (AO2): ONE session per processor. ListeningAgent already builds
+            // Endpoint.ListenerCount of these listeners, so mapping RequireSessions(n) onto each
+            // processor's MaxConcurrentSessions meant n listeners x n sessions = n-squared
+            // concurrent sessions -- 64 of them for RequireSessions(8). One per listener keeps the
+            // documented meaning of RequireSessions(n): n parallel sessions in total.
+            // A user may still override this in ConfigureSessionProcessor.
+            MaxConcurrentSessions = 1,
 
             // Preserve the in-session FIFO ordering the hand-rolled loop provided
             MaxConcurrentCallsPerSession = 1
         };
 
+        // GH-3494 (AO3): opt-in concurrency WITHIN a session. Left at 1 unless asked for, because
+        // raising it gives up the per-session FIFO ordering that is the whole point of sessions.
+        if (endpoint.MaximumConcurrentCalls.HasValue)
+        {
+            options.MaxConcurrentCallsPerSession = endpoint.MaximumConcurrentCalls.Value;
+        }
+
         endpoint.ConfigureSessionProcessor?.Invoke(options);
 
         // Reserved by Wolverine: the listener relies on the peek-lock model to explicitly complete,

src/Transports/Azure/Wolverine.AzureServiceBus/Internal/AzureServiceBusEndpoint.cs

@@ -84,6 +84,32 @@ public int PrefetchCount
         }
     }
 
+    private int? _maximumConcurrentCalls;
+
+    /// <summary>
+    ///     How many messages an <c>Inline</c> or session-processor listener for this endpoint hands
+    ///     to the handler pipeline at once. This maps to the Azure Service Bus SDK's
+    ///     <c>MaxConcurrentCalls</c>, which Wolverine never set -- so inline Azure Service Bus
+    ///     listeners ran strictly one message at a time on the SDK default, and the only way to
+    ///     change that was the raw <see cref="ConfigureProcessor" /> hook. Null keeps the SDK
+    ///     default of 1. Does not apply to the default Buffered/Durable batch receive loop, which
+    ///     scales through MaximumParallelMessages instead. See GH-3494.
+    /// </summary>
+    public int? MaximumConcurrentCalls
+    {
+        get => _maximumConcurrentCalls;
+        set
+        {
+            if (value is < 1)
+            {
+                throw new ArgumentOutOfRangeException(nameof(value), value,
+                    "MaximumConcurrentCalls must be at least 1");
+            }
+
+            _maximumConcurrentCalls = value;
+        }
+    }
+
     /// <summary>
     ///     Optional customization of the Azure Service Bus <see cref="ServiceBusProcessorOptions" /> used
     ///     by inline listeners for this endpoint. Wolverine reserves control of the properties it depends

src/Transports/Azure/Wolverine.AzureServiceBus/Internal/BatchedAzureServiceBusListener.cs

@@ -44,8 +44,20 @@ public BatchedAzureServiceBusListener(AzureServiceBusEndpoint endpoint, ILogger
         _complete = new RetryBlock<AzureServiceBusEnvelope>((e, _) => { return e.CompleteAsync(_cancellation.Token); },
             _logger, _cancellation.Token);
 
-        _defer = new RetryBlock<Envelope>(async (envelope, _) => { await _requeue.SendAsync(envelope); }, logger,
-            _cancellation.Token);
+        _defer = new RetryBlock<Envelope>(async (envelope, _) =>
+        {
+            // GH-3494 (AO8): settle the original before re-sending the copy, exactly like the
+            // inline listener already does. Leaving it unsettled meant the message stayed locked
+            // until the lock expired and Azure Service Bus redelivered it -- so every deferral
+            // produced a duplicate on top of the copy this block sends.
+            if (envelope is AzureServiceBusEnvelope e && !e.IsCompleted)
+            {
+                await e.CompleteAsync(_cancellation.Token);
+                e.IsCompleted = true;
+            }
+
+            await _requeue.SendAsync(envelope);
+        }, logger, _cancellation.Token);
 
         _deadLetter =
             new RetryBlock<AzureServiceBusEnvelope>(