427 lines
13 KiB
TypeScript
427 lines
13 KiB
TypeScript
/**
|
|
* Integration tests for ResilientTransportWrapper using real HTTP/SSE servers.
|
|
*
|
|
* These tests use the demo servers to simulate realistic network scenarios
|
|
* including heartbeat concurrency, reconnection, and timing issues.
|
|
*/
|
|
|
|
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
|
|
import { ResilientTransportWrapper } from '../src/resilient.js';
|
|
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js';
|
|
import { SSEClientTransport } from '@modelcontextprotocol/sdk/client/sse.js';
|
|
import { Client } from '@modelcontextprotocol/sdk/client/index.js';
|
|
import { createServer, type Server } from 'node:http';
|
|
import { randomUUID } from 'node:crypto';
|
|
|
|
// ---- Mock HTTP Server for testing ----
|
|
|
|
class MockHttpMcpServer {
|
|
private server: Server | null = null;
|
|
private port: number;
|
|
private requestDelay: number = 0;
|
|
private shouldFail: boolean = false;
|
|
private requestCount: number = 0;
|
|
|
|
constructor(port: number) {
|
|
this.port = port;
|
|
}
|
|
|
|
setRequestDelay(ms: number) {
|
|
this.requestDelay = ms;
|
|
}
|
|
|
|
setShouldFail(fail: boolean) {
|
|
this.shouldFail = fail;
|
|
}
|
|
|
|
getRequestCount(): number {
|
|
return this.requestCount;
|
|
}
|
|
|
|
start(): Promise<void> {
|
|
return new Promise((resolve) => {
|
|
this.server = createServer(async (req, res) => {
|
|
this.requestCount++;
|
|
|
|
// Simulate slow network
|
|
if (this.requestDelay > 0) {
|
|
await new Promise(r => setTimeout(r, this.requestDelay));
|
|
}
|
|
|
|
if (this.shouldFail) {
|
|
res.writeHead(500, { 'Content-Type': 'application/json' });
|
|
res.end(JSON.stringify({ error: 'Server error' }));
|
|
return;
|
|
}
|
|
|
|
// Read body
|
|
const chunks: Buffer[] = [];
|
|
for await (const chunk of req) chunks.push(chunk);
|
|
const body = chunks.length > 0 ? JSON.parse(Buffer.concat(chunks).toString()) : {};
|
|
|
|
// Handle MCP requests
|
|
if (body.method === 'initialize') {
|
|
res.writeHead(200, {
|
|
'Content-Type': 'application/json',
|
|
'mcp-session-id': randomUUID(),
|
|
});
|
|
res.end(JSON.stringify({
|
|
jsonrpc: '2.0',
|
|
id: body.id,
|
|
result: {
|
|
protocolVersion: '2024-11-05',
|
|
capabilities: {},
|
|
serverInfo: { name: 'mock-server', version: '1.0' },
|
|
},
|
|
}));
|
|
return;
|
|
}
|
|
|
|
if (body.method === 'tools/list') {
|
|
res.writeHead(200, { 'Content-Type': 'application/json' });
|
|
res.end(JSON.stringify({
|
|
jsonrpc: '2.0',
|
|
id: body.id,
|
|
result: {
|
|
tools: [
|
|
{ name: 'echo', description: 'Echo tool', inputSchema: {} },
|
|
],
|
|
},
|
|
}));
|
|
return;
|
|
}
|
|
|
|
// Default response
|
|
res.writeHead(200, { 'Content-Type': 'application/json' });
|
|
res.end(JSON.stringify({
|
|
jsonrpc: '2.0',
|
|
id: body.id,
|
|
result: {},
|
|
}));
|
|
});
|
|
|
|
this.server.listen(this.port, '127.0.0.1', () => resolve());
|
|
});
|
|
}
|
|
|
|
async stop(): Promise<void> {
|
|
return new Promise((resolve) => {
|
|
if (this.server) {
|
|
this.server.close(() => resolve());
|
|
} else {
|
|
resolve();
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
describe('ResilientTransportWrapper - Integration Tests', () => {
|
|
let mockServer: MockHttpMcpServer;
|
|
const TEST_PORT = 19080;
|
|
const TEST_URL = `http://127.0.0.1:${TEST_PORT}/mcp`;
|
|
|
|
beforeEach(() => {
|
|
mockServer = new MockHttpMcpServer(TEST_PORT);
|
|
vi.useFakeTimers();
|
|
});
|
|
|
|
afterEach(async () => {
|
|
await mockServer.stop();
|
|
vi.useRealTimers();
|
|
vi.restoreAllMocks();
|
|
});
|
|
|
|
describe('Heartbeat Concurrency Protection', () => {
|
|
it('should prevent multiple concurrent health checks with slow network', async () => {
|
|
// Start mock server with slow response
|
|
mockServer.setRequestDelay(500); // 500ms delay per request
|
|
await mockServer.start();
|
|
|
|
let healthCheckCalls = 0;
|
|
let concurrentCalls = 0;
|
|
let maxConcurrentCalls = 0;
|
|
|
|
const wrapper = new ResilientTransportWrapper({
|
|
name: 'concurrency-test',
|
|
pingIntervalMs: 1000,
|
|
pingTimeoutMs: 10000, // Long timeout to not interfere
|
|
connectParams: async () => {
|
|
return new StreamableHTTPClientTransport(new URL(TEST_URL));
|
|
},
|
|
healthCheckFn: async () => {
|
|
healthCheckCalls++;
|
|
concurrentCalls++;
|
|
maxConcurrentCalls = Math.max(maxConcurrentCalls, concurrentCalls);
|
|
|
|
// Simulate slow health check
|
|
await new Promise(r => setTimeout(r, 2000));
|
|
|
|
concurrentCalls--;
|
|
return true;
|
|
},
|
|
});
|
|
|
|
await wrapper.start();
|
|
wrapper.enableHeartbeat();
|
|
|
|
// Trigger first health check
|
|
await vi.advanceTimersByTimeAsync(1001);
|
|
|
|
// At this point, first health check is in progress (started but not completed)
|
|
expect(healthCheckCalls).toBe(1);
|
|
|
|
// Trigger more timer events while health check is in progress
|
|
// These should be blocked by healthCheckInProgress guard
|
|
await vi.advanceTimersByTimeAsync(1000); // Would trigger second check if not guarded
|
|
await vi.advanceTimersByTimeAsync(1000); // Would trigger third check
|
|
|
|
// Complete the first health check
|
|
await vi.advanceTimersByTimeAsync(100);
|
|
|
|
// Only 1 call should have started because of the guard
|
|
expect(maxConcurrentCalls).toBe(1);
|
|
|
|
// After first check completes, next one can be scheduled
|
|
await vi.advanceTimersByTimeAsync(1000);
|
|
expect(healthCheckCalls).toBe(2);
|
|
|
|
await wrapper.close();
|
|
});
|
|
|
|
it('should handle rapid heartbeat timer triggers without stacking', async () => {
|
|
await mockServer.start();
|
|
|
|
const healthCheckTimes: number[] = [];
|
|
let startTime = Date.now();
|
|
|
|
const wrapper = new ResilientTransportWrapper({
|
|
name: 'stacking-test',
|
|
pingIntervalMs: 500,
|
|
connectParams: async () => {
|
|
return new StreamableHTTPClientTransport(new URL(TEST_URL));
|
|
},
|
|
healthCheckFn: async () => {
|
|
healthCheckTimes.push(Date.now() - startTime);
|
|
return true;
|
|
},
|
|
});
|
|
|
|
await wrapper.start();
|
|
wrapper.enableHeartbeat();
|
|
|
|
// Run for 10 "heartbeat cycles"
|
|
for (let i = 0; i < 10; i++) {
|
|
await vi.advanceTimersByTimeAsync(500);
|
|
await vi.advanceTimersByTimeAsync(10); // Allow async to complete
|
|
}
|
|
|
|
expect(healthCheckTimes.length).toBe(10);
|
|
|
|
// Verify each check is ~500ms apart (response-driven scheduling)
|
|
for (let i = 1; i < healthCheckTimes.length; i++) {
|
|
const diff = healthCheckTimes[i] - healthCheckTimes[i - 1];
|
|
// Should be at least 500ms (pingIntervalMs), not stacked
|
|
expect(diff).toBeGreaterThanOrEqual(500);
|
|
}
|
|
|
|
await wrapper.close();
|
|
});
|
|
});
|
|
|
|
describe('Response-Driven Scheduling', () => {
|
|
it('should schedule next check only after current one completes', async () => {
|
|
await mockServer.start();
|
|
|
|
let checkCompleted = false;
|
|
let nextCheckScheduledBeforeComplete = false;
|
|
|
|
const wrapper = new ResilientTransportWrapper({
|
|
name: 'response-driven-test',
|
|
pingIntervalMs: 1000,
|
|
connectParams: async () => {
|
|
return new StreamableHTTPClientTransport(new URL(TEST_URL));
|
|
},
|
|
healthCheckFn: async () => {
|
|
// Check if next check was already scheduled (it shouldn't be)
|
|
// This is a bit indirect, but we can verify by timing
|
|
await new Promise(r => setTimeout(r, 500));
|
|
checkCompleted = true;
|
|
return true;
|
|
},
|
|
});
|
|
|
|
await wrapper.start();
|
|
wrapper.enableHeartbeat();
|
|
|
|
// Trigger first check
|
|
await vi.advanceTimersByTimeAsync(1001);
|
|
|
|
// Check is in progress, not completed yet
|
|
expect(checkCompleted).toBe(false);
|
|
|
|
// Complete the check
|
|
await vi.advanceTimersByTimeAsync(500);
|
|
expect(checkCompleted).toBe(true);
|
|
|
|
// Now next check should be scheduled 1000ms after completion
|
|
await vi.advanceTimersByTimeAsync(999);
|
|
// Should not have triggered yet
|
|
// (we'd need another counter to verify this properly)
|
|
|
|
await vi.advanceTimersByTimeAsync(2);
|
|
// Now it should trigger
|
|
|
|
await wrapper.close();
|
|
});
|
|
|
|
it('should not schedule next check when state changes during health check', async () => {
|
|
await mockServer.start();
|
|
|
|
let healthCheckCount = 0;
|
|
let shouldFailHealthCheck = false;
|
|
|
|
const wrapper = new ResilientTransportWrapper({
|
|
name: 'state-change-test',
|
|
pingIntervalMs: 1000,
|
|
maxConsecutiveFailures: 2,
|
|
reconnectDelayMs: 100,
|
|
connectParams: async () => {
|
|
return new StreamableHTTPClientTransport(new URL(TEST_URL));
|
|
},
|
|
healthCheckFn: async () => {
|
|
healthCheckCount++;
|
|
if (shouldFailHealthCheck) {
|
|
return false;
|
|
}
|
|
return true;
|
|
},
|
|
});
|
|
|
|
await wrapper.start();
|
|
wrapper.enableHeartbeat();
|
|
|
|
// First check passes
|
|
await vi.advanceTimersByTimeAsync(1001);
|
|
await vi.advanceTimersByTimeAsync(10);
|
|
expect(healthCheckCount).toBe(1);
|
|
|
|
// Make health checks fail
|
|
shouldFailHealthCheck = true;
|
|
|
|
// Second check fails
|
|
await vi.advanceTimersByTimeAsync(1000);
|
|
await vi.advanceTimersByTimeAsync(10);
|
|
expect(healthCheckCount).toBe(2);
|
|
|
|
// Third check fails, triggers reconnect
|
|
await vi.advanceTimersByTimeAsync(1000);
|
|
await vi.advanceTimersByTimeAsync(10);
|
|
expect(healthCheckCount).toBe(3);
|
|
|
|
// After max failures, state should be reconnecting
|
|
// No more health checks should run during reconnection
|
|
const countBeforeReconnect = healthCheckCount;
|
|
await vi.advanceTimersByTimeAsync(50); // During reconnection delay
|
|
|
|
// Should not have increased because state is reconnecting
|
|
expect(healthCheckCount).toBe(countBeforeReconnect);
|
|
|
|
await wrapper.close();
|
|
});
|
|
});
|
|
|
|
describe('Real-World Scenarios', () => {
|
|
it('should handle server restart with pending health checks', async () => {
|
|
await mockServer.start();
|
|
|
|
let healthCheckCount = 0;
|
|
const wrapper = new ResilientTransportWrapper({
|
|
name: 'restart-test',
|
|
pingIntervalMs: 1000,
|
|
maxConsecutiveFailures: 3,
|
|
reconnectDelayMs: 100,
|
|
connectParams: async () => {
|
|
return new StreamableHTTPClientTransport(new URL(TEST_URL));
|
|
},
|
|
healthCheckFn: async () => {
|
|
healthCheckCount++;
|
|
return true;
|
|
},
|
|
});
|
|
|
|
wrapper.onreconnect = async () => {
|
|
// Simulate reconnection handling
|
|
};
|
|
|
|
await wrapper.start();
|
|
wrapper.enableHeartbeat();
|
|
|
|
// Run a few successful heartbeats
|
|
await vi.advanceTimersByTimeAsync(1001);
|
|
await vi.advanceTimersByTimeAsync(10);
|
|
await vi.advanceTimersByTimeAsync(1000);
|
|
await vi.advanceTimersByTimeAsync(10);
|
|
|
|
expect(healthCheckCount).toBe(2);
|
|
|
|
// Simulate server going down
|
|
mockServer.setShouldFail(true);
|
|
|
|
// Wait for health checks to fail
|
|
await vi.advanceTimersByTimeAsync(3000);
|
|
await vi.advanceTimersByTimeAsync(10);
|
|
|
|
// Bring server back up
|
|
mockServer.setShouldFail(false);
|
|
|
|
// Wait for reconnection
|
|
await vi.advanceTimersByTimeAsync(500);
|
|
|
|
await wrapper.close();
|
|
});
|
|
|
|
it('should maintain heartbeat rhythm despite slow health checks', async () => {
|
|
await mockServer.start();
|
|
|
|
const checkTimes: number[] = [];
|
|
const startTime = Date.now();
|
|
|
|
const wrapper = new ResilientTransportWrapper({
|
|
name: 'rhythm-test',
|
|
pingIntervalMs: 1000,
|
|
connectParams: async () => {
|
|
return new StreamableHTTPClientTransport(new URL(TEST_URL));
|
|
},
|
|
healthCheckFn: async () => {
|
|
checkTimes.push(Date.now() - startTime);
|
|
// Simulate variable health check duration
|
|
await new Promise(r => setTimeout(r, Math.random() * 200));
|
|
return true;
|
|
},
|
|
});
|
|
|
|
await wrapper.start();
|
|
wrapper.enableHeartbeat();
|
|
|
|
// Run for 5 cycles
|
|
for (let i = 0; i < 5; i++) {
|
|
await vi.advanceTimersByTimeAsync(1000);
|
|
await vi.advanceTimersByTimeAsync(250); // Account for health check duration
|
|
}
|
|
|
|
expect(checkTimes.length).toBe(5);
|
|
|
|
// Verify that checks don't stack up despite variable duration
|
|
// Each check should start approximately 1000ms after the previous one completed
|
|
for (let i = 1; i < checkTimes.length; i++) {
|
|
const diff = checkTimes[i] - checkTimes[i - 1];
|
|
// Should be around 1000ms + health check duration
|
|
expect(diff).toBeGreaterThanOrEqual(1000);
|
|
expect(diff).toBeLessThan(1500); // Not stacked
|
|
}
|
|
|
|
await wrapper.close();
|
|
});
|
|
});
|
|
});
|