diff --git a/.changeset/storeevent-rejection-teardown.md b/.changeset/storeevent-rejection-teardown.md new file mode 100644 index 0000000000..519789b8a6 --- /dev/null +++ b/.changeset/storeevent-rejection-teardown.md @@ -0,0 +1,18 @@ +--- +'@modelcontextprotocol/server': patch +--- + +Fix a rejected event-store write leaking the request stream in +`WebStandardStreamableHTTPServerTransport`. + +`send()` awaits the user-supplied `eventStore.storeEvent()` before writing a response to +the per-request SSE stream. A rejection propagated straight out of `send()`, skipping +every teardown below it: the stream mapping and the request correlation stayed in their +maps, the keep-alive timer stayed armed, and the HTTP response body was never closed — so +the client waited on a stream that would never carry its response, and the server held the +request forever. + +The write is now wrapped so a rejection retires the request and closes its stream before +the error is rethrown. `send()` still rejects, so callers continue to see the failure. + +Also applies to `NodeStreamableHTTPServerTransport`, which wraps this transport. diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index c0f48560a2..1da46d7939 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -1154,7 +1154,24 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // interpreted as the client cancelling its request. let eventId: string | undefined; if (this._eventStore) { - eventId = await this._eventStore.storeEvent(streamId, message); + try { + eventId = await this._eventStore.storeEvent(streamId, message); + } catch (error) { + // The event store is user-supplied. A rejected write used to + // propagate straight out of send(), skipping every teardown + // below: the per-request SSE stream stayed open with its + // keep-alive timer armed, and the stream and correlation + // entries leaked. Retire the request first, then rethrow so + // the caller still sees the failure. + this._streamMapping.get(streamId)?.cleanup(); + for (const [id, mappedStreamId] of this._requestToStreamMapping) { + if (mappedStreamId === streamId) { + this._requestResponseMap.delete(id); + this._requestToStreamMapping.delete(id); + } + } + throw error; + } // Re-read after the await: a Last-Event-ID reconnect during // storeEvent() may have registered a resumed stream under this // streamId (mirrors the standalone path's post-await read). diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index 9ec6baf46c..30d55240f2 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -1406,6 +1406,44 @@ describe('Zod v4', () => { expect(cleanupCalls).toEqual(['stream-1']); }); }); + + describe('a rejected event store write on the response path', () => { + it('should retire the request and close its stream instead of leaking it', async () => { + const eventStore: EventStore = { + async storeEvent(): Promise { + throw new Error('event store write failed'); + }, + async replayEventsAfter(): Promise { + return ''; + } + }; + + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: undefined, + eventStore, + keepAliveMs: 20 + }); + transport.onmessage = () => {}; + + const res = await transport.handleRequest(createRequest('POST', { jsonrpc: '2.0', id: 1, method: 'ping' } as JSONRPCMessage)); + expect(res.status).toBe(200); + + await expect(transport.send({ jsonrpc: '2.0', id: 1, result: {} } as JSONRPCMessage)).rejects.toThrow( + 'event store write failed' + ); + + // @ts-expect-error accessing private map for test purposes + expect(transport._streamMapping.size).toBe(0); + // @ts-expect-error accessing private map for test purposes + expect(transport._requestToStreamMapping.size).toBe(0); + // @ts-expect-error accessing private map for test purposes + expect(transport._requestResponseMap.size).toBe(0); + + // The SSE body is terminated rather than held open by the keep-alive timer. + const reader = res.body!.getReader(); + await expect(reader.read()).resolves.toMatchObject({ done: true }); + }); + }); }); describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => {