Compression=true: a read that inflates to zero bytes is taken for the end of the stream ("Missing 4 bytes to fill total 4") #1293

// Minimal reproduction: FirebirdClient 10.3.4 with Compression=true over a TCP path that delivers the
// server's stream in small fragments. No Firebird change is needed; the relay stands in for a link that
// fragments the stream. Reads system tables only.
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using FirebirdSql.Data.FirebirdClient;

var password = args[0];
var database = args[1];
var sizes = args[2].Split(',').Select(int.Parse).ToArray();

foreach (var compression in new[] { false, true })
{
    foreach (var size in sizes)
    {
        var listener = new TcpListener(IPAddress.Loopback, 0);
        listener.Start();
        var port = ((IPEndPoint)listener.LocalEndpoint).Port;
        using var stop = new CancellationTokenSource();
        var relay = Task.Run(() => RunRelayAsync(listener, size, stop.Token));

        var builder = new FbConnectionStringBuilder
        {
            DataSource = "127.0.0.1",
            Port = port,
            Database = database,
            UserID = "SYSDBA",
            Password = password,
            Pooling = false,
            Compression = compression,
        };

        var clock = Stopwatch.StartNew();
        var rows = 0L;
        string outcome;
        var work = Task.Run(async () =>
        {
            await using var connection = new FbConnection(builder.ToString());
            await connection.OpenAsync();
            await using var command = connection.CreateCommand();
            command.CommandText = "select a.RDB$FIELD_NAME, a.RDB$RELATION_NAME, a.RDB$FIELD_SOURCE from RDB$RELATION_FIELDS a";
            await using var reader = await command.ExecuteReaderAsync();
            while (await reader.ReadAsync())
            {
                Interlocked.Increment(ref rows);
            }
        });

        if (await Task.WhenAny(work, Task.Delay(TimeSpan.FromSeconds(25))) != work)
        {
            outcome = "no answer in 25 s";
        }
        else if (work.IsFaulted)
        {
            var ex = work.Exception!.InnerException!;
            var root = ex;
            while (root.InnerException is not null)
            {
                root = root.InnerException;
            }

            outcome = $"FAILED {ex.GetType().Name} / {root.GetType().Name}: {root.Message}";
        }
        else
        {
            outcome = "OK";
        }

        stop.Cancel();
        listener.Stop();
        Console.WriteLine($"compression={compression,-5} fragment={(size == 0 ? "whole" : size + " B"),-8} rows={Interlocked.Read(ref rows),-6} {clock.ElapsedMilliseconds,6} ms  {outcome}");
    }
}

static async Task RunRelayAsync(TcpListener listener, int fragment, CancellationToken token)
{
    try
    {
        while (!token.IsCancellationRequested)
        {
            var client = await listener.AcceptTcpClientAsync(token);
            client.NoDelay = true;
            var server = new TcpClient { NoDelay = true };
            await server.ConnectAsync(IPAddress.Loopback, 3050, token);
            _ = PumpAsync(client.GetStream(), server.GetStream(), 0, token);
            _ = PumpAsync(server.GetStream(), client.GetStream(), fragment, token);
        }
    }
    catch (OperationCanceledException)
    {
    }
    catch (SocketException)
    {
    }
}

static async Task PumpAsync(NetworkStream from, NetworkStream to, int fragment, CancellationToken token)
{
    var buffer = new byte[65536];
    try
    {
        while (true)
        {
            var read = await from.ReadAsync(buffer, token);
            if (read == 0)
            {
                to.Close();
                return;
            }

            if (fragment == 0)
            {
                await to.WriteAsync(buffer.AsMemory(0, read), token);
                continue;
            }

            for (var offset = 0; offset < read; offset += fragment)
            {
                await to.WriteAsync(buffer.AsMemory(offset, Math.Min(fragment, read - offset)), token);
                await to.FlushAsync(token);

                // Long enough for the client to take the fragment as a read of its own.
                var spin = Stopwatch.StartNew();
                while (spin.Elapsed.TotalMicroseconds < 150)
                {
                    Thread.SpinWait(20);
                }
            }
        }
    }
    catch (Exception ex) when (ex is IOException or OperationCanceledException or ObjectDisposedException)
    {
        to.Close();
    }
}