From ec09628705595f5bd817f6103b5c68b2f2f723dd Mon Sep 17 00:00:00 2001 From: cliffhall Date: Sun, 13 Sep 2026 22:53:08 -0400 Subject: [PATCH] fix: address Copilot review round 3 on #2318 - fetchTracking: the close report is guarded so a throwing listener cannot become an unhandled rejection off the discarded watcher; a bare CR that is the last byte of the stream terminates its event; a moving event count is coalesced to one report per interval (default 250ms, 0 in tests), with the open and the close still immediate. - MessageTrackingTransport rolls back a request whose signal was already aborted before the send began, and suppresses the rollback only for an abort that lands during the send. - useTickingClock takes an `enabled` flag; Connection Info installs no interval when there is no activity section. - The diagnostics-callbacks test justifies its private-seam cast. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01APKSYHGxXBUpnR7ErFiLJn Signed-off-by: cliffhall --- .../ConnectionInfoContent.tsx | 9 +- clients/web/src/hooks/useTickingClock.test.ts | 22 +++++ clients/web/src/hooks/useTickingClock.ts | 12 ++- .../src/test/core/mcp/fetchTracking.test.ts | 93 ++++++++++++++++++- ...pectorClient-diagnostics-callbacks.test.ts | 6 ++ .../core/mcp/messageTrackingTransport.test.ts | 30 +++++- core/mcp/fetchTracking.ts | 58 ++++++++++-- core/mcp/messageTrackingTransport.ts | 16 +++- 8 files changed, 221 insertions(+), 25 deletions(-) diff --git a/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.tsx b/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.tsx index 25290f67..6c604880 100644 --- a/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.tsx +++ b/clients/web/src/components/groups/ConnectionInfoContent/ConnectionInfoContent.tsx @@ -337,8 +337,13 @@ export function ConnectionInfoContent({ : SERVER_INFO_NOT_REPORTED_LABEL; // The activity rows' clock: the snapshot's own on first paint (pure), then - // the wall clock once a second so "sent 5s ago" keeps counting. - const now = useTickingClock(diagnostics?.capturedAt ?? 0); + // the wall clock once a second so "sent 5s ago" keeps counting. No timer + // at all when there is no activity section to tick. + const now = useTickingClock( + diagnostics?.capturedAt ?? 0, + 1000, + diagnostics !== undefined, + ); const serverCaps = getServerCapabilityEntries(capabilities, protocolEra); const clientCaps = getCapabilityEntries( diff --git a/clients/web/src/hooks/useTickingClock.test.ts b/clients/web/src/hooks/useTickingClock.test.ts index 9242e1cb..dc459353 100644 --- a/clients/web/src/hooks/useTickingClock.test.ts +++ b/clients/web/src/hooks/useTickingClock.test.ts @@ -47,6 +47,28 @@ describe("useTickingClock", () => { expect(result.current).toBe(5_002_500); }); + it("installs no timer and returns the seed when disabled", () => { + vi.useFakeTimers(); + vi.setSystemTime(5_000_000); + const setSpy = vi.spyOn(globalThis, "setInterval"); + const { result, rerender } = renderHook( + ({ enabled }) => useTickingClock(1_000_000, 1000, enabled), + { initialProps: { enabled: false } }, + ); + expect(setSpy).not.toHaveBeenCalled(); + act(() => { + vi.advanceTimersByTime(3000); + }); + expect(result.current).toBe(1_000_000); + // Enabling later starts the clock. + rerender({ enabled: true }); + act(() => { + vi.advanceTimersByTime(1000); + }); + expect(result.current).toBe(5_004_000); + setSpy.mockRestore(); + }); + it("stops ticking on unmount", () => { vi.useFakeTimers(); const clearSpy = vi.spyOn(globalThis, "clearInterval"); diff --git a/clients/web/src/hooks/useTickingClock.ts b/clients/web/src/hooks/useTickingClock.ts index d82ac3f7..4dc47123 100644 --- a/clients/web/src/hooks/useTickingClock.ts +++ b/clients/web/src/hooks/useTickingClock.ts @@ -16,12 +16,20 @@ import { useEffect, useState } from "react"; * A timer is exactly the "synchronize with an external system" case an effect * exists for; the state is set only from the interval callback, never in the * effect body, so the first paint is not followed by a corrective re-render. + * `enabled: false` installs no timer at all and returns the seed. */ -export function useTickingClock(seed: number, intervalMs = 1000): number { +export function useTickingClock( + seed: number, + intervalMs = 1000, + enabled = true, +): number { const [now, setNow] = useState(seed); useEffect(() => { + // No timer for a consumer with nothing to tick — a component that + // renders no durations must not pay for a repeating interval. + if (!enabled) return; const id = setInterval(() => setNow(Date.now()), intervalMs); return () => clearInterval(id); - }, [intervalMs]); + }, [intervalMs, enabled]); return Math.max(now, seed); } diff --git a/clients/web/src/test/core/mcp/fetchTracking.test.ts b/clients/web/src/test/core/mcp/fetchTracking.test.ts index 3d834caa..fef4764c 100644 --- a/clients/web/src/test/core/mcp/fetchTracking.test.ts +++ b/clients/web/src/test/core/mcp/fetchTracking.test.ts @@ -748,6 +748,7 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => { { trackRequest: (entry) => tracked.push(entry), updateStream: (id, state) => updates.push({ id, stream: state }), + streamUpdateIntervalMs: 0, }, ); const response = await fetcher("https://example.com/mcp", { @@ -800,7 +801,10 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => { new Response(server.stream, { headers: { "content-type": "application/x-ndjson" }, })) as typeof fetch, - { updateStream: (id, state) => updates.push({ id, stream: state }) }, + { + updateStream: (id, state) => updates.push({ id, stream: state }), + streamUpdateIntervalMs: 0, + }, ); await fetcher("https://example.com/mcp", { method: "GET" }); server.push('{"jsonrpc":"2.0","method":"a"}\n\n'); @@ -846,7 +850,10 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => { new Response(server.stream, { headers: { "content-type": "Text/Event-Stream" }, })) as typeof fetch, - { updateStream: (id, state) => updates.push({ id, state }) }, + { + updateStream: (id, state) => updates.push({ id, state }), + streamUpdateIntervalMs: 0, + }, ); await fetcher("https://example.com/mcp", { method: "get" }); expect(updates).toHaveLength(1); @@ -867,6 +874,7 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => { trackRequest: (entry) => tracked.push(entry), updateResponseBody: (_id, body) => bodies.push(body), updateStream: (id, state) => updates.push({ id, state }), + streamUpdateIntervalMs: 0, }, ); await fetcher(new Request("https://example.com/mcp", { method: "POST" })); @@ -876,6 +884,80 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => { expect(updates).toEqual([]); }); + it("coalesces a moving count to one report per interval, and reports the close at once", async () => { + vi.useFakeTimers(); + try { + const server = controlledStream(); + const updates: Array<{ id: string; stream: FetchStreamState }> = []; + const fetcher = createFetchTracker( + (async () => + new Response(server.stream, { + headers: { "content-type": "text/event-stream" }, + })) as typeof fetch, + { + updateStream: (id, state) => updates.push({ id, stream: state }), + streamUpdateIntervalMs: 100, + }, + ); + await fetcher("https://example.com/mcp", { method: "GET" }); + // The open is immediate. + expect(updates.map((u) => u.stream.eventCount)).toEqual([0]); + server.push("data: a\n\ndata: b\n\ndata: c\n\n"); + await vi.advanceTimersByTimeAsync(0); + // Three events, no report yet — they are inside one window. + expect(updates.map((u) => u.stream.eventCount)).toEqual([0]); + await vi.advanceTimersByTimeAsync(100); + expect(updates.map((u) => u.stream.eventCount)).toEqual([0, 3]); + // A close cancels a pending window and reports immediately. + server.push("data: d\n\n"); + server.end(); + await vi.advanceTimersByTimeAsync(0); + expect(updates.map((u) => u.stream.eventCount)).toEqual([0, 3, 4]); + expect(updates.at(-1)!.stream.closedAt).toBeInstanceOf(Date); + await vi.advanceTimersByTimeAsync(200); + expect(updates).toHaveLength(3); + } finally { + vi.useRealTimers(); + } + }); + + it("counts an event whose bare-CR terminator is the last byte of the stream", async () => { + const server = controlledStream(); + const { updates } = await trackStream(server.stream); + server.push("data: x\r\r"); + await flush(); + // Held while the stream is open: the CR could be half of a CRLF. + expect(updates.map((u) => u.stream.eventCount)).toEqual([0]); + server.end(); + await flush(); + expect(updates.at(-1)!.stream).toMatchObject({ eventCount: 1 }); + expect(updates.at(-1)!.stream.closedAt).toBeInstanceOf(Date); + }); + + 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 + // would otherwise be an unhandled rejection — which fails the whole run. + const server = controlledStream(); + let reports = 0; + const fetcher = createFetchTracker( + (async () => + new Response(server.stream, { + headers: { "content-type": "text/event-stream" }, + })) as typeof fetch, + { + updateStream: (_id, state) => { + reports += 1; + if (state.closedAt) throw new Error("listener exploded"); + }, + streamUpdateIntervalMs: 0, + }, + ); + await fetcher("https://example.com/mcp", { method: "GET" }); + server.end(); + await flush(); + expect(reports).toBe(2); + }); + it("reports the close when the server ends the stream", async () => { const server = controlledStream(); const { updates, id } = await trackStream(server.stream); @@ -932,7 +1014,10 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => { new Response("data: x\n\n", { headers: { "content-type": "text/event-stream" }, })) as typeof fetch, - { updateStream: (id, state) => updates.push({ id, state }) }, + { + updateStream: (id, state) => updates.push({ id, state }), + streamUpdateIntervalMs: 0, + }, ); // A POST SSE reply is bounded and goes down the body-capture path. await fetcher("https://example.com/mcp", { method: "POST" }); @@ -950,6 +1035,7 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => { const updates: unknown[] = []; const fetcher = createFetchTracker((async () => response) as typeof fetch, { updateStream: (id, state) => updates.push({ id, state }), + streamUpdateIntervalMs: 0, }); await expect( fetcher("https://example.com/mcp", { method: "GET" }), @@ -971,6 +1057,7 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => { const updates: unknown[] = []; const fetcher = createFetchTracker((async () => response) as typeof fetch, { updateStream: (id, state) => updates.push({ id, state }), + streamUpdateIntervalMs: 0, }); await fetcher("https://example.com/mcp", { method: "GET" }); await flush(); diff --git a/clients/web/src/test/core/mcp/inspectorClient-diagnostics-callbacks.test.ts b/clients/web/src/test/core/mcp/inspectorClient-diagnostics-callbacks.test.ts index cae7981b..35adb3c6 100644 --- a/clients/web/src/test/core/mcp/inspectorClient-diagnostics-callbacks.test.ts +++ b/clients/web/src/test/core/mcp/inspectorClient-diagnostics-callbacks.test.ts @@ -23,6 +23,12 @@ describe("InspectorClient diagnostics tracking callbacks", () => { // Never connects; the callbacks are exercised directly. { environment: { transport: () => ({}) as never } }, ); + // `createMessageTrackingCallbacks` is private: it is the seam the client + // hands its transport, and driving it directly is what lets this test + // exercise the bookkeeping without a connection. The double cast is the + // documented test-only exception for reaching a private member — the + // same one `inspectorClient-raw-wire.test.ts` uses — and `Internals` + // names exactly the one method it reaches. const callbacks = ( client as unknown as Internals ).createMessageTrackingCallbacks(); diff --git a/clients/web/src/test/core/mcp/messageTrackingTransport.test.ts b/clients/web/src/test/core/mcp/messageTrackingTransport.test.ts index 4265186f..ea69da47 100644 --- a/clients/web/src/test/core/mcp/messageTrackingTransport.test.ts +++ b/clients/web/src/test/core/mcp/messageTrackingTransport.test.ts @@ -48,18 +48,38 @@ describe("MessageTrackingTransport.send", () => { await expect(failing.send(request)).rejects.toBe(boom); expect(callbacks.trackRequest).toHaveBeenCalledWith(request, "client"); expect(trackSendFailure).toHaveBeenCalledWith(request, boom); - // A caller's own abort is a cancellation, not a send failure: the + // A signal already aborted before the send began: the frame never went + // out, so this is a failed send and rolls back. + trackSendFailure.mockClear(); + const preAborted = new AbortController(); + preAborted.abort(); + const preAbortedRequest = { + jsonrpc: "2.0", + id: 2, + method: "tools/call", + } as const; + await expect( + failing.send(preAbortedRequest, { requestSignal: preAborted.signal }), + ).rejects.toBe(boom); + expect(trackSendFailure).toHaveBeenCalledWith(preAbortedRequest, boom); + // A signal aborted while the send is in flight is a cancellation: the // request may well have reached the server, and stays unanswered. trackSendFailure.mockClear(); - const controller = new AbortController(); - controller.abort(); + const midSend = new AbortController(); + base.send = async () => { + midSend.abort(); + throw boom; + }; await expect( failing.send( - { jsonrpc: "2.0", id: 2, method: "tools/call" }, - { requestSignal: controller.signal }, + { jsonrpc: "2.0", id: 3, method: "tools/call" }, + { requestSignal: midSend.signal }, ), ).rejects.toBe(boom); expect(trackSendFailure).not.toHaveBeenCalled(); + base.send = async () => { + throw boom; + }; // Only requests roll back: a failed notification has no pending entry. trackSendFailure.mockClear(); await expect( diff --git a/core/mcp/fetchTracking.ts b/core/mcp/fetchTracking.ts index fb3f117f..19b00d23 100644 --- a/core/mcp/fetchTracking.ts +++ b/core/mcp/fetchTracking.ts @@ -250,8 +250,21 @@ export interface FetchTrackingCallbacks { * server serializes requests behind it (#2318). */ updateStream?: (id: string, stream: FetchStreamState) => void; + /** + * How often, at most, `updateStream` reports a moving event count, in ms. + * The first report (open, zero events) and the close are immediate; counts + * in between are coalesced so a server pushing hundreds of notifications a + * second costs one update per window rather than one per event — each + * update re-publishes the whole fetch log and, on the web client, crosses + * the backend-to-browser event stream. Defaults to + * `DEFAULT_STREAM_UPDATE_INTERVAL_MS`; `0` reports every event, for tests. + */ + streamUpdateIntervalMs?: number; } +/** See `FetchTrackingCallbacks.streamUpdateIntervalMs`. */ +export const DEFAULT_STREAM_UPDATE_INTERVAL_MS = 250; + /** * The framing a long-lived stream's events arrive in — the two content types * `isLongLivedStreamResponse` accepts. SSE dispatches on a blank line; @@ -297,8 +310,10 @@ function watchLongLivedStream( framing: LongLivedStreamFraming, callbacks: FetchTrackingCallbacks, ): void { - const update = callbacks.updateStream; - if (!update) return; + const report = callbacks.updateStream; + if (!report) return; + const intervalMs = + callbacks.streamUpdateIntervalMs ?? DEFAULT_STREAM_UPDATE_INTERVAL_MS; let body: ReadableStream | null; try { body = response.clone().body; @@ -313,23 +328,37 @@ function watchLongLivedStream( let eventCount = 0; let pending = ""; let blockHasData = false; + // A moving count is reported at most once per interval (see + // `streamUpdateIntervalMs`); the open and the close go out at once. + let reportTimer: ReturnType | null = null; + const reportCount = (): void => { + if (intervalMs <= 0) { + report(id, { eventCount }); + return; + } + if (reportTimer !== null) return; + reportTimer = setTimeout(() => { + reportTimer = null; + report(id, { eventCount }); + }, intervalMs); + }; // 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 // stream with no state to show — the count only moves on an event. - update(id, { eventCount }); + report(id, { eventCount }); const consumeLine = (line: string): void => { if (framing === "ndjson") { if (line.trim() === "") return; eventCount += 1; - update(id, { eventCount }); + reportCount(); return; } if (line === "") { if (blockHasData) { eventCount += 1; blockHasData = false; - update(id, { eventCount }); + reportCount(); } return; } @@ -368,17 +397,30 @@ function watchLongLivedStream( pending += decoder.decode(value, { stream: true }); consumeCompleteLines(); } + // A CR held back as a possible half of a CRLF is, at end-of-stream, a + // whole line ending — `data: x\r\r` is a complete event. Anything after + // it is a partial block and is dropped, as above. + if (pending.endsWith("\r")) consumeLine(pending.slice(0, -1)); } catch { // An errored or aborted stream has still ended — fall through to report // the close; the entry's `error` field is for a fetch that never got a // response, which this one did. } - update(id, { eventCount, closedAt: new Date() }); + if (reportTimer !== null) { + clearTimeout(reportTimer); + reportTimer = null; + } + try { + 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 // now, and this watcher lives as long as the stream does. It owns its own - // failures — the read loop is wrapped in a `try` that treats an error as a - // close — so the promise cannot reject. + // failures — the read loop and the final report are each wrapped in a + // `try` — so the promise cannot reject. void pump(); } diff --git a/core/mcp/messageTrackingTransport.ts b/core/mcp/messageTrackingTransport.ts index 4a94df37..c7645dcc 100644 --- a/core/mcp/messageTrackingTransport.ts +++ b/core/mcp/messageTrackingTransport.ts @@ -30,10 +30,11 @@ export interface MessageTrackingCallbacks { * `trackRequest` already recorded it, so a consumer keeping "requests still * awaiting a response" can roll that entry back — a request that never went * out is not unanswered (#2318). Not fired when the caller aborted the - * request through `requestSignal`: the modern transport aborts an in-flight - * request's own stream to cancel it (and the SDK does the same on a - * per-request timeout), and a request that was sent and then given up on - * is exactly the unanswered kind. + * request through `requestSignal` *during* the send: the modern transport + * aborts an in-flight request's own stream to cancel it (and the SDK does + * the same on a per-request timeout), and a request that was sent and then + * given up on is exactly the unanswered kind. A signal already aborted + * before the send began is a failed send like any other. */ trackSendFailure?: (message: JSONRPCRequest, error: unknown) => void; } @@ -109,8 +110,13 @@ export class MessageTrackingTransport implements Transport { } else if ("method" in message) { const request = message as JSONRPCRequest; this.callbacks.trackRequest?.(request, "client"); + // A signal already aborted here means the frame will not go out at + // all — that is a failed send. One that becomes aborted while the + // send is in flight is a cancellation of a request that may well + // have reached the server, which stays unanswered. + const abortedBeforeSend = options?.requestSignal?.aborted === true; return this.baseTransport.send(message, options).catch((err) => { - if (!options?.requestSignal?.aborted) { + if (abortedBeforeSend || !options?.requestSignal?.aborted) { this.callbacks.trackSendFailure?.(request, err); } throw err;