GH-3533: ASB session-id pinning (5.x backport → 5.40.0)

Directory.Build.props

@@ -11,7 +11,7 @@
         <NoWarn>1570;1571;1572;1573;1574;1587;1591;1701;1702;1711;1735;0618;VSTHRD200</NoWarn>
         <ImplicitUsings>true</ImplicitUsings>
         <Nullable>enable</Nullable>
-        <Version>5.39.5</Version>
+        <Version>5.40.0</Version>
         <RepositoryUrl>$(PackageProjectUrl)</RepositoryUrl>
         <PublishRepositoryUrl>true</PublishRepositoryUrl>
         <EmbedUntrackedSources>true</EmbedUntrackedSources>

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

@@ -0,0 +1,133 @@
+using Azure.Messaging.ServiceBus;
+using JasperFx.Core;
+using Shouldly;
+using Wolverine.AzureServiceBus.Internal;
+using Wolverine.Configuration;
+using Xunit;
+
+namespace Wolverine.AzureServiceBus.Tests;
+
+public class configuring_session_processor_options
+{
+    [Fact]
+    public void configure_session_processor_stores_the_action_and_requires_sessions_on_the_queue()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+
+        var configuration = new AzureServiceBusQueueListenerConfiguration(queue);
+        configuration.ConfigureSessionProcessor(o => o.MaxConcurrentSessions = 4);
+
+        ((IDelayedEndpointConfiguration)configuration).Apply();
+
+        queue.ConfigureSessionProcessor.ShouldNotBeNull();
+        queue.Options.RequiresSession.ShouldBeTrue();
+    }
+
+    [Fact]
+    public void configure_session_processor_stores_the_action_and_requires_sessions_on_the_subscription()
+    {
+        var transport = new AzureServiceBusTransport();
+        var topic = transport.Topics["topic1"];
+        var subscription = topic.FindOrCreateSubscription("sub1");
+
+        var configuration = new AzureServiceBusSubscriptionListenerConfiguration(subscription);
+        configuration.ConfigureSessionProcessor(o => o.MaxConcurrentSessions = 4);
+
+        ((IDelayedEndpointConfiguration)configuration).Apply();
+
+        subscription.ConfigureSessionProcessor.ShouldNotBeNull();
+        subscription.Options.RequiresSession.ShouldBeTrue();
+    }
+
+    [Fact]
+    public void require_sessions_with_only_these_identifiers_populates_session_ids()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+
+        var configuration = new AzureServiceBusQueueListenerConfiguration(queue);
+        configuration.RequireSessionsWithOnlyTheseIdentifiers("A", "B");
+
+        ((IDelayedEndpointConfiguration)configuration).Apply();
+
+        queue.Options.RequiresSession.ShouldBeTrue();
+
+        var options = AzureServiceBusTransport.BuildSessionProcessorOptions(queue);
+        options.SessionIds.ShouldBe(new[] { "A", "B" });
+    }
+
+    [Fact]
+    public void build_session_processor_options_reasserts_peek_lock_and_disables_autocomplete()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+
+        // A user trying to break Wolverine's acknowledgement contract must not win
+        queue.ConfigureSessionProcessor = o =>
+        {
+            o.ReceiveMode = ServiceBusReceiveMode.ReceiveAndDelete;
+            o.AutoCompleteMessages = true;
+        };
+
+        var options = AzureServiceBusTransport.BuildSessionProcessorOptions(queue);
+
+        options.ReceiveMode.ShouldBe(ServiceBusReceiveMode.PeekLock);
+        options.AutoCompleteMessages.ShouldBeFalse();
+    }
+
+    [Fact]
+    public void build_session_processor_options_maps_listener_count_to_max_concurrent_sessions()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+
+        var configuration = new AzureServiceBusQueueListenerConfiguration(queue);
+        configuration.RequireSessions(8).ConfigureSessionProcessor(_ => { });
+
+        ((IDelayedEndpointConfiguration)configuration).Apply();
+
+        var options = AzureServiceBusTransport.BuildSessionProcessorOptions(queue);
+        options.MaxConcurrentSessions.ShouldBe(8);
+
+        // FIFO ordering per session is preserved
+        options.MaxConcurrentCallsPerSession.ShouldBe(1);
+    }
+
+    [Fact]
+    public void build_session_processor_options_composes_multiple_actions()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+
+        var configuration = new AzureServiceBusQueueListenerConfiguration(queue);
+
+        // The SessionIds sugar and an explicit customization must both apply
+        configuration
+            .RequireSessionsWithOnlyTheseIdentifiers("only-me")
+            .ConfigureSessionProcessor(o => o.MaxAutoLockRenewalDuration = 10.Minutes());
+
+        ((IDelayedEndpointConfiguration)configuration).Apply();
+
+        var options = AzureServiceBusTransport.BuildSessionProcessorOptions(queue);
+
+        options.SessionIds.ShouldBe(new[] { "only-me" });
+        options.MaxAutoLockRenewalDuration.ShouldBe(10.Minutes());
+    }
+
+    [Fact]
+    public void session_listener_defaults_to_the_legacy_loop_when_no_customization()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+
+        var configuration = new AzureServiceBusQueueListenerConfiguration(queue);
+        configuration.RequireSessions();
+
+        ((IDelayedEndpointConfiguration)configuration).Apply();
+
+        // Zero behavior change for existing session users: the processor path is opt-in only
+        queue.Options.RequiresSession.ShouldBeTrue();
+        queue.ConfigureSessionProcessor.ShouldBeNull();
+    }
+}

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

@@ -0,0 +1,90 @@
+using Azure.Messaging.ServiceBus;
+using IntegrationTests;
+using JasperFx.Core;
+using Microsoft.Extensions.Hosting;
+using Shouldly;
+using Wolverine.Tracking;
+using Xunit;
+
+namespace Wolverine.AzureServiceBus.Tests;
+
+// GH-3533: pinning a session-enabled listener to specific session identifiers turns the session id
+// into a broker-enforced routing key on a shared queue, so a listener pinned to "A" never sees the
+// messages meant for "B".
+[Trait("Category", "Flaky")]
+public class session_id_pinning : IAsyncLifetime
+{
+    private IHost _host = null!;
+
+    public async Task InitializeAsync()
+    {
+        _host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.UseAzureServiceBusTesting().AutoProvision().AutoPurgeOnStartup();
+
+                opts.ListenToAzureServiceBusQueue("shared-pinned")
+
+                    // Only ever lock the "A" session on this shared queue
+                    .RequireSessionsWithOnlyTheseIdentifiers("A")
+                    .Sequential();
+
+                opts.PublishMessage<PinnedMessage>().ToAzureServiceBusQueue("shared-pinned");
+            }).StartAsync();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await _host.StopAsync();
+        _host.Dispose();
+        await AzureServiceBusTesting.DeleteAllEmulatorObjectsAsync();
+    }
+
+    [Fact]
+    public async Task pinned_listener_only_receives_its_own_session()
+    {
+        await using var client = new ServiceBusClient(Servers.AzureServiceBusConnectionString);
+
+        // Seed a message destined for session "B" directly onto the shared queue, bypassing Wolverine
+        var sender = client.CreateSender("shared-pinned");
+        await sender.SendMessageAsync(new ServiceBusMessage("not for A")
+        {
+            SessionId = "B",
+            MessageId = Guid.NewGuid().ToString()
+        });
+
+        // Drive three "A" messages through Wolverine and confirm ONLY those are received
+        Func<IMessageContext, Task> sendAll = async bus =>
+        {
+            await bus.SendAsync(new PinnedMessage("A-1"), new DeliveryOptions { GroupId = "A" });
+            await bus.SendAsync(new PinnedMessage("A-2"), new DeliveryOptions { GroupId = "A" });
+            await bus.SendAsync(new PinnedMessage("A-3"), new DeliveryOptions { GroupId = "A" });
+        };
+
+        var tracked = await _host.TrackActivity()
+            .IncludeExternalTransports()
+            .Timeout(30.Seconds())
+            .ExecuteAndWaitAsync(sendAll);
+
+        // Every "A" message was delivered here (order is a separate FIFO concern), and nothing else
+        tracked.Received.MessagesOf<PinnedMessage>().Select(x => x.Name).OrderBy(x => x)
+            .ShouldBe(["A-1", "A-2", "A-3"]);
+
+        // The "B" session message must still be sitting on the shared queue, never delivered to the
+        // A-pinned listener.
+        await using var sessionReceiver = await client.AcceptSessionAsync("shared-pinned", "B");
+        var leftover = await sessionReceiver.ReceiveMessageAsync(5.Seconds());
+        leftover.ShouldNotBeNull();
+        leftover.SessionId.ShouldBe("B");
+    }
+}
+
+public record PinnedMessage(string Name);
+
+public static class PinnedMessageHandler
+{
+    public static void Handle(PinnedMessage message)
+    {
+        // no-op; tracking observes receipt
+    }
+}

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

@@ -1,3 +1,4 @@
+using Azure.Messaging.ServiceBus;
 using Azure.Messaging.ServiceBus.Administration;
 using Wolverine.AzureServiceBus.Internal;
 using Wolverine.Configuration;
@@ -142,6 +143,51 @@ public AzureServiceBusQueueListenerConfiguration RequireSessions(int? listenerCo
         return this;
     }
 
+    /// <summary>
+    ///     Customize the Azure Service Bus <see cref="ServiceBusSessionProcessorOptions" /> used by this
+    ///     session-enabled listener -- e.g. <c>MaxConcurrentSessions</c>, <c>MaxAutoLockRenewalDuration</c>,
+    ///     <c>SessionIdleTimeout</c>, or <c>SessionIds</c>. Calling this implies <see cref="RequireSessions" />
+    ///     and switches the session listener from the default AcceptNextSession loop to a
+    ///     <see cref="ServiceBusSessionProcessor" />. Multiple calls compose. Wolverine reserves control of the
+    ///     properties it depends on for message acknowledgement (<c>ReceiveMode</c>, <c>AutoCompleteMessages</c>),
+    ///     which are re-asserted after this action runs.
+    /// </summary>
+    /// <param name="configure"></param>
+    /// <returns></returns>
+    public AzureServiceBusQueueListenerConfiguration ConfigureSessionProcessor(
+        Action<ServiceBusSessionProcessorOptions> configure)
+    {
+        add(e =>
+        {
+            e.Options.RequiresSession = true;
+            // Compose rather than overwrite so the SessionIds sugar can coexist with an explicit hook
+            e.ConfigureSessionProcessor += configure;
+        });
+        return this;
+    }
+
+    /// <summary>
+    ///     Pin this listener to only the given session identifiers. On a shared queue this turns the session
+    ///     id into a broker-enforced routing key: competing consumers each pinned to their own id(s) never see
+    ///     each other's messages. Producers select the target by setting <c>DeliveryOptions.GroupId</c> to the
+    ///     session id. Delegates to <see cref="ConfigureSessionProcessor" /> by populating
+    ///     <c>ServiceBusSessionProcessorOptions.SessionIds</c>. (GH-3533)
+    /// </summary>
+    /// <param name="identifiers">The session identifiers this listener should exclusively lock</param>
+    /// <returns></returns>
+    public AzureServiceBusQueueListenerConfiguration RequireSessionsWithOnlyTheseIdentifiers(
+        params string[] identifiers)
+    {
+        RequireSessions();
+        return ConfigureSessionProcessor(options =>
+        {
+            foreach (var id in identifiers)
+            {
+                options.SessionIds.Add(id);
+            }
+        });
+    }
+
     /// <summary>
     /// Utilize custom envelope mapping for Amazon Service Bus interoperability with external non-Wolverine systems
     /// </summary>

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

@@ -1,3 +1,4 @@
+using Azure.Messaging.ServiceBus;
 using Azure.Messaging.ServiceBus.Administration;
 using Wolverine.AzureServiceBus.Internal;
 using Wolverine.Configuration;
@@ -128,6 +129,51 @@ public AzureServiceBusSubscriptionListenerConfiguration RequireSessions(int? lis
         return this;
     }
 
+    /// <summary>
+    ///     Customize the Azure Service Bus <see cref="ServiceBusSessionProcessorOptions" /> used by this
+    ///     session-enabled subscription listener -- e.g. <c>MaxConcurrentSessions</c>,
+    ///     <c>MaxAutoLockRenewalDuration</c>, <c>SessionIdleTimeout</c>, or <c>SessionIds</c>. Calling this
+    ///     implies <see cref="RequireSessions" /> and switches the session listener from the default
+    ///     AcceptNextSession loop to a <see cref="ServiceBusSessionProcessor" />. Multiple calls compose.
+    ///     Wolverine reserves control of the properties it depends on for message acknowledgement
+    ///     (<c>ReceiveMode</c>, <c>AutoCompleteMessages</c>), which are re-asserted after this action runs.
+    /// </summary>
+    /// <param name="configure"></param>
+    /// <returns></returns>
+    public AzureServiceBusSubscriptionListenerConfiguration ConfigureSessionProcessor(
+        Action<ServiceBusSessionProcessorOptions> configure)
+    {
+        add(e =>
+        {
+            e.Options.RequiresSession = true;
+            // Compose rather than overwrite so the SessionIds sugar can coexist with an explicit hook
+            e.ConfigureSessionProcessor += configure;
+        });
+        return this;
+    }
+
+    /// <summary>
+    ///     Pin this listener to only the given session identifiers. On a shared subscription this turns the
+    ///     session id into a broker-enforced routing key: competing consumers each pinned to their own id(s)
+    ///     never see each other's messages. Producers select the target by setting <c>DeliveryOptions.GroupId</c>
+    ///     to the session id. Delegates to <see cref="ConfigureSessionProcessor" /> by populating
+    ///     <c>ServiceBusSessionProcessorOptions.SessionIds</c>. (GH-3533)
+    /// </summary>
+    /// <param name="identifiers">The session identifiers this listener should exclusively lock</param>
+    /// <returns></returns>
+    public AzureServiceBusSubscriptionListenerConfiguration RequireSessionsWithOnlyTheseIdentifiers(
+        params string[] identifiers)
+    {
+        RequireSessions();
+        return ConfigureSessionProcessor(options =>
+        {
+            foreach (var id in identifiers)
+            {
+                options.SessionIds.Add(id);
+            }
+        });
+    }
+
     /// <summary>
     /// Utilize custom envelope mapping for Amazon Service Bus interoperability with external non-Wolverine systems
     /// </summary>