Skip to content

fix(client): stop listen() rejections escaping as unhandledRejection during an in-flight send #2642

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
claude wants to merge 1 commit into
base: main
Choose a base branch
from
+137 −10
Open

fix(client): stop listen() rejections escaping as unhandledRejection during an in-flight send #2642

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/listen-opening-unhandled-rejection.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
33 changes: 23 additions & 10 deletions packages/client/src/client/client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2106,21 +2106,34 @@
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;

Check notice on line 2136 in packages/client/src/client/client.ts

View check run for this annotation

Claude / Claude Code Review

McpSubscription.close() still hangs forever on a parked send (sibling path of the fixed hang) 

Pre-existing issue (unchanged lines, same state machine): `McpSubscription.close()` retains the same hang class this PR fixes for `listen()` — it serially `await wireTeardown()`, which ends in an un-abortable `transport.send` of `notifications/cancelled`, so under the same parked-send condition the new tests model (stdio write parked on 'drain'), `await sub.close()` suspends forever. Consider mirroring the fire-and-forget pattern the ack-timeout and caller-abort paths already use, or bounding th
Comment on lines +2109 to 2136

@claude claude Bot Aug 11, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟣 Pre-existing issue (unchanged lines, same state machine): McpSubscription.close() retains the same hang class this PR fixes for listen() — it serially await wireTeardown(), which ends in an un-abortable transport.send of notifications/cancelled, so under the same parked-send condition the new tests model (stdio write parked on 'drain'), await sub.close() suspends forever. Consider mirroring the fire-and-forget pattern the ack-timeout and caller-abort paths already use, or bounding the await — here or as a follow-up alongside #2552.

Extended reasoning...

The sibling path: close() still awaits a send that may never settle

This PR fixes listen() so it suspends on the opening state machine instead of the send promise — a send parked forever no longer hangs the caller. But the McpSubscription handle that listen() returns keeps the same hang class in its close() path, inside the very closure this PR edits:

const wireTeardown = async (): Promise<void> => {
    requestAbort.abort();
    await this.notification({ method: 'notifications/cancelled', params: { requestId: listenId } }).catch(() => {});
};

const close = async (): Promise<void> => {
    if (state === 'closed') return;
    settle({ cause: 'local' });
    await wireTeardown();   // <-- serial await on an un-abortable send
};

Protocol.notification() → _notificationViaCodec ends in await this._transport.send(jsonrpcNotification, options) (the debounce fast-path doesn't apply — notifications/cancelled carries params and debouncedNotificationMethods defaults to []). The .catch(() => {}) only guards rejection; it does nothing for a promise that never settles.

Concrete walk-through under the PR's own trigger scenario

  1. Stdio child stops reading stdin; the write buffer fills; write() returns false; StdioClientTransport.send() parks on 'drain' indefinitely (stdio ignores TransportSendOptions.requestSignal, per this PR's own comments and fix(client): reject stdio send() when the write fails instead of waiting for 'drain' #2552).
  2. Every subsequent write parks too — including wireTeardown's notifications/cancelled write.
  3. Caller runs await sub.close(). settle({ cause: 'local' }) completes all observable local teardown (clears _listenState, ack timer, resolves sub.closed with 'local').
  4. await wireTeardown() then suspends on the parked send forever. The promise close() returned never resolves, hanging the caller's cleanup/shutdown code exactly the way listen() used to hang.

Why nothing existing covers it

  • The other two locally-initiated teardown paths — the ack-timeout branch and the caller-signal-abort branch — deliberately use fire-and-forget void wireTeardown().catch(() => {}). Only the user-facing close() serially awaits, and since failures are swallowed anyway, the await conveys no error signal to the caller.
  • The named follow-up (fix(client): reject stdio send() when the write fails instead of waiting for 'drain' #2552, making stdio honor requestSignal) would not fix this: wireTeardown sends the cancel notification with no signal, after requestAbort has already been aborted.
  • This matches the repo's Completeness recurring catch: a PR fixing a failure mode shouldn't leave a sibling path in the same function/state machine with the very hang it eliminates.

Suggested fix

Mirror the fire-and-forget pattern (settle(...) already completes all observable local teardown before the await), or bound the await with a timeout race if there's value in usually confirming the cancel reached the wire before close() resolves. One existing test ("close() sends notifications/cancelled referencing the listen id") relies on the cancel being on the wire when close() resolves, so a bounded race may be the gentler shape.

Why pre-existing severity

The close()/wireTeardown lines are byte-for-byte untouched by this diff; the hang predates the PR and isn't widened by it. It's worth a follow-up (or an opportunistic fix here alongside #2552) rather than a blocker.

return { honoredFilter: honored, close, closed };
}

Expand Down
109 changes: 109 additions & 0 deletions packages/client/test/client/listen.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<void>): Promise<unknown[]> {
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<void>(() => {});
}
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<unknown, unknown> })._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<unknown, unknown> })._listenState.size).toBe(0);
});
expect(unhandled).toEqual([]);
});
});

describe('Client.listen() — ack timeout (fake timers)', () => {
beforeEach(() => vi.useFakeTimers());
afterEach(() => vi.useRealTimers());
Expand Down
Loading