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

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();
    }

    /// <summary>
    /// A bucket writes an object whole, so a retried upload starts again rather than resuming.
    /// </summary>
    /// <remarks>
    /// The failed attempt leaves a part object behind — a bucket keeps what was sent when the stream closes —
    /// and the retry used to ask for the offset that part had reached. A bucket refuses any offset, so every
    /// retry failed until the row was discarded, under a message saying it would start again.
    /// </remarks>
    [Fact]
    public async Task ARetriedUploadToAStoreThatCannotResume_StartsAgainAndLands()
    {
        var content = Bytes(300_000);
        await File.WriteAllBytesAsync(Local("artefact.tar"), content, Token);

        host.CanWriteFromAnOffset = false;
        host.FailWriteAfter = 100_000;

        await using var queue = NewQueue();

        var broken = await RunToCompletionAsync(
            queue, TransferDirection.Upload, Local("artefact.tar"), Remote("artefact.tar"), content.Length);

        broken.State.ShouldBe(TransferState.Failed);
        host.Exists(Remote("artefact.tar") + FileTransferQueue.PartSuffix).ShouldBeTrue("what a bucket keeps");

        var retried = await AwaitFinishAsync(queue, () => queue.Retry(broken.Id));

        retried.State.ShouldBe(TransferState.Completed, retried.Failure);
        host.Read(Remote("artefact.tar")).ShouldBe(content);
    }

    [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<IRemoteFileStore>(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");
    }

    /// <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);

    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(10), 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");
    }
}