From f6278a472e9b97579e9e9ef512303aa91a926a67 Mon Sep 17 00:00:00 2001 From: "robopipi[bot]" <336283572+robopipi[bot]@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:09:23 +0100 Subject: [PATCH] fix(mcp): read large stdio messages in linear time (#11791) `JsonRpcLineReader.readStdout` (used by the stdio transport and by `PipeTransport`) appended every stdout chunk to the buffered bytes with `Buffer.concat`, then searched the whole buffer from the start for a newline. A large message arriving in many chunks was therefore copied and scanned again on every chunk. A 12 MiB tool result in 64 KiB chunks took about 350 ms just to frame. Now the reader keeps the bytes of an incomplete message as a list of chunks, with a running byte count, and searches only the new chunk for a newline. When the newline arrives, the chunks are joined once. A line that fits inside a single chunk is sliced directly. The rest is unchanged: the `maxMessageBytes` limit (now checked against the running count), CRLF and blank-line handling, errors for invalid JSON and oversized messages, and the incomplete-message error in `end()`. Testing: - New `packages/mcp/test/lines.test.ts`: - a 1 MiB message fed in 4 KiB chunks calls `Buffer.concat` once (fails on the old code); - a mixed input of multi-byte UTF-8, CRLF, blank lines, an invalid line and several messages gives the same results at every chunk size from 1 byte to the whole input; - reading recovers after an oversized message; - a message without its newline is reported at the end. - `test/lines.test.ts`, `test/stdio.test.ts` and `test/pipe.test.ts` pass. `npm run check` passes. - Micro-benchmark, one message fed in 64 KiB chunks: - before: 1 MiB 5.4 ms, 4 MiB 49.9 ms, 12 MiB 348 ms; - after: 2.2 ms, 7.9 ms, 21.6 ms. Fixes #11761 Requested by @badlogic Co-authored-by: triagent --- packages/mcp/src/transports/lines.ts | 40 +++++++++++---- packages/mcp/test/lines.test.ts | 75 ++++++++++++++++++++++++++++ 2 files changed, 105 insertions(+), 10 deletions(-) create mode 100644 packages/mcp/test/lines.test.ts diff --git a/packages/mcp/src/transports/lines.ts b/packages/mcp/src/transports/lines.ts index 8d53d9c27..3a07be176 100644 --- a/packages/mcp/src/transports/lines.ts +++ b/packages/mcp/src/transports/lines.ts @@ -18,7 +18,9 @@ export interface JsonRpcLineReaderOptions { */ export class JsonRpcLineReader { private readonly options: JsonRpcLineReaderOptions; - private stdoutBuffer = Buffer.alloc(0); + /** The bytes of the incomplete message received so far. */ + private stdoutChunks: Buffer[] = []; + private stdoutPendingBytes = 0; private stderrBuffer = Buffer.alloc(0); constructor(options: JsonRpcLineReaderOptions) { @@ -30,21 +32,38 @@ export class JsonRpcLineReader { return this.stderrBuffer.toString("utf8"); } - /** Takes the next stdout bytes and reports every complete message, or the error parsing it. */ + /** + * Takes the next stdout bytes and reports every complete message, or the error parsing it. Bytes of an + * incomplete message are kept as chunks and joined once when its newline arrives, so a large message that + * arrives in many chunks is copied and scanned only once. + */ readStdout(chunk: Uint8Array): void { - this.stdoutBuffer = Buffer.concat([this.stdoutBuffer, chunk]); + const data = Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength); const maxMessageBytes = this.options.maxMessageBytes ?? DEFAULT_MAX_MESSAGE_BYTES; + let start = 0; while (true) { - const newline = this.stdoutBuffer.indexOf(0x0a); + const newline = data.indexOf(0x0a, start); if (newline < 0) { - if (this.stdoutBuffer.length > maxMessageBytes) { - this.stdoutBuffer = Buffer.alloc(0); + if (start < data.length) { + // Copied, since the caller may reuse the chunk's memory. + this.stdoutChunks.push(Buffer.from(data.subarray(start))); + this.stdoutPendingBytes += data.length - start; + } + if (this.stdoutPendingBytes > maxMessageBytes) { + this.stdoutChunks = []; + this.stdoutPendingBytes = 0; this.options.onError(new Error(`MCP stdio message exceeds ${maxMessageBytes} bytes`)); } return; } - const line = this.stdoutBuffer.subarray(0, newline); - this.stdoutBuffer = this.stdoutBuffer.subarray(newline + 1); + let line = data.subarray(start, newline); + if (this.stdoutChunks.length > 0) { + this.stdoutChunks.push(line); + line = Buffer.concat(this.stdoutChunks, this.stdoutPendingBytes + line.length); + this.stdoutChunks = []; + this.stdoutPendingBytes = 0; + } + start = newline + 1; if (line.length > maxMessageBytes) { this.options.onError(new Error(`MCP stdio message exceeds ${maxMessageBytes} bytes`)); continue; @@ -69,9 +88,10 @@ export class JsonRpcLineReader { /** Ends stdout: a message without its newline is reported as an error and dropped. */ end(): void { - if (this.stdoutBuffer.toString("utf8").trim()) { + if (Buffer.concat(this.stdoutChunks).toString("utf8").trim()) { this.options.onError(new Error("MCP stdio server closed with an incomplete JSON-RPC message")); } - this.stdoutBuffer = Buffer.alloc(0); + this.stdoutChunks = []; + this.stdoutPendingBytes = 0; } } diff --git a/packages/mcp/test/lines.test.ts b/packages/mcp/test/lines.test.ts new file mode 100644 index 000000000..f911c063c --- /dev/null +++ b/packages/mcp/test/lines.test.ts @@ -0,0 +1,75 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { JsonRpcMessage } from "../src/index.ts"; +import { JsonRpcLineReader } from "../src/transports/lines.ts"; + +function startReader(maxMessageBytes?: number) { + const messages: JsonRpcMessage[] = []; + const errors: string[] = []; + const reader = new JsonRpcLineReader({ + maxMessageBytes, + onMessage: (message) => messages.push(message), + onError: (error) => errors.push(error instanceof Error ? error.message : String(error)), + }); + return { reader, messages, errors }; +} + +function feed(reader: JsonRpcLineReader, bytes: Buffer, chunkSize: number): void { + for (let i = 0; i < bytes.length; i += chunkSize) reader.readStdout(bytes.subarray(i, i + chunkSize)); +} + +describe("JsonRpcLineReader", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it("joins a message split across many chunks once, when its newline arrives", () => { + const { reader, messages, errors } = startReader(); + const text = "x".repeat(1024 * 1024); + const bytes = Buffer.from(`${JSON.stringify({ jsonrpc: "2.0", id: 1, result: { text } })}\n`); + const concat = vi.spyOn(Buffer, "concat"); + feed(reader, bytes, 4096); + expect(concat).toHaveBeenCalledTimes(1); + expect(errors).toEqual([]); + expect(messages).toEqual([{ jsonrpc: "2.0", id: 1, result: { text } }]); + }); + + it("handles multi-byte characters, CRLF, blank lines, and several messages per chunk at any split", () => { + const input = Buffer.from( + [ + JSON.stringify({ jsonrpc: "2.0", id: 1, result: { text: "héllo 🌍" } }), + "", + `${JSON.stringify({ jsonrpc: "2.0", method: "notifications/x" })}\r`, + "not json", + JSON.stringify({ jsonrpc: "2.0", id: 2, result: {} }), + "", + ].join("\n"), + ); + for (let chunkSize = 1; chunkSize <= input.length; chunkSize++) { + const { reader, messages, errors } = startReader(); + feed(reader, input, chunkSize); + reader.end(); + expect(messages).toEqual([ + { jsonrpc: "2.0", id: 1, result: { text: "héllo 🌍" } }, + { jsonrpc: "2.0", method: "notifications/x" }, + { jsonrpc: "2.0", id: 2, result: {} }, + ]); + expect(errors).toHaveLength(1); + } + }); + + it("drops an oversized message and keeps reading the next one", () => { + const { reader, messages, errors } = startReader(64); + const next = JSON.stringify({ jsonrpc: "2.0", id: 2, result: {} }); + feed(reader, Buffer.from(`${"a".repeat(50)}`), 10); + feed(reader, Buffer.from(`${"a".repeat(50)}\n${next}\n`), 10); + expect(errors[0]).toBe("MCP stdio message exceeds 64 bytes"); + expect(messages).toEqual([{ jsonrpc: "2.0", id: 2, result: {} }]); + }); + + it("reports a message without its newline at the end", () => { + const { reader, errors } = startReader(); + feed(reader, Buffer.from('{"jsonrpc":"2.0",'), 4); + reader.end(); + expect(errors).toEqual(["MCP stdio server closed with an incomplete JSON-RPC message"]); + }); +});