Skip to content

Commit c06dc31

Browse files
authored
Merge pull request #2135 from modelcontextprotocol/v2/fix/2134-parsesse-chunk-boundary
fix(core): carry the in-progress SSE frame across chunk boundaries
2 parents ca2846d + 99f2a20 commit c06dc31

2 files changed

Lines changed: 141 additions & 29 deletions

File tree

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
/**
2+
* `parseSSE` cross-chunk frame loss.
3+
*
4+
* The reproduction needs no browser and no network: `parseSSE` takes a reader,
5+
* so a test supplies the chunk boundary directly instead of hoping a real
6+
* transport happens to produce one. That is what makes this defect testable at
7+
* all — on the wire the split is a function of payload size, TCP segmentation
8+
* and any proxy in between, none of which a test can steer.
9+
*
10+
* The bug: `currentEvent` / `currentData` are declared INSIDE the read loop, so
11+
* a frame whose `data:` line arrives in one chunk and whose terminating blank
12+
* line arrives in the next is discarded. `buffer` already carries a partial
13+
* *line* across reads; nothing carries a partial *frame*.
14+
*/
15+
16+
import { describe, expect, it } from "vitest";
17+
import { parseSSE } from "@inspector/core/mcp/remote/remoteClientTransport";
18+
19+
/** A reader that yields exactly the chunks given, in order. */
20+
function readerFrom(chunks: string[]): ReadableStreamDefaultReader<Uint8Array> {
21+
const encoder = new TextEncoder();
22+
let i = 0;
23+
return {
24+
read: async () =>
25+
i < chunks.length
26+
? { done: false as const, value: encoder.encode(chunks[i++]!) }
27+
: { done: true as const, value: undefined },
28+
} as ReadableStreamDefaultReader<Uint8Array>;
29+
}
30+
31+
async function collect(chunks: string[]) {
32+
const out: { event: string; data: string }[] = [];
33+
for await (const frame of parseSSE(readerFrom(chunks))) out.push(frame);
34+
return out;
35+
}
36+
37+
const FRAME = 'event: message\ndata: {"id":1,"result":"payload"}\n\n';
38+
39+
describe("parseSSE chunk boundaries", () => {
40+
it("delivers a frame that arrives whole (the case that hides the bug)", async () => {
41+
// Chromium and Firefox happen to produce this shape at our payload sizes,
42+
// which is why the defect below has never been observed in practice.
43+
expect(await collect([FRAME])).toEqual([
44+
{ event: "message", data: '{"id":1,"result":"payload"}' },
45+
]);
46+
});
47+
48+
it("delivers a frame split between its data line and its terminator", async () => {
49+
// THE BUG. Chunk 1 ends after the data line's newline; chunk 2 carries only
50+
// the blank line that closes the frame. Nothing is malformed — this is one
51+
// valid frame, delivered in two reads.
52+
const split = ['event: message\ndata: {"id":1,"result":"payload"}\n', "\n"];
53+
expect(split.join("")).toBe(FRAME); // same bytes as above, different reads
54+
expect(await collect(split)).toEqual([
55+
{ event: "message", data: '{"id":1,"result":"payload"}' },
56+
]);
57+
});
58+
59+
it("delivers a frame split mid-payload", async () => {
60+
const split = ['event: message\ndata: {"id":1,"res', 'ult":"payload"}\n\n'];
61+
expect(split.join("")).toBe(FRAME);
62+
expect(await collect(split)).toEqual([
63+
{ event: "message", data: '{"id":1,"result":"payload"}' },
64+
]);
65+
});
66+
67+
it("delivers every frame when a burst is split at each boundary", async () => {
68+
// Bytes for three frames, re-cut so each split lands between a data line
69+
// and its terminator — the shape a large response most plausibly produces.
70+
const chunks: string[] = [];
71+
for (const id of [1, 2, 3]) {
72+
chunks.push(`event: message\ndata: {"id":${id}}\n`, "\n");
73+
}
74+
expect(await collect(chunks)).toEqual([
75+
{ event: "message", data: '{"id":1}' },
76+
{ event: "message", data: '{"id":2}' },
77+
{ event: "message", data: '{"id":3}' },
78+
]);
79+
});
80+
81+
it("flushes a final frame when the stream ends without a blank line", async () => {
82+
expect(await collect(['event: message\ndata: {"id":9}\n'])).toEqual([
83+
{ event: "message", data: '{"id":9}' },
84+
]);
85+
});
86+
});

‎core/mcp/remote/remoteClientTransport.ts‎

Lines changed: 55 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -165,14 +165,55 @@ function legacySessionId(json: RemoteConnectResponse): string | undefined {
165165
}
166166

167167
/**
168-
* Parse SSE stream from a ReadableStream.
169-
* Yields { event, data } for each SSE message.
168+
* Parse an SSE stream from a ReadableStream.
169+
* Yields `{ event, data }` for each complete SSE message.
170+
*
171+
* ⚠️ **The in-progress event must live OUTSIDE the read loop.** A `read()`
172+
* returns an arbitrary slice of bytes, not a whole frame, so a frame's `data:`
173+
* line can arrive in one chunk and the blank line terminating it in the next.
174+
* Declaring `currentEvent` / `currentData` inside the loop — which this did
175+
* until #2134 — discards the accumulated payload at that boundary: the frame is
176+
* never yielded, the JSON-RPC response it carried never settles, and the caller
177+
* hangs forever with no error on any channel. `buffer` already carries a partial
178+
* *line* across reads; these carry the partial *frame*.
179+
*
180+
* It survived because Chromium and Firefox happen to deliver whole frames at the
181+
* payload sizes this transport sees. That is luck rather than a guarantee —
182+
* chunk boundaries are a function of payload size, TCP segmentation and any
183+
* intermediary — and the exposure grows with payload size, so a large
184+
* `resources/read` is the most likely first victim.
185+
*
186+
* Exported for `clients/web/src/test/core/mcp/remote/parseSSE.test.ts`, which is
187+
* the only way to reach the defect deterministically: driving a real transport
188+
* cannot steer where the chunk boundary lands, but a test that supplies the
189+
* reader can put it exactly on the terminator.
170190
*/
171-
async function* parseSSE(
191+
export async function* parseSSE(
172192
reader: ReadableStreamDefaultReader<Uint8Array>,
173193
): AsyncGenerator<{ event: string; data: string }> {
174194
const decoder = new TextDecoder();
175195
let buffer = "";
196+
// Carried ACROSS reads, not per chunk — see the header.
197+
let currentEvent = "message";
198+
let currentData: string[] = [];
199+
200+
const consume = (line: string): { event: string; data: string } | null => {
201+
if (line.startsWith("event:")) {
202+
currentEvent = line.slice(6).trim();
203+
} else if (line.startsWith("data:")) {
204+
currentData.push(line.slice(5).trimStart());
205+
} else if (line === "") {
206+
const complete =
207+
currentData.length > 0
208+
? { event: currentEvent, data: currentData.join("\n") }
209+
: null;
210+
currentEvent = "message";
211+
currentData = [];
212+
return complete;
213+
}
214+
// Anything else (a `:` keep-alive comment, `id:`, `retry:`) is ignored.
215+
return null;
216+
};
176217

177218
while (true) {
178219
const { done, value } = await reader.read();
@@ -183,36 +224,21 @@ async function* parseSSE(
183224
/* v8 ignore next -- String.split always returns a non-empty array, so pop() is never undefined; the ?? "" fallback is unreachable. */
184225
buffer = lines.pop() ?? "";
185226

186-
let currentEvent = "message";
187-
let currentData: string[] = [];
188-
189227
for (const line of lines) {
190-
if (line.startsWith("event:")) {
191-
currentEvent = line.slice(6).trim();
192-
} else if (line.startsWith("data:")) {
193-
currentData.push(line.slice(5).trimStart());
194-
} else if (line === "") {
195-
if (currentData.length > 0) {
196-
yield { event: currentEvent, data: currentData.join("\n") };
197-
}
198-
currentEvent = "message";
199-
currentData = [];
200-
}
228+
const complete = consume(line);
229+
if (complete) yield complete;
201230
}
202231
}
203232

204-
if (buffer.trim()) {
205-
const lines = buffer.split("\n");
206-
let currentEvent = "message";
207-
const currentData: string[] = [];
208-
for (const line of lines) {
209-
if (line.startsWith("event:")) currentEvent = line.slice(6).trim();
210-
else if (line.startsWith("data:"))
211-
currentData.push(line.slice(5).trimStart());
212-
}
213-
if (currentData.length > 0) {
214-
yield { event: currentEvent, data: currentData.join("\n") };
215-
}
233+
// Stream ended: flush the trailing partial line, then anything still held. A
234+
// server that closes without a final blank line has still delivered a complete
235+
// frame, and dropping it would lose the last message of every such stream.
236+
if (buffer) {
237+
const complete = consume(buffer);
238+
if (complete) yield complete;
239+
}
240+
if (currentData.length > 0) {
241+
yield { event: currentEvent, data: currentData.join("\n") };
216242
}
217243
}
218244

0 commit comments

Comments
 (0)