fix: address Copilot review round 4 on #2318
- fetchTracking: one non-throwing reporter fronts every stream update — the open (inside the fetch wrapper), each coalesced count (from a timer) and the close (off the discarded watcher). - rawWireRequest annotates a request timeout its transport's send rejects with, so the browser's relay timeout on a raw-wire call reads like every other. - RemoteClientTransport reports still-open streams closed on close(), since the backend watcher's own close cannot reach a channel that the disconnect has already torn down. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01APKSYHGxXBUpnR7ErFiLJn Signed-off-by: cliffhall <cliff@futurescale.com>
This commit is contained in:
parent
ec09628705
commit
abc4967298
6 changed files with 180 additions and 15 deletions
|
|
@ -934,6 +934,28 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => {
|
||||||
expect(updates.at(-1)!.stream.closedAt).toBeInstanceOf(Date);
|
expect(updates.at(-1)!.stream.closedAt).toBeInstanceOf(Date);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("returns the response even when the listener throws on the open report", async () => {
|
||||||
|
// The open is reported synchronously inside the fetch wrapper; a throw
|
||||||
|
// there must not turn a successful response into a rejected fetch.
|
||||||
|
const server = controlledStream();
|
||||||
|
const response = new Response(server.stream, {
|
||||||
|
headers: { "content-type": "text/event-stream" },
|
||||||
|
});
|
||||||
|
const fetcher = createFetchTracker((async () => response) as typeof fetch, {
|
||||||
|
updateStream: () => {
|
||||||
|
throw new Error("sink closed");
|
||||||
|
},
|
||||||
|
streamUpdateIntervalMs: 0,
|
||||||
|
});
|
||||||
|
await expect(
|
||||||
|
fetcher("https://example.com/mcp", { method: "GET" }),
|
||||||
|
).resolves.toBe(response);
|
||||||
|
// ...and a throw from a coalesced count does not escape the timer either.
|
||||||
|
server.push("data: a\n\n");
|
||||||
|
server.end();
|
||||||
|
await flush();
|
||||||
|
});
|
||||||
|
|
||||||
it("survives a stream-update listener that throws on the close", async () => {
|
it("survives a stream-update listener that throws on the close", async () => {
|
||||||
// The watcher's promise is discarded, so a throw from the final report
|
// The watcher's promise is discarded, so a throw from the final report
|
||||||
// would otherwise be an unhandled rejection — which fails the whole run.
|
// would otherwise be an unhandled rejection — which fails the whole run.
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,5 @@
|
||||||
import { describe, it, expect, vi } from "vitest";
|
import { describe, it, expect, vi } from "vitest";
|
||||||
import { SdkErrorCode } from "@modelcontextprotocol/client";
|
import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client";
|
||||||
import { InspectorClient } from "@inspector/core/mcp/inspectorClient.js";
|
import { InspectorClient } from "@inspector/core/mcp/inspectorClient.js";
|
||||||
import { ModernGetTaskResultSchema } from "@inspector/core/mcp/modernTaskSchemas.js";
|
import { ModernGetTaskResultSchema } from "@inspector/core/mcp/modernTaskSchemas.js";
|
||||||
|
|
||||||
|
|
@ -151,6 +151,46 @@ describe("InspectorClient raw-wire channel (#1631)", () => {
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("annotates a request timeout the transport's send rejects with (#2318)", async () => {
|
||||||
|
// The browser's remote transport awaits the response inside `send`, so
|
||||||
|
// its relay wait can expire before the local timer; it arrives here as
|
||||||
|
// the SDK's timeout shape and gets the same annotation.
|
||||||
|
const client = makeClient();
|
||||||
|
internals(client).transport = {
|
||||||
|
send: vi.fn().mockRejectedValue(
|
||||||
|
new SdkError(SdkErrorCode.RequestTimeout, "Request timed out", {
|
||||||
|
timeout: 60_000,
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
};
|
||||||
|
await expect(
|
||||||
|
internals(client).rawWireRequest(
|
||||||
|
"tasks/get",
|
||||||
|
{},
|
||||||
|
ModernGetTaskResultSchema,
|
||||||
|
),
|
||||||
|
).rejects.toMatchObject({
|
||||||
|
code: SdkErrorCode.RequestTimeout,
|
||||||
|
message: expect.stringMatching(
|
||||||
|
/^Request timed out after 1m00s \(tasks\/get\)\. /,
|
||||||
|
),
|
||||||
|
data: expect.objectContaining({ timeout: 60_000, method: "tasks/get" }),
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it("leaves a non-timeout send failure untouched", async () => {
|
||||||
|
const client = makeClient();
|
||||||
|
const boom = new Error("connection closed");
|
||||||
|
internals(client).transport = { send: vi.fn().mockRejectedValue(boom) };
|
||||||
|
await expect(
|
||||||
|
internals(client).rawWireRequest(
|
||||||
|
"tasks/get",
|
||||||
|
{},
|
||||||
|
ModernGetTaskResultSchema,
|
||||||
|
),
|
||||||
|
).rejects.toBe(boom);
|
||||||
|
});
|
||||||
|
|
||||||
it("rejects all pending requests on teardown", async () => {
|
it("rejects all pending requests on teardown", async () => {
|
||||||
const client = makeClient();
|
const client = makeClient();
|
||||||
internals(client).transport = {
|
internals(client).transport = {
|
||||||
|
|
|
||||||
|
|
@ -375,6 +375,63 @@ describe("RemoteClientTransport (focused branch coverage)", () => {
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("reports a still-open stream closed on close(), since the backend's own close cannot arrive (#2318)", async () => {
|
||||||
|
const onFetchStreamUpdate = vi.fn();
|
||||||
|
const { response, push } = createPushableEventStream();
|
||||||
|
const t = makeTransport(
|
||||||
|
{ events: () => response },
|
||||||
|
{ onFetchStreamUpdate, onFetchRequest: vi.fn() },
|
||||||
|
);
|
||||||
|
await t.start();
|
||||||
|
const entry = (id: string, method: string, contentType: string) =>
|
||||||
|
({
|
||||||
|
id,
|
||||||
|
method,
|
||||||
|
url: "http://x/mcp",
|
||||||
|
timestamp: new Date().toISOString(),
|
||||||
|
requestHeaders: {},
|
||||||
|
responseStatus: 200,
|
||||||
|
responseHeaders: { "Content-Type": contentType },
|
||||||
|
}) as never;
|
||||||
|
push({
|
||||||
|
type: "fetch_request",
|
||||||
|
data: entry("stream", "GET", "text/event-stream"),
|
||||||
|
});
|
||||||
|
push({
|
||||||
|
type: "fetch_request",
|
||||||
|
data: entry("post", "POST", "text/event-stream"),
|
||||||
|
});
|
||||||
|
push({
|
||||||
|
type: "fetch_stream_update",
|
||||||
|
data: { id: "stream", eventCount: 2 },
|
||||||
|
});
|
||||||
|
push({
|
||||||
|
type: "fetch_request",
|
||||||
|
data: entry("earlier", "GET", "text/event-stream"),
|
||||||
|
});
|
||||||
|
push({
|
||||||
|
type: "fetch_stream_update",
|
||||||
|
data: {
|
||||||
|
id: "earlier",
|
||||||
|
eventCount: 1,
|
||||||
|
closedAt: new Date().toISOString(),
|
||||||
|
},
|
||||||
|
});
|
||||||
|
await tick();
|
||||||
|
onFetchStreamUpdate.mockClear();
|
||||||
|
await t.close();
|
||||||
|
// Only the stream still open is reported closed, with its last count;
|
||||||
|
// the bounded POST and the already-closed stream are not.
|
||||||
|
expect(onFetchStreamUpdate).toHaveBeenCalledTimes(1);
|
||||||
|
expect(onFetchStreamUpdate).toHaveBeenCalledWith("stream", {
|
||||||
|
eventCount: 2,
|
||||||
|
closedAt: expect.any(Date),
|
||||||
|
});
|
||||||
|
// A second close reports nothing more.
|
||||||
|
await t.close();
|
||||||
|
expect(onFetchStreamUpdate).toHaveBeenCalledTimes(1);
|
||||||
|
});
|
||||||
|
|
||||||
it("drops fetch_stream_update events when no handler is registered", async () => {
|
it("drops fetch_stream_update events when no handler is registered", async () => {
|
||||||
const onFetchRequest = vi.fn();
|
const onFetchRequest = vi.fn();
|
||||||
const t = makeTransport(
|
const t = makeTransport(
|
||||||
|
|
|
||||||
|
|
@ -310,8 +310,21 @@ function watchLongLivedStream(
|
||||||
framing: LongLivedStreamFraming,
|
framing: LongLivedStreamFraming,
|
||||||
callbacks: FetchTrackingCallbacks,
|
callbacks: FetchTrackingCallbacks,
|
||||||
): void {
|
): void {
|
||||||
const report = callbacks.updateStream;
|
const listener = callbacks.updateStream;
|
||||||
if (!report) return;
|
if (!listener) return;
|
||||||
|
// Every report goes through here: the listener is consumer code (a relay
|
||||||
|
// sink that may already be closed, a state store), and it runs from three
|
||||||
|
// places none of which may throw outward — synchronously inside the fetch
|
||||||
|
// wrapper (the open), from a timer (a coalesced count), and off the
|
||||||
|
// discarded watcher promise (the close). A throw is the listener's
|
||||||
|
// problem; the stream's bookkeeping carries on.
|
||||||
|
const report = (state: FetchStreamState): void => {
|
||||||
|
try {
|
||||||
|
listener(id, state);
|
||||||
|
} catch {
|
||||||
|
// See above.
|
||||||
|
}
|
||||||
|
};
|
||||||
const intervalMs =
|
const intervalMs =
|
||||||
callbacks.streamUpdateIntervalMs ?? DEFAULT_STREAM_UPDATE_INTERVAL_MS;
|
callbacks.streamUpdateIntervalMs ?? DEFAULT_STREAM_UPDATE_INTERVAL_MS;
|
||||||
let body: ReadableStream<Uint8Array> | null;
|
let body: ReadableStream<Uint8Array> | null;
|
||||||
|
|
@ -333,19 +346,19 @@ function watchLongLivedStream(
|
||||||
let reportTimer: ReturnType<typeof setTimeout> | null = null;
|
let reportTimer: ReturnType<typeof setTimeout> | null = null;
|
||||||
const reportCount = (): void => {
|
const reportCount = (): void => {
|
||||||
if (intervalMs <= 0) {
|
if (intervalMs <= 0) {
|
||||||
report(id, { eventCount });
|
report({ eventCount });
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
if (reportTimer !== null) return;
|
if (reportTimer !== null) return;
|
||||||
reportTimer = setTimeout(() => {
|
reportTimer = setTimeout(() => {
|
||||||
reportTimer = null;
|
reportTimer = null;
|
||||||
report(id, { eventCount });
|
report({ eventCount });
|
||||||
}, intervalMs);
|
}, intervalMs);
|
||||||
};
|
};
|
||||||
// Announce the stream as open before a byte arrives. A stream that never
|
// Announce the stream as open before a byte arrives. A stream that never
|
||||||
// delivers anything is the #2187 tell, and it would otherwise be the one
|
// delivers anything is the #2187 tell, and it would otherwise be the one
|
||||||
// stream with no state to show — the count only moves on an event.
|
// stream with no state to show — the count only moves on an event.
|
||||||
report(id, { eventCount });
|
report({ eventCount });
|
||||||
|
|
||||||
const consumeLine = (line: string): void => {
|
const consumeLine = (line: string): void => {
|
||||||
if (framing === "ndjson") {
|
if (framing === "ndjson") {
|
||||||
|
|
@ -410,17 +423,12 @@ function watchLongLivedStream(
|
||||||
clearTimeout(reportTimer);
|
clearTimeout(reportTimer);
|
||||||
reportTimer = null;
|
reportTimer = null;
|
||||||
}
|
}
|
||||||
try {
|
report({ eventCount, closedAt: new Date() });
|
||||||
report(id, { eventCount, closedAt: new Date() });
|
|
||||||
} catch {
|
|
||||||
// A listener that throws must not become an unhandled rejection off a
|
|
||||||
// promise nothing awaits; the stream is closed either way.
|
|
||||||
}
|
|
||||||
};
|
};
|
||||||
// Deliberately not awaited: the fetch wrapper has to return the response
|
// Deliberately not awaited: the fetch wrapper has to return the response
|
||||||
// now, and this watcher lives as long as the stream does. It owns its own
|
// now, and this watcher lives as long as the stream does. It owns its own
|
||||||
// failures — the read loop and the final report are each wrapped in a
|
// failures — the read loop is wrapped in a `try`, and every report goes
|
||||||
// `try` — so the promise cannot reject.
|
// through the non-throwing `report` above — so the promise cannot reject.
|
||||||
void pump();
|
void pump();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -2980,7 +2980,17 @@ export class InspectorClient extends InspectorClientEventTarget {
|
||||||
clearTimeout(pending.timer);
|
clearTimeout(pending.timer);
|
||||||
this.pendingRawWireRequests.delete(id);
|
this.pendingRawWireRequests.delete(id);
|
||||||
}
|
}
|
||||||
reject(err instanceof Error ? err : new Error(String(err)));
|
// The browser's remote transport awaits the response inside `send`,
|
||||||
|
// so its relay wait can expire here first, as the SDK's timeout
|
||||||
|
// shape; annotate it exactly as the local timer above does (a
|
||||||
|
// non-timeout error passes through untouched).
|
||||||
|
reject(
|
||||||
|
annotateRequestTimeout(
|
||||||
|
err instanceof Error ? err : new Error(String(err)),
|
||||||
|
method,
|
||||||
|
this.getConnectionDiagnostics(),
|
||||||
|
),
|
||||||
|
);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
return resultSchema.parse(raw);
|
return resultSchema.parse(raw);
|
||||||
|
|
|
||||||
|
|
@ -12,6 +12,7 @@ import type {
|
||||||
MessageExtraInfo,
|
MessageExtraInfo,
|
||||||
} from "@modelcontextprotocol/client";
|
} from "@modelcontextprotocol/client";
|
||||||
import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client";
|
import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client";
|
||||||
|
import { findHeader, isLongLivedStreamResponse } from "../fetchTracking.js";
|
||||||
import type { InspectorServerSettings, StderrLogEntry } from "../types.js";
|
import type { InspectorServerSettings, StderrLogEntry } from "../types.js";
|
||||||
import type { FetchRequestEntryBase, FetchStreamState } from "../types.js";
|
import type { FetchRequestEntryBase, FetchStreamState } from "../types.js";
|
||||||
import type {
|
import type {
|
||||||
|
|
@ -265,6 +266,14 @@ export class RemoteClientTransport implements Transport {
|
||||||
private eventStreamConsumeTask: Promise<void> | null = null;
|
private eventStreamConsumeTask: Promise<void> | null = null;
|
||||||
private restartingEventStream = false;
|
private restartingEventStream = false;
|
||||||
private closed = false;
|
private closed = false;
|
||||||
|
/**
|
||||||
|
* Long-lived streams the backend has reported open, by fetch entry id, with
|
||||||
|
* the last event count seen for each. `close()` reports them closed
|
||||||
|
* itself: it aborts the event channel before asking the backend to
|
||||||
|
* disconnect, so the backend watcher's own close report lands on an
|
||||||
|
* unconsumed queue and is lost with the session (#2318).
|
||||||
|
*/
|
||||||
|
private openStreams = new Map<string, number>();
|
||||||
private readonly sseResponseWaits = new Map<
|
private readonly sseResponseWaits = new Map<
|
||||||
string | number,
|
string | number,
|
||||||
SseResponseWait
|
SseResponseWait
|
||||||
|
|
@ -548,6 +557,14 @@ export class RemoteClientTransport implements Transport {
|
||||||
this.options.onFetchRequest
|
this.options.onFetchRequest
|
||||||
) {
|
) {
|
||||||
const entry = parsed.data;
|
const entry = parsed.data;
|
||||||
|
if (
|
||||||
|
isLongLivedStreamResponse(
|
||||||
|
entry.method,
|
||||||
|
findHeader(entry.responseHeaders, "content-type"),
|
||||||
|
)
|
||||||
|
) {
|
||||||
|
this.openStreams.set(entry.id, 0);
|
||||||
|
}
|
||||||
this.options.onFetchRequest({
|
this.options.onFetchRequest({
|
||||||
...entry,
|
...entry,
|
||||||
timestamp:
|
timestamp:
|
||||||
|
|
@ -568,6 +585,10 @@ export class RemoteClientTransport implements Transport {
|
||||||
this.options.onFetchStreamUpdate
|
this.options.onFetchStreamUpdate
|
||||||
) {
|
) {
|
||||||
const { id, eventCount, closedAt } = parsed.data;
|
const { id, eventCount, closedAt } = parsed.data;
|
||||||
|
if (closedAt !== undefined) this.openStreams.delete(id);
|
||||||
|
else if (this.openStreams.has(id)) {
|
||||||
|
this.openStreams.set(id, eventCount);
|
||||||
|
}
|
||||||
this.options.onFetchStreamUpdate(id, {
|
this.options.onFetchStreamUpdate(id, {
|
||||||
eventCount,
|
eventCount,
|
||||||
...(closedAt !== undefined && { closedAt: new Date(closedAt) }),
|
...(closedAt !== undefined && { closedAt: new Date(closedAt) }),
|
||||||
|
|
@ -856,6 +877,13 @@ export class RemoteClientTransport implements Transport {
|
||||||
this.closed = true;
|
this.closed = true;
|
||||||
this.cancelAllSseWaits(new Error("Transport closed"));
|
this.cancelAllSseWaits(new Error("Transport closed"));
|
||||||
await this.stopEventStream();
|
await this.stopEventStream();
|
||||||
|
// The disconnect below ends every stream the backend still holds open,
|
||||||
|
// and its close reports cannot reach us any more (see `openStreams`).
|
||||||
|
const closedAt = new Date();
|
||||||
|
for (const [id, eventCount] of this.openStreams) {
|
||||||
|
this.options.onFetchStreamUpdate?.(id, { eventCount, closedAt });
|
||||||
|
}
|
||||||
|
this.openStreams.clear();
|
||||||
|
|
||||||
if (this._sessionId) {
|
if (this._sessionId) {
|
||||||
try {
|
try {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue