limit on retries

Author: ihrpr

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<void>
+    };
+    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);
+  });
+
 });

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<void> {
@@ -136,36 +173,101 @@ 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);
       throw error;
     }
   }
 
+  // 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<Uint8Array> | 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() {