From cfbc014ec195c4632168b1c23baf13166cad1253 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 16 Oct 2026 18:37:42 -0700 Subject: [PATCH 1/2] 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. --- docs/MCP.md | 11 +- src/mcp/client-resume-exhaustion.test.ts | 213 +++++++++++++++++++++++ src/mcp/client.ts | 86 ++++++++- 3 files changed, 298 insertions(+), 12 deletions(-) create mode 100644 src/mcp/client-resume-exhaustion.test.ts diff --git a/docs/MCP.md b/docs/MCP.md index 14ea76c01..11e3813de 100644 --- a/docs/MCP.md +++ b/docs/MCP.md @@ -141,10 +141,13 @@ than hanging or dispatching against a dead transport. An HTTP SSE drop does not start that second dial. The Streamable HTTP transport resumes the stream itself (`reconnectionOptions`: 1s start, doubling, -30s cap). Those options have no random factor and no exhaustion callback, so -the delay is not jittered and a stream that stops retrying stays on the -connected row instead of flipping to failed. Tools stay live; only a real -stdio death uses the fail-fast stubs and attempt count above. +30s cap, 1000 attempts). Those options have no random factor, so the delay is +not jittered; the SDK reports the end of that resume loop on the transport's +`onerror` (`Maximum reconnection attempts (1000) exceeded.`), and Corbits +routes exactly that terminal error to the owner's disconnect hook. The row +flips to **reconnecting** with the standard attempt count and fail-fast stubs, +and the owner loop redials with a fresh dial — a stream that stops retrying +no longer sits on the connected row. Pressing Enter on a failed or reconnecting row retries now with a single dial instead of waiting for backoff, reusing the saved settings row — it never diff --git a/src/mcp/client-resume-exhaustion.test.ts b/src/mcp/client-resume-exhaustion.test.ts new file mode 100644 index 000000000..1c9ccc47d --- /dev/null +++ b/src/mcp/client-resume-exhaustion.test.ts @@ -0,0 +1,213 @@ +import { describe, expect, test } from "bun:test"; +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; +import type { JSONRPCMessage } from "@modelcontextprotocol/sdk/types.js"; +import { + armResumeExhaustionSignal, + disarmTransportDeathSignal, + isStreamableResumeExhausted, + type LiveHttpTransport, +} from "./client.js"; + +describe("isStreamableResumeExhausted", () => { + test.each([ + "Maximum reconnection attempts (1000) exceeded.", + "Maximum reconnection attempts (1) exceeded.", + "Maximum reconnection attempts (999999) exceeded.", + ])("matches the SDK exhaustion copy (%j)", (message) => { + expect(isStreamableResumeExhausted(new Error(message))).toBe(true); + }); + + test.each([ + ["a transient stream error", new Error("Failed to open SSE stream")], + ["a reconnect-loop error", new Error("Failed to reconnect SSE stream")], + [ + "an exhaustion prefix without the tail", + new Error("Maximum reconnection attempts (1000)"), + ], + [ + "an exhaustion copy without the count", + new Error("Maximum reconnection attempts exceeded."), + ], + ["a non-error", "Maximum reconnection attempts (1000) exceeded."], + ["nothing", undefined], + ])("rejects %s", (_label, value) => { + expect(isStreamableResumeExhausted(value)).toBe(false); + }); +}); + +describe("armResumeExhaustionSignal", () => { + test("leaves the transport untouched without an owner hook", () => { + const transport: Pick<{ onerror?: (error: Error) => void }, "onerror"> = {}; + armResumeExhaustionSignal(transport, () => true, undefined); + expect(transport.onerror).toBeUndefined(); + }); + + test("routes exhaustion to the owner hook and keeps the prior handler", () => { + const seen: string[] = []; + let disconnects = 0; + const transport: { onerror?: (error: Error) => void } = { + onerror: (error) => { + seen.push(error.message); + }, + }; + armResumeExhaustionSignal( + transport, + () => true, + () => { + disconnects += 1; + }, + ); + transport.onerror?.( + new Error("Maximum reconnection attempts (1000) exceeded."), + ); + expect(seen).toEqual(["Maximum reconnection attempts (1000) exceeded."]); + expect(disconnects).toBe(1); + }); + + test("ignores non-exhaustion transport errors", () => { + let disconnects = 0; + const transport: { onerror?: (error: Error) => void } = {}; + armResumeExhaustionSignal( + transport, + () => true, + () => { + disconnects += 1; + }, + ); + transport.onerror?.(new Error("Failed to open SSE stream")); + expect(disconnects).toBe(0); + }); + + test("stays silent for a transport superseded by a recovery redial", () => { + let disconnects = 0; + const seen: string[] = []; + const transport: { onerror?: (error: Error) => void } = { + onerror: (error) => { + seen.push(error.message); + }, + }; + armResumeExhaustionSignal( + transport, + () => false, + () => { + disconnects += 1; + }, + ); + transport.onerror?.( + new Error("Maximum reconnection attempts (1000) exceeded."), + ); + expect(seen).toEqual(["Maximum reconnection attempts (1000) exceeded."]); + expect(disconnects).toBe(0); + }); +}); + +describe("disarmTransportDeathSignal", () => { + test("keeps protocol onclose and stays silent for the owner", () => { + let disconnects = 0; + let protocolCloses = 0; + const live: LiveHttpTransport = {}; + const transport: { + onclose?: () => void; + onerror?: (error: Error) => void; + } = { + onclose: () => { + protocolCloses += 1; + }, + }; + live.current = transport; + armResumeExhaustionSignal( + transport, + () => live.current === transport && live.disarmed !== true, + () => { + disconnects += 1; + }, + ); + + disarmTransportDeathSignal(live); + + live.current?.onclose?.(); + live.current?.onerror?.( + new Error("Maximum reconnection attempts (1000) exceeded."), + ); + expect(disconnects).toBe(0); + expect(protocolCloses).toBe(1); + expect(live.disarmed).toBe(true); + }); + + test("tolerates an empty handle", () => { + expect(() => disarmTransportDeathSignal({})).not.toThrow(); + }); +}); + +const RESUME_EXHAUSTED = "Maximum reconnection attempts (1000) exceeded."; + +/** Transport that answers initialize and then never completes a request. */ +class HungToolsTransport implements Transport { + onclose?: () => void; + onerror?: (error: Error) => void; + onmessage?: (message: JSONRPCMessage) => void; + + async start(): Promise {} + + async close(): Promise { + this.onclose?.(); + } + + async send(message: JSONRPCMessage): Promise { + if (!("method" in message) || !("id" in message)) return; + if (message.method !== "initialize") return; + this.onmessage?.({ + jsonrpc: "2.0", + id: message.id, + result: { + protocolVersion: "2025-11-25", + capabilities: { tools: {} }, + serverInfo: { name: "hung", version: "0" }, + }, + }); + } +} + +describe("resume exhaustion closes in-flight SDK requests", () => { + test("rejects listTools when the SDK reports resume exhaustion", async () => { + const client = new Client({ name: "resume-exhaustion", version: "0" }); + const transport = new HungToolsTransport(); + const live: LiveHttpTransport = {}; + let disconnects = 0; + armResumeExhaustionSignal( + transport, + () => live.current === transport && live.disarmed !== true, + () => { + disconnects += 1; + // Owner redial: MCPClient.close disarms, then Protocol.close. + disarmTransportDeathSignal(live); + void client.close(); + }, + ); + live.current = transport; + await client.connect(transport); + + const pending = client.listTools(); + transport.onerror?.(new Error(RESUME_EXHAUSTED)); + + const outcome = await Promise.race([ + pending.then( + () => "resolved" as const, + (err: unknown) => err, + ), + new Promise<"pending">((resolve) => { + setTimeout(() => resolve("pending"), 500); + }), + ]); + + expect(disconnects).toBe(1); + expect(outcome).not.toBe("pending"); + expect(outcome).not.toBe("resolved"); + expect(outcome).toBeInstanceOf(Error); + if (!(outcome instanceof Error)) return; + expect(outcome.message).toBe("MCP error -32000: Connection closed"); + expect("code" in outcome && outcome.code).toBe(-32000); + expect(client.transport).toBeUndefined(); + }); +}); diff --git a/src/mcp/client.ts b/src/mcp/client.ts index ef2fca151..b87723787 100644 --- a/src/mcp/client.ts +++ b/src/mcp/client.ts @@ -87,9 +87,11 @@ export interface MCPConnectOptions { */ onAuthorized?: (serverName: string) => void; signal?: AbortSignal; - // Wired to transport.onclose after connect so the owner learns the - // transport died under a live client; close() disarms it before - // intentional teardown. + // Wired to transport.onclose after connect, plus HTTP SSE-resume + // exhaustion via transport.onerror, so the owner learns the transport + // died under a live client. close() marks the live handle disarmed + // before intentional teardown; it must not replace onclose, or + // Protocol._onclose never rejects in-flight calls. onDisconnect?: () => void; // Session OS-sandbox handle: a model-reachable stdio server spawn runs // wrapped; an unwrappable command fails the connect (fail closed). Omitted @@ -406,6 +408,62 @@ export const MCP_HTTP_RECONNECTION_OPTIONS = { maxRetries: 1_000, } as const; +/** Live HTTP transport handle shared with finishClient so intentional + * teardown can suppress the owner death signal without replacing + * `onclose` (Protocol._onclose rejects in-flight calls from that hook). */ +export interface LiveHttpTransport { + current?: { + onclose?: () => void; + onerror?: (error: Error) => void; + }; + /** Intentional teardown. Owner hooks must consult this and stay quiet. */ + disarmed?: boolean; +} + +/** + * SDK SSE resume reports a dead stream on `transport.onerror`, never + * `onclose`: once the reconnection cap trips it emits + * `Maximum reconnection attempts (N) exceeded.` and stops. Without a hook + * the row would stay on the connected row forever, so exactly that terminal + * error routes to the owner's disconnect hook. The copy is pinned to the + * ^1.32.1 SDK range in package.json. + */ +const RESUME_EXHAUSTED_PATTERN = + /^Maximum reconnection attempts \(\d+\) exceeded\.$/; + +export function isStreamableResumeExhausted(error: unknown): boolean { + return error instanceof Error && RESUME_EXHAUSTED_PATTERN.test(error.message); +} + +/** + * Chain SSE-resume exhaustion onto the owner's disconnect hook so the row + * flips to reconnecting and the owner loop redials with a fresh dial. + * Stale transports (superseded by a recovery redial) stay silent; every + * other transport error passes through to the prior handler untouched. + */ +export function armResumeExhaustionSignal( + transport: Pick, + isLive: () => boolean, + onDisconnect?: () => void, +): void { + if (onDisconnect === undefined) return; + const prior = transport.onerror; + transport.onerror = (error: Error) => { + prior?.(error); + if (isLive() && isStreamableResumeExhausted(error)) onDisconnect(); + }; +} + +/** Suppress the owner death signal for intentional teardown. + * Do not replace `onclose`: Protocol.close only calls transport.close(), + * and the SDK rejects in-flight requests from the onclose wrapper + * installed at connect (`_onclose`, MCP -32000 Connection closed). A noop + * there leaves the HTTP RPC pending until the request timeout. + */ +export function disarmTransportDeathSignal(live: LiveHttpTransport): void { + live.disarmed = true; +} + export function streamableHTTPTransportOptions( authProvider: CorbitsOAuthProvider | undefined, signal: AbortSignal | undefined, @@ -644,7 +702,7 @@ async function finishClient( authContext?: HTTPAuthContext, signal?: AbortSignal, closeLifecycle?: () => void, - live?: { current?: { onclose?: () => void } }, + live?: LiveHttpTransport, ): Promise { const result = await withHTTPAuthorizationRecovery(authContext, () => signal === undefined @@ -693,7 +751,7 @@ async function finishClient( }, async close() { // Disarm first: the owner's hook reports unintentional death only. - if (live?.current !== undefined) live.current.onclose = () => undefined; + if (live !== undefined) disarmTransportDeathSignal(live); closeLifecycle?.(); authContext?.callback.close(); await closeQuietly(() => client.close()); @@ -746,21 +804,23 @@ async function connectStdio( const client = new Client({ name: MCP_CLIENT_NAME, version: "1.0.0" }); try { const transport = new StdioClientTransport(transportOptions); + const live: LiveHttpTransport = { current: transport }; await client.connect( transport, options.signal === undefined ? undefined : { signal: options.signal }, ); // Child exit is the only death the owner redials. Chain the protocol // handler so in-flight requests still reject; HTTP SSE resume stays - // inside the SDK and must not start a second dial. + // inside the SDK and must not start a second dial. Disarm skips the + // owner hook only — prior() is Protocol._onclose and must still run. if (options.onDisconnect !== undefined) { const prior = transport.onclose; transport.onclose = () => { prior?.(); + if (live.disarmed === true) return; options.onDisconnect?.(); }; } - const live = { current: transport }; return { ok: true, client: await finishClient( @@ -807,7 +867,7 @@ async function connectHttp( const normalizedURL = normalizeMCPServerURL(config.url); const url = new URL(normalizedURL); let authContext: HTTPAuthContext | undefined; - const live: { current?: { onclose?: () => void } } = {}; + const live: LiveHttpTransport = {}; const connectedClient = new Client({ name: MCP_CLIENT_NAME, version: "1.0.0", @@ -820,6 +880,11 @@ async function connectHttp( url, streamableHTTPTransportOptions(undefined, options.signal), ); + armResumeExhaustionSignal( + transport, + () => live.current === transport && live.disarmed !== true, + options.onDisconnect, + ); live.current = transport; return transport as unknown as Transport; }; @@ -841,6 +906,11 @@ async function connectHttp( url, streamableHTTPTransportOptions(authProvider, lifecycle.signal), ); + armResumeExhaustionSignal( + transport, + () => live.current === transport && live.disarmed !== true, + options.onDisconnect, + ); coordinator.transport = transport; live.current = transport; return transport as unknown as Transport; From 0c6d28d816a720fb025698337b96ad25009f56e4 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 16 Oct 2026 18:44:23 -0700 Subject: [PATCH 2/2] test(mcp): satisfy oxlint on the resume transport --- src/mcp/client-resume-exhaustion.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/mcp/client-resume-exhaustion.test.ts b/src/mcp/client-resume-exhaustion.test.ts index 1c9ccc47d..c16543e63 100644 --- a/src/mcp/client-resume-exhaustion.test.ts +++ b/src/mcp/client-resume-exhaustion.test.ts @@ -148,7 +148,9 @@ class HungToolsTransport implements Transport { onerror?: (error: Error) => void; onmessage?: (message: JSONRPCMessage) => void; - async start(): Promise {} + async start(): Promise { + return undefined; + } async close(): Promise { this.onclose?.();