From 76c50211af6bf033059b03207454093f060495ea Mon Sep 17 00:00:00 2001 From: rxits <132228481+rxits@users.noreply.github.com> Date: Sat, 1 Aug 2026 23:26:02 +0530 Subject: [PATCH 1/2] fix(server): tear down the request stream when an event store write rejects 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. 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 rethrowing. send() still rejects, so callers continue to see the failure. --- .changeset/storeevent-rejection-teardown.md | 18 +++++++++ packages/server/src/server/streamableHttp.ts | 19 +++++++++- .../server/test/server/streamableHttp.test.ts | 38 +++++++++++++++++++ 3 files changed, 74 insertions(+), 1 deletion(-) create mode 100644 .changeset/storeevent-rejection-teardown.md 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..90e7c10b96 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.entries()]) { + 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', () => { From 996b709a0878e44983985f31643d3d2a6888a75b Mon Sep 17 00:00:00 2001 From: rxits <132228481+rxits@users.noreply.github.com> Date: Mon, 3 Aug 2026 07:04:20 +0530 Subject: [PATCH 2/2] fix(server): drop redundant spread when retiring correlations for...of iterates a Map directly; the array copy was unnecessary and tripped unicorn/no-useless-spread. Deleting entries during Map iteration is well-defined, and the entries removed here are the ones the loop intends to retire. --- packages/server/src/server/streamableHttp.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 90e7c10b96..1da46d7939 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -1164,7 +1164,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // 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.entries()]) { + for (const [id, mappedStreamId] of this._requestToStreamMapping) { if (mappedStreamId === streamId) { this._requestResponseMap.delete(id); this._requestToStreamMapping.delete(id);