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

Author: niemyjski

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,12 +340,14 @@ 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);
                 _deadletterQueue.Enqueue(targetEntry);
+                Interlocked.Decrement(ref _pendingRetryCount);
             }
         }
     }
@@ -373,6 +381,7 @@ protected override Task DeleteQueueImplAsync()
         _completedCount = 0;
         _abandonedCount = 0;
         _workerErrorCount = 0;
+        _pendingRetryCount = 0;
 
         return Task.CompletedTask;
     }