fix: prime SSE streams so Firefox resolves the events fetch
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 flushed headers immediately and then stayed silent until there was something to report, which deadlocked `/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 hung on "Connecting..." forever with nothing in the UI, the console, or the node logs. Write an inert `:` SSE comment the instant each stream opens. Conforming parsers ignore a comment line, so this unblocks the read without inventing a wire event and needs no client-side protocol change. `X-Content-Type-Options: nosniff` was also proposed on the issue but does not fix it -- verified against Firefox 153, the fetch still never resolves with the header set and no body byte. The delay is not MIME sniffing. The priming write is deliberately ordered *after* each handler registers its consumer (the session event consumer; the file-watch subscriber plus `ensureWatcher()`). Callers treat the arrival of the stream's first bytes as proof they are subscribed, so priming first would hand out that proof across an `await` and drop an edit made in the gap. `/api/servers/events` gets the same treatment: its fetch was equally stuck on Firefox until the first real change event, which for an unedited `mcp.json` never comes. Its client-side reader counted any `\n\n` as a change, so the inert frame would have fired a spurious background re-fetch on every connection -- it now skips frames carrying no `event:`/`data:` field. The transport's own `parseSSE` already drops comment frames correctly and needed no change. Verified end to end in real Firefox against a live streamable-HTTP test server: before, stuck on "Connecting..." past 25s; after, initialize -> notifications/initialized -> tools/list, connected. Closes #1858 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YAt8rqxysNbhYWLhoRm3fU
This commit is contained in:
parent
097bfafdc1
commit
9644ac7fe8
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 () => {
|
it("falls back to globalThis.fetch when no fetchFn is provided", async () => {
|
||||||
// No fetchFn → the `doFetch = fetchFn ?? globalThis.fetch` default branch.
|
// No fetchFn → the `doFetch = fetchFn ?? globalThis.fetch` default branch.
|
||||||
const globalFetch = vi
|
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 { envSecretField } from "../../../auth/secret-fields.js";
|
||||||
import { ZodError } from "zod";
|
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).
|
* Shape of the initial config returned by GET /api/config (defaults for client).
|
||||||
*/
|
*/
|
||||||
|
|
@ -785,6 +806,7 @@ export function createRemoteApp(
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// Prime every SSE stream the instant it opens — see SSE_PRIMING_COMMENT.
|
||||||
app.get("/api/mcp/events", async (c) => {
|
app.get("/api/mcp/events", async (c) => {
|
||||||
const sessionId = c.req.query("sessionId");
|
const sessionId = c.req.query("sessionId");
|
||||||
if (!sessionId) {
|
if (!sessionId) {
|
||||||
|
|
@ -818,6 +840,11 @@ export function createRemoteApp(
|
||||||
return;
|
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(() => {
|
stream.onAbort(() => {
|
||||||
// Client disconnected - clear event consumer
|
// Client disconnected - clear event consumer
|
||||||
const shouldCleanup = session.clearEventConsumer();
|
const shouldCleanup = session.clearEventConsumer();
|
||||||
|
|
@ -2389,6 +2416,16 @@ export function createRemoteApp(
|
||||||
ensureWatcher();
|
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(() => {
|
stream.onAbort(() => {
|
||||||
serverEventSubscribers.delete(send);
|
serverEventSubscribers.delete(send);
|
||||||
void maybeStopWatcher();
|
void maybeStopWatcher();
|
||||||
|
|
|
||||||
|
|
@ -82,6 +82,18 @@ async function readErrorMessage(res: Response): Promise<string> {
|
||||||
return `HTTP ${res.status}`;
|
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 {
|
export function useServers(opts: UseServersOptions): UseServersResult {
|
||||||
const { baseUrl, authToken, fetchFn } = opts;
|
const { baseUrl, authToken, fetchFn } = opts;
|
||||||
const doFetch = fetchFn ?? globalThis.fetch;
|
const doFetch = fetchFn ?? globalThis.fetch;
|
||||||
|
|
@ -157,6 +169,10 @@ export function useServers(opts: UseServersOptions): UseServersResult {
|
||||||
// single background refresh per decode chunk. Two `change`
|
// single background refresh per decode chunk. Two `change`
|
||||||
// broadcasts landing in the same chunk become one re-fetch instead
|
// broadcasts landing in the same chunk become one re-fetch instead
|
||||||
// of two concurrent ones whose setState order is unspecified.
|
// 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
|
// Cross-chunk debounce is not added: `awaitWriteFinish`'s 100ms
|
||||||
// stability threshold already serializes external edits at the
|
// stability threshold already serializes external edits at the
|
||||||
// source, and chained fetches against the same GET endpoint are
|
// source, and chained fetches against the same GET endpoint are
|
||||||
|
|
@ -169,8 +185,9 @@ export function useServers(opts: UseServersOptions): UseServersResult {
|
||||||
let sawFrame = false;
|
let sawFrame = false;
|
||||||
let frameEnd = buffer.indexOf("\n\n");
|
let frameEnd = buffer.indexOf("\n\n");
|
||||||
while (frameEnd !== -1) {
|
while (frameEnd !== -1) {
|
||||||
|
const frame = buffer.slice(0, frameEnd);
|
||||||
buffer = buffer.slice(frameEnd + 2);
|
buffer = buffer.slice(frameEnd + 2);
|
||||||
sawFrame = true;
|
if (isSseDataFrame(frame)) sawFrame = true;
|
||||||
frameEnd = buffer.indexOf("\n\n");
|
frameEnd = buffer.indexOf("\n\n");
|
||||||
}
|
}
|
||||||
if (sawFrame) void refreshInternal(true);
|
if (sawFrame) void refreshInternal(true);
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue