DodoSSH/tests/DodoSSH.Client.Terminal.Tests/TerminalSessionPumpTests.cs

using System.Diagnostics;
using System.Globalization;
using DodoSSH.Client.Ssh;

namespace DodoSSH.Client.Terminal.Tests;

/// <summary>
/// The output path under load: coalescing, backpressure, and shutdown.
/// </summary>
/// <remarks>
/// The plan calls for a throughput harness built before any UI, and this is it. Terminal throughput is
/// where a client like this usually fails — <c>cat</c> on a large file either freezes the UI or grows
/// memory until the process dies — and neither symptom is diagnosable once a WebView is in the picture.
/// </remarks>
public sealed class TerminalSessionPumpTests
{
    private const uint SessionId = 7;

    [Fact]
    public async Task Output_ReachesTheTransportTaggedWithItsSession()
    {
        await using var session = new FakeShellSession(bytesToProduce: 64);
        var transport = new RecordingTransport();

        await using var pump = CreatePump(session, transport);
        transport.AutoAcknowledge = pump;

        await pump.RunAsync(TestContext.Current.CancellationToken);

        transport.OutputBytes().Length.ShouldBe(64);

        foreach (var frame in transport.Frames)
        {
            TerminalFrame.TryRead(frame, out _, out var sessionId, out _).ShouldBeTrue();
            sessionId.ShouldBe(SessionId);
        }
    }

    [Fact]
    public async Task ASessionOpenedFrame_PrecedesAnyOutput()
    {
        // The renderer has to attach a terminal before bytes arrive for it, or the first screenful is
        // written into nothing.
        await using var session = new FakeShellSession(bytesToProduce: 16);
        var transport = new RecordingTransport();

        await using var pump = CreatePump(session, transport);
        transport.AutoAcknowledge = pump;

        await pump.RunAsync(TestContext.Current.CancellationToken);

        TerminalFrame.TryRead(transport.Frames[0], out var opcode, out _, out _).ShouldBeTrue();
        opcode.ShouldBe((byte)TerminalServerOpcode.SessionOpened);
    }

    [Fact]
    public async Task SeveralReadsWithinOneInterval_BecomeASingleFrame()
    {
        // A terminal cannot render faster than the display refreshes, so sending each read separately
        // is work whose result is overwritten before anyone sees it.
        await using var session = new FakeShellSession(bytesToProduce: 8 * 1024);
        var transport = new RecordingTransport();

        await using var pump = CreatePump(
            session,
            transport,
            new TerminalPumpOptions
            {
                ReadBufferBytes = 512,
                FlushInterval = TimeSpan.FromMilliseconds(400),
            });

        transport.AutoAcknowledge = pump;

        var run = pump.RunAsync(TestContext.Current.CancellationToken);

        // Long enough for one flush, short enough that a second cannot have happened.
        await Task.Delay(TimeSpan.FromMilliseconds(600), TestContext.Current.CancellationToken);

        // 8 KiB arrived as sixteen 512-byte reads and left as one frame.
        transport.CountOf(TerminalServerOpcode.Output).ShouldBe(1);
        transport.OutputBytes().Length.ShouldBe(8 * 1024);

        await run;
    }

    [Fact]
    public async Task WhenCreditRunsOut_ThePumpStopsReading()
    {
        // The property the whole design rests on. With no acknowledgement the pump must read exactly
        // the window and then stop, so nothing accumulates between the socket and the screen. Left
        // unbounded, a remote running `yes` grows the client's memory until it dies.
        const int Window = 4 * 1024;

        await using var session = new FakeShellSession(bytesToProduce: long.MaxValue);
        var transport = new RecordingTransport();

        // Deliberately no AutoAcknowledge: this models a renderer that has stopped keeping up.
        await using var pump = CreatePump(
            session,
            transport,
            new TerminalPumpOptions
            {
                ReadBufferBytes = 1024,
                FlushInterval = TimeSpan.FromMilliseconds(5),
                WindowBytes = Window,
            });

        var run = pump.RunAsync(TestContext.Current.CancellationToken);

        await Task.Delay(TimeSpan.FromMilliseconds(300), TestContext.Current.CancellationToken);

        pump.BytesRead.ShouldBe(
            Window,
            "The pump read past its credit window, so output accumulates without bound.");

        // Still stalled a moment later, rather than merely slow.
        await Task.Delay(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);
        pump.BytesRead.ShouldBe(Window);

        pump.Credits.Available.ShouldBe(0);

        await pump.DisposeAsync();
        await run;
    }

    [Fact]
    public async Task AcknowledgingCredit_LetsThePumpResume()
    {
        const int Window = 4 * 1024;

        await using var session = new FakeShellSession(bytesToProduce: long.MaxValue);
        var transport = new RecordingTransport();

        await using var pump = CreatePump(
            session,
            transport,
            new TerminalPumpOptions
            {
                ReadBufferBytes = 1024,
                FlushInterval = TimeSpan.FromMilliseconds(5),
                WindowBytes = Window,
            });

        var run = pump.RunAsync(TestContext.Current.CancellationToken);

        await Task.Delay(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);
        pump.BytesRead.ShouldBe(Window);

        pump.Acknowledge(2048);

        await Task.Delay(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);

        pump.BytesRead.ShouldBe(Window + 2048);

        await pump.DisposeAsync();
        await run;
    }

    [Fact]
    public async Task AnOverLargeAcknowledgement_DoesNotWidenTheWindow()
    {
        // The acknowledgement comes from JavaScript. A page claiming to have rendered a gigabyte must
        // not be able to talk the host into an unbounded read.
        const int Window = 4 * 1024;

        await using var session = new FakeShellSession(bytesToProduce: long.MaxValue);
        var transport = new RecordingTransport();

        await using var pump = CreatePump(
            session,
            transport,
            new TerminalPumpOptions
            {
                ReadBufferBytes = 1024,
                FlushInterval = TimeSpan.FromMilliseconds(5),
                WindowBytes = Window,
            });

        var run = pump.RunAsync(TestContext.Current.CancellationToken);
        await Task.Delay(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);

        pump.Acknowledge(uint.MaxValue);

        await Task.Delay(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);

        // One further window's worth at most, never everything the endless producer could offer.
        pump.BytesRead.ShouldBeLessThanOrEqualTo(Window * 2);

        await pump.DisposeAsync();
        await run;
    }

    [Fact]
    public async Task WhenTheRemoteCloses_ASessionClosedFrameFollowsTheLastOutput()
    {
        // The last thing a remote writes is often the most important — an error, an exit code — and
        // dropping it makes a clean exit look like a crash.
        await using var session = new FakeShellSession(bytesToProduce: 4096);
        var transport = new RecordingTransport();

        await using var pump = CreatePump(
            session,
            transport,
            new TerminalPumpOptions { FlushInterval = TimeSpan.FromMilliseconds(20) });

        transport.AutoAcknowledge = pump;

        await pump.RunAsync(TestContext.Current.CancellationToken);

        transport.OutputBytes().Length.ShouldBe(4096);

        TerminalFrame.TryRead(transport.Frames[^1], out var opcode, out _, out var payload)
            .ShouldBeTrue();

        opcode.ShouldBe((byte)TerminalServerOpcode.SessionClosed);
        System.Text.Encoding.UTF8.GetString(payload).ShouldNotBeNullOrWhiteSpace();
    }

    /// <remarks>
    /// <c>exit</c>, as the pump sees it: the read loop runs out and nothing latched a transport failure on
    /// the way. This is the ending that must not be reported as a dropped connection, because it is the
    /// ordinary end of a session and there is nothing wrong with the machine.
    /// </remarks>
    [Fact]
    public async Task WhenTheShellExits_TheOutcomeIsTheRemoteClosingIt()
    {
        await using var session = new FakeShellSession(bytesToProduce: 32);
        var transport = new RecordingTransport();

        await using var pump = CreatePump(session, transport);
        transport.AutoAcknowledge = pump;

        await pump.RunAsync(TestContext.Current.CancellationToken);

        pump.Outcome.End.ShouldBe(SshSessionEnd.RemoteClosed);
        pump.Outcome.Description.ShouldBe("The remote closed the session.");
    }

    /// <remarks>
    /// ◆ A reboot, as the pump sees it. The shell's read ends exactly as it does for <c>exit</c> — there is
    /// no separate signal on that path — so the only thing telling the two apart is the failure the session
    /// latched off the transport before the read was woken. Without it, a machine that went away is reported
    /// as somebody having typed <c>exit</c>, which is the state this whole wave exists to stop drawing.
    /// </remarks>
    [Fact]
    public async Task WhenTheTransportFailed_TheOutcomeIsALostConnectionCarryingWhatFailed()
    {
        await using var session = new FakeShellSession(bytesToProduce: 32)
        {
            TransportFailure = "An established connection was aborted by the server.",
        };

        var transport = new RecordingTransport();

        await using var pump = CreatePump(session, transport);
        transport.AutoAcknowledge = pump;

        await pump.RunAsync(TestContext.Current.CancellationToken);

        pump.Outcome.End.ShouldBe(SshSessionEnd.ConnectionLost);
        pump.Outcome.Description.ShouldBe("An established connection was aborted by the server.");

        // And the renderer is told the same sentence, so the line in the scrollback and whatever the shell
        // draws around the tab cannot disagree.
        TerminalFrame.TryRead(transport.Frames[^1], out var opcode, out _, out var payload).ShouldBeTrue();

        opcode.ShouldBe((byte)TerminalServerOpcode.SessionClosed);
        System.Text.Encoding.UTF8.GetString(payload)
            .ShouldBe("An established connection was aborted by the server.");
    }

    /// <remarks>
    /// The third ending, and the one nothing draws a status for: this client asked. A tab being closed and a
    /// reconnect starting both arrive here, and neither is news to whoever caused it.
    /// </remarks>
    [Fact]
    public async Task WhenThisClientEndsIt_TheOutcomeSaysSoRatherThanBlamingTheRemote()
    {
        await using var session = new FakeShellSession(bytesToProduce: long.MaxValue);
        var transport = new RecordingTransport();

        var pump = CreatePump(session, transport);
        transport.AutoAcknowledge = pump;

        var run = pump.RunAsync(TestContext.Current.CancellationToken);

        await pump.DisposeAsync();
        await run;

        pump.Outcome.End.ShouldBe(SshSessionEnd.ClosedLocally);
    }

    [Fact]
    public async Task Input_ReachesTheRemote()
    {
        await using var session = new FakeShellSession();
        var transport = new RecordingTransport();

        await using var pump = CreatePump(session, transport);

        await pump.WriteInputAsync("ls -la\r"u8.ToArray(), TestContext.Current.CancellationToken);

        System.Text.Encoding.UTF8.GetString(session.Written).ShouldBe("ls -la\r");
    }

    [Fact]
    public async Task AResize_ReachesTheRemote()
    {
        await using var session = new FakeShellSession();
        var transport = new RecordingTransport();

        await using var pump = CreatePump(session, transport);

        pump.Resize(new TerminalSize(132, 43, 1320, 1075));

        session.ResizeCount.ShouldBe(1);
        session.LastResize.ShouldBe(new TerminalSize(132, 43, 1320, 1075));
    }

    [Fact]
    public async Task AZeroSizedResize_IsDropped()
    {
        // Arrives naturally when a pane collapses or a window minimises. Forwarding it leaves the
        // remote's idea of the terminal nonsensical until the next resize.
        await using var session = new FakeShellSession();
        var transport = new RecordingTransport();

        await using var pump = CreatePump(session, transport);

        pump.Resize(new TerminalSize(0, 0));

        session.ResizeCount.ShouldBe(0);
    }

    [Fact]
    public async Task ThroughputHarness_SustainsTenMegabytesPerSecondWithBoundedMemory()
    {
        // The plan's target: `yes` at full tilt must hold a fixed memory ceiling. Throughput is
        // asserted well below what the machine manages, because a CI runner under load is slower than
        // a desktop and a flaky performance test gets deleted rather than fixed. The bounded-memory
        // assertion is the one that carries the real meaning.
        const long Target = 32L * 1024 * 1024;
        const int Window = CreditWindow.DefaultWindowBytes;

        await using var session = new FakeShellSession(bytesToProduce: Target);
        var transport = new RecordingTransport();

        await using var pump = CreatePump(
            session,
            transport,
            new TerminalPumpOptions
            {
                ReadBufferBytes = 32 * 1024,
                FlushInterval = TimeSpan.FromMilliseconds(1),
                WindowBytes = Window,
            });

        transport.AutoAcknowledge = pump;

        using var monitor = new CancellationTokenSource();
        var sampler = SamplePeakOutstandingAsync(pump, monitor.Token);

        var stopwatch = Stopwatch.StartNew();
        await pump.RunAsync(TestContext.Current.CancellationToken);
        stopwatch.Stop();

        await monitor.CancelAsync();
        var peakOutstanding = await sampler;

        pump.BytesRead.ShouldBe(Target);
        transport.OutputBytes().Length.ShouldBe((int)Target);

        var megabytesPerSecond = Target / 1024d / 1024d / stopwatch.Elapsed.TotalSeconds;

        megabytesPerSecond.ShouldBeGreaterThan(
            10,
            string.Create(
                CultureInfo.InvariantCulture,
                $"Sustained {megabytesPerSecond:F1} MB/s, below the 10 MB/s target."));

        // The real assertion: in-flight bytes never exceeded the window, however fast the producer ran.
        peakOutstanding.ShouldBeLessThanOrEqualTo(
            Window,
            string.Create(
                CultureInfo.InvariantCulture,
                $"Peak in-flight was {peakOutstanding} bytes against a {Window}-byte window."));
    }

    /// <summary>Watches in-flight bytes for the duration of a run, returning the highest seen.</summary>
    private static async Task<int> SamplePeakOutstandingAsync(
        TerminalSessionPump pump,
        CancellationToken cancellationToken)
    {
        var peak = 0;

        while (!cancellationToken.IsCancellationRequested)
        {
            peak = Math.Max(peak, pump.Credits.Outstanding);

            try
            {
                await Task.Delay(1, cancellationToken);
            }
            catch (OperationCanceledException)
            {
                break;
            }
        }

        return peak;
    }

    private static TerminalSessionPump CreatePump(
        FakeShellSession session,
        ITerminalTransport transport,
        TerminalPumpOptions? options = null) =>
        new(
            SessionId,
            session,
            transport,
            TimeProvider.System,
            options ?? new TerminalPumpOptions { FlushInterval = TimeSpan.FromMilliseconds(10) });
}