diff --git a/src/DodoSSH.Client.Transfer/FileTransferQueue.cs b/src/DodoSSH.Client.Transfer/FileTransferQueue.cs index d1c2a1b..69a624f 100644 --- a/src/DodoSSH.Client.Transfer/FileTransferQueue.cs +++ b/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; + + /// Whether a pump owns the queue: set when one is started, cleared by the pump as it finds it empty. + /// + /// Guarded by , and the reason it exists rather than asking 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. + /// + private bool pumping; private bool disposed; /// @@ -451,13 +460,39 @@ public sealed class FileTransferQueue : IAsyncDisposable /// private void EnsurePumping() { - if (pump.IsCompleted && !disposed) + if (!pumping && !disposed) { + pumping = true; pump = Task.Run(() => PumpAsync(lifetime.Token), lifetime.Token); } } + /// + /// Leaves by one of two doors. The ordinary one is 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. + /// 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 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; } /// Takes the next queued transfer and marks it running. @@ -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; } diff --git a/tests/DodoSSH.Client.Transfer.Tests/FileTransferQueueTests.cs b/tests/DodoSSH.Client.Transfer.Tests/FileTransferQueueTests.cs index d6119fd..d77808f 100644 --- a/tests/DodoSSH.Client.Transfer.Tests/FileTransferQueueTests.cs +++ b/tests/DodoSSH.Client.Transfer.Tests/FileTransferQueueTests.cs @@ -304,6 +304,36 @@ public sealed class FileTransferQueueTests : IDisposable finished.Failure.ShouldBe("the host is not reachable"); } + /// + /// + /// 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. + /// + /// + /// 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. + /// + /// + [Fact] + public async Task ATransferQueuedAsTheLastOneFinishes_StillRuns() + { + await using var queue = new FileTransferQueue( + _ => Task.FromException(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(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 {