fix(web): accept CRLF-delimited SSE frames on the servers events stream
The `/api/servers/events` reader in `useServers` searched for `\n\n` and consumed exactly two characters as the separator. SSE permits CRLF line endings, so a `event: change\r\n\r\n` frame stayed buffered forever: `isSseDataFrame()` was never reached and no background refresh fired, leaving the server list stale until a manual refresh or an LF-delimited event arrived. Frame boundaries are now located with a CRLF-aware matcher that reports the separator's real length, and frame lines split on either terminator. A bare CR is deliberately not accepted as a terminator: it cannot be distinguished from the first half of a CRLF straddling a chunk boundary, so a buffer ending in `…\r\n\r` would read as a completed frame and the `\n` arriving next would open a bogus one. Two regressions cover it, both verified failing against the old parser. Each uses a referentially stable `fetchFn` and holds the frame behind a promise released after the disk write — with an inline `fetchFn` the hook re-subscribes every render and that re-subscription's own mount refresh picks the edit up regardless, so the test would pass unfixed. Closes #2006 Signed-off-by: cliffhall <cliff@futurescale.com>
This commit is contained in:
parent
6bc355df63
commit
07a7b0beb5
2 changed files with 183 additions and 7 deletions
|
|
@ -969,6 +969,152 @@ describe("useServers", () => {
|
||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("parses CRLF-delimited SSE frames as change notifications (#2006)", async () => {
|
||||||
|
// SSE permits CRLF line endings, and this hook is a hand-rolled parser
|
||||||
|
// (EventSource can't send `x-mcp-remote-auth`). Matching only `\n\n` left
|
||||||
|
// a CRLF frame buffered forever, so an external mcp.json edit never
|
||||||
|
// reached the list.
|
||||||
|
writeFileSync(
|
||||||
|
h.configPath,
|
||||||
|
JSON.stringify({
|
||||||
|
mcpServers: { seed: { type: "stdio", command: "s" } },
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
|
||||||
|
// The frame is held back until the test has mutated the file, so the
|
||||||
|
// resulting list can only come from a refresh this frame triggered.
|
||||||
|
let releaseFrame: (() => void) | undefined;
|
||||||
|
const framed = new Promise<void>((r) => {
|
||||||
|
releaseFrame = r;
|
||||||
|
});
|
||||||
|
let reads = 0;
|
||||||
|
const encoder = new TextEncoder();
|
||||||
|
// Referentially stable: an inline arrow re-subscribes on every render, and
|
||||||
|
// the re-subscription's own mount refresh would pick the edit up on its
|
||||||
|
// own — passing whether or not the frame was ever parsed.
|
||||||
|
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;
|
||||||
|
if (reads === 1) {
|
||||||
|
await framed;
|
||||||
|
return {
|
||||||
|
done: false,
|
||||||
|
value: encoder.encode("event: change\r\n\r\n"),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
return { done: true, value: undefined };
|
||||||
|
},
|
||||||
|
}),
|
||||||
|
};
|
||||||
|
return { ok: true, body } as unknown as Response;
|
||||||
|
}
|
||||||
|
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"]);
|
||||||
|
|
||||||
|
writeFileSync(
|
||||||
|
h.configPath,
|
||||||
|
JSON.stringify({
|
||||||
|
mcpServers: { aftercrlf: { type: "stdio", command: "x" } },
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
releaseFrame?.();
|
||||||
|
|
||||||
|
await waitFor(
|
||||||
|
() => {
|
||||||
|
expect(result.current.servers.map((s) => s.id)).toEqual(["aftercrlf"]);
|
||||||
|
},
|
||||||
|
{ timeout: 3000 },
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("does not split a CRLF frame separator straddling two chunks (#2006)", async () => {
|
||||||
|
// The bare-CR terminator is deliberately not accepted: a chunk ending in
|
||||||
|
// `…\r\n\r` would otherwise read as a completed frame, and the `\n`
|
||||||
|
// arriving next would start a bogus one. Deliver exactly that split and
|
||||||
|
// assert the frame is recognized once — only after both halves land.
|
||||||
|
writeFileSync(
|
||||||
|
h.configPath,
|
||||||
|
JSON.stringify({
|
||||||
|
mcpServers: { seed: { type: "stdio", command: "s" } },
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
|
||||||
|
let releaseFirstHalf: (() => void) | undefined;
|
||||||
|
const firstHalf = new Promise<void>((r) => {
|
||||||
|
releaseFirstHalf = r;
|
||||||
|
});
|
||||||
|
let releaseSecondHalf: (() => void) | undefined;
|
||||||
|
const secondHalf = new Promise<void>((r) => {
|
||||||
|
releaseSecondHalf = r;
|
||||||
|
});
|
||||||
|
let listGets = 0;
|
||||||
|
let reads = 0;
|
||||||
|
const encoder = new TextEncoder();
|
||||||
|
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;
|
||||||
|
if (reads === 1) {
|
||||||
|
await firstHalf;
|
||||||
|
return {
|
||||||
|
done: false,
|
||||||
|
value: encoder.encode("event: change\r\n\r"),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
if (reads === 2) {
|
||||||
|
await secondHalf;
|
||||||
|
return { done: false, value: encoder.encode("\n") };
|
||||||
|
}
|
||||||
|
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(listGets).toBe(1);
|
||||||
|
|
||||||
|
// Half a separator is not a frame: no refresh yet.
|
||||||
|
releaseFirstHalf?.();
|
||||||
|
await waitFor(() => expect(reads).toBeGreaterThanOrEqual(2));
|
||||||
|
await new Promise((r) => setTimeout(r, 100));
|
||||||
|
expect(listGets).toBe(1);
|
||||||
|
|
||||||
|
writeFileSync(
|
||||||
|
h.configPath,
|
||||||
|
JSON.stringify({
|
||||||
|
mcpServers: { aftersplit: { type: "stdio", command: "x" } },
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
releaseSecondHalf?.();
|
||||||
|
|
||||||
|
await waitFor(
|
||||||
|
() => {
|
||||||
|
expect(result.current.servers.map((s) => s.id)).toEqual(["aftersplit"]);
|
||||||
|
},
|
||||||
|
{ timeout: 3000 },
|
||||||
|
);
|
||||||
|
expect(listGets).toBe(2);
|
||||||
|
});
|
||||||
|
|
||||||
it("ignores the backend's inert priming comment frame (#1858)", async () => {
|
it("ignores the backend's inert priming comment frame (#1858)", async () => {
|
||||||
// The backend opens every SSE stream with a `:` comment so a streaming
|
// The backend opens every SSE stream with a `:` comment so a streaming
|
||||||
// fetch() resolves on Firefox at all. That frame carries no event/data
|
// fetch() resolves on Firefox at all. That frame carries no event/data
|
||||||
|
|
|
||||||
|
|
@ -90,10 +90,38 @@ async function readErrorMessage(res: Response): Promise<string> {
|
||||||
*/
|
*/
|
||||||
function isSseDataFrame(frame: string): boolean {
|
function isSseDataFrame(frame: string): boolean {
|
||||||
return frame
|
return frame
|
||||||
.split("\n")
|
.split(SSE_LINE_SEPARATOR)
|
||||||
.some((line) => line.startsWith("event:") || line.startsWith("data:"));
|
.some((line) => line.startsWith("event:") || line.startsWith("data:"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* SSE line terminators. The spec permits CRLF, LF, and a bare CR; we accept
|
||||||
|
* the first two. A bare CR is deliberately excluded because it can't be told
|
||||||
|
* apart from the first half of a CRLF that straddles a chunk boundary — a
|
||||||
|
* buffer ending in `…\r\n\r` would look like a completed frame and the `\n`
|
||||||
|
* arriving in the next chunk would be mis-read as the start of another. No
|
||||||
|
* SSE producer in the wild emits bare-CR frames, so the ambiguity buys
|
||||||
|
* nothing.
|
||||||
|
*
|
||||||
|
* A frame ends at a blank line, i.e. two terminators back to back.
|
||||||
|
*/
|
||||||
|
const SSE_LINE_SEPARATOR = /\r\n|\n/;
|
||||||
|
const SSE_FRAME_SEPARATOR = /(?:\r\n|\n){2}/;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Locates the next frame boundary in `buffer`. Returns the frame's end offset
|
||||||
|
* and the length of the blank-line separator that terminated it (2–4 chars,
|
||||||
|
* depending on which terminators the server used), or `null` when the buffer
|
||||||
|
* holds no complete frame yet.
|
||||||
|
*/
|
||||||
|
function findSseFrameEnd(
|
||||||
|
buffer: string,
|
||||||
|
): { end: number; separatorLength: number } | null {
|
||||||
|
const match = SSE_FRAME_SEPARATOR.exec(buffer);
|
||||||
|
if (!match) return null;
|
||||||
|
return { end: match.index, separatorLength: match[0].length };
|
||||||
|
}
|
||||||
|
|
||||||
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;
|
||||||
|
|
@ -163,7 +191,9 @@ export function useServers(opts: UseServersOptions): UseServersResult {
|
||||||
const reader = res.body.getReader();
|
const reader = res.body.getReader();
|
||||||
const decoder = new TextDecoder();
|
const decoder = new TextDecoder();
|
||||||
let buffer = "";
|
let buffer = "";
|
||||||
// SSE frames are separated by a blank line (`\n\n`). We don't parse
|
// SSE frames are separated by a blank line — LF- or CRLF-terminated,
|
||||||
|
// both of which the spec permits and `findSseFrameEnd` accepts (#2006).
|
||||||
|
// We don't parse
|
||||||
// the event type or data — `refreshInternal()` re-fetches the
|
// the event type or data — `refreshInternal()` re-fetches the
|
||||||
// canonical state regardless — so we just count frames and fire a
|
// canonical state regardless — so we just count frames and fire a
|
||||||
// single background refresh per decode chunk. Two `change`
|
// single background refresh per decode chunk. Two `change`
|
||||||
|
|
@ -183,12 +213,12 @@ export function useServers(opts: UseServersOptions): UseServersResult {
|
||||||
if (done) break;
|
if (done) break;
|
||||||
buffer += decoder.decode(value, { stream: true });
|
buffer += decoder.decode(value, { stream: true });
|
||||||
let sawFrame = false;
|
let sawFrame = false;
|
||||||
let frameEnd = buffer.indexOf("\n\n");
|
let boundary = findSseFrameEnd(buffer);
|
||||||
while (frameEnd !== -1) {
|
while (boundary !== null) {
|
||||||
const frame = buffer.slice(0, frameEnd);
|
const frame = buffer.slice(0, boundary.end);
|
||||||
buffer = buffer.slice(frameEnd + 2);
|
buffer = buffer.slice(boundary.end + boundary.separatorLength);
|
||||||
if (isSseDataFrame(frame)) sawFrame = true;
|
if (isSseDataFrame(frame)) sawFrame = true;
|
||||||
frameEnd = buffer.indexOf("\n\n");
|
boundary = findSseFrameEnd(buffer);
|
||||||
}
|
}
|
||||||
if (sawFrame) void refreshInternal(true);
|
if (sawFrame) void refreshInternal(true);
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue