From 857767842fa4ea4f6c855755e2387c5bcdd2e574 Mon Sep 17 00:00:00 2001 From: Sameh Khamis Date: Fri, 2 Oct 2026 19:47:37 -0700 Subject: [PATCH] fix(mcp): bound panel requests, request bodies and cancelled calls Panels on one MCP server share six requests in flight; further requests wait their turn (up to 256 waiting) instead of failing, and a waiting request leaves the queue when its caller cancels. Held responses belong to their own MCP server rather than to the process. /mcp request bodies must arrive within 30 seconds. Over stdio, a client's notifications/cancelled aborts the named request, which then gets no response. Co-Authored-By: Claude Opus 5.5 (1M context) --- server/mcp/plugin.js | 42 ++++++++-- server/mcp/stdio.js | 27 +++++- src/tools/localRoute.test.mjs | 41 ++++++++++ src/tools/localServer.test.mjs | 40 +++++++++ src/tools/queries/panelRequest.js | 104 +++++++++++++++++++----- src/tools/queries/panelRequest.test.mjs | 88 ++++++++++++++++++++ 6 files changed, 310 insertions(+), 32 deletions(-) diff --git a/server/mcp/plugin.js b/server/mcp/plugin.js index 974ca9abd..75ad2b558 100644 --- a/server/mcp/plugin.js +++ b/server/mcp/plugin.js @@ -14,6 +14,8 @@ import { createLocalMcpServer } from './server.js'; const LOOPBACK_HOSTS = new Set(['localhost', '127.0.0.1', '[::1]']); const LOOPBACK_ADDRESSES = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']); const MAX_BODY_BYTES = 1024 * 1024; +/** How long a client may take to send its request body. */ +const BODY_TIMEOUT_MS = 30 * 1000; /** * Whether a request may reach the local MCP server. `localPort` is the port @@ -56,7 +58,10 @@ export function isLocalMcpRequest({ } /** Vite plugin that mounts the local MCP server at /mcp. */ -export function localMcpPlugin({ createServer = createLocalMcpServer } = {}) { +export function localMcpPlugin({ + createServer = createLocalMcpServer, + bodyTimeoutMs = BODY_TIMEOUT_MS, +} = {}) { const handlers = new Map(); const handlerFor = (apiBase) => { if (!handlers.has(apiBase)) @@ -94,7 +99,10 @@ export function localMcpPlugin({ createServer = createLocalMcpServer } = {}) { }; res.on('close', onClose); try { - const body = req.method === 'POST' ? await readBody(req) : undefined; + const body = + req.method === 'POST' + ? await readBody(req, bodyTimeoutMs) + : undefined; const request = new Request(`http://${host}/mcp`, { method: req.method, headers: Object.entries(req.headers).flatMap(([name, value]) => @@ -109,13 +117,18 @@ export function localMcpPlugin({ createServer = createLocalMcpServer } = {}) { } catch (error) { if (disconnect.signal.aborted) return; if (res.headersSent) return res.destroy(); - res.writeHead(error?.status === 413 ? 413 : 400, { + res.writeHead([408, 413].includes(error?.status) ? error.status : 400, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store', }); res.end( JSON.stringify({ - error: error?.status === 413 ? 'Request too large' : 'Bad request', + error: + error?.status === 413 + ? 'Request too large' + : error?.status === 408 + ? 'Request timed out' + : 'Bad request', }), ); } finally { @@ -130,11 +143,21 @@ export function localMcpPlugin({ createServer = createLocalMcpServer } = {}) { }; } -function readBody(req) { +function readBody(req, timeoutMs = BODY_TIMEOUT_MS) { return new Promise((resolve, reject) => { const chunks = []; let size = 0; let tooLarge = false; + // A client that stops sending must not hold the request open. + const timer = setTimeout( + () => + reject(Object.assign(new Error('Request timed out'), { status: 408 })), + timeoutMs, + ); + const fail = (error) => { + clearTimeout(timer); + reject(error); + }; req.on('data', (chunk) => { size += chunk.length; // Keep draining without storing, so the 413 response can still be sent. @@ -144,9 +167,12 @@ function readBody(req) { } tooLarge = true; chunks.length = 0; - reject(Object.assign(new Error('Request too large'), { status: 413 })); + fail(Object.assign(new Error('Request too large'), { status: 413 })); }); - req.on('end', () => resolve(Buffer.concat(chunks))); - req.on('error', reject); + req.on('end', () => { + clearTimeout(timer); + resolve(Buffer.concat(chunks)); + }); + req.on('error', fail); }); } diff --git a/server/mcp/stdio.js b/server/mcp/stdio.js index c89407be3..26f1da8c4 100644 --- a/server/mcp/stdio.js +++ b/server/mcp/stdio.js @@ -30,13 +30,18 @@ export function describeRequest(message) { return typeof subject === 'string' ? `${method} ${subject}` : method; } -/** Serve newline-delimited JSON-RPC until `input` ends. */ +/** + * Serve newline-delimited JSON-RPC until `input` ends. A client's + * `notifications/cancelled` aborts the named request, which then gets no + * response, as the protocol asks. + */ export async function serveStdio(server, { input, output, log = () => {} }) { const lines = createInterface({ input, crlfDelay: Infinity }); const pending = new Set(); + const running = new Map(); for await (const line of lines) { if (!line.trim()) continue; - const work = respond(server, line, log).then((response) => { + const work = respond(server, line, log, running).then((response) => { if (response) output.write(`${JSON.stringify(response)}\n`); }); pending.add(work); @@ -45,16 +50,30 @@ export async function serveStdio(server, { input, output, log = () => {} }) { await Promise.all(pending); } -async function respond(server, line, log) { +async function respond(server, line, log, running) { let message; try { message = JSON.parse(line); } catch { return parseErrorResponse(); } + if (message?.method === 'notifications/cancelled') { + running.get(message.params?.requestId)?.abort(); + return null; + } const described = describeRequest(message); if (described) log(described); - const response = await server.handle(message); + const id = message?.id; + const cancellable = id !== undefined && id !== null && !running.has(id); + const controller = new AbortController(); + if (cancellable) running.set(id, controller); + let response; + try { + response = await server.handle(message, { signal: controller.signal }); + } finally { + if (cancellable) running.delete(id); + } + if (controller.signal.aborted) return null; // A failed tool call answers the client, not the log; say why here too. if (response?.result?.isError) log(` ${described} failed: ${response.result.content?.[0]?.text}`); diff --git a/src/tools/localRoute.test.mjs b/src/tools/localRoute.test.mjs index 9fe113c86..d02b3d014 100644 --- a/src/tools/localRoute.test.mjs +++ b/src/tools/localRoute.test.mjs @@ -195,3 +195,44 @@ test('a client that disconnects cancels its tool call', async (t) => { await aborted.promise; assert.equal(seen.aborted, true); }); + +test('a client that stops sending its request body is answered with a timeout', async (t) => { + let middleware; + const plugin = localMcpPlugin({ + bodyTimeoutMs: 20, + createServer: () => ({ handle: async () => null }), + }); + plugin.configureServer({ + middlewares: { use: (path, handler) => (middleware = handler) }, + }); + const http = createServer((req, res) => middleware(req, res)); + await new Promise((resolve) => http.listen(0, '127.0.0.1', resolve)); + t.after(() => http.close()); + const { port } = http.address(); + const status = await new Promise((resolve, reject) => { + const request = httpRequest( + { + host: '127.0.0.1', + port, + path: '/mcp', + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Content-Length': '100', + Host: `localhost:${port}`, + }, + }, + (response) => { + response.resume(); + resolve(response.statusCode); + request.destroy(); + }, + ); + request.on('error', (error) => { + if (error.code !== 'ECONNRESET') reject(error); + }); + // Part of the body, then nothing more. + request.write('{"jsonrpc"'); + }); + assert.equal(status, 408); +}); diff --git a/src/tools/localServer.test.mjs b/src/tools/localServer.test.mjs index 7fd47e251..ccdd5c69e 100644 --- a/src/tools/localServer.test.mjs +++ b/src/tools/localServer.test.mjs @@ -171,3 +171,43 @@ test("only the key in the server's own panel page opens panel requests", async ( }); assert.doesNotMatch(otherPage.result.contents[0].text, new RegExp(key)); }); + +test('a request the client cancels over stdio is aborted and gets no answer', async () => { + let seen; + const server = { + handle: (message, { signal } = {}) => { + if (message.id === 1) { + seen = signal; + return new Promise((resolve) => + signal.addEventListener( + 'abort', + () => resolve({ jsonrpc: '2.0', id: 1, result: {} }), + { once: true }, + ), + ); + } + return Promise.resolve( + message.id === undefined + ? null + : { jsonrpc: '2.0', id: message.id, result: {} }, + ); + }, + }; + const input = new PassThrough(); + const output = new PassThrough(); + let written = ''; + output.on('data', (chunk) => (written += chunk)); + const served = serveStdio(server, { input, output }); + input.write('{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{}}\n'); + input.write('{"jsonrpc":"2.0","id":2,"method":"ping"}\n'); + input.end( + '{"jsonrpc":"2.0","method":"notifications/cancelled","params":{"requestId":1}}\n', + ); + await served; + assert.equal(seen.aborted, true); + const ids = written + .trim() + .split('\n') + .map((line) => JSON.parse(line).id); + assert.deepEqual(ids, [2]); +}); diff --git a/src/tools/queries/panelRequest.js b/src/tools/queries/panelRequest.js index cd6e00158..8606d7809 100644 --- a/src/tools/queries/panelRequest.js +++ b/src/tools/queries/panelRequest.js @@ -48,9 +48,67 @@ const INCOMPRESSIBLE = /^(?:image\/(?!svg)|video\/|audio\/|font\/woff2)|zip|compressed/; const COMPRESS_MIN_BYTES = 1024; -// Bodies too large for one call, kept for the calls that read the rest. -const held = new Map(); -let heldBytes = 0; +/** + * Requests in flight at once for one server's panels. Several panels (one + * per conversation) share a server, so further requests wait their turn + * rather than fail; past `PANEL_QUEUED_REQUESTS` waiting, they are refused. + */ +export const PANEL_CONCURRENT_REQUESTS = 6; +export const PANEL_QUEUED_REQUESTS = 256; + +/** + * Each server's panel state: bodies too large for one call, kept for the + * calls that read the rest, and the requests in flight. Kept per server + * (its `services.app`), so one server's panel never reads another's. + */ +const panels = new WeakMap(); +function panelState(app) { + if (!panels.has(app)) + panels.set(app, { + held: new Map(), + heldBytes: 0, + inFlight: 0, + waiting: [], + }); + return panels.get(app); +} + +/** Take a place among the requests in flight, waiting for one if need be. */ +async function acquire(state, signal) { + if (state.inFlight < PANEL_CONCURRENT_REQUESTS) { + state.inFlight += 1; + return; + } + if (state.waiting.length >= PANEL_QUEUED_REQUESTS) + throw new ToolError( + 'retry_later', + 'The panel has too many requests waiting', + { + retryAfterSeconds: 1, + }, + ); + await new Promise((resolve, reject) => { + const waiter = { resolve }; + const abort = () => { + const index = state.waiting.indexOf(waiter); + if (index !== -1) state.waiting.splice(index, 1); + reject(signal.reason); + }; + waiter.resolve = () => { + signal?.removeEventListener('abort', abort); + resolve(); + }; + signal?.addEventListener('abort', abort, { once: true }); + state.waiting.push(waiter); + }); +} + +/** Give up a place, handing it straight to the next waiting request. */ +function releaseSlot(state) { + const next = state.waiting.shift(); + if (next) next.resolve(); + else state.inFlight -= 1; +} function base64(bytes) { let binary = ''; @@ -81,26 +139,27 @@ async function gzip(bytes) { return new Uint8Array(await new Response(stream).arrayBuffer()); } -function release(id) { - heldBytes -= held.get(id).bytes.length; - held.delete(id); +function release(state, id) { + state.heldBytes -= state.held.get(id).bytes.length; + state.held.delete(id); } /** Drop held responses nobody continued in time. */ -function forgetExpired(now = Date.now()) { - for (const [id, entry] of held) if (entry.expires <= now) release(id); +function forgetExpired(state, now = Date.now()) { + for (const [id, entry] of state.held) + if (entry.expires <= now) release(state, id); } /** Hold a response for its later parts, first making room for it. */ -function hold(id, entry) { - forgetExpired(); +function hold(state, id, entry) { + forgetExpired(state); // Oldest first; a response being read is renewed, so it is the newest. - for (const oldest of held.keys()) { - if (heldBytes + entry.bytes.length <= HELD_LIMIT_BYTES) break; - release(oldest); + for (const oldest of state.held.keys()) { + if (state.heldBytes + entry.bytes.length <= HELD_LIMIT_BYTES) break; + release(state, oldest); } - held.set(id, entry); - heldBytes += entry.bytes.length; + state.held.set(id, entry); + state.heldBytes += entry.bytes.length; } /** Read a response body, refusing one larger than the panel may load. */ @@ -214,13 +273,14 @@ export const panelRequest = defineTool({ 'invalid_arguments', "Only the God's Eye View panel may make this request", ); - forgetExpired(); + const state = panelState(services.app); + forgetExpired(state); if (args.id !== undefined) { - const entry = held.get(args.id); + const entry = state.held.get(args.id); if (entry) { // Still being read: renew it, and keep it newest. - held.delete(args.id); - held.set(args.id, { ...entry, expires: Date.now() + HELD_MS }); + state.held.delete(args.id); + state.held.set(args.id, { ...entry, expires: Date.now() + HELD_MS }); } if (!entry) throw new ToolError( @@ -241,6 +301,8 @@ export const panelRequest = defineTool({ if (FORWARDED_REQUEST_HEADERS.includes(name.toLowerCase())) headers[name] = String(value); } + signal?.throwIfAborted(); + await acquire(state, signal); const deadline = AbortSignal.timeout(REQUEST_TIMEOUT_MS); let answer; let bytes; @@ -260,6 +322,8 @@ export const panelRequest = defineTool({ if (signal?.aborted) throw error; if (error instanceof ToolError) throw error; throw new ToolError('unavailable', "The app's server did not answer"); + } finally { + releaseSlot(state); } const type = answer.headers.get('content-type') || ''; let encoding = 'identity'; @@ -280,7 +344,7 @@ export const panelRequest = defineTool({ }; if (bytes.length > PANEL_PART_BYTES) { response.id = crypto.randomUUID(); - hold(response.id, { + hold(state, response.id, { path, response, bytes, diff --git a/src/tools/queries/panelRequest.test.mjs b/src/tools/queries/panelRequest.test.mjs index fc9c07896..05bce98f9 100644 --- a/src/tools/queries/panelRequest.test.mjs +++ b/src/tools/queries/panelRequest.test.mjs @@ -2,6 +2,7 @@ import assert from 'node:assert/strict'; import test from 'node:test'; import { composeCatalog, coreTools } from '../index.js'; import { + PANEL_CONCURRENT_REQUESTS, PANEL_PART_BYTES, PANEL_RESPONSE_LIMIT_BYTES, } from './panelRequest.js'; @@ -251,3 +252,90 @@ test('routes the panel never loads are refused in any case or encoding', async ( ['/api/setups-help', '/mcpx.json'], ); }); + +test('requests past the limit wait their turn instead of failing', async () => { + const gates = []; + const { catalog, requests } = catalogWith( + () => + new Promise((resolve) => gates.push(() => resolve(new Response('ok')))), + ); + const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); + // Two panels on one server, each sending as many requests as it may. + const calls = Array.from({ length: PANEL_CONCURRENT_REQUESTS * 2 }, (_, n) => + catalog.call('panel_request', { key: KEY, path: `/panel/${n}.js` }), + ); + await tick(); + assert.equal(requests.length, PANEL_CONCURRENT_REQUESTS); + // Each finished request lets one waiting request start. + for (let started = PANEL_CONCURRENT_REQUESTS; gates.length;) { + gates.shift()(); + await tick(); + if (started < calls.length) { + started += 1; + assert.equal(requests.length, started); + } + } + const results = await Promise.all(calls); + assert.ok(results.every(({ data }) => data.status === 200)); +}); + +test('a waiting request its caller cancels leaves the queue', async () => { + const gates = []; + const { catalog, requests } = catalogWith( + () => + new Promise((resolve) => gates.push(() => resolve(new Response('ok')))), + ); + const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); + const busy = Array.from({ length: PANEL_CONCURRENT_REQUESTS }, (_, n) => + catalog.call('panel_request', { key: KEY, path: `/panel/${n}.js` }), + ); + const cancel = new AbortController(); + const waiting = catalog.call( + 'panel_request', + { key: KEY, path: '/panel/cancelled.js' }, + { signal: cancel.signal }, + ); + const after = catalog.call('panel_request', { + key: KEY, + path: '/panel/after.js', + }); + await tick(); + cancel.abort(); + await assert.rejects(waiting); + gates.shift()(); + await tick(); + // The cancelled request never reached the server; the next one did. + assert.equal(requests.at(-1).path, '/panel/after.js'); + while (gates.length) gates.shift()(); + await Promise.all([...busy, after]); + assert.ok( + !requests.some((request) => request.path === '/panel/cancelled.js'), + ); +}); + +test("one server's held responses are not readable through another's", async () => { + const big = new Uint8Array(PANEL_PART_BYTES * 2).fill(7); + const one = catalogWith( + () => new Response(big, { headers: { 'content-type': 'image/png' } }), + ); + const other = catalogWith(() => new Response('x')); + const { data } = await one.catalog.call('panel_request', { + key: KEY, + path: '/big.png', + }); + assert.ok(data.id); + await assert.rejects( + other.catalog.call('panel_request', { + key: KEY, + id: data.id, + offset: data.nextOffset, + }), + (error) => error.code === 'invalid_arguments', + ); + const rest = await one.catalog.call('panel_request', { + key: KEY, + id: data.id, + offset: data.nextOffset, + }); + assert.equal(rest.data.offset, data.nextOffset); +});