diff --git a/.changeset/fix-listen-abort-listener-after-complete.md b/.changeset/fix-listen-abort-listener-after-complete.md new file mode 100644 index 0000000000..06ec4b589e --- /dev/null +++ b/.changeset/fix-listen-abort-listener-after-complete.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/server': patch +--- + +`subscriptions/listen` no longer registers an `abort` listener on the request signal once the stream has already completed. A filter honored down to nothing is torn down inside the stream's `start()`, and teardown short-circuits once `closed` is set, so the listener that wiring then added could never be removed — not by a later teardown, and not by `closeAll()`. Completed listens that share one still-live signal therefore accumulated a listener, and its subscription closure, per request. diff --git a/packages/server/src/server/listenRouter.ts b/packages/server/src/server/listenRouter.ts index e58807d5c1..b276f2735a 100644 --- a/packages/server/src/server/listenRouter.ts +++ b/packages/server/src/server/listenRouter.ts @@ -238,7 +238,10 @@ export function createListenRouter(options: ListenRouterOptions): ListenRouter { } }); - if (signal !== undefined) { + // A stream honored down to nothing is already torn down inside `start()`, and + // `teardown()` short-circuits once `closed` is set — wiring `onAbort` to such a + // stream would strand it on the caller's signal, past `closeAll()`. + if (!closed && signal !== undefined) { if (signal.aborted) { teardown(false); } else { diff --git a/packages/server/test/server/listenRouterAbortCleanup.test.ts b/packages/server/test/server/listenRouterAbortCleanup.test.ts new file mode 100644 index 0000000000..073f2faede --- /dev/null +++ b/packages/server/test/server/listenRouterAbortCleanup.test.ts @@ -0,0 +1,76 @@ +/** + * `createListenRouter` — abort wiring for a listen stream that has already completed. + * + * When nothing is honored, `ReadableStream.start()` tears the stream down synchronously, so + * by the time the request-signal wiring runs the subscription is closed. Registering `onAbort` + * anyway strands a listener on the caller's signal: `teardown()` returns early once `closed` + * is set, so its `abortCleanup` never runs, and `closeAll()` cannot reach it either (the + * teardown is already out of the open set). One live signal serving many completed listens — + * the shape of a shared per-connection signal — therefore grows a listener per request. + */ +import { getEventListeners } from 'node:events'; +import type { JSONRPCRequest, ServerCapabilities } from '@modelcontextprotocol/core-internal'; +import { describe, expect, it } from 'vitest'; + +import { createListenRouter } from '../../src/server/listenRouter'; +import { InMemoryServerEventBus } from '../../src/server/serverEventBus'; + +const SERVER_INFO = { name: 'listen-abort-test', version: '1' }; + +function listenRequest(id: number, filter: Record): JSONRPCRequest { + return { jsonrpc: '2.0', id, method: 'subscriptions/listen', params: { notifications: filter } }; +} + +/** Capabilities that advertise nothing, so any requested filter is narrowed to empty. */ +const NO_CAPABILITIES: ServerCapabilities = {}; +/** Capabilities that advertise `tools.listChanged`, so an empty request honors nothing. */ +const TOOLS_LIST_CHANGED: ServerCapabilities = { tools: { listChanged: true } }; + +describe('createListenRouter — completed listens do not retain abort listeners', () => { + it('attaches no abort listener when capabilities narrow the filter to nothing', async () => { + const signal = new AbortController().signal; + const bus = new InMemoryServerEventBus(); + const router = createListenRouter({ bus, keepAliveMs: 0 }); + + for (let id = 1; id <= 3; id++) { + const response = router.serve(listenRequest(id, { toolsListChanged: true }), signal, NO_CAPABILITIES, SERVER_INFO); + await expect(response.text()).resolves.toContain('"resultType":"complete"'); + expect(router.openCount).toBe(0); + expect(bus.listenerCount).toBe(0); + expect(getEventListeners(signal, 'abort')).toHaveLength(0); + } + + router.closeAll(); + expect(getEventListeners(signal, 'abort')).toHaveLength(0); + }); + + it('attaches no abort listener when the requested filter is empty', async () => { + const signal = new AbortController().signal; + const router = createListenRouter({ bus: new InMemoryServerEventBus(), keepAliveMs: 0 }); + + const response = router.serve(listenRequest(7, {}), signal, TOOLS_LIST_CHANGED, SERVER_INFO); + await expect(response.text()).resolves.toContain('"resultType":"complete"'); + expect(getEventListeners(signal, 'abort')).toHaveLength(0); + + router.closeAll(); + expect(getEventListeners(signal, 'abort')).toHaveLength(0); + }); + + it('still wires and releases the abort listener for a stream that stays open', async () => { + const controller = new AbortController(); + const bus = new InMemoryServerEventBus(); + const router = createListenRouter({ bus, keepAliveMs: 0 }); + + const response = router.serve(listenRequest(9, { toolsListChanged: true }), controller.signal, TOOLS_LIST_CHANGED, SERVER_INFO); + expect(router.openCount).toBe(1); + expect(bus.listenerCount).toBe(1); + expect(getEventListeners(controller.signal, 'abort')).toHaveLength(1); + + controller.abort(); + expect(router.openCount).toBe(0); + expect(bus.listenerCount).toBe(0); + expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0); + + await response.text(); + }); +});