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

using System.Diagnostics;
using System.Globalization;
using System.Text;
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;
    }

    /// <summary>
    /// A renderer that has stopped acknowledging does not stop the pump reading, and does not grow it.
    /// </summary>
    /// <remarks>
    /// ◆ THIS TEST REPLACES ONE THAT ASSERTED THE OPPOSITE, and the reversal is the point rather than a
    /// relaxation. The old contract was that the pump reads exactly one credit window and then stops,
    /// which was believed to close SSH's receive window and block the remote. It does not — see
    /// <see cref="OutputBacklog"/> — so all the old behaviour bought was an unbounded buffer inside
    /// SSH.NET instead of a bounded one here. What has to be true now is both halves at once: reading
    /// continues, so nothing piles up underneath, <em>and</em> what is held stays inside the backlog.
    /// </remarks>
    [Fact]
    public async Task WhenCreditRunsOut_ThePumpKeepsReadingAndTheBacklogHoldsTheLine()
    {
        const int Window = 4 * 1024;
        const int Backlog = 64 * 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,
                BacklogBytes = Backlog,
            });

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

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

        var readByNow = pump.BytesRead;

        readByNow.ShouldBeGreaterThan(
            Window,
            "The pump stopped reading, which leaves the flood accumulating inside SSH.NET instead.");

        pump.BytesPending.ShouldBeLessThanOrEqualTo(
            Backlog,
            "The backlog grew past its capacity, which is the unbounded growth it exists to prevent.");

        pump.BytesDropped.ShouldBeGreaterThan(
            0,
            "Nothing was dropped, so something is still holding what the renderer never took.");

        // Still reading a moment later, rather than having quietly stalled once the backlog filled.
        await Task.Delay(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);

        pump.BytesRead.ShouldBeGreaterThan(readByNow);
        pump.BytesPending.ShouldBeLessThanOrEqualTo(Backlog);

        pump.Credits.Available.ShouldBe(0);

        await pump.DisposeAsync();
        await run;
    }

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

        // Endless, so the backlog never completes while this is being measured: once it does, the tail
        // goes out regardless of credit and there is nothing for the window to hold back. Drops along the
        // way change which bytes are sent, never how many — and the notice they raise is its own opcode.
        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);

        // One window's worth has gone out and stopped there, because nothing has been rendered.
        var sentBeforeAcknowledgement = transport.OutputBytes().Length;
        sentBeforeAcknowledgement.ShouldBe(Window);

        pump.Acknowledge(2048);

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

        transport.OutputBytes().Length.ShouldBe(sentBeforeAcknowledgement + 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 filling its own buffer without limit.
        const int Window = 4 * 1024;

        // Endless for the reason the test above gives: a backlog that completes stops the window mattering.
        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.
        transport.OutputBytes().Length.ShouldBeLessThanOrEqualTo(Window * 2);
        pump.Credits.Outstanding.ShouldBeLessThanOrEqualTo(Window);

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

        // Counting rather than recording, for the reason CountingTransport carries: a consumer that
        // copies every frame into a 32 MB list is the slowest thing in this pipeline, and a harness
        // whose consumer cannot keep up measures the harness.
        var transport = new CountingTransport();

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

        // ◆ Every byte read either reached the renderer or was counted as dropped, and none of it went
        // anywhere else. That is the invariant worth asserting here rather than "nothing was dropped":
        // this producer emits at memory bandwidth with no network under it, which is faster than any
        // real remote and faster than the pump can frame and send, so drops are the honest outcome and
        // the harness would be lying to forbid them. What must never happen is a byte that is neither
        // delivered nor accounted for — see NothingIsDropped_WhenTheRendererKeepsUpWithTheRemote for
        // the case where the producer is one a network could actually deliver.
        (transport.OutputBytes + pump.BytesDropped).ShouldBe(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>
    /// A renderer that keeps up loses nothing, however much the remote sends.
    /// </summary>
    /// <remarks>
    /// The other half of the trade the backlog makes, and the half a user actually lives in: dropping is
    /// for a terminal that cannot keep up, and a terminal that can must show every byte. The producer here
    /// is sized for a real link rather than for memory bandwidth — several times the backlog, delivered in
    /// pieces a network could carry — so a drop would mean the pump losing output it had room for.
    /// </remarks>
    [Fact]
    public async Task NothingIsDropped_WhenTheRendererKeepsUpWithTheRemote()
    {
        const int Backlog = 256 * 1024;
        const long Target = 512L * 1024;

        // 8 KiB a millisecond — 8 MB/s, a fast remote rather than an impossible one, and thirty-two
        // times the backlog's capacity per second so the pump has no room to be merely lucky.
        await using var session = new FakeShellSession(
            bytesToProduce: Target,
            pacePerRead: TimeSpan.FromMilliseconds(1));

        var transport = new CountingTransport();

        await using var pump = CreatePump(
            session,
            transport,
            new TerminalPumpOptions
            {
                ReadBufferBytes = 8 * 1024,
                FlushInterval = TimeSpan.FromMilliseconds(1),
                WindowBytes = CreditWindow.DefaultWindowBytes,
                BacklogBytes = Backlog,
            });

        transport.AutoAcknowledge = pump;

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

        pump.BytesDropped.ShouldBe(0);
        transport.OutputBytes.ShouldBe(Target);
    }

    /// <summary>
    /// A shell that ends against a renderer which stopped acknowledging still finishes.
    /// </summary>
    /// <remarks>
    /// ◆ THE HANG THIS DESIGN MAKES POSSIBLE, AND THE TEST THAT SAYS IT DOES NOT HAPPEN. The flush loop
    /// waits on credit when the renderer is behind, and completing the backlog is a different signal from
    /// returning credit — so a run whose reader finished while the flusher was parked on an
    /// acknowledgement that will never come would wait for ever. Nothing above would notice: the shell has
    /// ended, but the tab is told by <c>SessionEnded</c>, which fires off the run task completing, and the
    /// connection is given back at the same moment. A session in that state is invisible and permanent.
    /// </remarks>
    [Fact]
    public async Task WhenTheRemoteCloses_TheRunEndsEvenIfTheRendererNeverAcknowledged()
    {
        const int Window = 4 * 1024;

        // More than the window, so the flusher has certainly run out of credit and parked by the time the
        // producer reaches the end of what it has to give.
        await using var session = new FakeShellSession(bytesToProduce: Window * 4);

        // No AutoAcknowledge, ever: this renderer's process died holding the socket.
        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 Should.NotThrowAsync(() => run.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken));

        pump.Outcome.End.ShouldBe(SshSessionEnd.RemoteClosed);

        // And the tail went out rather than being abandoned with the flusher: the last thing a remote
        // writes is usually why it ended.
        transport.OutputBytes().Length.ShouldBe(Window * 4);
        transport.CountOf(TerminalServerOpcode.SessionClosed).ShouldBe(1);
    }

    /// <summary>
    /// A drop is announced in the pane rather than leaving a silent gap.
    /// </summary>
    /// <remarks>
    /// The thing that makes dropping defensible at all. A terminal quietly showing a different transcript
    /// from the one the remote sent is worse than one that says it could not keep up, because the second
    /// can be believed. Its own opcode, so the sentence is never counted as something the remote said —
    /// see <see cref="TerminalServerOpcode.OutputDropped"/>.
    /// </remarks>
    [Fact]
    public async Task WhenOutputIsDropped_TheUserIsToldInThePane()
    {
        const int Window = 4 * 1024;
        const int Backlog = 16 * 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,
                BacklogBytes = Backlog,
            });

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

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

        // Enough credit for one more frame, which is what carries the notice out.
        pump.Acknowledge((uint)Window);

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

        transport.CountOf(TerminalServerOpcode.OutputDropped).ShouldBeGreaterThan(0);

        var notice = Encoding.UTF8.GetString(transport.PayloadsOf(TerminalServerOpcode.OutputDropped));

        notice.ShouldContain("of output was dropped");
        notice.ShouldStartWith("\u001b[0m");

        await pump.DisposeAsync();
        await run;
    }

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