[Hold for next release] Use the shared QueueBase worker loop

.github/workflows/build.yml

@@ -3,7 +3,7 @@ on: [push, pull_request]
 
 jobs:
   build:
-    uses: FoundatioFx/Foundatio/.github/workflows/build-workflow.yml@main
+    uses: FoundatioFx/Foundatio/.github/workflows/build-workflow.yml@a28d0f736d9124431de09764c56aff77fbdc1d5d
     secrets: inherit
     with:
       new-test-runner: true

src/Foundatio.AzureServiceBus/Foundatio.AzureServiceBus.csproj

@@ -4,7 +4,7 @@
     <PackageReference Include="Azure.Messaging.ServiceBus" Version="7.21.0" />
     <PackageReference Include="Azure.Identity" Version="1.21.0" />
 
-    <PackageReference Include="Foundatio" Version="13.0.4" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
+    <PackageReference Include="Foundatio" Version="13.0.5-preview.queue-worker-loop.0.19" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
     <ProjectReference Include="..\..\..\Foundatio\src\Foundatio\Foundatio.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
   </ItemGroup>
 </Project>

src/Foundatio.AzureServiceBus/Queues/AzureServiceBusQueue.cs

@@ -26,7 +26,6 @@ public class AzureServiceBusQueue<T> : QueueBase<T, AzureServiceBusQueueOptions<
     private long _dequeuedCount;
     private long _completedCount;
     private long _abandonedCount;
-    private long _workerErrorCount;
 
     public AzureServiceBusQueue(AzureServiceBusQueueOptions<T> options) : base(options)
     {
@@ -176,7 +175,7 @@ protected override async Task DeleteQueueImplAsync()
         _dequeuedCount = 0;
         _completedCount = 0;
         _abandonedCount = 0;
-        _workerErrorCount = 0;
+        ResetWorkerErrorCount();
     }
 
     private async Task DrainQueueAsync()
@@ -249,7 +248,7 @@ protected override async Task<QueueStats> GetQueueStatsImplAsync()
                 Dequeued = _dequeuedCount,
                 Completed = _completedCount,
                 Abandoned = _abandonedCount,
-                Errors = _workerErrorCount,
+                Errors = WorkerErrorCount,
                 Timeouts = 0
             };
 
@@ -265,7 +264,7 @@ protected override async Task<QueueStats> GetQueueStatsImplAsync()
                 Dequeued = _dequeuedCount,
                 Completed = _completedCount,
                 Abandoned = _abandonedCount,
-                Errors = _workerErrorCount,
+                Errors = WorkerErrorCount,
                 Timeouts = 0
             };
         }
@@ -280,7 +279,7 @@ protected override async Task<QueueStats> GetQueueStatsImplAsync()
             Dequeued = _dequeuedCount,
             Completed = _completedCount,
             Abandoned = _abandonedCount,
-            Errors = _workerErrorCount,
+            Errors = WorkerErrorCount,
             Timeouts = 0
         };
     }
@@ -307,6 +306,8 @@ protected override Task<IEnumerable<T>> GetDeadletterItemsImplAsync(Cancellation
 
         if (options is AzureServiceBusQueueEntryOptions asbOptions && !String.IsNullOrEmpty(asbOptions.SessionId))
             message.SessionId = asbOptions.SessionId;
+        else if (!String.IsNullOrEmpty(options.GroupId))
+            message.SessionId = options.GroupId;
 
         if (options.DeliveryDelay.HasValue && options.DeliveryDelay.Value > TimeSpan.Zero)
             message.ScheduledEnqueueTime = DateTimeOffset.UtcNow.Add(options.DeliveryDelay.Value);
@@ -321,7 +322,10 @@ protected override Task<IEnumerable<T>> GetDeadletterItemsImplAsync(Cancellation
 
         await _queueSender!.SendMessageAsync(message).AnyContext();
 
-        var entry = new QueueEntry<T>(message.MessageId, message.CorrelationId, data, this, _timeProvider.GetUtcNow().UtcDateTime, 0);
+        var entry = new QueueEntry<T>(message.MessageId, message.CorrelationId, data, this, _timeProvider.GetUtcNow().UtcDateTime, 0)
+        {
+            GroupId = String.IsNullOrEmpty(message.SessionId) ? null : message.SessionId
+        };
 
         foreach (var property in message.ApplicationProperties)
         {
@@ -338,6 +342,8 @@ protected override Task<IEnumerable<T>> GetDeadletterItemsImplAsync(Cancellation
     // TODO: See if we can simplify this.
     protected override async Task<IQueueEntry<T>?> DequeueImplAsync(CancellationToken linkedCancellationToken)
     {
+        ThrowIfSessionsRequired();
+
         // Calculate timeout based on cancellation state - like SQS pattern
         var timeout = linkedCancellationToken.IsCancellationRequested
             ? TimeSpan.FromMilliseconds(100)
@@ -490,7 +496,8 @@ public override async Task AbandonAsync(IQueueEntry<T> queueEntry)
                 {
                     MessageId = entry.UnderlyingMessage.MessageId,
                     CorrelationId = entry.UnderlyingMessage.CorrelationId,
-                    ScheduledEnqueueTime = DateTimeOffset.UtcNow.Add(retryDelay)
+                    ScheduledEnqueueTime = DateTimeOffset.UtcNow.Add(retryDelay),
+                    SessionId = entry.UnderlyingMessage.SessionId
                 };
 
                 // Copy application properties (excluding SDK diagnostic properties)
@@ -503,10 +510,11 @@ public override async Task AbandonAsync(IQueueEntry<T> queueEntry)
                 // Store attempt count for next dequeue
                 retryMessage.ApplicationProperties["_attempts"] = entry.Attempts;
 
-                // Complete the original message and schedule the retry
-                await _queueReceiver!.CompleteMessageAsync(entry.UnderlyingMessage).AnyContext();
-
+                // Send the retry before completing the original: a failed send cannot lose the message, and a failed
+                // complete redelivers it, which is the documented at-least-once Service Bus contract.
                 await _queueSender!.SendMessageAsync(retryMessage).AnyContext();
+
+                await _queueReceiver!.CompleteMessageAsync(entry.UnderlyingMessage).AnyContext();
             }
             else
             {
@@ -527,47 +535,14 @@ public override async Task AbandonAsync(IQueueEntry<T> queueEntry)
 
     protected override void StartWorkingImpl(Func<IQueueEntry<T>, CancellationToken, Task> handler, bool autoComplete, CancellationToken cancellationToken)
     {
-        ArgumentNullException.ThrowIfNull(handler);
-
-        var linkedCancellationToken = GetLinkedDisposableCancellationTokenSource(cancellationToken);
-
-        Task.Run(async () =>
-        {
-            _logger.LogTrace("WorkerLoop Start {QueueName}", _options.Name);
-
-            while (!linkedCancellationToken.IsCancellationRequested)
-            {
-                _logger.LogTrace("WorkerLoop Signaled {QueueName}", _options.Name);
-
-                IQueueEntry<T>? entry = null;
-                try
-                {
-                    entry = await DequeueImplAsync(linkedCancellationToken.Token).AnyContext();
-                }
-                catch (OperationCanceledException) { }
-
-                if (linkedCancellationToken.IsCancellationRequested || entry == null)
-                    continue;
-
-                try
-                {
-                    await handler(entry, linkedCancellationToken.Token).AnyContext();
-
-                    if (autoComplete && !entry.IsAbandoned && !entry.IsCompleted && !linkedCancellationToken.IsCancellationRequested)
-                        await entry.CompleteAsync().AnyContext();
-                }
-                catch (Exception ex)
-                {
-                    Interlocked.Increment(ref _workerErrorCount);
-                    _logger.LogError(ex, "Worker error: {Message}", ex.Message);
-
-                    if (!entry.IsAbandoned && !entry.IsCompleted && !linkedCancellationToken.IsCancellationRequested)
-                        await entry.AbandonAsync().AnyContext();
-                }
-            }
+        ThrowIfSessionsRequired();
+        _ = StartWorker(handler, autoComplete, cancellationToken);
+    }
 
-            _logger.LogTrace("Worker exiting: {QueueName} IsCancellationRequested={IsCancellationRequested}", _options.Name, linkedCancellationToken.IsCancellationRequested);
-        }, linkedCancellationToken.Token).ContinueWith(_ => linkedCancellationToken.Dispose());
+    private void ThrowIfSessionsRequired()
+    {
+        if (_options.RequiresSession == true)
+            throw new NotSupportedException("Session-enabled queues are not supported for receiving by AzureServiceBusQueue; sending with SessionId works.");
     }
 
     private CreateQueueOptions CreateQueueOptions()

src/Foundatio.AzureServiceBus/Queues/AzureServiceBusQueueEntry.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Linq;
 using Azure.Messaging.ServiceBus;
 using Foundatio.AzureServiceBus.Utility;
@@ -21,6 +22,7 @@ public AzureServiceBusQueueEntry(ServiceBusReceivedMessage message, T value, IQu
         }
 
         UnderlyingMessage = message;
+        GroupId = String.IsNullOrEmpty(message.SessionId) ? null : message.SessionId;
     }
 
     /// <summary>

src/Foundatio.AzureServiceBus/Queues/AzureServiceBusQueueEntryOptions.cs

@@ -4,5 +4,8 @@ namespace Foundatio.AzureServiceBus.Queues;
 
 public record AzureServiceBusQueueEntryOptions : QueueEntryOptions
 {
+    /// <summary>
+    /// The Service Bus session id. Overrides <see cref="QueueEntryOptions.GroupId"/> when both are set.
+    /// </summary>
     public string? SessionId { get; set; }
 }

tests/Directory.Build.props

@@ -12,7 +12,7 @@
     <PackageReference Include="xunit.runner.visualstudio" Version="4.0.0" />
     <PackageReference Include="GitHubActionsTestLogger" Version="3.0.5" PrivateAssets="All" />
 
-    <PackageReference Include="Foundatio.TestHarness" Version="13.0.4" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
+    <PackageReference Include="Foundatio.TestHarness" Version="13.0.5-preview.queue-worker-loop.0.19" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
     <ProjectReference Include="..\..\..\Foundatio\src\Foundatio.TestHarness\Foundatio.TestHarness.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
   </ItemGroup>
 </Project>

tests/Foundatio.AzureServiceBus.Tests/Queues/AzureServiceBusQueueTests.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Threading.Tasks;
+using Foundatio.AzureServiceBus.Queues;
 using Foundatio.Queues;
 using Foundatio.Serializer;
 using Foundatio.Tests.Queue;
@@ -155,6 +156,24 @@ public override Task CanHandleErrorInWorkerAsync()
         return base.CanHandleErrorInWorkerAsync();
     }
 
+    [Fact]
+    public override Task StartWorkingAsync_WhenDequeueThrows_KeepsWorkingAsync()
+    {
+        return base.StartWorkingAsync_WhenDequeueThrows_KeepsWorkingAsync();
+    }
+
+    [Fact]
+    public override Task StartWorkingAsync_WhenAbandonThrows_KeepsWorkingAsync()
+    {
+        return base.StartWorkingAsync_WhenAbandonThrows_KeepsWorkingAsync();
+    }
+
+    [Fact]
+    public override Task StartWorkingAsync_WhenCancelled_StopsWithoutWorkerErrorsAsync()
+    {
+        return base.StartWorkingAsync_WhenCancelled_StopsWithoutWorkerErrorsAsync();
+    }
+
     [Fact]
     public override async Task CanHaveMultipleQueueInstancesAsync()
     {
@@ -242,6 +261,18 @@ public override Task DequeueAsync_AfterAbandonWithMutatedValue_ReturnsOriginalVa
         return base.DequeueAsync_AfterAbandonWithMutatedValue_ReturnsOriginalValueAsync();
     }
 
+    [Fact]
+    public override Task AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync()
+    {
+        return base.AbandonAsync_WithGroupId_PreservesGroupIdOnRetryAsync();
+    }
+
+    [Fact]
+    public override Task AbandonAsync_WithGroupIdAndRetryDelay_PreservesGroupIdOnRetryAsync()
+    {
+        return base.AbandonAsync_WithGroupIdAndRetryDelay_PreservesGroupIdOnRetryAsync();
+    }
+
     [Fact]
     public override Task DequeueAsync_WithDispose_AutoAbandonsEntryAsync()
     {
@@ -318,12 +349,122 @@ public override Task DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync(
         return base.DuplicateDetection_WithNullIdentifier_AcceptsAllItemsAsync();
     }
 
+    [Fact]
+    public override Task EnqueueAsync_WhenEnqueuingHandlerClearsGroupId_EnqueuesWithoutGroupAsync()
+    {
+        return base.EnqueueAsync_WhenEnqueuingHandlerClearsGroupId_EnqueuesWithoutGroupAsync();
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithEmptyGroupId_EnqueuesWithoutGroupAsync()
+    {
+        return base.EnqueueAsync_WithEmptyGroupId_EnqueuesWithoutGroupAsync();
+    }
+
+    [Fact]
+    public async Task EnqueueAsync_WithGroupId_DoesNotClaimGroupedDeliveryAsync()
+    {
+        // Arrange: the queue is not session-enabled and cannot receive from sessions, so GroupId cannot affect delivery
+        using var queue = GetQueue();
+        if (queue is null)
+            return;
+
+        try
+        {
+            await queue.DeleteQueueAsync();
+            await AssertEmptyQueueAsync(queue);
+
+            // Act
+            await queue.EnqueueAsync(new SimpleWorkItem { Data = "no-grouped-delivery" }, new QueueEntryOptions { GroupId = "tenant-123" });
+
+            // Assert
+            Assert.Contains(Log.LogEntries, e => e.LogLevel == LogLevel.Debug && e.Message.Contains("delivery order or fairness"));
+            var entry = await queue.DequeueAsync(TimeSpan.FromSeconds(5));
+            Assert.NotNull(entry);
+            await entry.CompleteAsync();
+        }
+        finally
+        {
+            await CleanupQueueAsync(queue);
+        }
+    }
+    [Fact]
+    public override Task EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync()
+    {
+        return base.EnqueueAsync_WithGroupId_RoundTripsGroupIdAsync();
+    }
+
+    [Fact]
+    public async Task EnqueueAsync_WithGroupId_SendsGroupIdAsSessionIdAsync()
+    {
+        // Arrange
+        using var queue = GetQueue();
+        if (queue is null)
+            return;
+
+        try
+        {
+            await queue.DeleteQueueAsync();
+            await AssertEmptyQueueAsync(queue);
+
+            // Act
+            await queue.EnqueueAsync(new SimpleWorkItem { Data = "session-id-test" }, new QueueEntryOptions { GroupId = "tenant-123" });
+
+            // Assert
+            var entry = await queue.DequeueAsync(TimeSpan.FromSeconds(5));
+            var serviceBusEntry = Assert.IsType<AzureServiceBusQueueEntry<SimpleWorkItem>>(entry);
+            Assert.Equal("tenant-123", serviceBusEntry.UnderlyingMessage.SessionId);
+            Assert.Equal("tenant-123", serviceBusEntry.GroupId);
+            await serviceBusEntry.CompleteAsync();
+        }
+        finally
+        {
+            await CleanupQueueAsync(queue);
+        }
+    }
+
     [Fact]
     public override Task EnqueueAsync_WithSerializationError_ThrowsAndLeavesQueueEmptyAsync()
     {
         return base.EnqueueAsync_WithSerializationError_ThrowsAndLeavesQueueEmptyAsync();
     }
 
+    [Fact]
+    public async Task EnqueueAsync_WithSessionIdAndGroupId_UsesSessionIdAsync()
+    {
+        // Arrange
+        using var queue = GetQueue();
+        if (queue is null)
+            return;
+
+        try
+        {
+            await queue.DeleteQueueAsync();
+            await AssertEmptyQueueAsync(queue);
+
+            // Act
+            await queue.EnqueueAsync(new SimpleWorkItem { Data = "session-id-test" },
+                new AzureServiceBusQueueEntryOptions { GroupId = "tenant-123", SessionId = "session-456" });
+
+            // Assert
+            var entry = await queue.DequeueAsync(TimeSpan.FromSeconds(5));
+            var serviceBusEntry = Assert.IsType<AzureServiceBusQueueEntry<SimpleWorkItem>>(entry);
+            Assert.Equal("session-456", serviceBusEntry.UnderlyingMessage.SessionId);
+            Assert.Equal("session-456", serviceBusEntry.GroupId);
+            await serviceBusEntry.CompleteAsync();
+        }
+        finally
+        {
+            await CleanupQueueAsync(queue);
+        }
+    }
+
+    [Fact]
+    public override Task EnqueueAsync_WithReusedOptions_DoesNotChangeCallerOptionsAsync()
+    {
+        return base.EnqueueAsync_WithReusedOptions_DoesNotChangeCallerOptionsAsync();
+    }
+
     [Fact]
     public override Task EnqueueAsync_WithUniqueId_UsesProvidedIdAsync()
     {