[Hold for next release] Shared queue worker loop in QueueBase

.agents/skills/foundatio/SKILL.md

@@ -319,6 +319,7 @@ public class OrderServiceTests : TestLoggerBase
 - **In-memory for tests**: In-memory implementations follow the same interfaces as production providers, but some behaviors differ (delivery delay limits, ordering, deduplication, group ids). Swap via DI for fast, isolated unit tests, and check [provider behavioral gaps](https://foundatio.dev/guide/provider-behavioral-gaps) before relying on provider-specific behavior.
 - **Queue `GroupId` is a hint**: It only changes delivery on SQS (fair queues on standard queues, strict ordering on `.fifo` queues) and session-enabled Azure Service Bus queues (ordered per group, one message at a time per group; `RetryDelay` is not applied, so an abandoned message retries ahead of the rest of its group; a group id of at most 128 characters is required). The other built-in queues store it as metadata and log one debug message. SQS group ids are max 128 characters with no spaces. Empty values are stored as `null`. A non-empty explicit `QueueEntryOptions.GroupId` overrides the queue's `GroupId(...)` resolver; null or empty invokes the resolver. Resolvers should return `null` for no group.
 - **`EnqueueAsync` copies `QueueEntryOptions`**: The caller's instance is never modified and can be reused. `Enqueuing` handlers get the per-message copy, with the resolved `GroupId` and the `Activity.Current` `CorrelationId`/`TraceState` already filled in. `Properties` is copied into a writable `Dictionary` per enqueue, keeping the key comparer of built-in dictionary types (`Dictionary`, `ConcurrentDictionary`, sorted and immutable dictionaries); other implementations use the default comparer.
+- **`QueueBase.StartWorker` handles worker errors**: Dequeue failures back off (about 1s doubling to 30s, with jitter); handler failures abandon the entry; abandon/auto-complete failures are logged and the lock expiry redelivers. Cancellation or disposal stops the worker. `InMemoryQueue` uses this loop; other `QueueBase` providers opt in by calling `StartWorker(...)` from `StartWorkingImpl` and reporting `WorkerErrorCount` as `QueueStats.Errors`.
 
 ## NuGet Packages
 

docs/guide/implementations/aws.md

@@ -178,6 +178,8 @@ Things to know:
 
 Deleting a queue takes up to 60 seconds in SQS, and a send during that window can succeed and then be lost with the queue ([DeleteQueue](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_DeleteQueue.html)). Stop producers and workers before `DeleteQueueAsync` and wait at least 60 seconds before reusing the name. After a `QueueDoesNotExistException`, the cached queue url is cleared so the next call recreates the queue.
 
+For shared-worker error handling, see [Worker Error Handling](../queues.md#worker-error-handling).
+
 ## Next Steps
 
 - [File Storage Guide](/guide/storage) - Usage patterns

docs/guide/queues.md

@@ -376,6 +376,19 @@ await queue.StartWorkingAsync(
 );
 ```
 
+#### Worker Error Handling
+
+`InMemoryQueue` workers started with `StartWorkingAsync` keep running through transient failures. Other providers get the same behavior when they delegate to `QueueBase.StartWorker`:
+
+| Failure | What the worker does | Counted in `QueueStats.Errors` |
+|---------|---------------------|--------------------------------|
+| Dequeue throws (network, provider outage, throwing `Dequeued` handler) | Logs the error and waits before the next dequeue: about 1s, doubling up to 30s, with jitter. The delay resets after a dequeue that doesn't throw. | Yes |
+| Your handler throws | Logs the error and abandons the entry (unless your handler already completed or abandoned it) so it is retried or dead-lettered. | Yes |
+| Abandon or auto-complete throws | Retries through the queue's resilience policy, then logs and keeps working. The entry becomes available again when its lock expires. | Auto-complete only |
+| The worker's cancellation token is cancelled, or the queue is disposed | Stops. A handler that throws `OperationCanceledException` because of that cancellation isn't counted, and its entry is abandoned. | No |
+
+Because entries can be redelivered after a failed complete or abandon, handlers should be idempotent.
+
 ## Queue Entry Options
 
 Configure enqueue behavior:
@@ -883,6 +896,7 @@ If you are writing a custom `IQueue<T>` implementation by extending `QueueBase<T
 - **`EnsureQueueCreatedAsync`** always receives `DisposedCancellationToken`. Use it for all setup operations (lock acquisition, API calls, etc.).
 - **`DequeueImplAsync`** receives a linked token (caller + disposal). Respect it for the wait/poll operation.
 - **`EnqueueImplAsync`** does not receive a cancellation token — keep enqueue fast and non-blocking.
+- **`StartWorkingImpl`** should call `StartWorker(handler, autoComplete, cancellationToken)` to use the shared loop (see [Worker Error Handling](#worker-error-handling)). Report `WorkerErrorCount` as `QueueStats.Errors` and call `ResetWorkerErrorCount()` when stats are reset.
 
 ## Best Practices
 

src/Foundatio.TestHarness/Queue/QueueTestBase.cs

@@ -961,6 +961,127 @@ await queue.StartWorkingAsync(w =>
         }
     }
 
+    public virtual async Task StartWorkingAsync_WhenDequeueThrows_KeepsWorkingAsync()
+    {
+        using var queue = GetQueue(retries: 1, retryDelay: TimeSpan.Zero);
+        if (queue == null)
+            return;
+
+        using var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+        try
+        {
+            await queue.DeleteQueueAsync();
+            await AssertEmptyQueueAsync(queue);
+
+            int dequeuedCount = 0;
+            using var dequeuedHandler = queue.Dequeued.AddSyncHandler((_, _) =>
+            {
+                if (Interlocked.Increment(ref dequeuedCount) == 1)
+                    throw new InvalidOperationException("Simulated dequeue failure");
+            });
+
+            var processed = new AsyncManualResetEvent(false);
+            await queue.StartWorkingAsync((entry, _) =>
+            {
+                processed.Set();
+                return Task.CompletedTask;
+            }, autoComplete: true, cancellationToken: cancellationTokenSource.Token);
+
+            await queue.EnqueueAsync(new SimpleWorkItem { Data = "First" });
+            await queue.EnqueueAsync(new SimpleWorkItem { Data = "Second" });
+
+            await processed.WaitAsync(cancellationTokenSource.Token);
+            Assert.True(dequeuedCount >= 2);
+        }
+        finally
+        {
+            await cancellationTokenSource.CancelAsync();
+            await CleanupQueueAsync(queue);
+        }
+    }
+
+    public virtual async Task StartWorkingAsync_WhenAbandonThrows_KeepsWorkingAsync()
+    {
+        using var queue = GetQueue(retries: 0, retryDelay: TimeSpan.Zero);
+        if (queue == null)
+            return;
+
+        using var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+        try
+        {
+            await queue.DeleteQueueAsync();
+            await AssertEmptyQueueAsync(queue);
+
+            int abandonedCount = 0;
+            using var abandonedHandler = queue.Abandoned.AddSyncHandler((_, _) =>
+            {
+                if (Interlocked.Increment(ref abandonedCount) == 1)
+                    throw new InvalidOperationException("Simulated abandon failure");
+            });
+
+            var processedSecond = new AsyncManualResetEvent(false);
+            await queue.StartWorkingAsync((entry, _) =>
+            {
+                if (entry.Value.Data == "First")
+                    throw new InvalidOperationException("Simulated handler failure");
+
+                processedSecond.Set();
+                return Task.CompletedTask;
+            }, autoComplete: true, cancellationToken: cancellationTokenSource.Token);
+
+            await queue.EnqueueAsync(new SimpleWorkItem { Data = "First" });
+            await queue.EnqueueAsync(new SimpleWorkItem { Data = "Second" });
+
+            await processedSecond.WaitAsync(cancellationTokenSource.Token);
+            Assert.True(abandonedCount >= 1);
+        }
+        finally
+        {
+            await cancellationTokenSource.CancelAsync();
+            await CleanupQueueAsync(queue);
+        }
+    }
+
+    public virtual async Task StartWorkingAsync_WhenCancelled_StopsWithoutWorkerErrorsAsync()
+    {
+        using var queue = GetQueue(retries: 0);
+        if (queue == null)
+            return;
+
+        try
+        {
+            await queue.DeleteQueueAsync();
+            await AssertEmptyQueueAsync(queue);
+
+            int handlerCalls = 0;
+            using var workerCancellationTokenSource = new CancellationTokenSource();
+            await queue.StartWorkingAsync((_, _) =>
+            {
+                Interlocked.Increment(ref handlerCalls);
+                return Task.CompletedTask;
+            }, autoComplete: true, cancellationToken: workerCancellationTokenSource.Token);
+
+            await Task.Delay(TimeSpan.FromMilliseconds(500), TestCancellationToken);
+            await workerCancellationTokenSource.CancelAsync();
+            await Task.Delay(TimeSpan.FromSeconds(1), TestCancellationToken);
+
+            await queue.EnqueueAsync(new SimpleWorkItem { Data = "Hello" });
+            await Task.Delay(TimeSpan.FromSeconds(2), TestCancellationToken);
+
+            Assert.Equal(0, handlerCalls);
+            if (_assertStats)
+            {
+                var stats = await queue.GetQueueStatsAsync();
+                Assert.Equal(0, stats.Errors);
+                Assert.Equal(1, stats.Queued);
+            }
+        }
+        finally
+        {
+            await CleanupQueueAsync(queue);
+        }
+    }
+
     public virtual async Task WorkItemsWillTimeoutAsync()
     {
         using var queue = GetQueue(retryDelay: TimeSpan.Zero, workItemTimeout: TimeSpan.FromSeconds(1));

src/Foundatio/Queues/InMemoryQueue.cs

@@ -24,7 +24,6 @@ public class InMemoryQueue<T> : QueueBase<T, InMemoryQueueOptions<T>> where T :
     private int _dequeuedCount;
     private int _completedCount;
     private int _abandonedCount;
-    private int _workerErrorCount;
     private int _workerItemTimeoutCount;
     private int _pendingRetryCount;
 
@@ -59,7 +58,7 @@ protected override QueueStats GetMetricsQueueStats()
             Dequeued = _dequeuedCount,
             Completed = _completedCount,
             Abandoned = _abandonedCount,
-            Errors = _workerErrorCount,
+            Errors = WorkerErrorCount,
             Timeouts = _workerItemTimeoutCount
         };
     }
@@ -127,71 +126,7 @@ public IReadOnlyCollection<QueueEntry<T>> GetDeadletterEntries()
 
     protected override void StartWorkingImpl(Func<IQueueEntry<T>, CancellationToken, Task> handler, bool autoComplete, CancellationToken cancellationToken)
     {
-        ArgumentNullException.ThrowIfNull(handler);
-
-        _logger.LogTrace("Queue {QueueName} start working", _options.Name);
-
-        var linkedCancellationTokenSource = GetLinkedDisposableCancellationTokenSource(cancellationToken);
-        _workers.Add(Task.Run(async () =>
-        {
-            using var _ = new DisposableAction(linkedCancellationTokenSource.Dispose);
-            _logger.LogTrace("WorkerLoop Start {QueueName}", _options.Name);
-
-            while (!linkedCancellationTokenSource.IsCancellationRequested)
-            {
-                _logger.LogTrace("WorkerLoop Signaled {QueueName}", _options.Name);
-
-                IQueueEntry<T>? queueEntry = null;
-                try
-                {
-                    queueEntry = await DequeueImplAsync(linkedCancellationTokenSource.Token).AnyContext();
-                }
-                catch (Exception ex)
-                {
-                    _logger.LogError(ex, "Error on Dequeue: {Message}", ex.Message);
-                }
-
-                if (linkedCancellationTokenSource.IsCancellationRequested || queueEntry is null)
-                    return;
-
-                try
-                {
-                    await handler(queueEntry, linkedCancellationTokenSource.Token).AnyContext();
-                }
-                catch (Exception ex)
-                {
-                    _logger.LogError(ex, "Worker error: {Message}", ex.Message);
-
-                    if (!queueEntry.IsAbandoned && !queueEntry.IsCompleted)
-                    {
-                        try
-                        {
-                            await _resiliencePolicy.ExecuteAsync(async _ => await queueEntry.AbandonAsync(), linkedCancellationTokenSource.Token).AnyContext();
-                        }
-                        catch (Exception abandonEx)
-                        {
-                            _logger.LogError(abandonEx, "Worker error abandoning queue entry: {Message}", abandonEx.Message);
-                        }
-                    }
-
-                    Interlocked.Increment(ref _workerErrorCount);
-                }
-
-                if (autoComplete && !queueEntry.IsAbandoned && !queueEntry.IsCompleted)
-                {
-                    try
-                    {
-                        await _resiliencePolicy.ExecuteAsync(async _ => await queueEntry.CompleteAsync(), linkedCancellationTokenSource.Token).AnyContext();
-                    }
-                    catch (Exception ex)
-                    {
-                        _logger.LogError(ex, "Worker error attempting to auto complete entry: {Message}", ex.Message);
-                    }
-                }
-            }
-
-            _logger.LogTrace("Worker exiting: {QueueName} Cancel Requested: {IsCancellationRequested}", _options.Name, linkedCancellationTokenSource.IsCancellationRequested);
-        }, linkedCancellationTokenSource.Token).ContinueWith(_ => linkedCancellationTokenSource.Dispose()));
+        _workers.Add(StartWorker(handler, autoComplete, cancellationToken));
     }
 
     protected override async Task<IQueueEntry<T>?> DequeueImplAsync(CancellationToken linkedCancellationToken)
@@ -380,7 +315,7 @@ protected override Task DeleteQueueImplAsync()
         _dequeuedCount = 0;
         _completedCount = 0;
         _abandonedCount = 0;
-        _workerErrorCount = 0;
+        ResetWorkerErrorCount();
         _pendingRetryCount = 0;
 
         return Task.CompletedTask;