From 1edbafd91edd32b4abf5a996391d12f540494bfb Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 05:17:06 +0000 Subject: [PATCH] fix(client): stop listen() rejections escaping as unhandledRejection during an in-flight send listen() suspended on transport.send() before awaiting the opening state-machine promise, so any settle() rejection that fired while the send was still pending (ack timeout, transport close, server cancel, caller abort) had no handler attached and escaped as a process-level unhandledRejection; a send that never settled also left listen() suspended forever. listen() now suspends on 'opening' directly and routes send failures into the state machine. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01DThnF41VH9fXQPjGmRySxa --- .../listen-opening-unhandled-rejection.md | 5 + packages/client/src/client/client.ts | 33 ++++-- packages/client/test/client/listen.test.ts | 109 ++++++++++++++++++ 3 files changed, 137 insertions(+), 10 deletions(-) create mode 100644 .changeset/listen-opening-unhandled-rejection.md diff --git a/.changeset/listen-opening-unhandled-rejection.md b/.changeset/listen-opening-unhandled-rejection.md new file mode 100644 index 0000000000..a951b4d51a --- /dev/null +++ b/.changeset/listen-opening-unhandled-rejection.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +Fix `Client.listen()` rejections escaping as process-level unhandled rejections. The internal `opening` promise could reject (ack timeout, transport close, server cancel, caller abort) while `listen()` was still serially awaiting `transport.send(...)`, so no rejection handler was attached yet — the rejection surfaced as an `unhandledRejection` that caller-side handling cannot prevent, and a send that never settles (e.g. a stdio write parked on `'drain'`) left `listen()` suspended forever even though the ack timer had already fired. `listen()` now suspends on the `opening` state machine directly and routes send failures into it, so every termination path rejects the returned promise and nothing escapes. diff --git a/packages/client/src/client/client.ts b/packages/client/src/client/client.ts index 0b386a63e8..a6665e4e15 100644 --- a/packages/client/src/client/client.ts +++ b/packages/client/src/client/client.ts @@ -2106,18 +2106,31 @@ export class Client extends Protocol { method: 'subscriptions/listen', params: { _meta: { ...this._outboundMetaEnvelope() }, notifications: filter } }; + // The send is NOT serially awaited: `opening` must be the promise + // listen() suspends on so that a settle() firing while the send is + // still in flight (ack timeout, transport close, server cancel, a + // stdio send parked on 'drain') rejects a promise whose handler is + // already attached — otherwise the rejection escapes as a + // process-level unhandledRejection the caller cannot prevent, and a + // send that never settles leaves listen() suspended forever even + // though the ack timer already fired. + const routeSendFailure = (error: unknown): void => { + // Send failure (including a per-request abort fired before + // response headers — `streamableHttp._send` rethrows with onerror + // suppressed). `settle()` is idempotent so a locally-aborted send + // hitting this path after `close()` is a no-op. + settle({ cause: 'remote', error: error instanceof Error ? error : new Error(String(error)) }); + }; try { - await this.transport.send(jsonrpcRequest, { - requestSignal: requestAbort.signal, - onRequestStreamEnd: () => settle({ cause: 'remote', error: new Error('subscriptions/listen: stream ended') }) - }); + this.transport + .send(jsonrpcRequest, { + requestSignal: requestAbort.signal, + onRequestStreamEnd: () => settle({ cause: 'remote', error: new Error('subscriptions/listen: stream ended') }) + }) + .catch(routeSendFailure); } catch (error) { - // Synchronous OR awaited send failure (including a per-request - // abort fired before response headers — `streamableHttp._send` - // rethrows with onerror suppressed). `settle()` is idempotent so - // a locally-aborted send hitting this path after `close()` is a - // no-op. - settle({ cause: 'remote', error: error instanceof Error ? error : new Error(String(error)) }); + // A synchronous throw from send() (before it returns a promise). + routeSendFailure(error); } const honored = await opening; diff --git a/packages/client/test/client/listen.test.ts b/packages/client/test/client/listen.test.ts index 60f96be47a..acc8e33d72 100644 --- a/packages/client/test/client/listen.test.ts +++ b/packages/client/test/client/listen.test.ts @@ -953,6 +953,115 @@ describe('_resetConnectionState() clears connection-scoped debounce timers (fake }); }); +describe('Client.listen() — settle while the send is still in flight (unhandled-rejection regression)', () => { + const sleep = (ms: number) => new Promise(r => setTimeout(r, ms)); + + /** + * Capture process-level unhandledRejection events for the duration of a + * test body. Every settle() path that rejects the `opening` promise while + * `transport.send()` is still pending (a stdio send parked on 'drain', a + * slow HTTP send) used to escape here — caller-side handling cannot + * prevent it because the rejection fires before `await opening` attaches. + */ + async function withUnhandledCapture(run: () => Promise): Promise { + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown): void => { + unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandled); + try { + await run(); + // Let any escaped rejection surface before uninstalling. + await sleep(30); + } finally { + process.off('unhandledRejection', onUnhandled); + } + return unhandled; + } + + /** A scripted modern connection whose `subscriptions/listen` send never settles. */ + async function parkedSendClient() { + const { clientTx, serverTx } = await scriptedModernNoAck(); + const realSend = clientTx.send.bind(clientTx); + clientTx.send = (m, opts) => { + if ((m as { method?: string }).method === 'subscriptions/listen') { + // Models a stdio send parked forever on 'drain' (stdio ignores + // TransportSendOptions.requestSignal) or any send that outlives + // the subscription. + return new Promise(() => {}); + } + return realSend(m, opts); + }; + const client = new Client({ name: 'c', version: '1' }, { versionNegotiation: { mode: 'auto' } }); + await client.connect(clientTx); + return { client, clientTx, serverTx }; + } + + it('ack timeout during a parked send: listen() rejects with RequestTimeout instead of hanging; nothing escapes', async () => { + let outcome: unknown; + const unhandled = await withUnhandledCapture(async () => { + const { client } = await parkedSendClient(); + const settled = client.listen({ toolsListChanged: true }, { timeout: 50 }).then( + () => 'resolved', + e => e + ); + outcome = await Promise.race([settled, sleep(1000).then(() => 'listen() hung')]); + expect((client as unknown as { _listenState: Map })._listenState.size).toBe(0); + }); + expect(outcome).toBeInstanceOf(SdkError); + expect((outcome as SdkError).code).toBe(SdkErrorCode.RequestTimeout); + expect(unhandled).toEqual([]); + }); + + it('transport close during a parked send: listen() rejects instead of hanging; nothing escapes', async () => { + let outcome: unknown; + const unhandled = await withUnhandledCapture(async () => { + const { client, serverTx } = await parkedSendClient(); + const settled = client.listen({ toolsListChanged: true }, { timeout: 60_000 }).then( + () => 'resolved', + e => e + ); + await flush(); + await serverTx.close(); + outcome = await Promise.race([settled, sleep(1000).then(() => 'listen() hung')]); + }); + expect(outcome).toBeInstanceOf(Error); + expect(outcome).not.toBe('listen() hung'); + expect(unhandled).toEqual([]); + }); + + it('caller-signal abort during a parked send: listen() rejects with the signal reason; nothing escapes', async () => { + let outcome: unknown; + const unhandled = await withUnhandledCapture(async () => { + const { client } = await parkedSendClient(); + const ac = new AbortController(); + const settled = client.listen({ toolsListChanged: true }, { timeout: 60_000, signal: ac.signal }).then( + () => 'resolved', + e => e + ); + await flush(); + ac.abort(new Error('caller gave up')); + outcome = await Promise.race([settled, sleep(1000).then(() => 'listen() hung')]); + }); + expect(outcome).toBeInstanceOf(Error); + expect((outcome as Error).message).toContain('caller gave up'); + expect(unhandled).toEqual([]); + }); + + it('an asynchronously-rejected send still rejects listen() with the send failure', async () => { + const unhandled = await withUnhandledCapture(async () => { + const { clientTx } = await scriptedModernNoAck(); + const client = new Client({ name: 'c', version: '1' }, { versionNegotiation: { mode: 'auto' } }); + await client.connect(clientTx); + clientTx.send = () => Promise.reject(new Error('wire dropped')); + const error = await client.listen({ toolsListChanged: true }).catch(e => e as Error); + expect((error as Error).message).toContain('wire dropped'); + expect((client as unknown as { _listenState: Map })._listenState.size).toBe(0); + }); + expect(unhandled).toEqual([]); + }); +}); + describe('Client.listen() — ack timeout (fake timers)', () => { beforeEach(() => vi.useFakeTimers()); afterEach(() => vi.useRealTimers());