dotnet: release oversized JSON-RPC receive buffers (#2047)

* Fix oversized JSON-RPC buffer retention

Bound the reusable receive buffer after a completed large frame while preserving any carried bytes for the next frame. Add a regression test that verifies the next read does not retain a multi-megabyte buffer.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Strengthen oversized buffer carry-over test

Keep the receive stream open, coalesce a second valid response with the oversized frame, and verify both carried-message processing and the subsequent bounded read buffer.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: adirh3

dotnet/src/JsonRpc.cs

@@ -29,6 +29,8 @@ internal sealed partial class JsonRpc : IDisposable
 {
     private const int ErrorCodeMethodNotFound = -32601;
     private const int ErrorCodeInternalError = -32603;
+    private const int InitialReadBufferSize = 256;
+    private const int MaximumRetainedReadBufferSize = 1024 * 1024;
 
     private readonly Stream _sendStream;
     private readonly Stream _receiveStream;
@@ -259,7 +261,7 @@ private static byte[] BuildFrame(ReadOnlySpan<byte> json, out int frameLen)
 
     private async Task ReadLoopAsync(CancellationToken cancellationToken)
     {
-        var buffer = new byte[256];
+        var buffer = new byte[InitialReadBufferSize];
         int carried = 0; // bytes in buffer carried over from previous read
         try
         {
@@ -298,6 +300,17 @@ private async Task ReadLoopAsync(CancellationToken cancellationToken)
                     Buffer.BlockCopy(buffer, contentLength, buffer, 0, carried);
                 }
 
+                if (buffer.Length > MaximumRetainedReadBufferSize)
+                {
+                    var retainedBuffer = new byte[Math.Max(InitialReadBufferSize, carried)];
+                    if (carried > 0)
+                    {
+                        Buffer.BlockCopy(buffer, 0, retainedBuffer, 0, carried);
+                    }
+
+                    buffer = retainedBuffer;
+                }
+
                 if (message is not { } parsed)
                 {
                     continue;

dotnet/test/Unit/JsonRpcTests.cs

@@ -3,6 +3,7 @@
  *--------------------------------------------------------------------------------------------*/
 
 using System.Reflection;
+using System.Text;
 using System.Text.Json;
 using System.Text.Json.Serialization.Metadata;
 using Xunit;
@@ -93,6 +94,57 @@ public async Task JsonRpc_Cancels_And_Disposes_Pending_Requests()
         await Assert.ThrowsAnyAsync<ObjectDisposedException>(() => pending);
     }
 
+    [Fact]
+    public async Task JsonRpc_Does_Not_Retain_Oversized_Receive_Buffer()
+    {
+        var oversizedFrame = CreateResponseFrame(
+            long.MaxValue,
+            "ignored",
+            headerPaddingLength: 1024 * 1024);
+        var carriedFrame = CreateResponseFrame(1, "carried");
+        using var receiveStream = new CoalescedFramesThenWaitStream(oversizedFrame, carriedFrame);
+        using var rpc = new JsonRpcReflection(Stream.Null, receiveStream);
+
+        var carriedResponse = rpc.InvokeAsync<string>("pending", args: null);
+        rpc.StartListening();
+
+        var responseCompleted = await Task.WhenAny(
+            carriedResponse,
+            Task.Delay(TimeSpan.FromSeconds(5)));
+        Assert.Same(carriedResponse, responseCompleted);
+        Assert.Equal("carried", await carriedResponse);
+        Assert.True(receiveStream.FramesWereCoalesced);
+
+        var readCompleted = await Task.WhenAny(
+            receiveStream.PostFrameReadBufferSize,
+            Task.Delay(TimeSpan.FromSeconds(5)));
+        Assert.Same(receiveStream.PostFrameReadBufferSize, readCompleted);
+        Assert.InRange(await receiveStream.PostFrameReadBufferSize, 1, 1024 * 1024);
+    }
+
+    private static byte[] CreateResponseFrame(long id, string result, int headerPaddingLength = 0)
+    {
+        using var bodyStream = new MemoryStream();
+        using (var writer = new Utf8JsonWriter(bodyStream))
+        {
+            writer.WriteStartObject();
+            writer.WriteString("jsonrpc", "2.0");
+            writer.WriteNumber("id", id);
+            writer.WriteString("result", result);
+            writer.WriteEndObject();
+        }
+
+        var body = bodyStream.ToArray();
+        var paddingHeader = headerPaddingLength > 0
+            ? $"X-Padding: {new string('x', headerPaddingLength)}\r\n"
+            : string.Empty;
+        var header = Encoding.ASCII.GetBytes($"{paddingHeader}Content-Length: {body.Length}\r\n\r\n");
+        var frame = new byte[header.Length + body.Length];
+        header.CopyTo(frame, 0);
+        body.CopyTo(frame, header.Length);
+        return frame;
+    }
+
     private static int GetRemoteErrorCode(Exception exception)
     {
         var property = exception.GetType().GetProperty("ErrorCode", BindingFlags.Instance | BindingFlags.Public);
@@ -170,12 +222,17 @@ private sealed class JsonRpcReflection : IDisposable
         private readonly object _instance;
 
         public JsonRpcReflection(Stream stream)
+            : this(stream, stream)
+        {
+        }
+
+        public JsonRpcReflection(Stream sendStream, Stream receiveStream)
         {
             _instance = Activator.CreateInstance(
                 JsonRpcType,
                 BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic,
                 binder: null,
-                args: [stream, stream, SerializerOptions, null],
+                args: [sendStream, receiveStream, SerializerOptions, null],
                 culture: null)!;
         }
 
@@ -198,6 +255,82 @@ public async Task<T> InvokeAsync<T>(string methodName, object?[]? args, Cancella
         public void Dispose() => ((IDisposable)_instance).Dispose();
     }
 
+    private sealed class CoalescedFramesThenWaitStream : Stream
+    {
+        private readonly TaskCompletionSource<int> _postFrameReadBufferSize =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly byte[] _frames;
+        private readonly int _firstFrameLength;
+        private int _offset;
+
+        public CoalescedFramesThenWaitStream(byte[] firstFrame, byte[] secondFrame)
+        {
+            _firstFrameLength = firstFrame.Length;
+            _frames = new byte[firstFrame.Length + secondFrame.Length];
+            firstFrame.CopyTo(_frames, 0);
+            secondFrame.CopyTo(_frames, firstFrame.Length);
+        }
+
+        public bool FramesWereCoalesced { get; private set; }
+
+        public Task<int> PostFrameReadBufferSize => _postFrameReadBufferSize.Task;
+
+        public override bool CanRead => true;
+
+        public override bool CanSeek => false;
+
+        public override bool CanWrite => false;
+
+        public override long Length => throw new NotSupportedException();
+
+        public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
+
+        public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+
+        public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) =>
+            ReadCoreAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask();
+
+#if NET8_0_OR_GREATER
+        public override
+#else
+        internal
+#endif
+        ValueTask<int> ReadAsync(Memory<byte> destination, CancellationToken cancellationToken = default) =>
+            ReadCoreAsync(destination, cancellationToken);
+
+        public override void Flush()
+        {
+        }
+
+        public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
+
+        public override void SetLength(long value) => throw new NotSupportedException();
+
+        public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+
+        private ValueTask<int> ReadCoreAsync(Memory<byte> destination, CancellationToken cancellationToken)
+        {
+            if (_offset >= _frames.Length)
+            {
+                _postFrameReadBufferSize.TrySetResult(destination.Length);
+                return new ValueTask<int>(WaitForCancellationAsync(cancellationToken));
+            }
+
+            var startingOffset = _offset;
+            var bytesRead = Math.Min(destination.Length, _frames.Length - _offset);
+            _frames.AsMemory(_offset, bytesRead).CopyTo(destination);
+            _offset += bytesRead;
+            FramesWereCoalesced |= startingOffset < _firstFrameLength && _offset == _frames.Length;
+            return new ValueTask<int>(bytesRead);
+        }
+
+        private static async Task<int> WaitForCancellationAsync(CancellationToken cancellationToken)
+        {
+            await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken).ConfigureAwait(false);
+            return 0;
+        }
+    }
+
     private sealed class InMemoryDuplexStream : Stream
     {
         private readonly Queue<byte> _buffer = new();