Transfers: stop a transfer queued as the last one finishes from waiting for ever

src/DodoSSH.Client.Transfer/FileTransferQueue.cs

@@ -194,6 +194,15 @@ public sealed class FileTransferQueue : IAsyncDisposable
     private readonly CancellationTokenSource lifetime = new();
 
     private Task pump = Task.CompletedTask;
+
+    /// <summary>Whether a pump owns the queue: set when one is started, cleared by the pump as it finds it empty.</summary>
+    /// <remarks>
+    /// Guarded by <see cref="gate"/>, and the reason it exists rather than asking <see cref="pump"/> whether it
+    /// has finished. The pump decides it is done under the gate, but its task is only marked complete after it
+    /// has let go — so a transfer queued in between found a pump that was still "running", started none, and
+    /// stayed queued for good. This is cleared in the same locked step that found nothing to take.
+    /// </remarks>
+    private bool pumping;
     private bool disposed;
 
     /// <param name="sessions">
@@ -451,13 +460,39 @@ public sealed class FileTransferQueue : IAsyncDisposable
     /// </remarks>
     private void EnsurePumping()
     {
-        if (pump.IsCompleted && !disposed)
+        if (!pumping && !disposed)
         {
+            pumping = true;
             pump = Task.Run(() => PumpAsync(lifetime.Token), lifetime.Token);
         }
     }
 
+    /// <remarks>
+    /// Leaves by one of two doors. The ordinary one is <see cref="TakeNext"/> finding the queue empty, which
+    /// hands the queue back in the same locked step. Anything else — a fault escaping a row — hands it back
+    /// on the way out, so the next transfer queued starts a pump rather than waiting on one that is gone.
+    /// </remarks>
     private async Task PumpAsync(CancellationToken cancellationToken)
+    {
+        var drained = false;
+
+        try
+        {
+            drained = await PumpUntilEmptyAsync(cancellationToken).ConfigureAwait(false);
+        }
+        finally
+        {
+            if (!drained)
+            {
+                lock (gate)
+                {
+                    pumping = false;
+                }
+            }
+        }
+    }
+
+    private async Task<bool> PumpUntilEmptyAsync(CancellationToken cancellationToken)
     {
         while (TakeNext() is { } entry)
         {
@@ -489,6 +524,8 @@ public sealed class FileTransferQueue : IAsyncDisposable
 
             await RunAsync(entry, session, cancellationToken).ConfigureAwait(false);
         }
+
+        return true;
     }
 
     /// <summary>Takes the next queued transfer and marks it running.</summary>
@@ -502,6 +539,10 @@ public sealed class FileTransferQueue : IAsyncDisposable
 
             if (next is null)
             {
+                // In the same locked step that found nothing, so a transfer queued from here on starts a
+                // pump of its own. See pumping.
+                pumping = false;
+
                 return null;
             }
 

tests/DodoSSH.Client.Transfer.Tests/FileTransferQueueTests.cs

@@ -304,6 +304,36 @@ public sealed class FileTransferQueueTests : IDisposable
         finished.Failure.ShouldBe("the host is not reachable");
     }
 
+    /// <remarks>
+    /// <para>
+    /// The pump decides the queue is empty under the lock and only finishes its task after letting go, so a
+    /// transfer queued in that gap used to find a pump that still looked busy, start none, and sit queued for
+    /// ever. Queuing each transfer the moment the last one reports finishing is what lands in that gap.
+    /// </para>
+    /// <para>
+    /// A race, so it is run many times rather than arranged once; on the code that had it, a few hundred
+    /// rounds strand a transfer every time. The session factory refuses at once, which keeps each round to
+    /// the queue's own bookkeeping — the part the race is in.
+    /// </para>
+    /// </remarks>
+    [Fact]
+    public async Task ATransferQueuedAsTheLastOneFinishes_StillRuns()
+    {
+        await using var queue = new FileTransferQueue(
+            _ => Task.FromException<IRemoteFileStore>(new IOException("refused at once")),
+            TimeProvider.System);
+
+        for (var round = 0; round < 500; round++)
+        {
+            var finished = await RunToCompletionAsync(
+                queue, TransferDirection.Download, Local($"round-{round}"), Remote($"round-{round}"), 1);
+
+            finished.State.ShouldBe(TransferState.Failed, $"round {round}");
+        }
+
+        queue.IsBusy.ShouldBeFalse();
+    }
+
     private FileTransferQueue NewQueue() =>
         new(_ => Task.FromResult<IRemoteFileStore>(host), TimeProvider.System);
 
@@ -371,7 +401,7 @@ public sealed class FileTransferQueueTests : IDisposable
         {
             start();
 
-            return await finished.Task.WaitAsync(TimeSpan.FromSeconds(30), Token);
+            return await finished.Task.WaitAsync(TimeSpan.FromSeconds(10), Token);
         }
         finally
         {