fix(server): tear down the request stream when an event store write rejects

.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.

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).

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<EventId> {
+                    throw new Error('event store write failed');
+                },
+                async replayEventsAfter(): Promise<StreamId> {
+                    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', () => {