mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
fix(mcp): cancel SSE response streams that fail the message-size limit (#11735)
When an SSE event went over `maxMessageBytes`, `consumeSseStream()` threw the size-limit error, but its `finally` only released the reader lock. Releasing the lock does not cancel the response body. If the server kept the response open, for example with heartbeats, the failed request held its HTTP connection until the server ended it or the whole client was closed. The same applies to the GET notification stream, which uses the same function. On error, `consumeSseStream()` now cancels the reader before it rethrows and releases the lock. A failure in `cancel()` is ignored, so the caller still gets the original error. Testing: - New test in `packages/mcp/test/streamable-http.test.ts`: it feeds an oversized event through a stream that stays open and checks that the stream's `cancel` callback runs. It fails without the change and passes with it. - Full mcp package tests: 48/48 pass. `npm run check` is clean. - Reproduction over loopback: a local `node:http` server sends a 512-byte event, then a heartbeat every 20 ms, and the client reads it through `fetch` with `maxEventBytes: 128`, three times. - Before: none of the three server responses closed, and the server made 28 more heartbeat writes. - After: all three closed, with no further writes. Fixes #11729 Requested by @badlogic Co-authored-by: triagent <triagent@localhost>
This commit is contained in:
parent
797e122cb1
commit
f409847473
2 changed files with 23 additions and 0 deletions
|
|
@ -103,6 +103,10 @@ export async function consumeSseStream(stream: ReadableStream<Uint8Array>, optio
|
|||
buffered += decoder.decode();
|
||||
if (buffered) processLine(buffered);
|
||||
dispatch();
|
||||
} catch (error) {
|
||||
// Releasing the lock alone leaves the response open; the original error wins over a failed cancel.
|
||||
await reader.cancel().catch(() => {});
|
||||
throw error;
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -118,6 +118,25 @@ describe("consumeSseStream", () => {
|
|||
);
|
||||
expect(sent).toBeLessThan(100);
|
||||
});
|
||||
|
||||
// Regression test for #11729: a failed stream must be cancelled, not left open.
|
||||
it("cancels the stream when an event exceeds the limit", async () => {
|
||||
const encoder = new TextEncoder();
|
||||
let cancelled = false;
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
// The server keeps the response open after the oversized event.
|
||||
controller.enqueue(encoder.encode(`data: ${"x".repeat(512)}\n\n`));
|
||||
},
|
||||
cancel() {
|
||||
cancelled = true;
|
||||
},
|
||||
});
|
||||
await expect(consumeSseStream(stream, { maxEventBytes: 128, onEvent: () => {} })).rejects.toThrow(
|
||||
"MCP SSE event exceeds 128 bytes",
|
||||
);
|
||||
expect(cancelled).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("StreamableHttpTransport", () => {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue