Skip to content

Commit cb23247

Browse files
TheGreatAxiosTheGreatAxios
authored
Merge pull request #1477 from corbitsdev/cl-10758-provider-mcp-residuals
fix(mcp): fail in-flight calls when resume exhaustion redials
2 parents 170480e + 0c6d28d commit cb23247

3 files changed

Lines changed: 300 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: 215 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,215 @@
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+
return undefined;
153+
}
154+
155+
async close(): Promise<void> {
156+
this.onclose?.();
157+
}
158+
159+
async send(message: JSONRPCMessage): Promise<void> {
160+
if (!("method" in message) || !("id" in message)) return;
161+
if (message.method !== "initialize") return;
162+
this.onmessage?.({
163+
jsonrpc: "2.0",
164+
id: message.id,
165+
result: {
166+
protocolVersion: "2025-11-25",
167+
capabilities: { tools: {} },
168+
serverInfo: { name: "hung", version: "0" },
169+
},
170+
});
171+
}
172+
}
173+
174+
describe("resume exhaustion closes in-flight SDK requests", () => {
175+
test("rejects listTools when the SDK reports resume exhaustion", async () => {
176+
const client = new Client({ name: "resume-exhaustion", version: "0" });
177+
const transport = new HungToolsTransport();
178+
const live: LiveHttpTransport = {};
179+
let disconnects = 0;
180+
armResumeExhaustionSignal(
181+
transport,
182+
() => live.current === transport && live.disarmed !== true,
183+
() => {
184+
disconnects += 1;
185+
// Owner redial: MCPClient.close disarms, then Protocol.close.
186+
disarmTransportDeathSignal(live);
187+
void client.close();
188+
},
189+
);
190+
live.current = transport;
191+
await client.connect(transport);
192+
193+
const pending = client.listTools();
194+
transport.onerror?.(new Error(RESUME_EXHAUSTED));
195+
196+
const outcome = await Promise.race([
197+
pending.then(
198+
() => "resolved" as const,
199+
(err: unknown) => err,
200+
),
201+
new Promise<"pending">((resolve) => {
202+
setTimeout(() => resolve("pending"), 500);
203+
}),
204+
]);
205+
206+
expect(disconnects).toBe(1);
207+
expect(outcome).not.toBe("pending");
208+
expect(outcome).not.toBe("resolved");
209+
expect(outcome).toBeInstanceOf(Error);
210+
if (!(outcome instanceof Error)) return;
211+
expect(outcome.message).toBe("MCP error -32000: Connection closed");
212+
expect("code" in outcome && outcome.code).toBe(-32000);
213+
expect(client.transport).toBeUndefined();
214+
});
215+
});

‎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,
@@ -647,7 +705,7 @@ async function finishClient(
647705
authContext?: HTTPAuthContext,
648706
signal?: AbortSignal,
649707
closeLifecycle?: () => void,
650-
live?: { current?: { onclose?: () => void } },
708+
live?: LiveHttpTransport,
651709
): Promise<MCPClient> {
652710
const result = await withHTTPAuthorizationRecovery(authContext, () =>
653711
signal === undefined
@@ -696,7 +754,7 @@ async function finishClient(
696754
},
697755
async close() {
698756
// Disarm first: the owner's hook reports unintentional death only.
699-
if (live?.current !== undefined) live.current.onclose = () => undefined;
757+
if (live !== undefined) disarmTransportDeathSignal(live);
700758
closeLifecycle?.();
701759
authContext?.callback.close();
702760
await closeQuietly(() => client.close());
@@ -749,21 +807,23 @@ async function connectStdio(
749807
const client = new Client({ name: MCP_CLIENT_NAME, version: "1.0.0" });
750808
try {
751809
const transport = new StdioClientTransport(transportOptions);
810+
const live: LiveHttpTransport = { current: transport };
752811
await client.connect(
753812
transport,
754813
options.signal === undefined ? undefined : { signal: options.signal },
755814
);
756815
// Child exit is the only death the owner redials. Chain the protocol
757816
// handler so in-flight requests still reject; HTTP SSE resume stays
758-
// inside the SDK and must not start a second dial.
817+
// inside the SDK and must not start a second dial. Disarm skips the
818+
// owner hook only — prior() is Protocol._onclose and must still run.
759819
if (options.onDisconnect !== undefined) {
760820
const prior = transport.onclose;
761821
transport.onclose = () => {
762822
prior?.();
823+
if (live.disarmed === true) return;
763824
options.onDisconnect?.();
764825
};
765826
}
766-
const live = { current: transport };
767827
return {
768828
ok: true,
769829
client: await finishClient(
@@ -810,7 +870,7 @@ async function connectHttp(
810870
const normalizedURL = normalizeMCPServerURL(config.url);
811871
const url = new URL(normalizedURL);
812872
let authContext: HTTPAuthContext | undefined;
813-
const live: { current?: { onclose?: () => void } } = {};
873+
const live: LiveHttpTransport = {};
814874
const connectedClient = new Client({
815875
name: MCP_CLIENT_NAME,
816876
version: "1.0.0",
@@ -823,6 +883,11 @@ async function connectHttp(
823883
url,
824884
streamableHTTPTransportOptions(undefined, options.signal),
825885
);
886+
armResumeExhaustionSignal(
887+
transport,
888+
() => live.current === transport && live.disarmed !== true,
889+
options.onDisconnect,
890+
);
826891
live.current = transport;
827892
return transport as unknown as Transport;
828893
};
@@ -844,6 +909,11 @@ async function connectHttp(
844909
url,
845910
streamableHTTPTransportOptions(authProvider, lifecycle.signal),
846911
);
912+
armResumeExhaustionSignal(
913+
transport,
914+
() => live.current === transport && live.disarmed !== true,
915+
options.onDisconnect,
916+
);
847917
coordinator.transport = transport;
848918
live.current = transport;
849919
return transport as unknown as Transport;

0 commit comments

Comments
 (0)