Skip to content

Commit cfbc014

Browse files
TheGreatAxiosTheGreatAxios
committed
fix(mcp): fail in-flight calls when resume exhaustion redials
Leave the SDK onclose wrapper in place so Protocol._onclose rejects an in-flight HTTP request with MCP -32000 when resume exhaustion redials.
1 parent 25998b8 commit cfbc014

3 files changed

Lines changed: 298 additions & 12 deletions

File tree

‎docs/MCP.md‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -141,10 +141,13 @@ than hanging or dispatching against a dead transport.
141141

142142
An HTTP SSE drop does not start that second dial. The Streamable HTTP
143143
transport resumes the stream itself (`reconnectionOptions`: 1s start, doubling,
144-
30s cap). Those options have no random factor and no exhaustion callback, so
145-
the delay is not jittered and a stream that stops retrying stays on the
146-
connected row instead of flipping to failed. Tools stay live; only a real
147-
stdio death uses the fail-fast stubs and attempt count above.
144+
30s cap, 1000 attempts). Those options have no random factor, so the delay is
145+
not jittered; the SDK reports the end of that resume loop on the transport's
146+
`onerror` (`Maximum reconnection attempts (1000) exceeded.`), and Corbits
147+
routes exactly that terminal error to the owner's disconnect hook. The row
148+
flips to **reconnecting** with the standard attempt count and fail-fast stubs,
149+
and the owner loop redials with a fresh dial — a stream that stops retrying
150+
no longer sits on the connected row.
148151

149152
Pressing Enter on a failed or reconnecting row retries now with a single dial
150153
instead of waiting for backoff, reusing the saved settings row — it never

‎src/mcp/client-resume-exhaustion.test.ts‎

Lines changed: 213 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,213 @@
1+
import { describe, expect, test } from "bun:test";
2+
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
3+
import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js";
4+
import type { JSONRPCMessage } from "@modelcontextprotocol/sdk/types.js";
5+
import {
6+
armResumeExhaustionSignal,
7+
disarmTransportDeathSignal,
8+
isStreamableResumeExhausted,
9+
type LiveHttpTransport,
10+
} from "./client.js";
11+
12+
describe("isStreamableResumeExhausted", () => {
13+
test.each([
14+
"Maximum reconnection attempts (1000) exceeded.",
15+
"Maximum reconnection attempts (1) exceeded.",
16+
"Maximum reconnection attempts (999999) exceeded.",
17+
])("matches the SDK exhaustion copy (%j)", (message) => {
18+
expect(isStreamableResumeExhausted(new Error(message))).toBe(true);
19+
});
20+
21+
test.each([
22+
["a transient stream error", new Error("Failed to open SSE stream")],
23+
["a reconnect-loop error", new Error("Failed to reconnect SSE stream")],
24+
[
25+
"an exhaustion prefix without the tail",
26+
new Error("Maximum reconnection attempts (1000)"),
27+
],
28+
[
29+
"an exhaustion copy without the count",
30+
new Error("Maximum reconnection attempts exceeded."),
31+
],
32+
["a non-error", "Maximum reconnection attempts (1000) exceeded."],
33+
["nothing", undefined],
34+
])("rejects %s", (_label, value) => {
35+
expect(isStreamableResumeExhausted(value)).toBe(false);
36+
});
37+
});
38+
39+
describe("armResumeExhaustionSignal", () => {
40+
test("leaves the transport untouched without an owner hook", () => {
41+
const transport: Pick<{ onerror?: (error: Error) => void }, "onerror"> = {};
42+
armResumeExhaustionSignal(transport, () => true, undefined);
43+
expect(transport.onerror).toBeUndefined();
44+
});
45+
46+
test("routes exhaustion to the owner hook and keeps the prior handler", () => {
47+
const seen: string[] = [];
48+
let disconnects = 0;
49+
const transport: { onerror?: (error: Error) => void } = {
50+
onerror: (error) => {
51+
seen.push(error.message);
52+
},
53+
};
54+
armResumeExhaustionSignal(
55+
transport,
56+
() => true,
57+
() => {
58+
disconnects += 1;
59+
},
60+
);
61+
transport.onerror?.(
62+
new Error("Maximum reconnection attempts (1000) exceeded."),
63+
);
64+
expect(seen).toEqual(["Maximum reconnection attempts (1000) exceeded."]);
65+
expect(disconnects).toBe(1);
66+
});
67+
68+
test("ignores non-exhaustion transport errors", () => {
69+
let disconnects = 0;
70+
const transport: { onerror?: (error: Error) => void } = {};
71+
armResumeExhaustionSignal(
72+
transport,
73+
() => true,
74+
() => {
75+
disconnects += 1;
76+
},
77+
);
78+
transport.onerror?.(new Error("Failed to open SSE stream"));
79+
expect(disconnects).toBe(0);
80+
});
81+
82+
test("stays silent for a transport superseded by a recovery redial", () => {
83+
let disconnects = 0;
84+
const seen: string[] = [];
85+
const transport: { onerror?: (error: Error) => void } = {
86+
onerror: (error) => {
87+
seen.push(error.message);
88+
},
89+
};
90+
armResumeExhaustionSignal(
91+
transport,
92+
() => false,
93+
() => {
94+
disconnects += 1;
95+
},
96+
);
97+
transport.onerror?.(
98+
new Error("Maximum reconnection attempts (1000) exceeded."),
99+
);
100+
expect(seen).toEqual(["Maximum reconnection attempts (1000) exceeded."]);
101+
expect(disconnects).toBe(0);
102+
});
103+
});
104+
105+
describe("disarmTransportDeathSignal", () => {
106+
test("keeps protocol onclose and stays silent for the owner", () => {
107+
let disconnects = 0;
108+
let protocolCloses = 0;
109+
const live: LiveHttpTransport = {};
110+
const transport: {
111+
onclose?: () => void;
112+
onerror?: (error: Error) => void;
113+
} = {
114+
onclose: () => {
115+
protocolCloses += 1;
116+
},
117+
};
118+
live.current = transport;
119+
armResumeExhaustionSignal(
120+
transport,
121+
() => live.current === transport && live.disarmed !== true,
122+
() => {
123+
disconnects += 1;
124+
},
125+
);
126+
127+
disarmTransportDeathSignal(live);
128+
129+
live.current?.onclose?.();
130+
live.current?.onerror?.(
131+
new Error("Maximum reconnection attempts (1000) exceeded."),
132+
);
133+
expect(disconnects).toBe(0);
134+
expect(protocolCloses).toBe(1);
135+
expect(live.disarmed).toBe(true);
136+
});
137+
138+
test("tolerates an empty handle", () => {
139+
expect(() => disarmTransportDeathSignal({})).not.toThrow();
140+
});
141+
});
142+
143+
const RESUME_EXHAUSTED = "Maximum reconnection attempts (1000) exceeded.";
144+
145+
/** Transport that answers initialize and then never completes a request. */
146+
class HungToolsTransport implements Transport {
147+
onclose?: () => void;
148+
onerror?: (error: Error) => void;
149+
onmessage?: (message: JSONRPCMessage) => void;
150+
151+
async start(): Promise<void> {}
152+
153+
async close(): Promise<void> {
154+
this.onclose?.();
155+
}
156+
157+
async send(message: JSONRPCMessage): Promise<void> {
158+
if (!("method" in message) || !("id" in message)) return;
159+
if (message.method !== "initialize") return;
160+
this.onmessage?.({
161+
jsonrpc: "2.0",
162+
id: message.id,
163+
result: {
164+
protocolVersion: "2025-11-25",
165+
capabilities: { tools: {} },
166+
serverInfo: { name: "hung", version: "0" },
167+
},
168+
});
169+
}
170+
}
171+
172+
describe("resume exhaustion closes in-flight SDK requests", () => {
173+
test("rejects listTools when the SDK reports resume exhaustion", async () => {
174+
const client = new Client({ name: "resume-exhaustion", version: "0" });
175+
const transport = new HungToolsTransport();
176+
const live: LiveHttpTransport = {};
177+
let disconnects = 0;
178+
armResumeExhaustionSignal(
179+
transport,
180+
() => live.current === transport && live.disarmed !== true,
181+
() => {
182+
disconnects += 1;
183+
// Owner redial: MCPClient.close disarms, then Protocol.close.
184+
disarmTransportDeathSignal(live);
185+
void client.close();
186+
},
187+
);
188+
live.current = transport;
189+
await client.connect(transport);
190+
191+
const pending = client.listTools();
192+
transport.onerror?.(new Error(RESUME_EXHAUSTED));
193+
194+
const outcome = await Promise.race([
195+
pending.then(
196+
() => "resolved" as const,
197+
(err: unknown) => err,
198+
),
199+
new Promise<"pending">((resolve) => {
200+
setTimeout(() => resolve("pending"), 500);
201+
}),
202+
]);
203+
204+
expect(disconnects).toBe(1);
205+
expect(outcome).not.toBe("pending");
206+
expect(outcome).not.toBe("resolved");
207+
expect(outcome).toBeInstanceOf(Error);
208+
if (!(outcome instanceof Error)) return;
209+
expect(outcome.message).toBe("MCP error -32000: Connection closed");
210+
expect("code" in outcome && outcome.code).toBe(-32000);
211+
expect(client.transport).toBeUndefined();
212+
});
213+
});

‎src/mcp/client.ts‎

Lines changed: 78 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -87,9 +87,11 @@ export interface MCPConnectOptions {
8787
*/
8888
onAuthorized?: (serverName: string) => void;
8989
signal?: AbortSignal;
90-
// Wired to transport.onclose after connect so the owner learns the
91-
// transport died under a live client; close() disarms it before
92-
// intentional teardown.
90+
// Wired to transport.onclose after connect, plus HTTP SSE-resume
91+
// exhaustion via transport.onerror, so the owner learns the transport
92+
// died under a live client. close() marks the live handle disarmed
93+
// before intentional teardown; it must not replace onclose, or
94+
// Protocol._onclose never rejects in-flight calls.
9395
onDisconnect?: () => void;
9496
// Session OS-sandbox handle: a model-reachable stdio server spawn runs
9597
// wrapped; an unwrappable command fails the connect (fail closed). Omitted
@@ -406,6 +408,62 @@ export const MCP_HTTP_RECONNECTION_OPTIONS = {
406408
maxRetries: 1_000,
407409
} as const;
408410

411+
/** Live HTTP transport handle shared with finishClient so intentional
412+
* teardown can suppress the owner death signal without replacing
413+
* `onclose` (Protocol._onclose rejects in-flight calls from that hook). */
414+
export interface LiveHttpTransport {
415+
current?: {
416+
onclose?: () => void;
417+
onerror?: (error: Error) => void;
418+
};
419+
/** Intentional teardown. Owner hooks must consult this and stay quiet. */
420+
disarmed?: boolean;
421+
}
422+
423+
/**
424+
* SDK SSE resume reports a dead stream on `transport.onerror`, never
425+
* `onclose`: once the reconnection cap trips it emits
426+
* `Maximum reconnection attempts (N) exceeded.` and stops. Without a hook
427+
* the row would stay on the connected row forever, so exactly that terminal
428+
* error routes to the owner's disconnect hook. The copy is pinned to the
429+
* ^1.32.1 SDK range in package.json.
430+
*/
431+
const RESUME_EXHAUSTED_PATTERN =
432+
/^Maximum reconnection attempts \(\d+\) exceeded\.$/;
433+
434+
export function isStreamableResumeExhausted(error: unknown): boolean {
435+
return error instanceof Error && RESUME_EXHAUSTED_PATTERN.test(error.message);
436+
}
437+
438+
/**
439+
* Chain SSE-resume exhaustion onto the owner's disconnect hook so the row
440+
* flips to reconnecting and the owner loop redials with a fresh dial.
441+
* Stale transports (superseded by a recovery redial) stay silent; every
442+
* other transport error passes through to the prior handler untouched.
443+
*/
444+
export function armResumeExhaustionSignal(
445+
transport: Pick<StreamableHTTPClientTransport, "onerror">,
446+
isLive: () => boolean,
447+
onDisconnect?: () => void,
448+
): void {
449+
if (onDisconnect === undefined) return;
450+
const prior = transport.onerror;
451+
transport.onerror = (error: Error) => {
452+
prior?.(error);
453+
if (isLive() && isStreamableResumeExhausted(error)) onDisconnect();
454+
};
455+
}
456+
457+
/** Suppress the owner death signal for intentional teardown.
458+
* Do not replace `onclose`: Protocol.close only calls transport.close(),
459+
* and the SDK rejects in-flight requests from the onclose wrapper
460+
* installed at connect (`_onclose`, MCP -32000 Connection closed). A noop
461+
* there leaves the HTTP RPC pending until the request timeout.
462+
*/
463+
export function disarmTransportDeathSignal(live: LiveHttpTransport): void {
464+
live.disarmed = true;
465+
}
466+
409467
export function streamableHTTPTransportOptions(
410468
authProvider: CorbitsOAuthProvider | undefined,
411469
signal: AbortSignal | undefined,
@@ -644,7 +702,7 @@ async function finishClient(
644702
authContext?: HTTPAuthContext,
645703
signal?: AbortSignal,
646704
closeLifecycle?: () => void,
647-
live?: { current?: { onclose?: () => void } },
705+
live?: LiveHttpTransport,
648706
): Promise<MCPClient> {
649707
const result = await withHTTPAuthorizationRecovery(authContext, () =>
650708
signal === undefined
@@ -693,7 +751,7 @@ async function finishClient(
693751
},
694752
async close() {
695753
// Disarm first: the owner's hook reports unintentional death only.
696-
if (live?.current !== undefined) live.current.onclose = () => undefined;
754+
if (live !== undefined) disarmTransportDeathSignal(live);
697755
closeLifecycle?.();
698756
authContext?.callback.close();
699757
await closeQuietly(() => client.close());
@@ -746,21 +804,23 @@ async function connectStdio(
746804
const client = new Client({ name: MCP_CLIENT_NAME, version: "1.0.0" });
747805
try {
748806
const transport = new StdioClientTransport(transportOptions);
807+
const live: LiveHttpTransport = { current: transport };
749808
await client.connect(
750809
transport,
751810
options.signal === undefined ? undefined : { signal: options.signal },
752811
);
753812
// Child exit is the only death the owner redials. Chain the protocol
754813
// handler so in-flight requests still reject; HTTP SSE resume stays
755-
// inside the SDK and must not start a second dial.
814+
// inside the SDK and must not start a second dial. Disarm skips the
815+
// owner hook only — prior() is Protocol._onclose and must still run.
756816
if (options.onDisconnect !== undefined) {
757817
const prior = transport.onclose;
758818
transport.onclose = () => {
759819
prior?.();
820+
if (live.disarmed === true) return;
760821
options.onDisconnect?.();
761822
};
762823
}
763-
const live = { current: transport };
764824
return {
765825
ok: true,
766826
client: await finishClient(
@@ -807,7 +867,7 @@ async function connectHttp(
807867
const normalizedURL = normalizeMCPServerURL(config.url);
808868
const url = new URL(normalizedURL);
809869
let authContext: HTTPAuthContext | undefined;
810-
const live: { current?: { onclose?: () => void } } = {};
870+
const live: LiveHttpTransport = {};
811871
const connectedClient = new Client({
812872
name: MCP_CLIENT_NAME,
813873
version: "1.0.0",
@@ -820,6 +880,11 @@ async function connectHttp(
820880
url,
821881
streamableHTTPTransportOptions(undefined, options.signal),
822882
);
883+
armResumeExhaustionSignal(
884+
transport,
885+
() => live.current === transport && live.disarmed !== true,
886+
options.onDisconnect,
887+
);
823888
live.current = transport;
824889
return transport as unknown as Transport;
825890
};
@@ -841,6 +906,11 @@ async function connectHttp(
841906
url,
842907
streamableHTTPTransportOptions(authProvider, lifecycle.signal),
843908
);
909+
armResumeExhaustionSignal(
910+
transport,
911+
() => live.current === transport && live.disarmed !== true,
912+
options.onDisconnect,
913+
);
844914
coordinator.transport = transport;
845915
live.current = transport;
846916
return transport as unknown as Transport;

0 commit comments

Comments
 (0)