diff --git a/src/client/streamableHttp.test.ts b/src/client/streamableHttp.test.ts index 8d9a6a9c..a73927ac 100644 --- a/src/client/streamableHttp.test.ts +++ b/src/client/streamableHttp.test.ts @@ -1,4 +1,4 @@ -import { StreamableHTTPClientTransport } from "./streamableHttp.js"; +import { StreamableHTTPClientTransport, StreamableHTTPReconnectionOptions } from "./streamableHttp.js"; import { JSONRPCMessage } from "../types.js"; @@ -275,59 +275,62 @@ describe("StreamableHTTPClientTransport", () => { })).toBe(true); }); - it("should handle reconnection with last-event-id for resumability", async () => { - // Set up a stream that will send an event with ID then error - const encoder = new TextEncoder(); - const stream = new ReadableStream({ - start(controller) { - const event = "id: event-123\nevent: message\ndata: {\"jsonrpc\": \"2.0\", \"method\": \"serverNotification\", \"params\": {}}\n\n"; - controller.enqueue(encoder.encode(event)); - // Simulate network error - setTimeout(() => { - controller.error(new Error("Network error")); - }, 10); + it("should support custom reconnection options", () => { + // Create a transport with custom reconnection options + transport = new StreamableHTTPClientTransport(new URL("http://localhost:1234/mcp"), { + reconnectionOptions: { + initialReconnectionDelay: 500, + maxReconnectionDelay: 10000, + reconnectionDelayGrowFactor: 2, + maxRetries: 5, } }); - // Mock the initial connection - (global.fetch as jest.Mock).mockResolvedValueOnce({ + // Verify options were set correctly (checking implementation details) + // Access private properties for testing + const transportInstance = transport as unknown as { + _defaultReconnectionOptions: StreamableHTTPReconnectionOptions; + }; + expect(transportInstance._defaultReconnectionOptions.initialReconnectionDelay).toBe(1000); + expect(transportInstance._defaultReconnectionOptions.maxRetries).toBe(2); + }); + + it("should pass lastEventId when reconnecting", async () => { + // Create a fresh transport + transport = new StreamableHTTPClientTransport(new URL("http://localhost:1234/mcp")); + + // Mock fetch to verify headers sent + const fetchSpy = global.fetch as jest.Mock; + fetchSpy.mockReset(); + fetchSpy.mockResolvedValue({ ok: true, status: 200, headers: new Headers({ "content-type": "text/event-stream" }), - body: stream + body: new ReadableStream() }); - const errorSpy = jest.fn(); - transport.onerror = errorSpy; - + // Call the reconnect method directly with a lastEventId await transport.start(); - await transport["_startOrAuthStandaloneSSE"](); - - // Let the stream process and error - await new Promise(resolve => setTimeout(resolve, 50)); - - // Verify error was caught - expect(errorSpy).toHaveBeenCalled(); - - // Mock the reconnection (the transport should try to reconnect after error) - (global.fetch as jest.Mock).mockResolvedValueOnce({ - ok: true, - status: 200, - headers: new Headers({ "content-type": "text/event-stream" }), - body: null - }); - - // Allow time for automatic reconnection - await new Promise(resolve => setTimeout(resolve, 1100)); // > 1 second delay - - // Verify that the client attempted to reconnect with the last-event-id - const calls = (global.fetch as jest.Mock).mock.calls; - const lastCall = calls[calls.length - 1]; - expect(lastCall[1].method).toBe("GET"); - expect(lastCall[1].headers.get("last-event-id")).toBe("event-123"); + // Type assertion to access private method + const transportWithPrivateMethods = transport as unknown as { + _startOrAuthStandaloneSSE: (lastEventId?: string) => Promise + }; + await transportWithPrivateMethods._startOrAuthStandaloneSSE("test-event-id"); + + // Verify fetch was called with the lastEventId header + expect(fetchSpy).toHaveBeenCalled(); + const fetchCall = fetchSpy.mock.calls[0]; + const headers = fetchCall[1].headers; + expect(headers.get("last-event-id")).toBe("test-event-id"); }); it("should throw error when invalid content-type is received", async () => { + // Clear any previous state from other tests + jest.clearAllMocks(); + + // Create a fresh transport instance + transport = new StreamableHTTPClientTransport(new URL("http://localhost:1234/mcp")); + const message: JSONRPCMessage = { jsonrpc: "2.0", method: "test", @@ -337,7 +340,7 @@ describe("StreamableHTTPClientTransport", () => { const stream = new ReadableStream({ start(controller) { - controller.enqueue("invalid text response"); + controller.enqueue(new TextEncoder().encode("invalid text response")); controller.close(); } }); @@ -389,4 +392,38 @@ describe("StreamableHTTPClientTransport", () => { expect(global.fetch).toHaveBeenCalledTimes(2); }); + + + it("should have exponential backoff with configurable maxRetries", () => { + // This test verifies the maxRetries and backoff calculation directly + + // Create transport with specific options for testing + transport = new StreamableHTTPClientTransport(new URL("http://localhost:1234/mcp"), { + reconnectionOptions: { + initialReconnectionDelay: 100, + maxReconnectionDelay: 5000, + reconnectionDelayGrowFactor: 2, + maxRetries: 3, + } + }); + + // Get access to the internal implementation + const getDelay = transport["_getNextReconnectionDelay"].bind(transport); + + // First retry - should use initial delay + expect(getDelay(0)).toBe(100); + + // Second retry - should double (2^1 * 100 = 200) + expect(getDelay(1)).toBe(200); + + // Third retry - should double again (2^2 * 100 = 400) + expect(getDelay(2)).toBe(400); + + // Fourth retry - should double again (2^3 * 100 = 800) + expect(getDelay(3)).toBe(800); + + // Tenth retry - should be capped at maxReconnectionDelay + expect(getDelay(10)).toBe(5000); + }); + }); diff --git a/src/client/streamableHttp.ts b/src/client/streamableHttp.ts index 9b3c7c20..9451e8d9 100644 --- a/src/client/streamableHttp.ts +++ b/src/client/streamableHttp.ts @@ -1,3 +1,4 @@ +import { log } from "node:console"; import { Transport } from "../shared/transport.js"; import { isJSONRPCNotification, JSONRPCMessage, JSONRPCMessageSchema } from "../types.js"; import { auth, AuthResult, OAuthClientProvider, UnauthorizedError } from "./auth.js"; @@ -12,6 +13,35 @@ export class StreamableHTTPError extends Error { } } +/** + * Configuration options for reconnection behavior of the StreamableHTTPClientTransport. + */ +export interface StreamableHTTPReconnectionOptions { + /** + * Maximum backoff time between reconnection attempts in milliseconds. + * Default is 30000 (30 seconds). + */ + maxReconnectionDelay: number; + + /** + * Initial backoff time between reconnection attempts in milliseconds. + * Default is 1000 (1 second). + */ + initialReconnectionDelay: number; + + /** + * The factor by which the reconnection delay increases after each attempt. + * Default is 1.5. + */ + reconnectionDelayGrowFactor: number; + + /** + * Maximum number of reconnection attempts before giving up. + * Default is 0 (unlimited). + */ + maxRetries: number; +} + /** * Configuration options for the `StreamableHTTPClientTransport`. */ @@ -36,6 +66,11 @@ export type StreamableHTTPClientTransportOptions = { * Customizes HTTP requests to the server. */ requestInit?: RequestInit; + + /** + * Options to configure the reconnection behavior. + */ + reconnectionOptions?: StreamableHTTPReconnectionOptions; }; /** @@ -49,6 +84,7 @@ export class StreamableHTTPClientTransport implements Transport { private _requestInit?: RequestInit; private _authProvider?: OAuthClientProvider; private _sessionId?: string; + private _reconnectionOptions: StreamableHTTPReconnectionOptions; onclose?: () => void; onerror?: (error: Error) => void; @@ -61,6 +97,7 @@ export class StreamableHTTPClientTransport implements Transport { this._url = url; this._requestInit = opts?.requestInit; this._authProvider = opts?.authProvider; + this._reconnectionOptions = opts?.reconnectionOptions || this._defaultReconnectionOptions; } private async _authThenStart(): Promise { @@ -136,7 +173,7 @@ export class StreamableHTTPClientTransport implements Transport { `Failed to open SSE stream: ${response.statusText}`, ); } - // Successful connection, handle the SSE stream as a standalone listener + this._handleSseStream(response.body); } catch (error) { this.onerror?.(error as Error); @@ -144,28 +181,93 @@ export class StreamableHTTPClientTransport implements Transport { } } + // Default reconnection options + private readonly _defaultReconnectionOptions: StreamableHTTPReconnectionOptions = { + initialReconnectionDelay: 1000, + maxReconnectionDelay: 30000, + reconnectionDelayGrowFactor: 1.5, + maxRetries: 2, + }; + + // We no longer need global reconnection state as it will be maintained per stream + + /** + * Calculates the next reconnection delay using exponential backoff algorithm + * with jitter for more effective reconnections in high load scenarios. + * + * @param attempt Current reconnection attempt count for the specific stream + * @returns Time to wait in milliseconds before next reconnection attempt + */ + private _getNextReconnectionDelay(attempt: number): number { + // Access default values directly, ensuring they're never undefined + const initialDelay = this._reconnectionOptions.initialReconnectionDelay; + const growFactor = this._reconnectionOptions.reconnectionDelayGrowFactor; + const maxDelay = this._reconnectionOptions.maxReconnectionDelay; + + // Cap at maximum delay + return Math.min(initialDelay * Math.pow(growFactor, attempt), maxDelay); + + } + + /** + * Schedule a reconnection attempt with exponential backoff + * + * @param lastEventId The ID of the last received event for resumability + * @param attemptCount Current reconnection attempt count for this specific stream + */ + private _scheduleReconnection(lastEventId: string, attemptCount = 0): void { + // Use provided options or default options + const maxRetries = this._reconnectionOptions.maxRetries; + + // Check if we've exceeded maximum retry attempts + if (maxRetries > 0 && attemptCount >= maxRetries) { + this.onerror?.(new Error(`Maximum reconnection attempts (${maxRetries}) exceeded.`)); + return; + } + + // Calculate next delay based on current attempt count + const delay = this._getNextReconnectionDelay(attemptCount); + log(`Reconnection attempt ${attemptCount + 1} in ${delay}ms...`); + + // Schedule the reconnection + setTimeout(() => { + // Use the last event ID to resume where we left off + this._startOrAuthStandaloneSSE(lastEventId).catch(error => { + this.onerror?.(new Error(`Failed to reconnect SSE stream: ${error instanceof Error ? error.message : String(error)}`)); + // Schedule another attempt if this one failed, incrementing the attempt counter + this._scheduleReconnection(lastEventId, attemptCount + 1); + }); + }, delay); + } + private _handleSseStream(stream: ReadableStream | null): void { if (!stream) { return; } let lastEventId: string | undefined; - const processStream = async () => { - // Create a pipeline: binary stream -> text decoder -> SSE parser - const eventStream = stream - .pipeThrough(new TextDecoderStream()) - .pipeThrough(new EventSourceParserStream()); - + // this is the closest we can get to trying to cath network errors + // if something happens reader will throw try { - for await (const event of eventStream) { + // Create a pipeline: binary stream -> text decoder -> SSE parser + const reader = stream + .pipeThrough(new TextDecoderStream()) + .pipeThrough(new EventSourceParserStream()) + .getReader(); + + + while (true) { + const { value: event, done } = await reader.read(); + if (done) { + break; + } + // Update last event ID if provided if (event.id) { lastEventId = event.id; } - // Handle message events (default event type is undefined per docs) - // or explicit 'message' event type if (!event.event || event.event === "message") { try { const message = JSONRPCMessageSchema.parse(JSON.parse(event.data)); @@ -179,31 +281,22 @@ export class StreamableHTTPClientTransport implements Transport { // Handle stream errors - likely a network disconnect this.onerror?.(new Error(`SSE stream disconnected: ${error instanceof Error ? error.message : String(error)}`)); - // Attempt to reconnect if the stream disconnects unexpectedly - // Wait a short time before reconnecting to avoid rapid reconnection loops + // Attempt to reconnect if the stream disconnects unexpectedly and we aren't closing if (this._abortController && !this._abortController.signal.aborted) { - setTimeout(() => { - // Use the last event ID to resume where we left off - this._startOrAuthStandaloneSSE(lastEventId).catch(reconnectError => { - this.onerror?.(new Error(`Failed to reconnect SSE stream: ${reconnectError instanceof Error ? reconnectError.message : String(reconnectError)}`)); - }); - }, 1000); // 1 second delay before reconnection attempt + // Use the exponential backoff reconnection strategy + if (lastEventId !== undefined) { + try { + this._scheduleReconnection(lastEventId, 0); + } + catch (error) { + this.onerror?.(new Error(`Failed to reconnect: ${error instanceof Error ? error.message : String(error)}`)); + + } + } } } }; - - processStream().catch(err => { - this.onerror?.(err); - - // Try to reconnect on unexpected errors - if (this._abortController && !this._abortController.signal.aborted) { - setTimeout(() => { - this._startOrAuthStandaloneSSE(lastEventId).catch(reconnectError => { - this.onerror?.(new Error(`Failed to reconnect SSE stream: ${reconnectError instanceof Error ? reconnectError.message : String(reconnectError)}`)); - }); - }, 1000); - } - }); + processStream(); } async start() { diff --git a/src/examples/client/simpleStreamableHttp.ts b/src/examples/client/simpleStreamableHttp.ts index 42ffa28e..a687eccd 100644 --- a/src/examples/client/simpleStreamableHttp.ts +++ b/src/examples/client/simpleStreamableHttp.ts @@ -52,7 +52,6 @@ function printHelp(): void { console.log(' greet [name] - Call the greet tool'); console.log(' multi-greet [name] - Call the multi-greet tool with notifications'); console.log(' start-notifications [interval] [count] - Start periodic notifications'); - console.log(' simulate-disconnect [delay] - Simulate network disconnection after delay seconds'); console.log(' list-prompts - List available prompts'); console.log(' get-prompt [name] [args] - Get a prompt with optional JSON arguments'); console.log(' list-resources - List available resources'); @@ -115,12 +114,6 @@ function commandLoop(): void { break; } - case 'simulate-disconnect': { - const delay = args[1] ? parseInt(args[1], 10) : 5; - simulateDisconnect(delay); - break; - } - case 'list-prompts': await listPrompts(); break; @@ -188,6 +181,9 @@ async function connect(url?: string): Promise { name: 'example-client', version: '1.0.0' }); + client.onerror = (error) => { + console.error('\x1b[31mClient error:', error, '\x1b[0m'); + } transport = new StreamableHTTPClientTransport( new URL(serverUrl) @@ -324,29 +320,6 @@ async function startNotifications(interval: number, count: number): Promise { - if (!transport) return; - - console.log('\nSimulating network disconnect...'); - await transport.close(); - console.log('Transport closed. Client should automatically attempt to reconnect...'); - - // Keep the client object but null the transport - // This simulates a network disconnection without explicit user disconnect - transport = null; - - // Re-display the prompt - process.stdout.write('> '); - }, delaySeconds * 1000); -} - async function listPrompts(): Promise { if (!client) { console.log('Not connected to server.'); diff --git a/src/server/streamableHttp.ts b/src/server/streamableHttp.ts index 0474fc21..b8be6d15 100644 --- a/src/server/streamableHttp.ts +++ b/src/server/streamableHttp.ts @@ -237,13 +237,17 @@ export class StreamableHTTPServerTransport implements Transport { const events = await this._eventStore.getEventsAfter(lastEventId); const streamId = this._eventStore.getStreamIdFromEventId(lastEventId); - const oldResStream = this._streamMapping.get(streamId); this._streamMapping.set(streamId, res); - if (oldResStream) { - // If we have an old response stream, close it - oldResStream.end(); - } + const headers: Record = { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache, no-transform", + Connection: "keep-alive", + }; + if (this.sessionId !== undefined) { + headers["mcp-session-id"] = this.sessionId; + } + res.writeHead(200, headers).flushHeaders(); for (const { eventId, message } of events) { this.writeSSEEvent(res, message, eventId); }