Use explicit message settlement via ServiceBusMessageActions

src/IntegrationTest.Billing/BillingFunctions.cs

@@ -12,7 +12,7 @@ public partial class BillingFunctions
     [Function("BillingApi")]
     [NServiceBusFunction]
     public partial Task BillingApi(
-        [ServiceBusTrigger("billing-api", Connection = "AzureWebJobsServiceBus", AutoCompleteMessages = true)]
+        [ServiceBusTrigger("billing-api", Connection = "AzureWebJobsServiceBus", AutoCompleteMessages = false)]
         ServiceBusReceivedMessage message,
         ServiceBusMessageActions messageActions,
         FunctionContext functionContext,
@@ -27,7 +27,7 @@ public static void ConfigureBillingApi(EndpointConfiguration configuration)
     [Function("BillingBackend")]
     [NServiceBusFunction]
     public partial Task BillingBackend(
-        [ServiceBusTrigger("billing-backend", Connection = "AzureWebJobsServiceBus", AutoCompleteMessages = true)]
+        [ServiceBusTrigger("billing-backend", Connection = "AzureWebJobsServiceBus", AutoCompleteMessages = false)]
         ServiceBusReceivedMessage message,
         ServiceBusMessageActions messageActions,
         FunctionContext functionContext,

src/IntegrationTest.Sales/SalesEndpoint.cs

@@ -4,14 +4,15 @@ namespace IntegrationTest.Sales;
 using IntegrationTest.Shared;
 using Microsoft.Azure.Functions.Worker;
 using Microsoft.Extensions.DependencyInjection;
+using NServiceBus.AzureFunctions.AzureServiceBus;
 
 // Cleanest pattern for single-function endpoints
 [NServiceBusFunction]
 public partial class SalesEndpoint
 {
     [Function("Sales")]
     public partial Task Sales(
-        [ServiceBusTrigger("sales", Connection = "AzureWebJobsServiceBus", AutoCompleteMessages = true)]
+        [ServiceBusTrigger("sales", Connection = "AzureWebJobsServiceBus", AutoCompleteMessages = false)]
         ServiceBusReceivedMessage message,
         ServiceBusMessageActions messageActions,
         FunctionContext functionContext,
@@ -23,5 +24,17 @@ public static void ConfigureSales(EndpointConfiguration configuration)
 
         configuration.RegisterComponents(services => services.AddSingleton(new MyComponent("Sales")));
         configuration.AddHandler<Handlers.SubmitOrderHandler>();
+        configuration.AuditProcessedMessagesTo("audit");
+
+        // Use the dead letter queue for failures
+        configuration.Recoverability().CustomPolicy((_, context) =>
+        {
+            if (context.ImmediateProcessingFailures < 3)
+            {
+                return RecoverabilityAction.ImmediateRetry();
+            }
+
+            return RecoverabilityAction.DeadLetter(context.Exception);
+        });
     }
 }

src/IntegrationTest.Shared/Infrastructure/TestStorage.cs

@@ -25,9 +25,9 @@ public Payload CreatePayload(string testName)
 
         var sortedMessages = list
             .OrderBy(m => m.Order)
-                .ThenBy(m => m.MessageType)
-                .ThenBy(m => m.SendingEndpoint)
-                .ThenBy(m => m.ReceivingEndpoint)
+            .ThenBy(m => m.MessageType)
+            .ThenBy(m => m.SendingEndpoint)
+            .ThenBy(m => m.ReceivingEndpoint)
             .ToArray();
 
         return new Payload(sortedMessages);
@@ -39,8 +39,9 @@ public class TestStorage(string endpointName, GlobalTestStorage globalStorage)
     public void LogMessage<T>(string testName, T message, IMessageHandlerContext context)
         where T : class
     {
-        var sendingEndpoint = context.MessageHeaders[Headers.OriginatingEndpoint] ?? "<unknown>";
+        var sendingEndpoint = context.MessageHeaders.GetValueOrDefault(Headers.OriginatingEndpoint, "<unknown>");
         var storageOrder = context.Extensions.Get<int>("TestStorageOrder");
+
         var rec = new MessageReceived(message.GetType().FullName!, storageOrder, sendingEndpoint, endpointName);
         globalStorage.Add(testName, rec);
     }

src/IntegrationTest.Shipping/ShippingEndpoint.cs

@@ -9,7 +9,7 @@ public partial class ShippingEndpoint
 {
     [Function(nameof(Shipping))]
     public partial Task Shipping(
-        [ServiceBusTrigger("shipping", AutoCompleteMessages = true)]
+        [ServiceBusTrigger("shipping", AutoCompleteMessages = false)]
         ServiceBusReceivedMessage message,
         ServiceBusMessageActions messageActions,
         FunctionContext functionContext,

src/IntegrationTest/Program.cs

@@ -1,4 +1,3 @@
-using System.Text.Json;
 using IntegrationTest;
 using IntegrationTest.Shared;
 using IntegrationTest.Shared.Infrastructure;
@@ -10,18 +9,14 @@
 var builder = FunctionsApplication.CreateBuilder(args);
 builder.UseWhen<ExceptionTrackingMiddleware>(_ => true);
 
-builder.Logging.ClearProviders();
-builder.Logging.AddJsonConsole(o =>
-{
-    o.IncludeScopes = true;
-    o.JsonWriterOptions = new JsonWriterOptions { Indented = true };
-});
-builder.Logging.SetMinimumLevel(LogLevel.Information);
+builder.Logging.AddSimpleConsole(options => options.IncludeScopes = true);
+builder.Logging.SetMinimumLevel(LogLevel.Warning);
 
 builder.Services.AddSingleton<GlobalTestStorage>();
 builder.Services.AddSingleton(new MyComponent("global"));
 
 builder.AddNServiceBusFunctions();
+
 builder.AddSendOnlyNServiceBusEndpoint("client", configuration =>
 {
     configuration.RegisterComponents(services => services.AddSingleton(new MyComponent("client")));

src/NServiceBus.AzureFunctions.Analyzer/DiagnosticIds.cs

@@ -2,47 +2,56 @@ namespace NServiceBus.AzureFunctions.Analyzer;
 
 using Microsoft.CodeAnalysis;
 
-static class DiagnosticIds
+public static class DiagnosticIds
 {
     public const string ClassMustBePartial = "NSBFUNC001";
     public const string ShouldNotImplementIHandleMessages = "NSBFUNC002";
     public const string MethodMustBePartial = "NSBFUNC003";
-    public const string MultipleConfigureMethods = "NSBFUNC005";
     public const string MissingAddNServiceBusFunctionsCall = "NSBFUNC004";
+    public const string MultipleConfigureMethods = "NSBFUNC005";
+    public const string AutoCompleteEnabled = "NSBFUNC006";
 
-    public static readonly DiagnosticDescriptor ClassMustBePartialDescriptor = new(
+    internal static readonly DiagnosticDescriptor ClassMustBePartialDescriptor = new(
         id: ClassMustBePartial,
         title: "Class containing [NServiceBusFunction] must be partial",
         messageFormat: "Class '{0}' must be declared as partial to use [NServiceBusFunction]",
         category: "NServiceBus.AzureFunctions",
         defaultSeverity: DiagnosticSeverity.Error,
         isEnabledByDefault: true);
 
-    public static readonly DiagnosticDescriptor ShouldNotImplementIHandleMessagesDescriptor = new(
+    internal static readonly DiagnosticDescriptor ShouldNotImplementIHandleMessagesDescriptor = new(
         id: ShouldNotImplementIHandleMessages,
         title: "Function class should not implement IHandleMessages<T>",
         messageFormat: "Class '{0}' should not implement IHandleMessages<T>; message handlers should be registered separately via IEndpointConfiguration",
         category: "NServiceBus.AzureFunctions",
         defaultSeverity: DiagnosticSeverity.Warning,
         isEnabledByDefault: true);
 
-    public static readonly DiagnosticDescriptor MethodMustBePartialDescriptor = new(
+    internal static readonly DiagnosticDescriptor MethodMustBePartialDescriptor = new(
         id: MethodMustBePartial,
         title: "Method with [NServiceBusFunction] must be partial",
         messageFormat: "Method '{0}' must be declared as partial to use [NServiceBusFunction]",
         category: "NServiceBus.AzureFunctions",
         defaultSeverity: DiagnosticSeverity.Error,
         isEnabledByDefault: true);
 
-    public static readonly DiagnosticDescriptor MultipleConfigureMethodsDescriptor = new(
+    internal static readonly DiagnosticDescriptor MultipleConfigureMethodsDescriptor = new(
         id: MultipleConfigureMethods,
         title: "Multiple configuration methods found",
         messageFormat: "Multiple '{0}' configuration methods found on class '{1}'",
         category: "NServiceBus.AzureFunctions",
         defaultSeverity: DiagnosticSeverity.Error,
         isEnabledByDefault: true);
 
-    public static readonly DiagnosticDescriptor MissingAddNServiceBusFunctionsCallDescriptor = new(
+    internal static readonly DiagnosticDescriptor AutoCompleteMustBeExplicitlyDisabled = new(
+        id: AutoCompleteEnabled,
+        title: "Message auto completion must be explicitly disabled",
+        messageFormat: "The auto complete property on the service bus trigger for method '{0}' must be explicitly set to false",
+        category: "NServiceBus.AzureFunctions",
+        defaultSeverity: DiagnosticSeverity.Error,
+        isEnabledByDefault: true);
+
+    internal static readonly DiagnosticDescriptor MissingAddNServiceBusFunctionsCallDescriptor = new(
         id: MissingAddNServiceBusFunctionsCall,
         title: "AddNServiceBusFunctions() is not called",
         messageFormat: "This project has NServiceBus endpoint registrations but does not call builder.AddNServiceBusFunctions(). Endpoints will not be started.",

src/NServiceBus.AzureFunctions.Analyzer/FunctionEndpointGenerator.Parser.cs

@@ -125,12 +125,24 @@ static FunctionSpecs ExtractFromMethod(IMethodSymbol methodSymbol, FunctionEndpo
                             queueName = pAttr.ConstructorArguments[0].Value as string;
                         }
 
+                        var autoCompleteEnabled = true;
                         foreach (var namedArg in pAttr.NamedArguments)
                         {
                             if (namedArg.Key == "Connection")
                             {
                                 connectionName = namedArg.Value.Value as string;
                             }
+
+                            if (namedArg.Key == "AutoCompleteMessages")
+                            {
+                                var autoComplete = namedArg.Value.Value as bool?;
+                                autoCompleteEnabled = autoComplete!.Value;
+                            }
+                        }
+
+                        if (autoCompleteEnabled)
+                        {
+                            diagnostics.Add(CreateDiagnostic(DiagnosticIds.AutoCompleteMustBeExplicitlyDisabled, method, method.Name));
                         }
 
                         messageParamName = param.Name;

src/NServiceBus.AzureFunctions.AzureServiceBus/AzureServiceBusMessageProcessor.cs

@@ -7,7 +7,6 @@ namespace NServiceBus.AzureFunctions.AzureServiceBus;
 
 public class AzureServiceBusMessageProcessor(AzureServiceBusServerlessTransport transport, string endpointName)
 {
-    //NOTE: Message actions and function context is here to be ready for future features like native dlq support without having to change the end user api.
     public async Task Process(ServiceBusReceivedMessage message, ServiceBusMessageActions messageActions, FunctionContext functionContext, CancellationToken cancellationToken = default)
     {
         if (transport.MessageProcessor is null)
@@ -16,6 +15,6 @@ public async Task Process(ServiceBusReceivedMessage message, ServiceBusMessageAc
             throw new InvalidOperationException($"Endpoint {endpointName} cannot process messages because it is configured in send-only mode.");
         }
 
-        await transport.MessageProcessor.Process(message, cancellationToken).ConfigureAwait(false);
+        await transport.MessageProcessor.Process(message, messageActions, cancellationToken).ConfigureAwait(false);
     }
 }

src/NServiceBus.AzureFunctions.AzureServiceBus/AzureServiceBusServerlessTransport.cs

@@ -6,10 +6,12 @@ namespace NServiceBus;
 using System.Threading;
 using System.Threading.Tasks;
 using Azure.Core;
+using AzureFunctions.AzureServiceBus;
+using BitFaster.Caching.Lru;
 using Microsoft.Extensions.Azure;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
-using NServiceBus.AzureFunctions.AzureServiceBus.Serverless.TransportWrapper;
+using Microsoft.Extensions.Logging;
 using NServiceBus.Transport;
 
 public class AzureServiceBusServerlessTransport(TopicTopology topology) : TransportDefinition(TransportTransactionMode.ReceiveOnly,
@@ -53,7 +55,7 @@ public override async Task<TransportInfrastructure> Initialize(
             .ConfigureAwait(false);
 
         var serverlessTransportInfrastructure = new ServerlessTransportInfrastructure(baseTransportInfrastructure,
-            static receiver => new PipelineInvokingMessageProcessor(receiver));
+            receiver => new PipelineInvokingMessageProcessor(receiver, new FastConcurrentLru<string, bool>(1_000), hostSettings.ServiceProvider.GetRequiredService<ILogger<PipelineInvokingMessageProcessor>>()));
 
         var isSendOnly = hostSettings.CoreSettings.GetOrDefault<bool>(SendOnlyConfigKey);
 

src/NServiceBus.AzureFunctions.AzureServiceBus/DeadLetterMessage.cs

@@ -0,0 +1,23 @@
+namespace NServiceBus.AzureFunctions.AzureServiceBus;
+
+using Pipeline;
+using Transport;
+
+public sealed class DeadLetterMessage : RecoverabilityAction
+{
+    internal DeadLetterMessage(string deadLetterReason, string deadLetterErrorDescription, Dictionary<string, object>? propertiesToModify = null) =>
+        deadLetterRequest = new DeadLetterRequest(deadLetterReason, deadLetterErrorDescription, propertiesToModify);
+
+    internal DeadLetterMessage(Exception exception) =>
+        deadLetterRequest = new DeadLetterRequest(exception);
+
+    public override IReadOnlyCollection<IRoutingContext> GetRoutingContexts(IRecoverabilityActionContext context)
+    {
+        context.Extensions.Get<TransportTransaction>().Set(deadLetterRequest);
+        return [];
+    }
+
+    public override ErrorHandleResult ErrorHandleResult => ErrorHandleResult.Handled;
+
+    readonly DeadLetterRequest deadLetterRequest;
+}

src/NServiceBus.AzureFunctions.AzureServiceBus/DeadLetterRequest.cs

@@ -0,0 +1,17 @@
+namespace NServiceBus.AzureFunctions.AzureServiceBus;
+
+class DeadLetterRequest(string deadLetterReason, string deadLetterErrorDescription, Dictionary<string, object>? propertiesToModify = null)
+{
+    public string DeadLetterReason { get; } = Truncate(deadLetterReason, 1024);
+    public string DeadLetterErrorDescription { get; } = Truncate(deadLetterErrorDescription, 1024);
+    public Dictionary<string, object> PropertiesToModify { get; } = propertiesToModify ?? [];
+
+    public DeadLetterRequest(Exception exception, Dictionary<string, object>? propertiesToModify = null) : this(
+        $"{exception.GetType().FullName} - {exception.Message}",
+        exception.StackTrace ?? "No stack trace available",
+        propertiesToModify)
+    {
+    }
+
+    static string Truncate(string value, int maxLength) => string.IsNullOrEmpty(value) ? value : value.Length <= maxLength ? value : value[..maxLength];
+}

src/NServiceBus.AzureFunctions.AzureServiceBus/GuidHelper.cs

@@ -0,0 +1,29 @@
+namespace NServiceBus.AzureFunctions.AzureServiceBus;
+
+using System;
+using System.Buffers.Binary;
+
+/// <summary>
+/// Provides helper methods for working with <see cref="Guid"/>.
+/// </summary>
+/// <remarks>
+/// Inspired by <a href="https://github.com/bgrainger/NGuid">NGuid</a> by Bradley Grainger,
+/// used under the <a href="https://github.com/bgrainger/NGuid/blob/master/LICENSE">MIT License</a>.
+/// </remarks>
+static class GuidHelper
+{
+    /// <summary>
+    /// Creates a Version 8 UUID with a v7-style layout: <paramref name="timestamp"/> as Unix
+    /// milliseconds in bytes 0–7 (time-sortable, stable across redeliveries) and
+    /// <paramref name="sequenceNumber"/> in bytes 8–15, both big-endian.
+    /// </summary>
+    public static Guid CreateVersion8(DateTimeOffset timestamp, long sequenceNumber)
+    {
+        Span<byte> guidBytes = stackalloc byte[16];
+        BinaryPrimitives.WriteInt64BigEndian(guidBytes, timestamp.ToUnixTimeMilliseconds());
+        BinaryPrimitives.WriteInt64BigEndian(guidBytes[8..], sequenceNumber);
+        guidBytes[6] = (byte)(0x80 | (guidBytes[6] & 0xF));
+        guidBytes[8] = (byte)(0x80 | (guidBytes[8] & 0x3F));
+        return new Guid(guidBytes, bigEndian: true);
+    }
+}