mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
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 <triagent@localhost>
This commit is contained in:
parent
f4427f8a3e
commit
f6278a472e
2 changed files with 105 additions and 10 deletions
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
75
packages/mcp/test/lines.test.ts
Normal file
75
packages/mcp/test/lines.test.ts
Normal file
|
|
@ -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"]);
|
||||
});
|
||||
});
|
||||
Loading…
Add table
Add a link
Reference in a new issue