fix(core): resync ReadBuffer at the next message boundary after an oversized message

.changeset/readbuffer-resync-after-overflow.md

@@ -0,0 +1,7 @@
+---
+'@modelcontextprotocol/core-internal': patch
+'@modelcontextprotocol/client': patch
+'@modelcontextprotocol/server': patch
+---
+
+Resynchronise `ReadBuffer` at the next message boundary after an oversized message. On overflow the buffer was cleared and reading continued, but the rest of the oversized message was still arriving: it landed in the empty buffer and was fed to the parser as if it were the start of a new message, and a large enough remainder accumulated until it overflowed a second time. The remainder is now dropped, unbuffered, up to and including the newline that ends it, and parsing resumes with whatever follows. One oversized message now produces one error.

packages/core-internal/src/shared/stdio.ts

@@ -9,18 +9,60 @@ export const STDIO_DEFAULT_MAX_BUFFER_SIZE = 10 * 1024 * 1024;
 export class ReadBuffer {
     private _buffer?: Buffer;
     private _maxBufferSize: number;
+    /**
+     * Set after an oversized message: its remaining bytes are still arriving
+     * and are dropped, unbuffered, until the newline that ends it.
+     */
+    private _discardingToNewline = false;
 
     constructor(options?: { maxBufferSize?: number }) {
         this._maxBufferSize = options?.maxBufferSize ?? STDIO_DEFAULT_MAX_BUFFER_SIZE;
     }
 
     append(chunk: Buffer): void {
+        if (this._discardingToNewline) {
+            const newline = chunk.indexOf('\n');
+            if (newline === -1) {
+                return;
+            }
+            this._discardingToNewline = false;
+            chunk = chunk.subarray(newline + 1);
+        }
+
         const newSize = (this._buffer?.length ?? 0) + chunk.length;
         if (newSize > this._maxBufferSize) {
-            this.clear();
+            this._resyncAfterOverflow(chunk);
             throw new Error(`ReadBuffer exceeded maximum size of ${this._maxBufferSize} bytes`);
         }
-        this._buffer = this._buffer ? Buffer.concat([this._buffer, chunk]) : chunk;
+        if (chunk.length > 0) {
+            this._buffer = this._buffer ? Buffer.concat([this._buffer, chunk]) : chunk;
+        }
+    }
+
+    /**
+     * Drop the message that overflowed and resume at the next message boundary.
+     *
+     * Clearing the buffer alone is not enough: the rest of the oversized
+     * message is still in flight, and appending it into an empty buffer would
+     * parse a mid-message tail as if it were the start of a new one — a second
+     * error for the same message, or worse, a tail that happens to be valid
+     * JSON. So the remainder is skipped up to and including its newline, and
+     * anything after that newline is kept as the start of the next message.
+     * If that remainder is itself over the limit it is dropped the same way.
+     */
+    private _resyncAfterOverflow(chunk: Buffer): void {
+        this._buffer = undefined;
+        const newline = chunk.indexOf('\n');
+        if (newline === -1) {
+            this._discardingToNewline = true;
+            return;
+        }
+        const rest = chunk.subarray(newline + 1);
+        if (rest.length > this._maxBufferSize) {
+            this._resyncAfterOverflow(rest);
+        } else if (rest.length > 0) {
+            this._buffer = rest;
+        }
     }
 
     readMessage(): JSONRPCMessage | null {
@@ -50,6 +92,7 @@ export class ReadBuffer {
 
     clear(): void {
         this._buffer = undefined;
+        this._discardingToNewline = false;
     }
 }
 

packages/core-internal/test/shared/stdio.test.ts

@@ -143,6 +143,69 @@ describe('buffer size limit', () => {
         expect(readBuffer.readMessage()).toBeNull();
     });
 
+    describe('resync after an oversized message', () => {
+        const line = JSON.stringify(testMessage) + '\n';
+
+        test('resumes at the next message boundary, not mid-message', () => {
+            const readBuffer = new ReadBuffer({ maxBufferSize: 100 });
+            readBuffer.append(Buffer.alloc(60, 0x41));
+            expect(() => readBuffer.append(Buffer.alloc(60, 0x41))).toThrow(/ReadBuffer exceeded maximum size/);
+
+            // The tail of the oversized message, then a real one. Before, the
+            // tail landed in an empty buffer and was fed to the parser as if it
+            // were a message of its own.
+            readBuffer.append(Buffer.from('AAAA"}]}}\n' + line));
+            expect(readBuffer.readMessage()).toEqual(testMessage);
+            expect(readBuffer.readMessage()).toBeNull();
+        });
+
+        test('drops the remainder without buffering it, however many chunks it spans', () => {
+            const readBuffer = new ReadBuffer({ maxBufferSize: 100 });
+            expect(() => readBuffer.append(Buffer.alloc(101, 0x41))).toThrow();
+
+            // 240 more bytes of the same message. Before, these accumulated in
+            // the cleared buffer and overflowed a second time.
+            for (let i = 0; i < 3; i++) {
+                expect(() => readBuffer.append(Buffer.alloc(80, 0x41))).not.toThrow();
+            }
+            expect(readBuffer.readMessage()).toBeNull();
+
+            readBuffer.append(Buffer.from('\n' + line));
+            expect(readBuffer.readMessage()).toEqual(testMessage);
+        });
+
+        test('keeps what follows the boundary inside the chunk that overflowed', () => {
+            const readBuffer = new ReadBuffer({ maxBufferSize: 100 });
+            readBuffer.append(Buffer.alloc(60, 0x41));
+            expect(() => readBuffer.append(Buffer.from('A'.repeat(50) + '\n' + line))).toThrow();
+
+            expect(readBuffer.readMessage()).toEqual(testMessage);
+            expect(readBuffer.readMessage()).toBeNull();
+        });
+
+        test('never hands the parser more than the limit, even after the boundary', () => {
+            const readBuffer = new ReadBuffer({ maxBufferSize: 100 });
+            // Two oversized messages in one chunk: the second is dropped too,
+            // rather than kept as an unchecked buffer larger than the limit.
+            const chunk = Buffer.from('A'.repeat(150) + '\n' + 'B'.repeat(150) + '\n' + line);
+            expect(() => readBuffer.append(chunk)).toThrow();
+
+            expect(readBuffer.readMessage()).toEqual(testMessage);
+            expect(readBuffer.readMessage()).toBeNull();
+        });
+
+        test('clear() abandons the resync', () => {
+            const readBuffer = new ReadBuffer({ maxBufferSize: 100 });
+            expect(() => readBuffer.append(Buffer.alloc(101, 0x41))).toThrow();
+            readBuffer.clear();
+
+            // A fresh stream after clear() must not be skipped as if it were
+            // the tail of the old message.
+            readBuffer.append(Buffer.from(line));
+            expect(readBuffer.readMessage()).toEqual(testMessage);
+        });
+    });
+
     test('should allow appending up to exactly the max size', () => {
         const readBuffer = new ReadBuffer({ maxBufferSize: 100 });
         // Should not throw — exactly at limit