limit on retries
This commit is contained in:
ihrpr
parent
3bd28829a5
commit
6374f43390
4 changed files with 216 additions and 109 deletions
|
|
@ -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);
|
||||
});
|
||||
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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,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<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() {
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
|||
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<void
|
|||
await callTool('start-notification-stream', { interval, count });
|
||||
}
|
||||
|
||||
function simulateDisconnect(delaySeconds: number): void {
|
||||
if (!transport) {
|
||||
console.log('Not connected.');
|
||||
return;
|
||||
}
|
||||
|
||||
console.log(`Will simulate network disconnect in ${delaySeconds} seconds...`);
|
||||
setTimeout(async () => {
|
||||
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<void> {
|
||||
if (!client) {
|
||||
console.log('Not connected to server.');
|
||||
|
|
|
|||
|
|
@ -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<string, string> = {
|
||||
"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);
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue