diff --git a/.changeset/connect-signal-probe.md b/.changeset/connect-signal-probe.md new file mode 100644 index 0000000000..7b8a3600a9 --- /dev/null +++ b/.changeset/connect-signal-probe.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +`Client.connect()` now honors its `signal` during the `server/discover` probe in `auto` and pinned `versionNegotiation` modes. Aborting the signal rejects the connect promptly with the same error the legacy `initialize` path raises, instead of waiting out the probe timeout, and cancels the probe's HTTP request. A signal that is already aborted rejects before anything is sent. With the SDK's stdio transport the rejection waits for the probe sibling process to be reaped (up to about a second if it ignores `SIGTERM`), and the session process is never started. diff --git a/packages/client/src/client/client.ts b/packages/client/src/client/client.ts index 096a8660e2..ac91c341c3 100644 --- a/packages/client/src/client/client.ts +++ b/packages/client/src/client/client.ts @@ -1154,7 +1154,8 @@ export class Client extends Protocol { clientInfo: this._clientInfo, capabilities: this._capabilities, environment: detectProbeEnvironment(), - defaultTimeoutMs: options?.timeout ?? DEFAULT_REQUEST_TIMEOUT_MSEC + defaultTimeoutMs: options?.timeout ?? DEFAULT_REQUEST_TIMEOUT_MSEC, + signal: options?.signal }; // The SDK's stdio transport probes on a disposable sibling process, // so the caller's transport spends its one child life on the diff --git a/packages/client/src/client/versionNegotiation.ts b/packages/client/src/client/versionNegotiation.ts index 5dfced8c7f..bde0edeeec 100644 --- a/packages/client/src/client/versionNegotiation.ts +++ b/packages/client/src/client/versionNegotiation.ts @@ -174,7 +174,8 @@ type RawProbeReply = | { kind: 'response'; result?: unknown; error?: { code: number; message: string; data?: unknown } } | { kind: 'send-error'; error: unknown } | { kind: 'closed' } - | { kind: 'timeout' }; + | { kind: 'timeout' } + | { kind: 'aborted'; reason: unknown }; /** * Temporary ownership of a raw transport for the negotiation exchange, before @@ -255,8 +256,12 @@ class ProbeWindow { /** * Send one probe request and await its reply. Probe ids are strings, so they * never collide with Protocol's numeric ids (e.g. on a shared stdio pipe). + * An abort of `signal` settles the exchange and tears down the request. */ - async exchange(buildRequest: (id: string) => JSONRPCRequest, timeoutMs: number): Promise { + async exchange(buildRequest: (id: string) => JSONRPCRequest, timeoutMs: number, signal?: AbortSignal): Promise { + if (signal?.aborted) { + return { kind: 'aborted', reason: signal.reason }; + } const id = `server-discover-probe-${++this._probeCounter}`; return new Promise(resolve => { let settled = false; @@ -264,14 +269,20 @@ class ProbeWindow { if (settled) return; settled = true; clearTimeout(timer); + signal?.removeEventListener('abort', onAbort); if (this._pending?.id === id) { this._pending = undefined; } resolve(reply); }; + const onAbort = () => settle({ kind: 'aborted', reason: signal?.reason }); const timer = setTimeout(() => settle({ kind: 'timeout' }), timeoutMs); this._pending = { id, resolve: settle }; - this._transport.send(buildRequest(id)).catch((error: unknown) => settle({ kind: 'send-error', error })); + // Registered before send() so an abort fired during the send still settles here. + signal?.addEventListener('abort', onAbort, { once: true }); + const request = buildRequest(id); + const sent = signal === undefined ? this._transport.send(request) : this._transport.send(request, { requestSignal: signal }); + sent.catch((error: unknown) => settle({ kind: 'send-error', error })); }); } @@ -384,7 +395,7 @@ function isUnusableReplyError(error: unknown): boolean { return error instanceof Error && error.name === 'SyntaxError'; } -function normalizeReply(reply: RawProbeReply, timeoutMs: number): ProbeOutcome { +function normalizeReply(reply: Exclude, timeoutMs: number): ProbeOutcome { switch (reply.kind) { case 'response': { return reply.error === undefined ? { kind: 'result', result: reply.result } : { kind: 'rpc-error', ...reply.error }; @@ -434,6 +445,11 @@ function normalizeReply(reply: RawProbeReply, timeoutMs: number): ProbeOutcome { } } +/** The error a connect-signal abort raises; mirrors Protocol.request (update with #3046). */ +function abortError(reason: unknown): SdkError { + return reason instanceof SdkError ? reason : new SdkError(SdkErrorCode.RequestTimeout, String(reason)); +} + export interface NegotiationDeps { transport: Transport; clientInfo: Implementation; @@ -451,6 +467,8 @@ export interface NegotiationDeps { * fallback stream and stays a typed error. */ disposableProbe?: boolean; + /** The caller's connect signal: an abort rejects the probe promptly. */ + signal?: AbortSignal; } export type NegotiationResult = { era: 'modern'; version: string; discover: DiscoverResult } | { era: 'legacy' }; @@ -471,6 +489,9 @@ export async function negotiateEra( const clientModernVersions = negotiation.kind === 'pin' ? [negotiation.version] : negotiation.modernVersions; const fallbackAvailable = negotiation.kind === 'auto' && negotiation.fallbackAvailable; + if (deps.signal?.aborted) { + throw abortError(deps.signal.reason); + } const window = await ProbeWindow.open(deps.transport); const probe = async (): Promise => { @@ -485,8 +506,16 @@ export async function negotiateEra( for (;;) { const reply = await window.exchange( id => buildProbeRequest(id, requestedVersion, deps.clientInfo, deps.capabilities), - timeoutMs + timeoutMs, + deps.signal ); + if (reply.kind === 'aborted') { + throw abortError(reply.reason); + } + // An abort landing in the same tick as a reply still wins: never a verdict. + if (deps.signal?.aborted) { + throw abortError(deps.signal.reason); + } if (reply.kind === 'timeout' && timeoutRetriesRemaining > 0) { timeoutRetriesRemaining--; @@ -571,6 +600,11 @@ export async function negotiateEra( window.detach(); throw error; } + if (deps.signal?.aborted) { + // The probe settled, but the abort beat the handover; leave the transport untouched too. + window.detach(); + throw abortError(deps.signal.reason); + } window.release(); return result; } @@ -660,6 +694,10 @@ export async function negotiateStdioViaSibling( // disposal window aborts: the session transport is never started. throw callerCloseAbortError(); } + if (deps.signal?.aborted) { + // Same window for the caller's signal: an abort during disposal must not start the session. + throw abortError(deps.signal.reason); + } return result; } diff --git a/packages/client/test/client/connectSignalProbe.test.ts b/packages/client/test/client/connectSignalProbe.test.ts new file mode 100644 index 0000000000..e8123cc094 --- /dev/null +++ b/packages/client/test/client/connectSignalProbe.test.ts @@ -0,0 +1,463 @@ +/** + * `connect(transport, { signal })` during the `server/discover` probe: an abort + * rejects promptly in auto and pin mode, in place and on the stdio sibling, the + * same way the legacy `initialize` path does. Cases follow the reproduction in + * https://github.com/modelcontextprotocol/typescript-sdk/issues/3133. + */ +import type { Server } from 'node:http'; +import { createServer } from 'node:http'; +import type { AddressInfo } from 'node:net'; + +import type { JSONRPCMessage, Transport, TransportSendOptions } from '@modelcontextprotocol/core-internal'; +import { isJSONRPCRequest, SdkError, SdkErrorCode } from '@modelcontextprotocol/core-internal'; +import { afterEach, describe, expect, test, vi } from 'vitest'; + +import { Client } from '../../src/client/client'; +import { StreamableHTTPClientTransport } from '../../src/client/streamableHttp'; +import { negotiateEra } from '../../src/client/versionNegotiation'; + +const MODERN = '2026-07-28'; +/** The code an abort maps to, as in `Protocol.request` (see #3046). */ +const ABORT_CODE = SdkErrorCode.RequestTimeout; +/** Long enough that only the signal can settle `connect()` within PROMPT_MS. */ +const PROBE_TIMEOUT_MS = 30_000; +const PROMPT_MS = 1000; +const ABORT_AFTER_MS = 50; + +type Settled = { state: 'connected' } | { state: 'rejected'; error: unknown } | { state: 'pending' }; + +/** How `connect()` settled within PROMPT_MS. */ +async function settleWithin(connect: Promise): Promise { + connect.catch(() => {}); + let timer: ReturnType | undefined; + const pending = new Promise(resolve => { + timer = setTimeout(() => resolve({ state: 'pending' }), PROMPT_MS); + }); + try { + return await Promise.race([ + connect.then( + (): Settled => ({ state: 'connected' }), + (error: unknown): Settled => ({ state: 'rejected', error }) + ), + pending + ]); + } finally { + clearTimeout(timer); + } +} + +function expectAbortRejection(settled: Settled, message: RegExp): void { + expect(settled.state).toBe('rejected'); + const error = (settled as { error: unknown }).error; + expect(error).toBeInstanceOf(SdkError); + expect((error as SdkError).code).toBe(ABORT_CODE); + expect((error as SdkError).message).toMatch(message); +} + +const discoverResult = { + supportedVersions: [MODERN], + capabilities: {}, + _meta: { 'io.modelcontextprotocol/serverInfo': { name: 'modern-server', version: '1.0.0' } } +}; + +type Script = (message: JSONRPCMessage, transport: FakeHttpTransport) => void; + +/** A server that never answers. */ +const silent: Script = () => {}; + +/** A 2026 server: answers server/discover. */ +const modern: Script = (message, t) => { + if (isJSONRPCRequest(message) && message.method === 'server/discover') { + t.onmessage?.({ jsonrpc: '2.0', id: message.id, result: discoverResult }); + } +}; + +/** A 2025 server: -32601 for server/discover (legacy evidence), a plain result for initialize. */ +const legacy: Script = (message, t) => { + if (!isJSONRPCRequest(message)) return; + if (message.method === 'initialize') { + t.onmessage?.({ + jsonrpc: '2.0', + id: message.id, + result: { protocolVersion: '2025-11-25', capabilities: {}, serverInfo: { name: 'legacy-server', version: '1.0.0' } } + }); + } else { + t.onmessage?.({ jsonrpc: '2.0', id: message.id, error: { code: -32_601, message: 'Method not found' } }); + } +}; + +/** An HTTP-class transport (no `stderr`/`pid`) whose replies come from a script; `close()` always fires `onclose`. */ +class FakeHttpTransport implements Transport { + onclose?: () => void; + onerror?: (error: Error) => void; + onmessage?: (message: JSONRPCMessage) => void; + sessionId?: string; + + startCalls = 0; + sent: JSONRPCMessage[] = []; + sendOptions: (TransportSendOptions | undefined)[] = []; + /** Runs synchronously inside `start()`. */ + onStart?: () => void; + /** Runs synchronously inside `send()`, before the scripted reply is queued. */ + onSend?: (message: JSONRPCMessage) => void; + + constructor(private readonly script: Script = silent) {} + + async start(): Promise { + this.startCalls++; + this.onStart?.(); + } + + async send(message: JSONRPCMessage, options?: TransportSendOptions): Promise { + this.sent.push(message); + this.sendOptions.push(options); + this.onSend?.(message); + queueMicrotask(() => this.script(message, this)); + } + + async close(): Promise { + this.onclose?.(); + } + + setProtocolVersion(): void {} +} + +/** + * An SDK-shaped stdio transport (`_serverParams` with a command, an own + * `_dispose`, `stderr`/`pid`), so `connect()` probes on a disposable sibling. + * Its "child" answers server/discover when `behavior` is 'modern' and never + * otherwise; `close()` fires `onclose` only for a started child. + */ +class FakeStdioTransport implements Transport { + static instances: FakeStdioTransport[] = []; + /** Test hook: runs when sibling disposal begins (reset after use). */ + static onDisposeStart: (() => void) | undefined; + + onclose?: () => void; + onerror?: (error: Error) => void; + onmessage?: (message: JSONRPCMessage) => void; + + readonly _serverParams: Record; + startCalls = 0; + sent: JSONRPCMessage[] = []; + disposed = false; + private _alive = false; + + constructor(params: Record) { + this._serverParams = params; + FakeStdioTransport.instances.push(this); + } + + get stderr(): null { + return null; + } + get pid(): number | null { + return this._alive ? 4242 : null; + } + + async start(): Promise { + this._alive = true; + this.startCalls++; + } + + async send(message: JSONRPCMessage): Promise { + this.sent.push(message); + queueMicrotask(() => { + if (!this._alive || !isJSONRPCRequest(message)) return; + if (message.method === 'server/discover' && this._serverParams.behavior === 'modern') { + this.onmessage?.({ jsonrpc: '2.0', id: message.id, result: discoverResult }); + } + }); + } + + async close(): Promise { + if (this._alive) { + this._alive = false; + this.onclose?.(); + } + } + + private async _dispose(): Promise { + FakeStdioTransport.onDisposeStart?.(); + this.disposed = true; + this._alive = false; + } +} + +const autoClient = (maxRetries?: number) => + new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'auto', probe: { timeoutMs: PROBE_TIMEOUT_MS, maxRetries } } }); + +const abortSoon = (reason: unknown = new Error('aborted by caller')): AbortSignal => { + const abort = new AbortController(); + setTimeout(() => abort.abort(reason), ABORT_AFTER_MS); + return abort.signal; +}; + +const discoverCount = (sent: JSONRPCMessage[]) => sent.filter(m => isJSONRPCRequest(m) && m.method === 'server/discover').length; + +describe('connect() signal during the server/discover probe', () => { + afterEach(() => { + FakeStdioTransport.instances = []; + FakeStdioTransport.onDisposeStart = undefined; + }); + + test('control: legacy mode, an abort mid-initialize rejects promptly', async () => { + const client = new Client({ name: 'c', version: '0' }, { versionNegotiation: { mode: 'legacy' } }); + + const settled = await settleWithin(client.connect(new FakeHttpTransport(), { signal: abortSoon(), timeout: PROBE_TIMEOUT_MS })); + + expectAbortRejection(settled, /aborted by caller/); + }); + + test('auto mode: an abort mid-probe rejects promptly with the legacy-path error', async () => { + const transport = new FakeHttpTransport(); + + const settled = await settleWithin(autoClient().connect(transport, { signal: abortSoon() })); + + expectAbortRejection(settled, /aborted by caller/); + expect(discoverCount(transport.sent)).toBe(1); + expect(transport.sent.some(m => isJSONRPCRequest(m) && m.method === 'initialize')).toBe(false); + }); + + test('pin mode: an abort mid-probe rejects promptly', async () => { + const client = new Client( + { name: 'c', version: '0' }, + { versionNegotiation: { mode: { pin: MODERN }, probe: { timeoutMs: PROBE_TIMEOUT_MS } } } + ); + + const settled = await settleWithin(client.connect(new FakeHttpTransport(), { signal: abortSoon() })); + + expectAbortRejection(settled, /aborted by caller/); + }); + + test('an SdkError abort reason is rethrown as is', async () => { + const reason = new SdkError(SdkErrorCode.ConnectionClosed, 'caller shut down'); + + const settled = await settleWithin(autoClient().connect(new FakeHttpTransport(), { signal: abortSoon(reason) })); + + expect(settled).toEqual({ state: 'rejected', error: reason }); + expect((settled as { error: unknown }).error).toBe(reason); + }); + + test('a non-SdkError abort reason is wrapped in an SdkError carrying its text', async () => { + const settled = await settleWithin(autoClient().connect(new FakeHttpTransport(), { signal: abortSoon('plain string reason') })); + + expectAbortRejection(settled, /plain string reason/); + }); + + test('the caller signal is forwarded to transport.send as requestSignal', async () => { + const transport = new FakeHttpTransport(modern); + const abort = new AbortController(); + + await autoClient().connect(transport, { signal: abort.signal }); + + expect(discoverCount(transport.sent)).toBe(1); + expect(transport.sendOptions[0]?.requestSignal).toBe(abort.signal); + }); + + test('after a normal reply the abort listener is removed, and a later abort touches nothing', async () => { + const transport = new FakeHttpTransport(modern); + const abort = new AbortController(); + const signal = abort.signal; + const added = vi.spyOn(signal, 'addEventListener'); + const removed = vi.spyOn(signal, 'removeEventListener'); + const abortCalls = (spy: typeof added | typeof removed) => spy.mock.calls.filter(([type]) => type === 'abort').length; + const client = autoClient(); + let closes = 0; + const errors: Error[] = []; + client.onclose = () => closes++; + client.onerror = error => errors.push(error); + + await client.connect(transport, { signal }); + expect(abortCalls(added)).toBeGreaterThan(0); + expect(abortCalls(removed)).toBe(abortCalls(added)); + + abort.abort(new Error('too late')); + await new Promise(resolve => setTimeout(resolve, 10)); + expect(closes).toBe(0); + expect(errors).toEqual([]); + expect(client.getServerVersion()?.name).toBe('modern-server'); + }); + + describe('reply and abort in the same tick', () => { + test.each([ + ['modern', modern], + ['legacy', legacy] + ] as const)('a %s verdict that lands with the abort still rejects, and no session handshake starts', async (_name, script) => { + const abort = new AbortController(); + const transport = new FakeHttpTransport((message, t) => { + script(message, t); + abort.abort(new Error('aborted with the reply')); + }); + + const settled = await settleWithin(autoClient().connect(transport, { signal: abort.signal })); + + expectAbortRejection(settled, /aborted with the reply/); + expect(transport.sent.some(m => isJSONRPCRequest(m) && m.method === 'initialize')).toBe(false); + }); + }); + + test('an abort fired inside send() with maxRetries: 2 sends exactly one server/discover', async () => { + const transport = new FakeHttpTransport(); + const abort = new AbortController(); + transport.onSend = () => abort.abort(new Error('aborted during send')); + + const settled = await settleWithin(autoClient(2).connect(transport, { signal: abort.signal })); + + expectAbortRejection(settled, /aborted during send/); + expect(discoverCount(transport.sent)).toBe(1); + }); + + test('an abort fired while the transport starts rejects before any probe is sent', async () => { + const transport = new FakeHttpTransport(modern); + const abort = new AbortController(); + transport.onStart = () => abort.abort(new Error('aborted during start')); + + const settled = await settleWithin(autoClient().connect(transport, { signal: abort.signal })); + + expectAbortRejection(settled, /aborted during start/); + expect(transport.startCalls).toBe(1); + expect(transport.sent).toEqual([]); + }); + + test('negotiateEra: a modern reply landing with the abort rejects instead of returning a verdict', async () => { + const abort = new AbortController(); + const transport = new FakeHttpTransport((message, t) => { + modern(message, t); + abort.abort(new Error('aborted with the reply')); + }); + + const negotiated = negotiateEra( + { kind: 'auto', modernVersions: [MODERN], fallbackAvailable: true, probe: { timeoutMs: PROBE_TIMEOUT_MS } }, + { + transport, + clientInfo: { name: 'c', version: '0' }, + capabilities: {}, + environment: 'node', + transportKind: 'http', + defaultTimeoutMs: PROBE_TIMEOUT_MS, + signal: abort.signal + } + ); + + await expect(negotiated).rejects.toThrow(/aborted with the reply/); + expect(transport.onmessage).toBeUndefined(); + }); + + test('an abort landing after the verdict but before the handover leaves the transport reusable', async () => { + // Three microtasks after the reply lands between the probe loop's own + // abort check (which has already returned the verdict) and the window's + // handover of the transport. Fewer hit the loop check, more arrive after + // connect() has succeeded. The failed connect must leave start() + // untouched: a second connect on the same transport starts it again. + const abort = new AbortController(); + const transport = new FakeHttpTransport((message, t) => { + modern(message, t); + queueMicrotask(() => queueMicrotask(() => queueMicrotask(() => abort.abort(new Error('aborted before handover'))))); + }); + const client = autoClient(); + + const settled = await settleWithin(client.connect(transport, { signal: abort.signal })); + + expectAbortRejection(settled, /aborted before handover/); + expect(transport.sent.some(m => isJSONRPCRequest(m) && m.method === 'initialize')).toBe(false); + expect(transport.startCalls).toBe(1); + + await autoClient().connect(transport); + expect(transport.startCalls).toBe(2); + }); + + test('after an aborted connect, the same client connects on a fresh transport', async () => { + const client = autoClient(); + const aborted = await settleWithin(client.connect(new FakeHttpTransport(), { signal: abortSoon() })); + expect(aborted.state).toBe('rejected'); + + await client.connect(new FakeHttpTransport(modern)); + + expect(client.getServerVersion()?.name).toBe('modern-server'); + }); + + describe('already-aborted signal', () => { + test('HTTP-class transport: rejects without starting or sending; the cleanup close fires onclose once', async () => { + const transport = new FakeHttpTransport(modern); + let closes = 0; + transport.onclose = () => closes++; + const abort = new AbortController(); + abort.abort(new Error('aborted before connect')); + + const settled = await settleWithin(autoClient().connect(transport, { signal: abort.signal })); + + expectAbortRejection(settled, /aborted before connect/); + expect(transport.sent).toEqual([]); + expect(transport.startCalls).toBe(0); + expect(closes).toBe(1); + }); + + test('stdio sibling: no sibling is started and nothing is sent; the unstarted session fires no onclose', async () => { + const session = new FakeStdioTransport({ command: 'srv', behavior: 'modern' }); + let closes = 0; + session.onclose = () => closes++; + const abort = new AbortController(); + abort.abort(new Error('aborted before connect')); + + const settled = await settleWithin(autoClient().connect(session, { signal: abort.signal })); + + expectAbortRejection(settled, /aborted before connect/); + for (const instance of FakeStdioTransport.instances) { + expect(instance.startCalls).toBe(0); + expect(instance.sent).toEqual([]); + } + expect(closes).toBe(0); + }); + }); + + describe('stdio sibling probe', () => { + test('an abort mid-probe rejects promptly; the sibling is reaped and the session never starts', async () => { + const session = new FakeStdioTransport({ command: 'srv', behavior: 'silent' }); + + const settled = await settleWithin(autoClient().connect(session, { signal: abortSoon() })); + + expectAbortRejection(settled, /aborted by caller/); + expect(FakeStdioTransport.instances).toHaveLength(2); + const sibling = FakeStdioTransport.instances[1]!; + expect(discoverCount(sibling.sent)).toBe(1); + expect(sibling.disposed).toBe(true); + expect(session.startCalls).toBe(0); + }); + + test('an abort landing during sibling disposal still rejects: the session transport is never started', async () => { + const session = new FakeStdioTransport({ command: 'srv', behavior: 'modern' }); + const abort = new AbortController(); + FakeStdioTransport.onDisposeStart = () => abort.abort(new Error('aborted during disposal')); + + const settled = await settleWithin(autoClient().connect(session, { signal: abort.signal })); + + expectAbortRejection(settled, /aborted during disposal/); + expect(session.startCalls).toBe(0); + }); + }); + + describe('over real HTTP', () => { + let server: Server | undefined; + + afterEach(async () => { + server?.closeAllConnections(); + await new Promise(resolve => (server ? server.close(() => resolve()) : resolve())); + server = undefined; + }); + + test('StreamableHTTPClientTransport against a server that never answers: an abort mid-probe rejects promptly', async () => { + server = createServer(req => { + req.resume(); + }); + await new Promise(resolve => server!.listen(0, '127.0.0.1', resolve)); + const { port } = server.address() as AddressInfo; + const transport = new StreamableHTTPClientTransport(new URL(`http://127.0.0.1:${port}/mcp`)); + + const settled = await settleWithin(autoClient().connect(transport, { signal: abortSoon() })); + await transport.close().catch(() => {}); + + expectAbortRejection(settled, /aborted by caller/); + }); + }); +});