[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.AzureStorage/Foundatio.AzureStorage.csproj

@@ -7,7 +7,7 @@
     <PackageReference Include="Azure.Storage.Blobs" Version="12.29.2" />
     <PackageReference Include="Azure.Storage.Queues" Version="12.27.1" />
 
-    <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.AzureStorage/Queues/AzureStorageQueue.cs

@@ -22,7 +22,6 @@ public class AzureStorageQueue<T> : QueueBase<T, AzureStorageQueueOptions<T>> wh
     private long _dequeuedCount;
     private long _completedCount;
     private long _abandonedCount;
-    private long _workerErrorCount;
     private bool _queueCreated;
 
     public AzureStorageQueue(AzureStorageQueueOptions<T> options) : base(options)
@@ -87,6 +86,7 @@ await Task.WhenAll(
             {
                 CorrelationId = options.CorrelationId,
                 Properties = options.Properties,
+                GroupId = options.GroupId,
                 Data = data
             };
             messageBodyBytes = _serializer.SerializeToBytes(envelope);
@@ -103,7 +103,10 @@ await Task.WhenAll(
             visibilityTimeout: options.DeliveryDelay,
             cancellationToken: CancellationToken.None).AnyContext();
 
-        var entry = new QueueEntry<T>(response.Value.MessageId, null, data, this, _timeProvider.GetLocalNow().UtcDateTime, 0);
+        var entry = new QueueEntry<T>(response.Value.MessageId, null, data, this, _timeProvider.GetLocalNow().UtcDateTime, 0)
+        {
+            GroupId = _options.CompatibilityMode == AzureStorageQueueCompatibilityMode.Default ? options.GroupId : null
+        };
         await OnEnqueuedAsync(entry).AnyContext();
 
         _logger.LogTrace("Enqueued message {MessageId}", response.Value.MessageId);
@@ -179,6 +182,7 @@ await Task.WhenAll(
 
         T? data;
         string? correlationId = null;
+        string? groupId = null;
         IDictionary<string, string>? properties = null;
         Exception? deserializeException = null;
 
@@ -194,6 +198,7 @@ await Task.WhenAll(
                     {
                         data = envelope.Data;
                         correlationId = envelope.CorrelationId;
+                        groupId = envelope.GroupId;
                         properties = envelope.Properties;
                     }
                     else
@@ -231,7 +236,7 @@ await Task.WhenAll(
             return null;
         }
 
-        var entry = new AzureStorageQueueEntry<T>(message, correlationId, properties, data, this);
+        var entry = new AzureStorageQueueEntry<T>(message, correlationId, properties, data, this) { GroupId = groupId };
 
         if (entry.Attempts > _options.Retries + 1)
         {
@@ -340,7 +345,7 @@ protected override async Task<QueueStats> GetQueueStatsImplAsync()
                 Dequeued = _dequeuedCount,
                 Completed = _completedCount,
                 Abandoned = _abandonedCount,
-                Errors = _workerErrorCount,
+                Errors = WorkerErrorCount,
                 Timeouts = 0
             };
 
@@ -360,7 +365,7 @@ protected override async Task<QueueStats> GetQueueStatsImplAsync()
             Dequeued = _dequeuedCount,
             Completed = _completedCount,
             Abandoned = _abandonedCount,
-            Errors = _workerErrorCount,
+            Errors = WorkerErrorCount,
             Timeouts = 0 // Azure handles visibility timeout natively; client-side tracking not meaningful
         };
     }
@@ -377,7 +382,7 @@ protected override QueueStats GetMetricsQueueStats()
                 Dequeued = _dequeuedCount,
                 Completed = _completedCount,
                 Abandoned = _abandonedCount,
-                Errors = _workerErrorCount,
+                Errors = WorkerErrorCount,
                 Timeouts = 0
             };
 
@@ -396,7 +401,7 @@ protected override QueueStats GetMetricsQueueStats()
             Dequeued = _dequeuedCount,
             Completed = _completedCount,
             Abandoned = _abandonedCount,
-            Errors = _workerErrorCount,
+            Errors = WorkerErrorCount,
             Timeouts = 0
         };
     }
@@ -414,57 +419,15 @@ await Task.WhenAll(
         _dequeuedCount = 0;
         _completedCount = 0;
         _abandonedCount = 0;
-        _workerErrorCount = 0;
+        ResetWorkerErrorCount();
 
         sw.Stop();
         _logger.LogTrace("Deleting queue took {Elapsed:g}", sw.Elapsed);
     }
 
     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>? queueEntry = null;
-                try
-                {
-                    queueEntry = await DequeueImplAsync(linkedCancellationToken.Token).AnyContext();
-                }
-                catch (OperationCanceledException)
-                {
-                    // Ignore cancellation
-                }
-
-                if (linkedCancellationToken.IsCancellationRequested || queueEntry == null)
-                    continue;
-
-                try
-                {
-                    await handler(queueEntry, linkedCancellationToken.Token).AnyContext();
-                    if (autoComplete && !queueEntry.IsAbandoned && !queueEntry.IsCompleted)
-                        await queueEntry.CompleteAsync().AnyContext();
-                }
-                catch (Exception ex)
-                {
-                    Interlocked.Increment(ref _workerErrorCount);
-                    _logger.LogError(ex, "Worker error: {Message}", ex.Message);
-
-                    if (!queueEntry.IsAbandoned && !queueEntry.IsCompleted)
-                        await queueEntry.AbandonAsync().AnyContext();
-                }
-            }
-
-            _logger.LogTrace("Worker exiting: {QueueName} Cancel Requested: {IsCancellationRequested}", _options.Name, linkedCancellationToken.IsCancellationRequested);
-        }, linkedCancellationToken.Token).ContinueWith(t => linkedCancellationToken.Dispose());
+        _ = StartWorker(handler, autoComplete, cancellationToken);
     }
 
     private async Task DeadLetterMessageAsync(AzureStorageQueueEntry<T> entry)
@@ -482,6 +445,7 @@ private async Task DeadLetterMessageAsync(AzureStorageQueueEntry<T> entry)
             {
                 CorrelationId = entry.CorrelationId,
                 Properties = entry.Properties,
+                GroupId = entry.GroupId,
                 Data = entry.Value
             };
             messageBody = new BinaryData(_serializer.SerializeToBytes(envelope));
@@ -520,6 +484,11 @@ internal record QueueMessageEnvelope<T> where T : class
     /// </summary>
     public IDictionary<string, string>? Properties { get; init; }
 
+    /// <summary>
+    /// Logical group or tenant key the message was enqueued with
+    /// </summary>
+    public string? GroupId { get; init; }
+
     /// <summary>
     /// The actual message payload
     /// </summary>

src/Foundatio.AzureStorage/Queues/AzureStorageQueueOptions.cs

@@ -10,12 +10,13 @@ public enum AzureStorageQueueCompatibilityMode
 {
     /// <summary>
     /// Default mode: Uses message envelope for full metadata support.
-    /// Supports CorrelationId and Properties. New installations should use this mode.
+    /// Supports GroupId, CorrelationId and Properties. New installations should use this mode.
     /// </summary>
     Default = 0,
 
     /// <summary>
     /// Legacy mode: Raw message body with Base64 encoding for v11 SDK compatibility.
+    /// GroupId, CorrelationId and Properties are not persisted or included in the Enqueued entry.
     /// Use this for backward compatibility with existing queues that have messages
     /// written with the v11 Microsoft.Azure.Storage.Queue SDK.
     /// </summary>
@@ -41,7 +42,7 @@ public class AzureStorageQueueOptions<T> : SharedQueueOptions<T> where T : class
 
     /// <summary>
     /// Controls message format compatibility.
-    /// Default mode uses an envelope wrapper that supports CorrelationId and Properties.
+    /// Default mode uses an envelope wrapper that supports GroupId, CorrelationId and Properties.
     /// Legacy mode is provided for backward compatibility with existing queues that have messages
     /// written with the v11 Microsoft.Azure.Storage.Queue SDK (uses Base64 encoding and raw payload).
     ///

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>