From 8ab1236edfa8ba2c059313916fca52d64e577a97 Mon Sep 17 00:00:00 2001 From: Nick Christensen Date: Thu, 16 Jul 2026 16:10:03 -0500 Subject: [PATCH] Fix MCP HTTP connection cleanup --- .../plugins/mcp/src/sdk/connection.test.ts | 94 +++++++++++++++++++ packages/plugins/mcp/src/sdk/connection.ts | 30 ++++-- 2 files changed, 116 insertions(+), 8 deletions(-) create mode 100644 packages/plugins/mcp/src/sdk/connection.test.ts diff --git a/packages/plugins/mcp/src/sdk/connection.test.ts b/packages/plugins/mcp/src/sdk/connection.test.ts new file mode 100644 index 000000000..d514573ab --- /dev/null +++ b/packages/plugins/mcp/src/sdk/connection.test.ts @@ -0,0 +1,94 @@ +import { describe, it } from "@effect/vitest"; +import { Effect, Layer, Option, Predicate, Schema } from "effect"; +import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"; + +import { createMcpConnector } from "./connection"; + +const JsonRpcId = Schema.Union([Schema.String, Schema.Number, Schema.Null]); +const JsonRpcRequest = Schema.Struct({ + id: Schema.optional(JsonRpcId), + method: Schema.String, +}); +type JsonRpcRequest = typeof JsonRpcRequest.Type; + +const decodeJsonRpcRequest = Schema.decodeUnknownOption(Schema.fromJsonString(JsonRpcRequest)); + +const initializeResponse = (request: HttpClientRequest.HttpClientRequest): Response => { + const body = Predicate.isTagged(request.body, "Uint8Array") + ? new TextDecoder().decode(request.body.body) + : ""; + const rpc = Option.getOrElse(decodeJsonRpcRequest(body), () => ({ + id: undefined, + method: "", + })); + if (rpc.method !== "initialize") return new Response(null, { status: 202 }); + return new Response( + JSON.stringify({ + jsonrpc: "2.0", + id: rpc.id ?? null, + result: { + protocolVersion: "2025-06-18", + capabilities: {}, + serverInfo: { name: "connection-fixture", version: "1.0.0" }, + }, + }), + { + headers: { + "content-type": "application/json", + "mcp-session-id": "connection-fixture-session", + }, + }, + ); +}; + +describe("MCP HTTP connection cleanup", () => { + it.effect("aborts an open GET SSE response stream when the connection closes", () => + Effect.gen(function* () { + let resolveGetStarted!: () => void; + const getStarted = new Promise((resolve) => { + resolveGetStarted = resolve; + }); + let resolveCancelled!: () => void; + const cancelled = new Promise((resolve) => { + resolveCancelled = resolve; + }); + const sseBody = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(": priming\n\n")); + }, + cancel() { + resolveCancelled(); + }, + }); + const httpClientLayer = Layer.succeed(HttpClient.HttpClient)( + HttpClient.make((request) => + Effect.sync(() => { + if (request.method === "POST") { + return HttpClientResponse.fromWeb(request, initializeResponse(request)); + } + if (request.method === "GET") { + resolveGetStarted(); + return HttpClientResponse.fromWeb( + request, + new Response(sseBody, { + headers: { "content-type": "text/event-stream" }, + }), + ); + } + return HttpClientResponse.fromWeb(request, new Response(null, { status: 405 })); + }), + ), + ); + const connection = yield* createMcpConnector({ + transport: "remote", + endpoint: "https://connection-fixture.example/mcp", + remoteTransport: "streamable-http", + httpClientLayer, + }); + + yield* Effect.promise(() => getStarted); + yield* Effect.promise(() => connection.close()); + yield* Effect.promise(() => cancelled); + }), + ); +}); diff --git a/packages/plugins/mcp/src/sdk/connection.ts b/packages/plugins/mcp/src/sdk/connection.ts index 4cad5f767..20dde26ab 100644 --- a/packages/plugins/mcp/src/sdk/connection.ts +++ b/packages/plugins/mcp/src/sdk/connection.ts @@ -123,6 +123,17 @@ const abortError = (signal: AbortSignal): unknown => { return error; }; +const awaitAbort = (signal: AbortSignal): Effect.Effect => + Effect.callback((resume) => { + if (signal.aborted) { + resume(Effect.void); + return; + } + const onAbort = () => resume(Effect.void); + signal.addEventListener("abort", onAbort, { once: true }); + return Effect.sync(() => signal.removeEventListener("abort", onAbort)); + }); + const fetchFromHttpClientLayer = ( httpClientLayer: Layer.Layer, ): FetchLike => { @@ -142,7 +153,11 @@ const fetchFromHttpClientLayer = ( const body = response.status === 204 || response.status === 205 || response.status === 304 ? null - : Stream.toReadableStream(response.stream); + : Stream.toReadableStream( + init?.signal + ? Stream.interruptWhen(response.stream, awaitAbort(init.signal)) + : response.stream, + ); return new Response(body, { status: response.status, headers: responseHeaders, @@ -156,7 +171,10 @@ const fetchFromHttpClientLayer = ( // tagged error from the fetch adapter (a true runtime edge: the SDK // consumes promise rejections) so it reaches the invoke/connect catch // sites verbatim. - const promise = Effect.runPromise(effect).then((response) => { + const signal = init?.signal; + // oxlint-disable-next-line executor/no-promise-reject -- boundary: Fetch-compatible adapter mirrors abort rejection semantics + if (signal?.aborted) return Promise.reject(abortError(signal)); + const promise = Effect.runPromise(effect, { signal: signal ?? undefined }).then((response) => { if (response.status === 403) { const challenge = response.headers.get("www-authenticate"); if ( @@ -172,14 +190,10 @@ const fetchFromHttpClientLayer = ( } return response; }); - if (!init?.signal) return promise; - // oxlint-disable-next-line executor/no-promise-reject -- boundary: Fetch-compatible adapter mirrors abort rejection semantics - if (init.signal.aborted) return Promise.reject(abortError(init.signal)); + if (!signal) return promise; const aborted = new Promise((_, reject) => { // oxlint-disable-next-line executor/no-promise-reject -- boundary: Fetch-compatible adapter races the Effect request against AbortSignal - init.signal?.addEventListener("abort", () => reject(abortError(init.signal!)), { - once: true, - }); + signal.addEventListener("abort", () => reject(abortError(signal)), { once: true }); }); return Promise.race([promise, aborted]); };