Skip to content

Commit f46d957

Browse files
authored
fix(everything): clean up subscriptions on session disconnect (#4712)
Export removeSubscriber() to drop a session from all URI subscriber sets when a transport session closes, and call it from cleanup() alongside the existing interval and logging teardown. Fixes #4710 Co-authored-by: arimu1 <19286898+arimu1@users.noreply.github.com>
1 parent d73f99e commit f46d957

3 files changed

Lines changed: 102 additions & 0 deletions

File tree

‎src/everything/__tests__/resources.test.ts‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,14 @@ import {
1818
registerSessionResource,
1919
} from '../resources/session.js';
2020
import { registerFileResources } from '../resources/files.js';
21+
import {
22+
SubscribeRequestSchema,
23+
} from '@modelcontextprotocol/sdk/types.js';
2124
import {
2225
setSubscriptionHandlers,
2326
beginSimulatedResourceUpdates,
2427
stopSimulatedResourceUpdates,
28+
removeSubscriber,
2529
} from '../resources/subscriptions.js';
2630

2731
describe('Resource Templates', () => {
@@ -298,10 +302,87 @@ describe('Subscriptions', () => {
298302
});
299303
});
300304

305+
describe('removeSubscriber', () => {
306+
const testUri = 'demo://resource/dynamic/text/1';
307+
const sessionId = 'disconnect-test-session';
308+
309+
let subscribeHandler: (
310+
request: { params: { uri: string } },
311+
extra: { sessionId: string }
312+
) => Promise<unknown>;
313+
314+
beforeEach(() => {
315+
const handlers = new Map<unknown, typeof subscribeHandler>();
316+
const mockServer = {
317+
server: {
318+
setRequestHandler: vi.fn((schema, handler) => {
319+
handlers.set(schema, handler);
320+
}),
321+
notification: vi.fn(),
322+
},
323+
sendLoggingMessage: vi.fn(),
324+
} as unknown as McpServer;
325+
326+
setSubscriptionHandlers(mockServer);
327+
subscribeHandler = handlers.get(SubscribeRequestSchema)!;
328+
});
329+
330+
afterEach(() => {
331+
stopSimulatedResourceUpdates(sessionId);
332+
removeSubscriber(sessionId);
333+
});
334+
335+
it('should drop a disconnected session from all subscriptions', async () => {
336+
const notification = vi.fn();
337+
const mockServer = {
338+
server: {
339+
notification,
340+
},
341+
} as unknown as McpServer;
342+
343+
await subscribeHandler({ params: { uri: testUri } }, { sessionId });
344+
345+
beginSimulatedResourceUpdates(mockServer, sessionId);
346+
expect(notification).toHaveBeenCalled();
347+
348+
notification.mockClear();
349+
removeSubscriber(sessionId);
350+
stopSimulatedResourceUpdates(sessionId);
351+
352+
beginSimulatedResourceUpdates(mockServer, sessionId);
353+
expect(notification).not.toHaveBeenCalled();
354+
});
355+
356+
it('should not affect other sessions subscribed to the same URI', async () => {
357+
const otherSessionId = 'other-session';
358+
const notification = vi.fn();
359+
const mockServer = {
360+
server: {
361+
notification,
362+
},
363+
} as unknown as McpServer;
364+
365+
await subscribeHandler({ params: { uri: testUri } }, { sessionId });
366+
await subscribeHandler(
367+
{ params: { uri: testUri } },
368+
{ sessionId: otherSessionId }
369+
);
370+
371+
removeSubscriber(sessionId);
372+
373+
beginSimulatedResourceUpdates(mockServer, otherSessionId);
374+
expect(notification).toHaveBeenCalled();
375+
376+
stopSimulatedResourceUpdates(otherSessionId);
377+
removeSubscriber(otherSessionId);
378+
});
379+
});
380+
301381
describe('simulated resource updates lifecycle', () => {
302382
afterEach(() => {
303383
// Clean up any intervals
304384
stopSimulatedResourceUpdates('lifecycle-test-session');
385+
removeSubscriber('lifecycle-test-session');
305386
});
306387

307388
it('should start and stop updates without errors', () => {

‎src/everything/resources/subscriptions.ts‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -166,3 +166,22 @@ export const stopSimulatedResourceUpdates = (sessionId?: string) => {
166166
subsUpdateIntervals.delete(sessionId);
167167
}
168168
};
169+
170+
/**
171+
* Removes a session from every URI's subscriber set, dropping any URI entry
172+
* that ends up with no remaining subscribers.
173+
*
174+
* A session that disconnects without explicitly unsubscribing otherwise stays
175+
* in `subscriptions` for the life of the process. Call this from the
176+
* transport's `cleanup(sessionId)` when a session ends.
177+
*
178+
* @param {string} [sessionId]
179+
*/
180+
export const removeSubscriber = (sessionId?: string) => {
181+
for (const [uri, subscribers] of subscriptions) {
182+
subscribers.delete(sessionId);
183+
if (subscribers.size === 0) {
184+
subscriptions.delete(uri);
185+
}
186+
}
187+
};

‎src/everything/server/index.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import {
66
import {
77
setSubscriptionHandlers,
88
stopSimulatedResourceUpdates,
9+
removeSubscriber,
910
} from "../resources/subscriptions.js";
1011
import { registerConditionalTools, registerTools } from "../tools/index.js";
1112
import { registerResources, readInstructions } from "../resources/index.js";
@@ -110,6 +111,7 @@ export const createServer: () => ServerFactoryResponse = () => {
110111
// Stop any simulated logging or resource updates that may have been initiated.
111112
stopSimulatedLogging(sessionId);
112113
stopSimulatedResourceUpdates(sessionId);
114+
removeSubscriber(sessionId);
113115
// Clean up task store timers
114116
taskStore.cleanup();
115117
if (initializeTimeout) clearTimeout(initializeTimeout);

0 commit comments

Comments
 (0)