Public Access
Move files to and from a host over SFTP
M2's file transfer, built bottom-up: an SFTP session on the SSH layer, a transfer queue in a project of its own, and the two-pane browser the design asked for replacing the screen that said it did not exist. Remote listings carry names, sizes, modification times and a real drwxr-xr-x — nothing in this repository could render a POSIX mode before — and the queue moves one file at a time with progress, throughput and resume. The design import assumed this would be an SFTP subsystem channel on ISshConnection, beside the shell on a transport that is already up. SSH.NET does not offer that: SftpClient derives from BaseClient and owns its own transport, and there is no supported way to hand it an SshClient's session. So file transfer opens a second authenticated connection, and it is named for that rather than dressed up as a channel — OpenSftpAsync is on ISftpSessionFactory, not on a connection. The difference is visible to a user: the host records a second login, and a host whose password is typed each time asks for it again on this screen. It goes through the same host key gate, the same pin and the same two refusals a shell does, so a fingerprint approved for a terminal is approved here and one approved here reaches the other machines with the next sync. docs/design-import-gaps.md is corrected, and marked as the one row where what shipped differs from what it predicted. Nothing is written at its final name until it is complete. Every transfer goes to a .dodossh-part file beside its destination and is renamed into place at the end, so an interrupted transfer can never be mistaken for a finished one — which matters most for what this screen is actually for, which is copying a build artefact onto a server and then running it. A destination that already exists is refused outright rather than overwritten: the queue has no way to ask, and silently replacing a file somebody's process is serving is the worse of the two failures. The remote pane has DELETE and MKDIR so that refusal is not a dead end. A test against the container pins the assumption underneath all of this — that SFTP's rename does not clobber. Resume works within a run of the application and not across a restart, and the limit is deliberate rather than unfinished. Nothing records which source wrote a part file, and resuming one on the strength of its name matching is how a corrupt artefact gets delivered with nothing reporting a failure; a part file found at startup is started over. Making it survive a restart needs the preferences store this client still has not got. The offset a resume starts at is the part file's own length rather than the transfer's recorded progress: a cancellation can land between a write completing and the counter moving, and only one of those two is a fact about the bytes that are there. The queue and its connection outlive a lock, as shells do. LockAsync already argues that locking must not destroy work in flight — it is what somebody does when they walk away from the machine, which is exactly when a long transfer is most likely to be running — so TransfersViewModel is created once and the vault is attached on unlock and detached on lock. What locking takes is the host list, and it has to: those rows carry decrypted secrets. DodoSSH.Client.Transfer is a new project rather than more of Client.Ssh. The two answer different questions — one is about reaching a host, the other about moving bytes and what to do when moving them stops halfway — and this is the only client project that deliberately touches the local filesystem. Three defects the tests found, none of which review would have. SftpPath.Name answered an empty string for the root. NavigateRemoteAsync wrapped itself in the busy guard, so navigating from inside another command did nothing at all and the remote pane simply stayed empty after connecting, with no failure anywhere to explain it. And opening an SFTP session per test made two handshakes per test — this client learns a host key by being refused — which pushed the SSH assembly past sshd's MaxStartups and failed a different few unrelated tests each run; the session is shared through the fixture now, with the reason written where the next person will hit it. 1004 tests green across 18 projects, 24 of them new: the SFTP subsystem against the OpenSSH container, the queue against a real temporary directory and a fake host, and three more layout measurements because a screen this window has never laid out is a screen never checked. Not verified: the screen has not been looked at running. The layout harness measures it at the window's minimum in three shapes, which is the class of defect that has shipped here before, but reaching it in the application needs the compose stack, the migrations, the API and a browser sign-in. What is still absent — the status bar's transfer count, dragging between the panes, transferring a directory, and sftp over a bastion — is in docs/design-import-gaps.md.
This commit is contained in:
@@ -0,0 +1,365 @@
|
||||
using DodoSSH.Client.Ssh;
|
||||
|
||||
namespace DodoSSH.Client.Transfer.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// The transfer queue: what ends up on disk, and what happens when a transfer stops halfway.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Every assertion here is about a promise the queue makes in prose — nothing is written at its final name
|
||||
/// until it is complete, a destination that already exists is refused rather than overwritten, an
|
||||
/// interrupted transfer carries on rather than starting again, and a partial file left by something else is
|
||||
/// never resumed from. Those are the four ways a file transfer can quietly deliver the wrong bytes.
|
||||
/// </remarks>
|
||||
public sealed class FileTransferQueueTests : IDisposable
|
||||
{
|
||||
private const string RemoteDirectory = "/home/dodo";
|
||||
|
||||
private readonly FakeSftpSession host = new();
|
||||
private readonly string workspace = Directory.CreateTempSubdirectory("dodossh-transfer-").FullName;
|
||||
|
||||
private static CancellationToken Token => TestContext.Current.CancellationToken;
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Dispose()
|
||||
{
|
||||
try
|
||||
{
|
||||
Directory.Delete(workspace, recursive: true);
|
||||
}
|
||||
catch (IOException)
|
||||
{
|
||||
// A handle the runtime has not released yet. The directory is under the system temporary path
|
||||
// and failing a passing test over it would be the wrong trade.
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ADownload_LandsAtItsFinalNameWithNoPartFileLeftBehind()
|
||||
{
|
||||
var content = Bytes(300_000);
|
||||
host.Seed(Remote("artefact.tar"), content);
|
||||
|
||||
await using var queue = NewQueue();
|
||||
|
||||
var finished = await RunToCompletionAsync(
|
||||
queue, TransferDirection.Download, Local("artefact.tar"), Remote("artefact.tar"), content.Length);
|
||||
|
||||
finished.State.ShouldBe(TransferState.Completed);
|
||||
finished.Transferred.ShouldBe(content.Length);
|
||||
|
||||
(await File.ReadAllBytesAsync(Local("artefact.tar"), Token)).ShouldBe(content);
|
||||
File.Exists(Local("artefact.tar") + FileTransferQueue.PartSuffix).ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ADownloadOntoAFileThatIsAlreadyThere_IsRefusedAndLeavesItAlone()
|
||||
{
|
||||
host.Seed(Remote("artefact.tar"), Bytes(1_000));
|
||||
await File.WriteAllTextAsync(Local("artefact.tar"), "something else entirely", Token);
|
||||
|
||||
await using var queue = NewQueue();
|
||||
|
||||
var finished = await RunToCompletionAsync(
|
||||
queue, TransferDirection.Download, Local("artefact.tar"), Remote("artefact.tar"), 1_000);
|
||||
|
||||
finished.State.ShouldBe(TransferState.Failed);
|
||||
finished.Failure.ShouldNotBeNull();
|
||||
|
||||
// The whole point of the refusal: what was there is still there, unread and unreplaced.
|
||||
(await File.ReadAllTextAsync(Local("artefact.tar"), Token)).ShouldBe("something else entirely");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task AnInterruptedDownload_CarriesOnFromWhereItStopped()
|
||||
{
|
||||
var content = Bytes(300_000);
|
||||
host.Seed(Remote("artefact.tar"), content);
|
||||
host.FailReadAfter = 100_000;
|
||||
|
||||
await using var queue = NewQueue();
|
||||
|
||||
var broken = await RunToCompletionAsync(
|
||||
queue, TransferDirection.Download, Local("artefact.tar"), Remote("artefact.tar"), content.Length);
|
||||
|
||||
broken.State.ShouldBe(TransferState.Failed);
|
||||
broken.Transferred.ShouldBeInRange(1, content.Length - 1);
|
||||
broken.CanResume.ShouldBeTrue();
|
||||
|
||||
// The part file is what makes this a resume rather than a restart, and it is deliberately not at the
|
||||
// destination's name — nothing may look like a finished download until it is one.
|
||||
var part = Local("artefact.tar") + FileTransferQueue.PartSuffix;
|
||||
File.Exists(part).ShouldBeTrue();
|
||||
File.Exists(Local("artefact.tar")).ShouldBeFalse();
|
||||
|
||||
var partLength = new FileInfo(part).Length;
|
||||
|
||||
var resumed = await AwaitFinishAsync(queue, () => queue.Retry(broken.Id));
|
||||
|
||||
resumed.State.ShouldBe(TransferState.Completed);
|
||||
|
||||
// Asked for the second time at the offset the part file had reached, which is the only direct
|
||||
// evidence that the bytes already moved were not moved again.
|
||||
host.ReadOffsets.ShouldBe([0, partLength]);
|
||||
|
||||
(await File.ReadAllBytesAsync(Local("artefact.tar"), Token)).ShouldBe(content);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task AFreshDownload_WillNotResumeFromAPartFileItDidNotWrite()
|
||||
{
|
||||
var content = Bytes(200_000);
|
||||
host.Seed(Remote("artefact.tar"), content);
|
||||
|
||||
// Litter from something else entirely — an earlier run of the application, or an earlier attempt at
|
||||
// a file of the same name that has since changed. Resuming on the strength of the name matching is
|
||||
// how a corrupt artefact gets delivered with nothing reporting a failure.
|
||||
await File.WriteAllBytesAsync(
|
||||
Local("artefact.tar") + FileTransferQueue.PartSuffix, Bytes(50_000, seed: 7), Token);
|
||||
|
||||
await using var queue = NewQueue();
|
||||
|
||||
var finished = await RunToCompletionAsync(
|
||||
queue, TransferDirection.Download, Local("artefact.tar"), Remote("artefact.tar"), content.Length);
|
||||
|
||||
finished.State.ShouldBe(TransferState.Completed);
|
||||
|
||||
host.ReadOffsets.ShouldBe([0]);
|
||||
(await File.ReadAllBytesAsync(Local("artefact.tar"), Token)).ShouldBe(content);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task AnUpload_LandsAtItsFinalNameOnTheHost()
|
||||
{
|
||||
var content = Bytes(150_000);
|
||||
await File.WriteAllBytesAsync(Local("artefact.tar"), content, Token);
|
||||
|
||||
await using var queue = NewQueue();
|
||||
|
||||
var finished = await RunToCompletionAsync(
|
||||
queue, TransferDirection.Upload, Local("artefact.tar"), Remote("artefact.tar"), content.Length);
|
||||
|
||||
finished.State.ShouldBe(TransferState.Completed);
|
||||
|
||||
host.Read(Remote("artefact.tar")).ShouldBe(content);
|
||||
host.Exists(Remote("artefact.tar") + FileTransferQueue.PartSuffix).ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task AnUploadOntoAFileThatIsAlreadyThere_IsRefusedAndLeavesItAlone()
|
||||
{
|
||||
var existing = "the running deployment"u8.ToArray();
|
||||
host.Seed(Remote("artefact.tar"), existing);
|
||||
await File.WriteAllBytesAsync(Local("artefact.tar"), Bytes(1_000), Token);
|
||||
|
||||
await using var queue = NewQueue();
|
||||
|
||||
var finished = await RunToCompletionAsync(
|
||||
queue, TransferDirection.Upload, Local("artefact.tar"), Remote("artefact.tar"), 1_000);
|
||||
|
||||
finished.State.ShouldBe(TransferState.Failed);
|
||||
host.Read(Remote("artefact.tar")).ShouldBe(existing);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task TransfersRunOneAtATime()
|
||||
{
|
||||
for (var i = 0; i < 4; i++)
|
||||
{
|
||||
host.Seed(Remote($"file-{i}"), Bytes(200_000, seed: i));
|
||||
}
|
||||
|
||||
await using var queue = NewQueue();
|
||||
|
||||
var running = new HashSet<Guid>();
|
||||
var most = 0;
|
||||
var gate = new Lock();
|
||||
|
||||
queue.Changed += (_, e) =>
|
||||
{
|
||||
lock (gate)
|
||||
{
|
||||
if (e.Transfer.State is TransferState.Running)
|
||||
{
|
||||
running.Add(e.Transfer.Id);
|
||||
}
|
||||
else
|
||||
{
|
||||
running.Remove(e.Transfer.Id);
|
||||
}
|
||||
|
||||
most = Math.Max(most, running.Count);
|
||||
}
|
||||
};
|
||||
|
||||
var ids = new List<Guid>();
|
||||
|
||||
for (var i = 0; i < 4; i++)
|
||||
{
|
||||
ids.Add(queue.Enqueue(
|
||||
TransferDirection.Download, Local($"file-{i}"), Remote($"file-{i}"), 200_000));
|
||||
}
|
||||
|
||||
await WaitForQuietAsync(queue);
|
||||
|
||||
queue.Snapshot().ShouldAllBe(transfer => transfer.State == TransferState.Completed);
|
||||
|
||||
// The claim the whole design rests on: one channel, one transfer, so the throughput on a row is the
|
||||
// throughput of the link rather than a share of it.
|
||||
lock (gate)
|
||||
{
|
||||
most.ShouldBe(1);
|
||||
}
|
||||
|
||||
ids.Count.ShouldBe(4);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task CancellingAQueuedTransfer_TakesItOutOfTheQueueWithoutStartingIt()
|
||||
{
|
||||
host.Seed(Remote("slow"), Bytes(2_000_000));
|
||||
host.Seed(Remote("never"), Bytes(1_000));
|
||||
|
||||
await using var queue = NewQueue();
|
||||
|
||||
queue.Enqueue(TransferDirection.Download, Local("slow"), Remote("slow"), 2_000_000);
|
||||
var second = queue.Enqueue(TransferDirection.Download, Local("never"), Remote("never"), 1_000);
|
||||
|
||||
queue.Cancel(second);
|
||||
|
||||
await WaitForQuietAsync(queue);
|
||||
|
||||
var cancelled = queue.Snapshot().Single(transfer => transfer.Id == second);
|
||||
|
||||
cancelled.State.ShouldBe(TransferState.Cancelled);
|
||||
cancelled.Transferred.ShouldBe(0);
|
||||
File.Exists(Local("never")).ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task DiscardingAStoppedDownload_TakesItsPartFileWithIt()
|
||||
{
|
||||
var content = Bytes(200_000);
|
||||
host.Seed(Remote("artefact.tar"), content);
|
||||
host.FailReadAfter = 60_000;
|
||||
|
||||
await using var queue = NewQueue();
|
||||
|
||||
var broken = await RunToCompletionAsync(
|
||||
queue, TransferDirection.Download, Local("artefact.tar"), Remote("artefact.tar"), content.Length);
|
||||
|
||||
var part = Local("artefact.tar") + FileTransferQueue.PartSuffix;
|
||||
File.Exists(part).ShouldBeTrue();
|
||||
|
||||
(await queue.DiscardAsync(broken.Id, Token)).ShouldBeTrue();
|
||||
|
||||
// The row was the only thing that knew the part file existed, so removing one without the other
|
||||
// would leave bytes on disk nobody could attribute.
|
||||
queue.Snapshot().ShouldBeEmpty();
|
||||
File.Exists(part).ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ATransferWithNoSessionToRunOn_FailsOnItsOwnRowRatherThanSilently()
|
||||
{
|
||||
await using var queue = new FileTransferQueue(
|
||||
_ => Task.FromException<ISftpSession>(new IOException("the host is not reachable")),
|
||||
TimeProvider.System);
|
||||
|
||||
var finished = await RunToCompletionAsync(
|
||||
queue, TransferDirection.Download, Local("artefact.tar"), Remote("artefact.tar"), 10);
|
||||
|
||||
finished.State.ShouldBe(TransferState.Failed);
|
||||
finished.Failure.ShouldBe("the host is not reachable");
|
||||
}
|
||||
|
||||
private FileTransferQueue NewQueue() =>
|
||||
new(_ => Task.FromResult<ISftpSession>(host), TimeProvider.System);
|
||||
|
||||
private string Local(string name) => Path.Combine(workspace, name);
|
||||
|
||||
private static string Remote(string name) => SftpPath.Combine(RemoteDirectory, name);
|
||||
|
||||
/// <remarks>
|
||||
/// Repeatable rather than random, so a failure can be reproduced, and not all one byte, so a resumed
|
||||
/// transfer that started from the wrong offset produces a file that differs rather than one that happens
|
||||
/// to match.
|
||||
/// </remarks>
|
||||
private static byte[] Bytes(int length, int seed = 0)
|
||||
{
|
||||
var content = new byte[length];
|
||||
|
||||
for (var i = 0; i < length; i++)
|
||||
{
|
||||
content[i] = (byte)((i + seed) % 251);
|
||||
}
|
||||
|
||||
return content;
|
||||
}
|
||||
|
||||
private static Task<TransferSnapshot> RunToCompletionAsync(
|
||||
FileTransferQueue queue,
|
||||
TransferDirection direction,
|
||||
string localPath,
|
||||
string remotePath,
|
||||
long length) =>
|
||||
AwaitFinishAsync(queue, () => queue.Enqueue(direction, localPath, remotePath, length));
|
||||
|
||||
/// <summary>
|
||||
/// Subscribes, starts a transfer, and completes when a transfer reaches a state it will not leave.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// The subscription goes on <em>before</em> the transfer starts, because the pump runs on a thread-pool
|
||||
/// thread: a small transfer can be finished before <c>Enqueue</c> has returned its id, so subscribing
|
||||
/// afterwards would wait for an event that has already happened.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Which is also why it does not match on the id — there is nothing to match against yet. Every caller
|
||||
/// has exactly one transfer in flight, which is what makes "a transfer finished" and "this transfer
|
||||
/// finished" the same statement. The one test with several running uses
|
||||
/// <see cref="WaitForQuietAsync"/> instead.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
private static async Task<TransferSnapshot> AwaitFinishAsync(FileTransferQueue queue, Action start)
|
||||
{
|
||||
var finished = new TaskCompletionSource<TransferSnapshot>(
|
||||
TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
|
||||
void OnChanged(object? sender, TransferChangedEventArgs e)
|
||||
{
|
||||
if (e.Transfer.IsFinished)
|
||||
{
|
||||
finished.TrySetResult(e.Transfer);
|
||||
}
|
||||
}
|
||||
|
||||
queue.Changed += OnChanged;
|
||||
|
||||
try
|
||||
{
|
||||
start();
|
||||
|
||||
return await finished.Task.WaitAsync(TimeSpan.FromSeconds(30), Token);
|
||||
}
|
||||
finally
|
||||
{
|
||||
queue.Changed -= OnChanged;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Waits until nothing is queued or running.</summary>
|
||||
private static async Task WaitForQuietAsync(FileTransferQueue queue)
|
||||
{
|
||||
var deadline = TimeSpan.FromSeconds(30);
|
||||
var waited = TimeSpan.Zero;
|
||||
|
||||
while (queue.IsBusy && waited < deadline)
|
||||
{
|
||||
await Task.Delay(20, Token);
|
||||
waited += TimeSpan.FromMilliseconds(20);
|
||||
}
|
||||
|
||||
queue.IsBusy.ShouldBeFalse("the queue should have drained");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user