Merge pull request #1946 from modelcontextprotocol/v2/fix/firefox-sse-stream-hang
fix: prime SSE streams so Firefox resolves the events fetch
This commit is contained in:
commit
9e074aa6f2
4 changed files with 282 additions and 1 deletions
|
|
@ -969,6 +969,74 @@ describe("useServers", () => {
|
|||
);
|
||||
});
|
||||
|
||||
it("ignores the backend's inert priming comment frame (#1858)", async () => {
|
||||
// The backend opens every SSE stream with a `:` comment so a streaming
|
||||
// fetch() resolves on Firefox at all. That frame carries no event/data
|
||||
// field and must NOT be read as a change — otherwise every connection
|
||||
// would fire a spurious background re-fetch on open.
|
||||
writeFileSync(
|
||||
h.configPath,
|
||||
JSON.stringify({
|
||||
mcpServers: { seed: { type: "stdio", command: "s" } },
|
||||
}),
|
||||
);
|
||||
|
||||
// Count list GETs rather than watching the rendered list: a spurious
|
||||
// refresh fires the instant the priming frame is read, so it would race
|
||||
// ahead of any later disk mutation and land the same data — invisible in
|
||||
// the output but a real extra round-trip on every connection.
|
||||
let listGets = 0;
|
||||
let reads = 0;
|
||||
let releaseSecondRead: (() => void) | undefined;
|
||||
const secondRead = new Promise<void>((r) => {
|
||||
releaseSecondRead = r;
|
||||
});
|
||||
const encoder = new TextEncoder();
|
||||
// Must be referentially stable: the hook's SSE effect keys off `fetchFn`,
|
||||
// so an inline arrow would re-subscribe (and re-refresh) every render and
|
||||
// inflate the very count this test asserts on.
|
||||
const fetchFn: typeof fetch = async (input, init) => {
|
||||
const url = input instanceof Request ? input.url : String(input);
|
||||
if (url.endsWith("/api/servers/events")) {
|
||||
const body = {
|
||||
getReader: () => ({
|
||||
read: async () => {
|
||||
reads += 1;
|
||||
// Priming comment only — no `event:` / `data:` line.
|
||||
if (reads === 1) {
|
||||
return { done: false, value: encoder.encode(":\n\n") };
|
||||
}
|
||||
// Hold the stream open so the loop can't end and let a
|
||||
// teardown-time settle hide a queued refresh.
|
||||
await secondRead;
|
||||
return { done: true, value: undefined };
|
||||
},
|
||||
}),
|
||||
};
|
||||
return { ok: true, body } as unknown as Response;
|
||||
}
|
||||
if (url.endsWith("/api/servers")) listGets += 1;
|
||||
return h.fetchFn(url, init);
|
||||
};
|
||||
const { result } = renderHook(() =>
|
||||
useServers({ baseUrl: "http://test.local", fetchFn }),
|
||||
);
|
||||
await waitFor(() => expect(result.current.loading).toBe(false));
|
||||
expect(result.current.servers.map((s) => s.id)).toEqual(["seed"]);
|
||||
|
||||
// The priming frame has been consumed (the loop is parked on read #2).
|
||||
await waitFor(() => expect(reads).toBeGreaterThanOrEqual(2));
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
|
||||
// Exactly one GET: the hook's mount refresh. The comment frame added none.
|
||||
expect(listGets).toBe(1);
|
||||
|
||||
await act(async () => {
|
||||
releaseSecondRead?.();
|
||||
await Promise.resolve();
|
||||
});
|
||||
});
|
||||
|
||||
it("falls back to globalThis.fetch when no fetchFn is provided", async () => {
|
||||
// No fetchFn → the `doFetch = fetchFn ?? globalThis.fetch` default branch.
|
||||
const globalFetch = vi
|
||||
|
|
|
|||
159
clients/web/src/test/integration/mcp/remote/sse-priming.test.ts
Normal file
159
clients/web/src/test/integration/mcp/remote/sse-priming.test.ts
Normal file
|
|
@ -0,0 +1,159 @@
|
|||
/**
|
||||
* Regression tests for #1858 — the web UI hanging forever on "Connecting…"
|
||||
* in Firefox.
|
||||
*
|
||||
* Firefox does not hand a streaming `fetch()` response to JS until the first
|
||||
* *body* byte arrives (Chromium resolves on headers). Both SSE endpoints used
|
||||
* to flush headers and then stay silent until there was something to report,
|
||||
* which deadlocked `/api/mcp/events`: the browser transport awaits that fetch
|
||||
* before the MCP client sends `initialize`, so nothing was ever reported and
|
||||
* the fetch never resolved.
|
||||
*
|
||||
* The fix is a priming SSE comment written the instant each stream opens.
|
||||
* These tests assert the bytes land on an otherwise-idle stream — the
|
||||
* behavior the browser depends on — rather than any particular payload.
|
||||
*/
|
||||
|
||||
import { describe, it, expect, afterEach, beforeEach } from "vitest";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { serve } from "@hono/node-server";
|
||||
import type { ServerType } from "@hono/node-server";
|
||||
import { createRemoteApp } from "@inspector/core/mcp/remote/node/server.js";
|
||||
import type { MCPServerConfig } from "@inspector/core/mcp/types.js";
|
||||
import { closeHarnessServer } from "./harnessTeardown.js";
|
||||
|
||||
interface Harness {
|
||||
baseUrl: string;
|
||||
server: ServerType;
|
||||
storageDir: string;
|
||||
}
|
||||
|
||||
async function setup(): Promise<Harness> {
|
||||
const storageDir = mkdtempSync(join(tmpdir(), "sse-priming-"));
|
||||
const { app } = createRemoteApp({
|
||||
dangerouslyOmitAuth: true,
|
||||
storageDir,
|
||||
initialConfig: { defaultEnvironment: {} },
|
||||
});
|
||||
return new Promise((resolve, reject) => {
|
||||
const server = serve(
|
||||
{ fetch: app.fetch, port: 0, hostname: "127.0.0.1" },
|
||||
(info) => {
|
||||
const port =
|
||||
info && typeof info === "object" && "port" in info
|
||||
? (info as { port: number }).port
|
||||
: 0;
|
||||
resolve({ baseUrl: `http://127.0.0.1:${port}`, server, storageDir });
|
||||
},
|
||||
);
|
||||
server.on("error", reject);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the first body chunk off a streaming response, or reject if none
|
||||
* arrives within `timeoutMs`. Pre-fix, an idle stream produced no chunk at
|
||||
* all — which is exactly what stalled Firefox — so the timeout is the
|
||||
* assertion that matters here.
|
||||
*/
|
||||
async function firstChunk(res: Response, timeoutMs = 3000): Promise<string> {
|
||||
if (!res.body) throw new Error("response has no body");
|
||||
const reader = res.body.getReader();
|
||||
try {
|
||||
const read = reader.read().then(({ value }) => {
|
||||
return value ? new TextDecoder().decode(value) : "";
|
||||
});
|
||||
const timeout = new Promise<never>((_, reject) => {
|
||||
setTimeout(
|
||||
() => reject(new Error(`no body bytes within ${timeoutMs}ms`)),
|
||||
timeoutMs,
|
||||
);
|
||||
});
|
||||
return await Promise.race([read, timeout]);
|
||||
} finally {
|
||||
await reader.cancel().catch(() => {
|
||||
/* stream already torn down */
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/** The inert SSE comment frame the backend primes each stream with. */
|
||||
const PRIMING_FRAME = ":\n\n";
|
||||
|
||||
describe("SSE streams are primed on open (#1858)", () => {
|
||||
let h: Harness;
|
||||
|
||||
beforeEach(async () => {
|
||||
h = await setup();
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await closeHarnessServer(h.server);
|
||||
rmSync(h.storageDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("GET /api/mcp/events flushes bytes before any MCP traffic", async () => {
|
||||
// A subprocess that spawns and stays alive but never speaks — the state
|
||||
// the session is in between `connect` and the client's `initialize`.
|
||||
// This is precisely the window in which the deadlock occurred.
|
||||
const config: MCPServerConfig = {
|
||||
type: "stdio",
|
||||
command: process.execPath,
|
||||
args: ["-e", "setInterval(() => {}, 1000);"],
|
||||
};
|
||||
|
||||
const connectRes = await fetch(`${h.baseUrl}/api/mcp/connect`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ config }),
|
||||
});
|
||||
expect(connectRes.status).toBe(200);
|
||||
const { sessionId } = (await connectRes.json()) as { sessionId: string };
|
||||
|
||||
const controller = new AbortController();
|
||||
try {
|
||||
const eventsRes = await fetch(
|
||||
`${h.baseUrl}/api/mcp/events?sessionId=${sessionId}`,
|
||||
{ signal: controller.signal },
|
||||
);
|
||||
expect(eventsRes.status).toBe(200);
|
||||
|
||||
expect(await firstChunk(eventsRes)).toBe(PRIMING_FRAME);
|
||||
} finally {
|
||||
controller.abort();
|
||||
await fetch(`${h.baseUrl}/api/mcp/disconnect`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ sessionId }),
|
||||
}).catch(() => {
|
||||
/* best-effort teardown */
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
it("GET /api/servers/events flushes bytes before any file change", async () => {
|
||||
const controller = new AbortController();
|
||||
try {
|
||||
const res = await fetch(`${h.baseUrl}/api/servers/events`, {
|
||||
signal: controller.signal,
|
||||
});
|
||||
expect(res.status).toBe(200);
|
||||
|
||||
expect(await firstChunk(res)).toBe(PRIMING_FRAME);
|
||||
} finally {
|
||||
controller.abort();
|
||||
}
|
||||
});
|
||||
|
||||
it("primes with an SSE comment, which carries no event or data field", () => {
|
||||
// The priming frame must be inert: a conforming parser drops it, so no
|
||||
// client needs to know about it. Guards against it ever being changed
|
||||
// into something a consumer would mistake for a real event.
|
||||
const lines = PRIMING_FRAME.split("\n");
|
||||
expect(lines.some((l) => l.startsWith("data:"))).toBe(false);
|
||||
expect(lines.some((l) => l.startsWith("event:"))).toBe(false);
|
||||
expect(lines[0]).toBe(":");
|
||||
});
|
||||
});
|
||||
|
|
@ -78,6 +78,27 @@ import { formatClientConfigLoadError } from "../../../client/config-parse.js";
|
|||
import { envSecretField } from "../../../auth/secret-fields.js";
|
||||
import { ZodError } from "zod";
|
||||
|
||||
/**
|
||||
* Written to every SSE stream the instant it opens, before anything else.
|
||||
*
|
||||
* Firefox does not hand a streaming `fetch()` response to JS until the first
|
||||
* *body* byte arrives; Chromium resolves the promise as soon as the headers
|
||||
* do. Both SSE endpoints here flush headers immediately and then stay silent
|
||||
* until there is something to report, which deadlocks Firefox on
|
||||
* `/api/mcp/events`: `RemoteClientTransport.openEventStream()` awaits that
|
||||
* fetch *before* the MCP client sends `initialize`, so no `initialize` → no
|
||||
* event to report → no body byte → the fetch never resolves → the web UI
|
||||
* hangs on "Connecting…" forever with no error anywhere (#1858).
|
||||
*
|
||||
* A `:` comment line is inert per the SSE spec — conforming parsers ignore it
|
||||
* — so priming with one unblocks the read without inventing a wire event.
|
||||
*
|
||||
* `X-Content-Type-Options: nosniff` does **not** fix this. Verified against
|
||||
* Firefox 153: with the header and no body byte, the fetch still never
|
||||
* resolves. The delay is not MIME sniffing.
|
||||
*/
|
||||
const SSE_PRIMING_COMMENT = ":\n\n";
|
||||
|
||||
/**
|
||||
* Shape of the initial config returned by GET /api/config (defaults for client).
|
||||
*/
|
||||
|
|
@ -791,6 +812,7 @@ export function createRemoteApp(
|
|||
}
|
||||
});
|
||||
|
||||
// Prime every SSE stream the instant it opens — see SSE_PRIMING_COMMENT.
|
||||
app.get("/api/mcp/events", async (c) => {
|
||||
const sessionId = c.req.query("sessionId");
|
||||
if (!sessionId) {
|
||||
|
|
@ -824,6 +846,11 @@ export function createRemoteApp(
|
|||
return;
|
||||
}
|
||||
|
||||
// Prime only after the consumer is registered, so nothing observable
|
||||
// to the client happens before this stream can actually report
|
||||
// events. See SSE_PRIMING_COMMENT.
|
||||
await stream.write(SSE_PRIMING_COMMENT);
|
||||
|
||||
stream.onAbort(() => {
|
||||
// Client disconnected - clear event consumer
|
||||
const shouldCleanup = session.clearEventConsumer();
|
||||
|
|
@ -2395,6 +2422,16 @@ export function createRemoteApp(
|
|||
ensureWatcher();
|
||||
}
|
||||
|
||||
// Prime the stream so the client's fetch() actually resolves — on
|
||||
// Firefox it otherwise stays pending until the first real change
|
||||
// event, which for an unedited `mcp.json` is never. See
|
||||
// SSE_PRIMING_COMMENT. Deliberately *after* the subscriber is
|
||||
// registered and the watcher started: callers treat the arrival of
|
||||
// this stream's first bytes as proof they are subscribed, so priming
|
||||
// first would hand out that proof across an `await`, before the
|
||||
// watcher exists, and drop an edit made in the gap.
|
||||
await stream.write(SSE_PRIMING_COMMENT);
|
||||
|
||||
stream.onAbort(() => {
|
||||
serverEventSubscribers.delete(send);
|
||||
void maybeStopWatcher();
|
||||
|
|
|
|||
|
|
@ -82,6 +82,18 @@ async function readErrorMessage(res: Response): Promise<string> {
|
|||
return `HTTP ${res.status}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether an SSE frame carries an actual event rather than only comment
|
||||
* lines. The backend primes each stream with an inert `:` comment so a
|
||||
* streaming `fetch()` resolves on Firefox at all (#1858); that frame must
|
||||
* not be mistaken for a change notification.
|
||||
*/
|
||||
function isSseDataFrame(frame: string): boolean {
|
||||
return frame
|
||||
.split("\n")
|
||||
.some((line) => line.startsWith("event:") || line.startsWith("data:"));
|
||||
}
|
||||
|
||||
export function useServers(opts: UseServersOptions): UseServersResult {
|
||||
const { baseUrl, authToken, fetchFn } = opts;
|
||||
const doFetch = fetchFn ?? globalThis.fetch;
|
||||
|
|
@ -157,6 +169,10 @@ export function useServers(opts: UseServersOptions): UseServersResult {
|
|||
// single background refresh per decode chunk. Two `change`
|
||||
// broadcasts landing in the same chunk become one re-fetch instead
|
||||
// of two concurrent ones whose setState order is unspecified.
|
||||
// Frames carrying no `event:`/`data:` field are skipped: the backend
|
||||
// opens the stream with an inert `:` comment frame so Firefox
|
||||
// resolves this fetch at all (see SSE_PRIMING_COMMENT in the remote
|
||||
// server), and that must not read as a change.
|
||||
// Cross-chunk debounce is not added: `awaitWriteFinish`'s 100ms
|
||||
// stability threshold already serializes external edits at the
|
||||
// source, and chained fetches against the same GET endpoint are
|
||||
|
|
@ -169,8 +185,9 @@ export function useServers(opts: UseServersOptions): UseServersResult {
|
|||
let sawFrame = false;
|
||||
let frameEnd = buffer.indexOf("\n\n");
|
||||
while (frameEnd !== -1) {
|
||||
const frame = buffer.slice(0, frameEnd);
|
||||
buffer = buffer.slice(frameEnd + 2);
|
||||
sawFrame = true;
|
||||
if (isSseDataFrame(frame)) sawFrame = true;
|
||||
frameEnd = buffer.indexOf("\n\n");
|
||||
}
|
||||
if (sawFrame) void refreshInternal(true);
|
||||
|
|
|
|||
Loading…
Reference in a new issue