Port of #2239 (v1.x) to v2 (#2275)

packages/client/src/client/stdio.ts

@@ -38,6 +38,14 @@ export type StdioServerParameters = {
      * If not specified, the current working directory will be inherited.
      */
     cwd?: string;
+
+    /**
+     * Maximum size of the read buffer in bytes. If a single message exceeds
+     * this size the transport will emit an error and close.
+     *
+     * Defaults to 10 MB.
+     */
+    maxBufferSize?: number;
 };
 
 /**
@@ -92,7 +100,7 @@ export function getDefaultEnvironment(): Record<string, string> {
  */
 export class StdioClientTransport implements Transport {
     private _process?: ChildProcess;
-    private _readBuffer: ReadBuffer = new ReadBuffer();
+    private _readBuffer: ReadBuffer;
     private _serverParams: StdioServerParameters;
     private _stderrStream: PassThrough | null = null;
 
@@ -102,6 +110,7 @@ export class StdioClientTransport implements Transport {
 
     constructor(server: StdioServerParameters) {
         this._serverParams = server;
+        this._readBuffer = new ReadBuffer({ maxBufferSize: server.maxBufferSize });
         if (server.stderr === 'pipe' || server.stderr === 'overlapped') {
             this._stderrStream = new PassThrough();
         }
@@ -149,8 +158,13 @@ export class StdioClientTransport implements Transport {
             });
 
             this._process.stdout?.on('data', chunk => {
-                this._readBuffer.append(chunk);
-                this.processReadBuffer();
+                try {
+                    this._readBuffer.append(chunk);
+                    this.processReadBuffer();
+                } catch (error) {
+                    this.onerror?.(error as Error);
+                    this.close().catch(() => {});
+                }
             });
 
             this._process.stdout?.on('error', error => {

packages/client/test/client/stdio.test.ts

@@ -77,3 +77,44 @@ test('should return child process pid', async () => {
     await client.close();
     expect(client.pid).toBeNull();
 });
+
+test('should respect custom maxBufferSize option', async () => {
+    const client = new StdioClientTransport({
+        command: 'node',
+        args: ['-e', 'process.stdout.write(Buffer.alloc(200, 0x41))'],
+        maxBufferSize: 100
+    });
+
+    const errorReceived = new Promise<Error>(resolve => {
+        client.onerror = resolve;
+    });
+    const closed = new Promise<void>(resolve => {
+        client.onclose = () => resolve();
+    });
+
+    await client.start();
+
+    const error = await errorReceived;
+    expect(error.message).toMatch(/ReadBuffer exceeded maximum size/);
+    await closed;
+});
+
+test('should fire onerror and close when ReadBuffer overflows', async () => {
+    const client = new StdioClientTransport({
+        command: 'node',
+        args: ['-e', 'process.stdout.write(Buffer.alloc(11 * 1024 * 1024, 0x41))']
+    });
+
+    const errorReceived = new Promise<Error>(resolve => {
+        client.onerror = resolve;
+    });
+    const closed = new Promise<void>(resolve => {
+        client.onclose = () => resolve();
+    });
+
+    await client.start();
+
+    const error = await errorReceived;
+    expect(error.message).toMatch(/ReadBuffer exceeded maximum size/);
+    await closed;
+});