From f109cc4bdbb0b439d590b3863131e4d0be821d44 Mon Sep 17 00:00:00 2001 From: Dmitriy Vasyura Date: Sun, 20 Sep 2026 08:05:05 -0700 Subject: [PATCH] mcp: handle HTTP stream cancellation during disposal Disposing an MCP HTTP transport already aborts its fetch response. Cancelling the errored reader again creates a rejected promise that was previously discarded, surfacing an unhandled AbortError during normal stop/disposal. Skip that redundant cancellation after disposal and await cancellation for active read failures. Add a failing-first disposal regression and a control that keeps genuine stream failures visible. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/vs/workbench/api/common/extHostMcp.ts | 5 +- .../api/test/common/extHostMcp.test.ts | 81 ++++++++++++++++++- 2 files changed, 79 insertions(+), 7 deletions(-) diff --git a/src/vs/workbench/api/common/extHostMcp.ts b/src/vs/workbench/api/common/extHostMcp.ts index bcce7dd46bbce..29c55273ca213 100644 --- a/src/vs/workbench/api/common/extHostMcp.ts +++ b/src/vs/workbench/api/common/extHostMcp.ts @@ -710,12 +710,11 @@ export class McpHTTPHandle extends Disposable { try { chunk = await raceCancellationError(reader.read(), this._cts.token); } catch (err) { - reader.cancel(); if (this._store.isDisposed) { return; - } else { - throw err; } + await reader.cancel(); + throw err; } if (chunk.value) { diff --git a/src/vs/workbench/api/test/common/extHostMcp.test.ts b/src/vs/workbench/api/test/common/extHostMcp.test.ts index e18bb75ce4a02..d7fc9ddf72424 100644 --- a/src/vs/workbench/api/test/common/extHostMcp.test.ts +++ b/src/vs/workbench/api/test/common/extHostMcp.test.ts @@ -5,9 +5,15 @@ import * as assert from 'assert'; import * as sinon from 'sinon'; -import { LogLevel } from '../../../../platform/log/common/log.js'; -import { createAuthMetadata, CommonResponse, IAuthMetadata } from '../../common/extHostMcp.js'; +import { DeferredPromise } from '../../../../base/common/async.js'; +import { toDisposable } from '../../../../base/common/lifecycle.js'; +import { URI } from '../../../../base/common/uri.js'; +import { mock } from '../../../../base/test/common/mock.js'; +import { LogLevel, NullLogService } from '../../../../platform/log/common/log.js'; +import { MainThreadMcpShape } from '../../common/extHost.protocol.js'; +import { createAuthMetadata, CommonRequestInit, CommonResponse, IAuthMetadata, McpHTTPHandle } from '../../common/extHostMcp.js'; import { ensureNoDisposablesAreLeakedInTestSuite } from '../../../../base/test/common/utils.js'; +import { McpServerTransportType } from '../../../contrib/mcp/common/mcpTypes.js'; // Test constants to avoid magic strings const TEST_MCP_URL = 'https://example.com/mcp'; @@ -100,7 +106,75 @@ async function createTestAuthMetadata(options: { } suite('ExtHostMcp', () => { - ensureNoDisposablesAreLeakedInTestSuite(); + const store = ensureNoDisposablesAreLeakedInTestSuite(); + + suite('McpHTTPHandle stream cleanup', () => { + teardown(() => sinon.restore()); + + const createStreamingHandle = async () => { + const reading = new DeferredPromise(); + const backchannelStopped = new DeferredPromise(); + let controller!: ReadableStreamDefaultController; + const stream = new ReadableStream({ + start: value => { controller = value; }, + pull: () => { void reading.complete(); }, + }, { highWaterMark: 0 }); + const reader = stream.getReader(); + sinon.stub(stream, 'getReader').returns(reader); + const cancel = sinon.spy(reader, 'cancel'); + const warnings: string[] = []; + const proxy = new class extends mock() { + override $onDidChangeState(): void { } + override $onDidPublishLog(_id: number, level: LogLevel, message: string): void { + if (level === LogLevel.Warning) { + warnings.push(message); + } + if (message.startsWith('405 status connecting')) { + void backchannelStopped.complete(); + } + } + }; + const handle = store.add(new class extends McpHTTPHandle { + protected override async _fetchInternal(_url: string, init?: CommonRequestInit): Promise { + if (init?.method === 'GET') { + return createMockResponse({ status: 405 }); + } + assert.strictEqual(init?.method, 'POST'); + const signal = init?.signal; + assert.ok(signal); + const onAbort = () => controller.error(signal.reason); + signal.addEventListener('abort', onAbort, { once: true }); + store.add(toDisposable(() => signal.removeEventListener('abort', onAbort))); + return { ...createMockResponse({ headers: { 'Content-Type': 'text/event-stream' } }), body: stream }; + } + }(1, { type: McpServerTransportType.HTTP, uri: URI.parse(TEST_MCP_URL), headers: [] }, proxy, store.add(new NullLogService()))); + store.add(toDisposable(() => reader.releaseLock())); + const sending = handle.send(JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'initialize', params: {} })); + await Promise.all([reading.p, backchannelStopped.p]); + return { handle, controller, cancel, sending, warnings }; + }; + + test('does not cancel an already-aborted reader when disposed', async () => { + const { handle, cancel, sending, warnings } = await createStreamingHandle(); + handle.dispose(); + await sending; + // Observe any redundant cancellation rejection so the regression fails on the assertion below. + await Promise.all(cancel.returnValues.map(result => assert.rejects(result, { name: 'AbortError' }))); + assert.deepStrictEqual({ cancellations: cancel.callCount, warnings }, { cancellations: 0, warnings: [] }); + }); + + test('still reports a stream read failure while the handle is active', async () => { + const { controller, cancel, sending, warnings } = await createStreamingHandle(); + const error = new Error('fixture stream read failed'); + controller.error(error); + await sending; + await Promise.all(cancel.returnValues.map(result => assert.rejects(result, candidate => candidate === error))); + assert.deepStrictEqual({ cancellations: cancel.callCount, warnings }, { + cancellations: 1, + warnings: ['Error reading SSE stream: Error: fixture stream read failed'], + }); + }); + }); suite('IAuthMetadata', () => { suite('properties', () => { @@ -733,4 +807,3 @@ suite('ExtHostMcp', () => { }); }); }); -