Configure ServiceBusProcessorOptions for Azure Service Bus listeners (supersedes #3286)

Directory.Build.targets

@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="utf-8"?>
+<Project>
+
+    <!--
+      Security remediation for GHSA-v5pm-xwqc-g5wc (uncontrolled recursion / stack overflow
+      when parsing OpenAPI documents with circular schema references).
+
+      Microsoft.AspNetCore.OpenApi 10.0.0 (net10.0) pulls in the vulnerable Microsoft.OpenApi
+      2.0.0 transitively. With TreatWarningsAsErrors=true the NU1903 restore audit turns that
+      into a hard build break. Add a direct reference to the patched 2.7.5 (same 2.x major, so
+      API-compatible with AspNetCore.OpenApi 10.x; version pinned centrally in
+      Directory.Packages.props) so it wins over the transitive 2.0.0.
+
+      Scoped deliberately to (net10.0 AND references Microsoft.AspNetCore.OpenApi):
+        * net9.0 resolves Microsoft.OpenApi 1.6.x, which is not in the affected range — leave it.
+        * Swashbuckle-only web apps resolve Microsoft.OpenApi 1.2.x — forcing 2.x would break
+          Swashbuckle 6.5.0, and they aren't vulnerable, so leave them too.
+      Non-web test projects that reference these web apps inherit the pin transitively.
+    -->
+    <ItemGroup Condition=" '$(TargetFramework)' == 'net10.0' And @(PackageReference->AnyHaveMetadataValue('Identity', 'Microsoft.AspNetCore.OpenApi')) ">
+        <PackageReference Include="Microsoft.OpenApi" />
+    </ItemGroup>
+
+</Project>

Directory.Packages.props

@@ -100,6 +100,10 @@
     <PackageVersion Include="Spectre.Console" Version="0.55.0" />
     <PackageVersion Include="StackExchange.Redis" Version="2.9.17" />
     <PackageVersion Include="StronglyTypedId" Version="1.0.0-beta08" />
+    <!-- Transitively pinned (see CentralPackageTransitivePinningEnabled above): Swashbuckle 9.x
+         and Microsoft.AspNetCore.OpenApi 10.x pull Microsoft.OpenApi 2.0.0, which carries a high
+         severity advisory (GHSA-v5pm-xwqc-g5wc, patched in 2.7.5). -->
+    <PackageVersion Include="Microsoft.OpenApi" Version="2.7.5" />
     <PackageVersion Include="Swashbuckle.AspNetCore" Version="6.5.0" />
     <PackageVersion Include="Swashbuckle.AspNetCore.Swagger" Version="9.0.1" />
     <PackageVersion Include="Swashbuckle.AspNetCore.SwaggerGen" Version="9.0.1" />

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

@@ -45,6 +45,62 @@ await host.StartAsync();
 <sup><a href='https://github.com/JasperFx/wolverine/blob/main/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/DocumentationSamples.cs#L87-L119' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_configuring_an_azure_service_bus_listener' title='Start of snippet'>anchor</a></sup>
 <!-- endSnippet -->
 
+## Configuring the ServiceBusProcessor
+
+When a listener runs [inline](/guide/messaging/endpoints.html#endpoint-types) (`ProcessInline()`), Wolverine
+creates an Azure Service Bus `ServiceBusProcessor` for the queue or subscription. You can customize the
+underlying `ServiceBusProcessorOptions` with `ConfigureProcessor()` on either a queue or a subscription
+listener.
+
+The most common reason to reach for this is a long-running inline handler. The Azure SDK only renews a
+message's lock for five minutes by default (`MaxAutoLockRenewalDuration`). Any inline handler that runs
+longer than that loses the lock, so the completion acknowledgement silently fails and Azure Service Bus
+redelivers the message — resulting in duplicate work until the message is finally dead-lettered. Raising
+`MaxAutoLockRenewalDuration` keeps the lock alive for the length of your handler:
+
+<!-- snippet: sample_configuring_azure_service_bus_processor_options -->
+<a id='snippet-sample_configuring_azure_service_bus_processor_options'></a>
+```cs
+var builder = Host.CreateApplicationBuilder();
+builder.UseWolverine(opts =>
+{
+    // One way or another, you're probably pulling the Azure Service Bus
+    // connection string out of configuration
+    var azureServiceBusConnectionString = builder
+        .Configuration
+        .GetConnectionString("azure-service-bus")!;
+
+    opts.UseAzureServiceBus(azureServiceBusConnectionString).AutoProvision();
+
+    opts.ListenToAzureServiceBusQueue("incoming")
+
+        // Inline listeners create an Azure Service Bus ServiceBusProcessor. By default the
+        // Azure SDK only renews the message lock for five minutes, so an inline handler that
+        // runs longer than that loses its lock and the message is redelivered. Raise the
+        // renewal window here so long-running inline handlers keep their lock.
+        .ConfigureProcessor(processorOptions =>
+        {
+            processorOptions.MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(30);
+        })
+
+        // Run the handler inline against the ServiceBusProcessor
+        .ProcessInline();
+});
+
+using var host = builder.Build();
+await host.StartAsync();
+```
+<sup><a href='https://github.com/JasperFx/wolverine/blob/main/src/Transports/Azure/Wolverine.AzureServiceBus.Tests/DocumentationSamples.cs#L124-L153' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_configuring_azure_service_bus_processor_options' title='Start of snippet'>anchor</a></sup>
+<!-- endSnippet -->
+
+::: warning
+Wolverine's inline listener acknowledges, defers, and dead letters messages against the message lock, so it
+requires the peek-lock receive model. The `ReceiveMode` you set in `ConfigureProcessor()` is therefore
+ignored and always re-asserted to `ServiceBusReceiveMode.PeekLock`. All other `ServiceBusProcessorOptions`
+properties are honored. Session-based listeners (`RequireSessions()`) do not use a `ServiceBusProcessor` and
+are not affected by this option.
+:::
+
 ## Conventional Listener Configuration
 
 In the case of listening to a large number of queues, it may be beneficial

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

@@ -119,6 +119,41 @@ public async Task configuring_a_listener()
         #endregion
     }
 
+    public async Task configuring_processor_options()
+    {
+        #region sample_configuring_azure_service_bus_processor_options
+        var builder = Host.CreateApplicationBuilder();
+        builder.UseWolverine(opts =>
+        {
+            // One way or another, you're probably pulling the Azure Service Bus
+            // connection string out of configuration
+            var azureServiceBusConnectionString = builder
+                .Configuration
+                .GetConnectionString("azure-service-bus")!;
+
+            opts.UseAzureServiceBus(azureServiceBusConnectionString).AutoProvision();
+
+            opts.ListenToAzureServiceBusQueue("incoming")
+
+                // Inline listeners create an Azure Service Bus ServiceBusProcessor. By default the
+                // Azure SDK only renews the message lock for five minutes, so an inline handler that
+                // runs longer than that loses its lock and the message is redelivered. Raise the
+                // renewal window here so long-running inline handlers keep their lock.
+                .ConfigureProcessor(processorOptions =>
+                {
+                    processorOptions.MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(30);
+                })
+
+                // Run the handler inline against the ServiceBusProcessor
+                .ProcessInline();
+        });
+
+        using var host = builder.Build();
+        await host.StartAsync();
+
+        #endregion
+    }
+
     public async Task configure_buffered_listener()
     {
         var builder = Host.CreateApplicationBuilder();

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

@@ -0,0 +1,85 @@
+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_processor_options
+{
+    [Fact]
+    public void configure_processor_action_is_stored_on_the_queue_endpoint()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+
+        var configuration = new AzureServiceBusQueueListenerConfiguration(queue);
+        configuration.ConfigureProcessor(o => o.MaxAutoLockRenewalDuration = 30.Minutes());
+
+        // Apply the delayed configuration as Wolverine would at bootstrapping time
+        ((IDelayedEndpointConfiguration)configuration).Apply();
+
+        queue.ConfigureProcessor.ShouldNotBeNull();
+    }
+
+    [Fact]
+    public void configure_processor_action_is_stored_on_the_subscription_endpoint()
+    {
+        var transport = new AzureServiceBusTransport();
+        var topic = transport.Topics["topic1"];
+        var subscription = topic.FindOrCreateSubscription("sub1");
+
+        var configuration = new AzureServiceBusSubscriptionListenerConfiguration(subscription);
+        configuration.ConfigureProcessor(o => o.MaxAutoLockRenewalDuration = 30.Minutes());
+
+        ((IDelayedEndpointConfiguration)configuration).Apply();
+
+        subscription.ConfigureProcessor.ShouldNotBeNull();
+    }
+
+    [Fact]
+    public void build_processor_options_applies_the_configured_action()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+        queue.ConfigureProcessor = o =>
+        {
+            o.MaxAutoLockRenewalDuration = 30.Minutes();
+            o.PrefetchCount = 12;
+        };
+
+        var options = AzureServiceBusTransport.BuildProcessorOptions(queue);
+
+        options.MaxAutoLockRenewalDuration.ShouldBe(30.Minutes());
+        options.PrefetchCount.ShouldBe(12);
+    }
+
+    [Fact]
+    public void build_processor_options_reasserts_peek_lock_receive_mode()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+
+        // A user trying to switch to ReceiveAndDelete would break Wolverine's
+        // inline acknowledgement, so the transport must re-assert PeekLock.
+        queue.ConfigureProcessor = o => o.ReceiveMode = ServiceBusReceiveMode.ReceiveAndDelete;
+
+        var options = AzureServiceBusTransport.BuildProcessorOptions(queue);
+
+        options.ReceiveMode.ShouldBe(ServiceBusReceiveMode.PeekLock);
+    }
+
+    [Fact]
+    public void build_processor_options_is_valid_when_no_action_is_configured()
+    {
+        var transport = new AzureServiceBusTransport();
+        var queue = transport.Queues["incoming"];
+
+        var options = AzureServiceBusTransport.BuildProcessorOptions(queue);
+
+        options.ShouldNotBeNull();
+        options.ReceiveMode.ShouldBe(ServiceBusReceiveMode.PeekLock);
+    }
+}

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;
@@ -64,6 +65,22 @@ public AzureServiceBusQueueListenerConfiguration ConfigureQueue(Action<CreateQue
         return this;
     }
 
+    /// <summary>
+    ///     Customize the Azure Service Bus <see cref="ServiceBusProcessorOptions" /> used by this
+    ///     listener when running in the inline (<c>ProcessInline()</c>) mode. This is the way to raise
+    ///     <see cref="ServiceBusProcessorOptions.MaxAutoLockRenewalDuration" /> for inline handlers that
+    ///     run longer than the Azure SDK's default of five minutes. Wolverine reserves control of the
+    ///     properties it depends on for message acknowledgement (currently <c>ReceiveMode</c>), which are
+    ///     re-asserted after this action runs.
+    /// </summary>
+    /// <param name="configure"></param>
+    /// <returns></returns>
+    public AzureServiceBusQueueListenerConfiguration ConfigureProcessor(Action<ServiceBusProcessorOptions> configure)
+    {
+        add(e => e.ConfigureProcessor = configure);
+        return this;
+    }
+
     /// <summary>
     ///     The maximum number of messages to receive in a single batch when listening
     ///     in either buffered or durable modes. The default is 20.

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;
@@ -63,6 +64,22 @@ public AzureServiceBusSubscriptionListenerConfiguration ConfigureSubscription(Ac
         return this;
     }
 
+    /// <summary>
+    ///     Customize the Azure Service Bus <see cref="ServiceBusProcessorOptions" /> used by this
+    ///     subscription listener when running in the inline (<c>ProcessInline()</c>) mode. This is the way
+    ///     to raise <see cref="ServiceBusProcessorOptions.MaxAutoLockRenewalDuration" /> for inline handlers
+    ///     that run longer than the Azure SDK's default of five minutes. Wolverine reserves control of the
+    ///     properties it depends on for message acknowledgement (currently <c>ReceiveMode</c>), which are
+    ///     re-asserted after this action runs.
+    /// </summary>
+    /// <param name="configure"></param>
+    /// <returns></returns>
+    public AzureServiceBusSubscriptionListenerConfiguration ConfigureProcessor(Action<ServiceBusProcessorOptions> configure)
+    {
+        add(e => e.ConfigureProcessor = configure);
+        return this;
+    }
+
     /// <summary>
     ///     Configure the underlying Azure Service Bus Subscription rule. This is only applicable when
     ///     Wolverine is creating the Subscription.

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

@@ -66,7 +66,7 @@ private async Task<IListener> buildListenerForQueue(IWolverineRuntime runtime, I
 
         if (queue.Mode == EndpointMode.Inline)
         {
-            var messageProcessor = BusClient.CreateProcessor(queue.QueueName);
+            var messageProcessor = BusClient.CreateProcessor(queue.QueueName, BuildProcessorOptions(queue));
 
             var inlineListener = new InlineAzureServiceBusListener(queue,
                 runtime.LoggerFactory.CreateLogger<InlineAzureServiceBusListener>(), messageProcessor, receiver,
@@ -122,7 +122,7 @@ private async Task<IListener> buildListenerForSubscription(IWolverineRuntime run
 
         if (subscription.Mode == EndpointMode.Inline)
         {
-            var messageProcessor = BusClient.CreateProcessor(subscription.Topic.TopicName, subscription.SubscriptionName);
+            var messageProcessor = BusClient.CreateProcessor(subscription.Topic.TopicName, subscription.SubscriptionName, BuildProcessorOptions(subscription));
             var inlineListener = new InlineAzureServiceBusListener(subscription,
                 runtime.LoggerFactory.CreateLogger<InlineAzureServiceBusListener>(), messageProcessor, receiver, mapper,  requeue
             );
@@ -138,4 +138,23 @@ private async Task<IListener> buildListenerForSubscription(IWolverineRuntime run
 
         return listener;
     }
+
+    // Builds the ServiceBusProcessorOptions for an inline listener, applying any user supplied
+    // customization and then re-asserting the properties that Wolverine's inline acknowledgement
+    // logic depends on. InlineAzureServiceBusListener explicitly completes, defers, and dead
+    // letters messages against the message lock, so the processor must run in PeekLock mode; the
+    // ReceiveAndDelete mode would remove messages before Wolverine can process them and would break
+    // dead lettering and deferral.
+    internal static ServiceBusProcessorOptions BuildProcessorOptions(AzureServiceBusEndpoint endpoint)
+    {
+        var options = new ServiceBusProcessorOptions();
+
+        endpoint.ConfigureProcessor?.Invoke(options);
+
+        // Reserved by Wolverine: the inline listener relies on the peek-lock model to complete,
+        // defer, and dead letter messages, so this cannot be honored from user configuration.
+        options.ReceiveMode = ServiceBusReceiveMode.PeekLock;
+
+        return options;
+    }
 }

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

@@ -49,6 +49,15 @@ public AzureServiceBusEndpoint(AzureServiceBusTransport parent, Uri uri, Endpoin
     /// </summary>
     public TimeSpan MaximumWaitTime { get; set; } = 5.Seconds();
 
+    /// <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
+    ///     on for correct message acknowledgement (see AzureServiceBusTransport.Listening), so those will
+    ///     be re-asserted after this action runs.
+    /// </summary>
+    [IgnoreDescription]
+    public Action<ServiceBusProcessorOptions>? ConfigureProcessor { get; set; }
+
     public abstract ValueTask<bool> CheckAsync();
     public abstract ValueTask TeardownAsync(ILogger logger);
     public abstract ValueTask SetupAsync(ILogger logger);