Fix race condition in InMemoryQueue.AbandonAsync causing flaky RunUntilEmptyAsync (#470)

* Fix race condition in InMemoryQueue causing RunUntilEmptyAsync to exit prematurely

During AbandonAsync, there is a window between removing an entry from
_dequeued and re-enqueueing it for retry where the item exists in
neither collection. If RunUntilEmptyAsync checks queue stats during
this gap, it sees Queued=0 + Working=0 and terminates the job loop
while retryable items are still in flight.

Add a _pendingRetryCount that bridges the gap: incremented before
TryRemove, decremented after the item lands in its destination
(re-queued, deadlettered, or scheduled for delayed retry). The count
is included in the Queued stat so the continuation callback sees
items in transit.

For delayed retries (RetryDelay > 0), the counter is decremented
immediately after scheduling since the item is intentionally parked
and RunUntilEmptyAsync should not spin-wait for it.

Fixes flaky CanRunQueueJobWithLockFailAsync test.

Made-with: Cursor

* Move _pendingRetryCount decrement before _deadletterQueue.Enqueue

In the deadletter path, the item moves out of Queued/Working entirely,
so decrementing _pendingRetryCount before enqueuing to the deadletter
queue avoids a transient overcount where the item appears in both the
Queued stat (via _pendingRetryCount) and the Deadletter stat (via
_deadletterQueue.Count).

The synchronous retry path intentionally keeps the current ordering
(Retry then Decrement) because decrementing first would re-open the
race window this PR fixes: between the decrement and _queue.Enqueue
inside Retry(), both _pendingRetryCount and _queue.Count would be
zero, allowing RunUntilEmptyAsync to exit prematurely.

Made-with: Cursor

Author: niemyjski

src/Foundatio/Extensions/CacheClientExtensions.cs

@@ -16,7 +16,7 @@ public static class CacheClientExtensions
     /// below any real-world cache TTL.
     /// </summary>
     public static readonly TimeSpan MinimumExpiration = TimeSpan.FromMilliseconds(5);
-    
+
     public static async Task<T> GetAsync<T>(this ICacheClient client, string key, T defaultValue)
     {
         var cacheValue = await client.GetAsync<T>(key).AnyContext();

src/Foundatio/Queues/InMemoryQueue.cs

@@ -26,6 +26,7 @@ public class InMemoryQueue<T> : QueueBase<T, InMemoryQueueOptions<T>> where T :
     private int _abandonedCount;
     private int _workerErrorCount;
     private int _workerItemTimeoutCount;
+    private int _pendingRetryCount;
 
     public InMemoryQueue() : this(o => o) { }
 
@@ -51,7 +52,7 @@ protected override QueueStats GetMetricsQueueStats()
     {
         return new QueueStats
         {
-            Queued = _queue.Count,
+            Queued = _queue.Count + _pendingRetryCount,
             Working = _dequeued.Count,
             Deadletter = _deadletterQueue.Count,
             Enqueued = _enqueuedCount,
@@ -292,8 +293,12 @@ public override async Task AbandonAsync(IQueueEntry<T> queueEntry)
         if (queueEntry.IsAbandoned || queueEntry.IsCompleted)
             throw new InvalidOperationException("Queue entry has already been completed or abandoned");
 
+        Interlocked.Increment(ref _pendingRetryCount);
+
         if (!_dequeued.TryRemove(queueEntry.Id, out var targetEntry) || targetEntry == null)
         {
+            Interlocked.Decrement(ref _pendingRetryCount);
+
             foreach (var kvp in _queue)
             {
                 if (kvp.Id == queueEntry.Id)
@@ -323,8 +328,9 @@ public override async Task AbandonAsync(IQueueEntry<T> queueEntry)
                 var retryEntry = targetEntry.CreateRetryEntry();
                 if (_options.RetryDelay > TimeSpan.Zero)
                 {
+                    Interlocked.Decrement(ref _pendingRetryCount);
                     _logger.LogTrace("Adding item to wait list for future retry: {QueueEntryId} Attempts: {QueueEntryAttempts}", queueEntry.Id, queueEntry.Attempts);
-                    var unawaited = Run.DelayedAsync(GetRetryDelay(targetEntry.Attempts), () =>
+                    _ = Run.DelayedAsync(GetRetryDelay(targetEntry.Attempts), () =>
                     {
                         Retry(retryEntry);
                         return Task.CompletedTask;
@@ -334,11 +340,13 @@ public override async Task AbandonAsync(IQueueEntry<T> queueEntry)
                 {
                     _logger.LogTrace("Adding item back to queue for retry: {QueueEntryId} Attempts: {QueueEntryAttempts}", queueEntry.Id, queueEntry.Attempts);
                     Retry(retryEntry);
+                    Interlocked.Decrement(ref _pendingRetryCount);
                 }
             }
             else
             {
                 _logger.LogInformation("Exceeded retry limit ({Attempts}/{Retries}), moving message {QueueEntryId} to dead letter", targetEntry.Attempts, _options.Retries, queueEntry.Id);
+                Interlocked.Decrement(ref _pendingRetryCount);
                 _deadletterQueue.Enqueue(targetEntry);
             }
         }
@@ -373,6 +381,7 @@ protected override Task DeleteQueueImplAsync()
         _completedCount = 0;
         _abandonedCount = 0;
         _workerErrorCount = 0;
+        _pendingRetryCount = 0;
 
         return Task.CompletedTask;
     }

src/Foundatio/Utility/FoundatioDiagnostics.cs

@@ -69,4 +69,4 @@ public static Activity AddException(this Activity activity, Exception exception,
         return activity.AddEvent(new ActivityEvent(ExceptionEventName, timestamp, exceptionTags));
     }
 }
-#endif
+#endif