b2dfd4099a
Copilot review round 3 (no new code findings, three stale-comment ones): - `openModernListenStreamOnConnect`'s generation guard was justified by a `connect` listener racing it — which the previous commit made impossible by dispatching `connect` after this call. The guard is still right, for reasons that survive the reorder: `statusChange` has already fired, and a concurrent `subscribeToResource` or `disconnect()` (whose `resetSubscriptionStream` bumps the generation) can supersede it. - `reconcileModernStreamStateAfterFailedRefresh`'s JSDoc still described its branches as "nothing subscribed" vs "URIs live"; they are the empty vs non-empty *filter* now, which differ exactly on the listChanged-only path. - The superseded-failure test carried the same stale `connect` rationale. Comments only; no behavior change. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HDdo1rNRVQnVRRdsqSrvG3
991 lines
40 KiB
TypeScript
991 lines
40 KiB
TypeScript
import { describe, it, expect, afterEach, vi } from "vitest";
|
|
import { InspectorClient } from "@inspector/core/mcp/inspectorClient.js";
|
|
import { createTransportNode } from "@inspector/core/mcp/node/transport.js";
|
|
import { eraToVersionNegotiation } from "@inspector/core/mcp/types.js";
|
|
import type { McpSubscription } from "@modelcontextprotocol/client";
|
|
import {
|
|
createTestServerHttp,
|
|
type TestServerHttp,
|
|
createTestServerInfo,
|
|
createNumberedResources,
|
|
createEchoTool,
|
|
} from "@modelcontextprotocol/inspector-test-server";
|
|
import type { ServerConfig } from "@modelcontextprotocol/inspector-test-server";
|
|
import type { MessageEntry } from "@inspector/core/mcp/types.js";
|
|
|
|
/**
|
|
* Live coverage of the resource-subscription era fork (#1630). On the legacy era
|
|
* each subscription is a `resources/subscribe` request; on the modern
|
|
* (2026-07-28) era subscriptions are a filter over one `subscriptions/listen`
|
|
* stream. Both are exercised against a real server over a real transport.
|
|
*/
|
|
describe("resource subscriptions era fork (#1630)", () => {
|
|
let client: InspectorClient | null = null;
|
|
let server: TestServerHttp | null = null;
|
|
|
|
const RESOURCE_URI = "test://resource_0";
|
|
/**
|
|
* Every list-change opt-in off. The listen filter is config ∩ capability and
|
|
* the SDK server advertises `listChanged` for every list it registers, so this
|
|
* is what leaves the filter empty when nothing is subscribed — i.e. the
|
|
* pre-#1920 world these lifecycle tests were written against.
|
|
*/
|
|
const NO_LIST_CHANGED = { tools: false, resources: false, prompts: false };
|
|
const RESOURCE_URI_2 = "test://resource_1";
|
|
|
|
afterEach(async () => {
|
|
if (client) {
|
|
try {
|
|
await client.disconnect();
|
|
} catch {
|
|
// ignore
|
|
}
|
|
client = null;
|
|
}
|
|
if (server) {
|
|
try {
|
|
await server.stop();
|
|
} catch {
|
|
// ignore
|
|
}
|
|
server = null;
|
|
}
|
|
});
|
|
|
|
async function startServer(
|
|
modern: ServerConfig["modern"] | undefined,
|
|
): Promise<TestServerHttp> {
|
|
const started = createTestServerHttp({
|
|
serverInfo: createTestServerInfo("subscriptions-era-test", "1.0.0"),
|
|
resources: createNumberedResources(2),
|
|
listChanged: { resources: true },
|
|
subscriptions: true,
|
|
...(modern ? { modern } : {}),
|
|
});
|
|
await started.start();
|
|
server = started;
|
|
return started;
|
|
}
|
|
|
|
/**
|
|
* @param listChangedNotifications the Inspector-side opt-ins. The SDK server
|
|
* advertises `listChanged` for every list it registers, so this — not the
|
|
* server config — is how a test gets an *empty* listen filter: the filter is
|
|
* config ∩ capability, and with every opt-in off only a subscribed URI can
|
|
* open the stream (#1920).
|
|
*/
|
|
async function connect(
|
|
url: string,
|
|
era: "legacy" | "modern",
|
|
listChangedNotifications?: {
|
|
tools?: boolean;
|
|
resources?: boolean;
|
|
prompts?: boolean;
|
|
},
|
|
): Promise<{ connected: InspectorClient; messages: MessageEntry[] }> {
|
|
const connected = new InspectorClient(
|
|
{ type: "streamable-http", url },
|
|
{
|
|
environment: { transport: createTransportNode },
|
|
versionNegotiation: eraToVersionNegotiation(era),
|
|
...(listChangedNotifications ? { listChangedNotifications } : {}),
|
|
},
|
|
);
|
|
const messages: MessageEntry[] = [];
|
|
connected.addEventListener("message", (event) => {
|
|
messages.push(event.detail);
|
|
});
|
|
await connected.connect();
|
|
client = connected;
|
|
return { connected, messages };
|
|
}
|
|
|
|
function methodsSent(messages: MessageEntry[]): string[] {
|
|
return messages
|
|
.filter((m) => m.direction === "request")
|
|
.map((m) => ("method" in m.message ? m.message.method : ""))
|
|
.filter(Boolean);
|
|
}
|
|
|
|
/** Params of the last `subscriptions/listen` request captured in the log. */
|
|
function lastListenFilter(
|
|
messages: MessageEntry[],
|
|
): Record<string, unknown> | undefined {
|
|
const listen = messages
|
|
.filter(
|
|
(m) =>
|
|
m.direction === "request" &&
|
|
"method" in m.message &&
|
|
m.message.method === "subscriptions/listen",
|
|
)
|
|
.at(-1);
|
|
if (!listen || !("params" in listen.message)) return undefined;
|
|
const params = listen.message.params as { notifications?: unknown };
|
|
return params.notifications as Record<string, unknown> | undefined;
|
|
}
|
|
|
|
/**
|
|
* The private members these tests drive directly. `InspectorClient` declares
|
|
* every one of them, so the shape below is a faithful mirror rather than a
|
|
* reinterpretation — the double cast is only to reach past `private`, which no
|
|
* single `as` can do (the two types share no public overlap). It is confined to
|
|
* `internals()` so there is one such cast in the file, and it is unavoidable:
|
|
* these are lifecycle branches no public API can reach against a healthy server
|
|
* (there is no way to ask for a *remote* stream close, or for a `listen()` that
|
|
* fails).
|
|
*/
|
|
interface StreamInternals {
|
|
client: { listen: (...args: unknown[]) => Promise<McpSubscription> };
|
|
modernSubscription: McpSubscription | null;
|
|
modernListenGeneration: number;
|
|
modernReconnectAttempts: number;
|
|
subscribedResources: Set<string>;
|
|
refreshModernSubscription(fromReconnect?: boolean): Promise<void>;
|
|
onModernSubscriptionClosed(
|
|
subscription: McpSubscription,
|
|
reason: "local" | "graceful" | "remote",
|
|
generation: number,
|
|
): void;
|
|
}
|
|
|
|
function internals(c: InspectorClient): StreamInternals {
|
|
return c as unknown as StreamInternals;
|
|
}
|
|
|
|
describe("modern era", () => {
|
|
it("opens an acknowledged listen stream on subscribe (no resources/subscribe)", async () => {
|
|
const started = await startServer({});
|
|
const { connected, messages } = await connect(started.url, "modern");
|
|
expect(connected.getProtocolEra()).toBe("modern");
|
|
expect(connected.supportsResourceSubscriptions()).toBe(true);
|
|
|
|
messages.length = 0;
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
|
|
const streamState = connected.getResourceSubscriptionStreamState();
|
|
expect(streamState.active).toBe(true);
|
|
expect(streamState.status).toBe("acknowledged");
|
|
expect(streamState.honoredUris).toContain(RESOURCE_URI);
|
|
expect(connected.getSubscribedResources()).toEqual([RESOURCE_URI]);
|
|
|
|
const methods = methodsSent(messages);
|
|
expect(methods).toContain("subscriptions/listen");
|
|
expect(methods).not.toContain("resources/subscribe");
|
|
});
|
|
|
|
it("closes the stream when the last subscription is removed", async () => {
|
|
// Nothing else in the filter (no list-change opt-ins), so the last URI
|
|
// leaving empties it and the stream closes (#1920).
|
|
const started = await startServer({});
|
|
const { connected, messages } = await connect(
|
|
started.url,
|
|
"modern",
|
|
NO_LIST_CHANGED,
|
|
);
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
expect(connected.getResourceSubscriptionStreamState().active).toBe(true);
|
|
|
|
messages.length = 0;
|
|
await connected.unsubscribeFromResource(RESOURCE_URI);
|
|
const streamState = connected.getResourceSubscriptionStreamState();
|
|
expect(streamState.active).toBe(false);
|
|
expect(connected.getSubscribedResources()).toEqual([]);
|
|
// No re-listen: there is nothing left to listen for.
|
|
expect(methodsSent(messages)).not.toContain("subscriptions/listen");
|
|
});
|
|
|
|
it("keeps the stream open past the last subscription when a listChanged opt-in remains", async () => {
|
|
// The other half of the above: with an advertised listChanged the filter
|
|
// is still non-empty, so the last unsubscribe re-lists rather than
|
|
// closing — and the state goes inactive because the *Subscriptions*
|
|
// section has nothing to report, not because the stream is gone (#1920).
|
|
const started = await startServer({});
|
|
const { connected, messages } = await connect(started.url, "modern");
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
|
|
messages.length = 0;
|
|
await connected.unsubscribeFromResource(RESOURCE_URI);
|
|
|
|
expect(connected.getSubscribedResources()).toEqual([]);
|
|
expect(connected.getResourceSubscriptionStreamState().active).toBe(false);
|
|
const filter = lastListenFilter(messages);
|
|
expect(filter?.resourcesListChanged).toBe(true);
|
|
expect(filter?.resourceSubscriptions).toBeUndefined();
|
|
});
|
|
|
|
it("re-lists (stream stays open) when one of several URIs is removed", async () => {
|
|
const started = await startServer({});
|
|
const { connected, messages } = await connect(started.url, "modern");
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
await connected.subscribeToResource(RESOURCE_URI_2);
|
|
expect(connected.getSubscribedResources()).toEqual([
|
|
RESOURCE_URI,
|
|
RESOURCE_URI_2,
|
|
]);
|
|
|
|
messages.length = 0;
|
|
await connected.unsubscribeFromResource(RESOURCE_URI);
|
|
|
|
// A fresh listen re-established the reduced filter; the stream is still up.
|
|
const streamState = connected.getResourceSubscriptionStreamState();
|
|
expect(streamState.active).toBe(true);
|
|
expect(streamState.status).toBe("acknowledged");
|
|
expect(streamState.honoredUris).toEqual([RESOURCE_URI_2]);
|
|
expect(methodsSent(messages)).toContain("subscriptions/listen");
|
|
});
|
|
|
|
it("folds the subscribed URIs and listChanged opt-ins into the listen filter", async () => {
|
|
const started = await startServer({});
|
|
const { connected, messages } = await connect(started.url, "modern");
|
|
messages.length = 0;
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
|
|
const filter = lastListenFilter(messages);
|
|
expect(filter?.resourceSubscriptions).toEqual([RESOURCE_URI]);
|
|
// The server advertises resources.listChanged, so the single stream also
|
|
// opts into it (one listen stream carries every opted-in type).
|
|
expect(filter?.resourcesListChanged).toBe(true);
|
|
});
|
|
|
|
it("skips a redundant re-list when re-subscribing an already-subscribed URI", async () => {
|
|
const started = await startServer({});
|
|
const { connected, messages } = await connect(started.url, "modern");
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
|
|
// A second subscribe of the same URI leaves the filter unchanged, so it
|
|
// must not re-list (which would needlessly churn the server stream).
|
|
messages.length = 0;
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
expect(methodsSent(messages)).not.toContain("subscriptions/listen");
|
|
expect(connected.getSubscribedResources()).toEqual([RESOURCE_URI]);
|
|
expect(connected.getResourceSubscriptionStreamState().active).toBe(true);
|
|
});
|
|
|
|
it("skips a re-list when unsubscribing a URI that isn't subscribed", async () => {
|
|
const started = await startServer({});
|
|
const { connected, messages } = await connect(started.url, "modern");
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
|
|
messages.length = 0;
|
|
await connected.unsubscribeFromResource("test://not-subscribed");
|
|
expect(methodsSent(messages)).not.toContain("subscriptions/listen");
|
|
// The real subscription is untouched.
|
|
expect(connected.getSubscribedResources()).toEqual([RESOURCE_URI]);
|
|
});
|
|
});
|
|
|
|
// The stream used to be reachable only by subscribing to a resource, so a
|
|
// server with no resources could never open it — and its `tools.listChanged`
|
|
// notifications, which ride the same stream, could never arrive (#1920).
|
|
describe("modern era: opening the stream for listChanged alone (#1920)", () => {
|
|
async function startToolsOnlyServer(
|
|
listChanged: ServerConfig["listChanged"],
|
|
): Promise<TestServerHttp> {
|
|
const started = createTestServerHttp({
|
|
serverInfo: createTestServerInfo("tools-only-test", "1.0.0"),
|
|
tools: [createEchoTool()],
|
|
listChanged,
|
|
modern: {},
|
|
});
|
|
await started.start();
|
|
server = started;
|
|
return started;
|
|
}
|
|
|
|
it("opens the stream on connect against a tools-only server", async () => {
|
|
const started = await startToolsOnlyServer({ tools: true });
|
|
const { connected, messages } = await connect(started.url, "modern");
|
|
|
|
expect(connected.getCapabilities()?.resources).toBeUndefined();
|
|
expect(methodsSent(messages)).toContain("subscriptions/listen");
|
|
const filter = lastListenFilter(messages);
|
|
expect(filter?.toolsListChanged).toBe(true);
|
|
// Nothing is subscribed, so the filter carries no URIs at all.
|
|
expect(filter?.resourceSubscriptions).toBeUndefined();
|
|
// …and the Subscriptions section still has nothing to report.
|
|
expect(connected.getSubscribedResources()).toEqual([]);
|
|
expect(connected.getResourceSubscriptionStreamState().active).toBe(false);
|
|
});
|
|
|
|
it("leaves the stream closed when the opt-in is disabled in config", async () => {
|
|
// The filter is config ∩ capability, so turning the handler off in the
|
|
// Inspector's own options is enough to keep the stream closed.
|
|
const started = await startToolsOnlyServer({ tools: true });
|
|
const connected = new InspectorClient(
|
|
{ type: "streamable-http", url: started.url },
|
|
{
|
|
environment: { transport: createTransportNode },
|
|
versionNegotiation: eraToVersionNegotiation("modern"),
|
|
listChangedNotifications: { tools: false },
|
|
},
|
|
);
|
|
const messages: MessageEntry[] = [];
|
|
connected.addEventListener("message", (event) => {
|
|
messages.push(event.detail);
|
|
});
|
|
await connected.connect();
|
|
client = connected;
|
|
|
|
expect(methodsSent(messages)).not.toContain("subscriptions/listen");
|
|
});
|
|
|
|
it("opens no stream on the legacy era", async () => {
|
|
// Legacy has no `subscriptions/listen` at all — list-change notifications
|
|
// arrive on the session's own channel.
|
|
const started = createTestServerHttp({
|
|
serverInfo: createTestServerInfo("tools-only-legacy-test", "1.0.0"),
|
|
tools: [createEchoTool()],
|
|
listChanged: { tools: true },
|
|
});
|
|
await started.start();
|
|
server = started;
|
|
const { connected, messages } = await connect(started.url, "legacy");
|
|
|
|
expect(connected.getProtocolEra()).toBe("legacy");
|
|
expect(methodsSent(messages)).not.toContain("subscriptions/listen");
|
|
expect(connected.getResourceSubscriptionStreamState().active).toBe(false);
|
|
});
|
|
|
|
it("reconnects a dropped listChanged-only stream (no subscriptions to keep it alive)", async () => {
|
|
const started = await startToolsOnlyServer({ tools: true });
|
|
const { connected } = await connect(started.url, "modern");
|
|
|
|
const int = internals(connected);
|
|
const dropped = int.modernSubscription;
|
|
expect(dropped).not.toBeNull();
|
|
if (!dropped) return;
|
|
|
|
int.onModernSubscriptionClosed(
|
|
dropped,
|
|
"remote",
|
|
int.modernListenGeneration,
|
|
);
|
|
// Reconnect-by-re-listen runs on the filter, not on the subscribed set —
|
|
// which is empty here, and used to be the reason to give up.
|
|
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
|
"reconnecting",
|
|
);
|
|
await vi.waitFor(() => {
|
|
expect(int.modernSubscription).not.toBeNull();
|
|
});
|
|
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
|
"acknowledged",
|
|
);
|
|
});
|
|
|
|
it("has the stream up before the connect event fires", async () => {
|
|
// The managed list states start their initial `refresh()` from the
|
|
// `connect` event, so the stream has to be established (or its retry
|
|
// armed) first: dispatching earlier would let `tools/list` go out ahead of
|
|
// `subscriptions/listen`, and a list the server changed in that window
|
|
// would notify nobody.
|
|
const started = await startToolsOnlyServer({ tools: true });
|
|
const connected = new InspectorClient(
|
|
{ type: "streamable-http", url: started.url },
|
|
{
|
|
environment: { transport: createTransportNode },
|
|
versionNegotiation: eraToVersionNegotiation("modern"),
|
|
},
|
|
);
|
|
let streamAtConnectEvent: boolean | undefined;
|
|
connected.addEventListener("connect", () => {
|
|
streamAtConnectEvent = internals(connected).modernSubscription !== null;
|
|
});
|
|
await connected.connect();
|
|
client = connected;
|
|
|
|
expect(streamAtConnectEvent).toBe(true);
|
|
});
|
|
|
|
it("connects anyway when the connect-time listen fails, and retries", async () => {
|
|
// The handshake succeeded; every request-scoped feature works without the
|
|
// stream, so the failure is handed to the reconnect machinery instead of
|
|
// failing `connect()`.
|
|
const started = await startToolsOnlyServer({ tools: true });
|
|
const connected = new InspectorClient(
|
|
{ type: "streamable-http", url: started.url },
|
|
{
|
|
environment: { transport: createTransportNode },
|
|
versionNegotiation: eraToVersionNegotiation("modern"),
|
|
},
|
|
);
|
|
// Shadow the private refresh on the instance so the connect-time open
|
|
// fails without a live client to reach into (there is none until
|
|
// `connect()` builds one). The bump mirrors the real method's contract —
|
|
// it claims the generation synchronously before it can fail — which is
|
|
// what the caller's guard reads to tell "our refresh failed" from "a newer
|
|
// one took over".
|
|
const int = internals(connected);
|
|
int.refreshModernSubscription = () => {
|
|
int.modernListenGeneration++;
|
|
return Promise.reject(new Error("listen boom"));
|
|
};
|
|
|
|
await expect(connected.connect()).resolves.toBeUndefined();
|
|
client = connected;
|
|
|
|
expect(connected.getStatus()).toBe("connected");
|
|
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
|
"reconnecting",
|
|
);
|
|
});
|
|
|
|
it("leaves a superseded connect-time failure to the refresh that owns the stream", async () => {
|
|
// Nothing about the ordering above makes this call the only refresh that
|
|
// can be in flight: `statusChange` has already fired, and a concurrent
|
|
// `subscribeToResource` — or a `disconnect()`, whose
|
|
// `resetSubscriptionStream` bumps the generation too — can supersede this
|
|
// one while its `listen()` is pending. Reconciling anyway would arm a
|
|
// reconnect against a stream that is healthy (or a session that is gone),
|
|
// so this makes the same ownership test the subscribe/unsubscribe paths
|
|
// make. Driven here by a stub that bumps twice, standing in for "someone
|
|
// else advanced it".
|
|
const started = await startToolsOnlyServer({ tools: true });
|
|
const connected = new InspectorClient(
|
|
{ type: "streamable-http", url: started.url },
|
|
{
|
|
environment: { transport: createTransportNode },
|
|
versionNegotiation: eraToVersionNegotiation("modern"),
|
|
},
|
|
);
|
|
const int = internals(connected);
|
|
// Two bumps: this call's own, plus the newer refresh that superseded it
|
|
// before its failure landed.
|
|
int.refreshModernSubscription = () => {
|
|
int.modernListenGeneration += 2;
|
|
return Promise.reject(new Error("listen boom"));
|
|
};
|
|
|
|
await expect(connected.connect()).resolves.toBeUndefined();
|
|
client = connected;
|
|
|
|
// No reconnect armed: the state is the newer refresh's to write.
|
|
expect(connected.getResourceSubscriptionStreamState().status).not.toBe(
|
|
"reconnecting",
|
|
);
|
|
});
|
|
});
|
|
|
|
describe("legacy era", () => {
|
|
it("subscribes via resources/subscribe with no listen stream", async () => {
|
|
const started = await startServer(undefined);
|
|
const { connected, messages } = await connect(started.url, "legacy");
|
|
expect(connected.getProtocolEra()).toBe("legacy");
|
|
expect(connected.supportsResourceSubscriptions()).toBe(true);
|
|
|
|
messages.length = 0;
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
|
|
const methods = methodsSent(messages);
|
|
expect(methods).toContain("resources/subscribe");
|
|
expect(methods).not.toContain("subscriptions/listen");
|
|
|
|
// No persistent stream on the legacy era.
|
|
expect(connected.getResourceSubscriptionStreamState().active).toBe(false);
|
|
expect(connected.getSubscribedResources()).toEqual([RESOURCE_URI]);
|
|
|
|
messages.length = 0;
|
|
await connected.unsubscribeFromResource(RESOURCE_URI);
|
|
expect(methodsSent(messages)).toContain("resources/unsubscribe");
|
|
expect(connected.getSubscribedResources()).toEqual([]);
|
|
});
|
|
});
|
|
|
|
describe("guards", () => {
|
|
it("rejects subscribe when the server does not support subscriptions", async () => {
|
|
const started = createTestServerHttp({
|
|
serverInfo: createTestServerInfo("no-subscribe-test", "1.0.0"),
|
|
resources: createNumberedResources(1),
|
|
// subscriptions omitted → capability not advertised
|
|
modern: {},
|
|
});
|
|
await started.start();
|
|
server = started;
|
|
const { connected } = await connect(started.url, "modern");
|
|
expect(connected.supportsResourceSubscriptions()).toBe(false);
|
|
await expect(connected.subscribeToResource(RESOURCE_URI)).rejects.toThrow(
|
|
/does not support resource subscriptions/,
|
|
);
|
|
});
|
|
});
|
|
|
|
// Modern stream lifecycle branches that the public surface can't reach against
|
|
// a healthy server (an unexpected drop, a failed `listen()`). These reach into
|
|
// the client's private state — the pattern used across the InspectorClient
|
|
// coverage-backfill suite — to drive them deterministically.
|
|
describe("modern stream internals", () => {
|
|
/** A controllable fake `McpSubscription` whose `closed` we resolve on demand. */
|
|
function makeFakeSub(): {
|
|
sub: McpSubscription;
|
|
drop: (reason: "local" | "graceful" | "remote") => void;
|
|
} {
|
|
let drop: (reason: "local" | "graceful" | "remote") => void = () => {};
|
|
const closed = new Promise<"local" | "graceful" | "remote">((resolve) => {
|
|
drop = resolve;
|
|
});
|
|
const sub = {
|
|
honoredFilter: { resourceSubscriptions: [RESOURCE_URI] },
|
|
close: async () => {},
|
|
closed,
|
|
} as McpSubscription;
|
|
return { sub, drop };
|
|
}
|
|
|
|
/**
|
|
* Replace the client's live listen stream with a controllable fake and close
|
|
* the real one, so tests that drive `onModernSubscriptionClosed` by hand
|
|
* don't leave a real stream open (which would reject "Connection closed" on
|
|
* teardown). Returns the installed fake.
|
|
*/
|
|
async function installFakeSubscription(
|
|
int: StreamInternals,
|
|
): Promise<ReturnType<typeof makeFakeSub>> {
|
|
const real = int.modernSubscription;
|
|
const fake = makeFakeSub();
|
|
int.modernSubscription = fake.sub;
|
|
// real.closed fires onModernSubscriptionClosed, but modernSubscription is
|
|
// now the fake, so it's a no-op guard-wise; this just tears down the wire.
|
|
await real?.close().catch(() => {});
|
|
return fake;
|
|
}
|
|
|
|
for (const [label, close] of [
|
|
[
|
|
"throws synchronously",
|
|
(): Promise<void> => {
|
|
throw new Error("close blew up");
|
|
},
|
|
],
|
|
[
|
|
"returns a rejected promise",
|
|
(): Promise<void> => Promise.reject(new Error("close blew up")),
|
|
],
|
|
] as const) {
|
|
it(`re-lists when the superseded stream's close() ${label}`, async () => {
|
|
// A re-list drops its reference to the previous stream *before* closing
|
|
// it, so an escaping failure would both abandon a stream that may still
|
|
// be open on the server and abort the refresh before its replacement
|
|
// `listen()` — leaving a non-empty subscription set with no stream. The
|
|
// close is best-effort against both failure modes for that reason; this
|
|
// is the site where being best-effort against the *synchronous* one
|
|
// changed behaviour (a `.catch()` alone never caught it).
|
|
const started = await startServer({});
|
|
const { connected } = await connect(started.url, "modern");
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
|
|
const int = internals(connected);
|
|
// Swap the live stream for a poisoned one, closing the real stream so
|
|
// the wire is torn down (as `installFakeSubscription` does).
|
|
const real = int.modernSubscription;
|
|
int.modernSubscription = { close } as unknown as McpSubscription;
|
|
await real?.close().catch(() => {});
|
|
|
|
// A second URI changes the filter, so this re-lists over the poisoned
|
|
// stream.
|
|
await expect(
|
|
connected.subscribeToResource(RESOURCE_URI_2),
|
|
).resolves.toBeUndefined();
|
|
|
|
expect(connected.getSubscribedResources()).toEqual([
|
|
RESOURCE_URI,
|
|
RESOURCE_URI_2,
|
|
]);
|
|
const state = connected.getResourceSubscriptionStreamState();
|
|
expect(state.active).toBe(true);
|
|
expect(state.status).toBe("acknowledged");
|
|
expect(int.modernSubscription).not.toBeNull();
|
|
});
|
|
}
|
|
|
|
it("keeps the optimistic add when a newer refresh superseded the failure", async () => {
|
|
// The rollback's premise is that the server filter is unchanged, which is
|
|
// exactly what supersession falsifies: the newer refresh built its filter
|
|
// from the set *including* this URI, so on success the server honors it.
|
|
// Rolling back anyway would leave the set missing a URI the live stream
|
|
// carries — and with only one subscribed, an empty set with an active
|
|
// stream, the combination `resetSubscriptionStream` exists to prevent.
|
|
// Reachable without any close() failure: a rejecting `listen()` does it,
|
|
// which is what a user subscribing while the reconnect timer re-lists
|
|
// (i.e. exactly when the server is flaky) can produce.
|
|
//
|
|
// No list-change opt-ins, so connect leaves the stream closed (#1920) and
|
|
// the first subscribe's `listen()` is reached synchronously — with a
|
|
// stream to tear down first, the swap below would land after this call
|
|
// already read `client.listen`.
|
|
const started = await startServer({});
|
|
const { connected } = await connect(
|
|
started.url,
|
|
"modern",
|
|
NO_LIST_CHANGED,
|
|
);
|
|
const int = internals(connected);
|
|
|
|
const real = int.client.listen;
|
|
// The first listen hangs, so the second can supersede it before it fails.
|
|
let failFirst: (error: Error) => void = () => {};
|
|
int.client.listen = () =>
|
|
new Promise<McpSubscription>((_, reject) => {
|
|
failFirst = reject;
|
|
});
|
|
const first = connected.subscribeToResource(RESOURCE_URI);
|
|
// Attached before the supersession so the rejection is never unhandled.
|
|
const firstSettled = expect(first).rejects.toThrow(/listen boom/);
|
|
|
|
// A newer refresh starts and acknowledges, carrying both URIs.
|
|
int.client.listen = real;
|
|
await connected.subscribeToResource(RESOURCE_URI_2);
|
|
|
|
failFirst(new Error("listen boom"));
|
|
await firstSettled;
|
|
|
|
// The superseded call still reports its failure, but leaves the filter to
|
|
// the refresh that owns it.
|
|
expect(connected.getSubscribedResources()).toEqual([
|
|
RESOURCE_URI,
|
|
RESOURCE_URI_2,
|
|
]);
|
|
const state = connected.getResourceSubscriptionStreamState();
|
|
expect(state.active).toBe(true);
|
|
expect(state.status).toBe("acknowledged");
|
|
});
|
|
|
|
it("retries rather than stranding the URIs when both overlapping subscribes fail", async () => {
|
|
// The other half of the gate: skipping the superseded call's rollback is
|
|
// only safe if the superseding call reconciles when *it* fails too.
|
|
// Otherwise the set keeps the first URI while the state keeps the
|
|
// optimistic "connecting" — a badge that never changes, over a
|
|
// subscription no server ever honored, with nothing armed to fix it
|
|
// (neither `onModernSubscriptionClosed` nor `onModernReconnectFailed`
|
|
// ran). The reconcile hands it to the reconnect machinery instead.
|
|
const started = await startServer({});
|
|
const { connected } = await connect(started.url, "modern");
|
|
const int = internals(connected);
|
|
|
|
let failFirst: (error: Error) => void = () => {};
|
|
int.client.listen = () =>
|
|
new Promise<McpSubscription>((_, reject) => {
|
|
failFirst = reject;
|
|
});
|
|
const first = connected.subscribeToResource(RESOURCE_URI);
|
|
const firstSettled = expect(first).rejects.toThrow(/listen boom/);
|
|
|
|
// The superseding call fails on its own listen.
|
|
int.client.listen = () => Promise.reject(new Error("listen boom 2"));
|
|
await expect(
|
|
connected.subscribeToResource(RESOURCE_URI_2),
|
|
).rejects.toThrow(/listen boom 2/);
|
|
|
|
failFirst(new Error("listen boom"));
|
|
await firstSettled;
|
|
|
|
// Its own URI rolled back; the superseded one survives — and rather than
|
|
// sitting at "connecting" forever it is now a retry in progress.
|
|
expect(connected.getSubscribedResources()).toEqual([RESOURCE_URI]);
|
|
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
|
"reconnecting",
|
|
);
|
|
|
|
// And the retry makes the state true: the next re-listen acknowledges.
|
|
int.client.listen = () => Promise.resolve(makeFakeSub().sub);
|
|
await vi.waitFor(() => {
|
|
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
|
"acknowledged",
|
|
);
|
|
});
|
|
});
|
|
|
|
it("retries rather than stranding the URIs when the unsubscribe's re-listen fails", async () => {
|
|
// `unsubscribeFromResource` keeps its removal when the re-listen fails,
|
|
// so nothing is rolled back — but the stream is gone with URIs still
|
|
// subscribed, and without the reconcile the badge would keep reporting
|
|
// the previous success ("acknowledged"/Listening) over no stream at all.
|
|
const started = await startServer({});
|
|
const { connected } = await connect(started.url, "modern");
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
await connected.subscribeToResource(RESOURCE_URI_2);
|
|
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
|
"acknowledged",
|
|
);
|
|
|
|
const int = internals(connected);
|
|
int.client.listen = () => Promise.reject(new Error("re-listen boom"));
|
|
await expect(
|
|
connected.unsubscribeFromResource(RESOURCE_URI_2),
|
|
).rejects.toThrow(/re-listen boom/);
|
|
|
|
expect(connected.getSubscribedResources()).toEqual([RESOURCE_URI]);
|
|
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
|
"reconnecting",
|
|
);
|
|
});
|
|
|
|
it("never announces an empty subscription set with an active stream", async () => {
|
|
// The pair `resetSubscriptionStream` orders its own writes to prevent,
|
|
// asserted from the consumer's side at the two sites that empty the set
|
|
// outside it: the last-URI unsubscribe (happy path, where the re-listen
|
|
// that would set INACTIVE is a round-trip away) and the last-URI rollback
|
|
// of a failed subscribe (where the state still reads "connecting").
|
|
const started = await startServer({});
|
|
const { connected } = await connect(started.url, "modern");
|
|
const seen: { size: number; active: boolean }[] = [];
|
|
connected.addEventListener("resourceSubscriptionsChange", () => {
|
|
seen.push({
|
|
size: connected.getSubscribedResources().length,
|
|
active: connected.getResourceSubscriptionStreamState().active,
|
|
});
|
|
});
|
|
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
await connected.unsubscribeFromResource(RESOURCE_URI);
|
|
|
|
const int = internals(connected);
|
|
int.client.listen = () => Promise.reject(new Error("listen boom"));
|
|
await expect(connected.subscribeToResource(RESOURCE_URI)).rejects.toThrow(
|
|
/listen boom/,
|
|
);
|
|
|
|
expect(seen.length).toBeGreaterThan(0);
|
|
expect(seen.filter((s) => s.size === 0 && s.active)).toEqual([]);
|
|
});
|
|
|
|
it("rolls back the optimistic add when listen() fails", async () => {
|
|
const started = await startServer({});
|
|
const { connected } = await connect(started.url, "modern");
|
|
const real = internals(connected).client.listen;
|
|
internals(connected).client.listen = () =>
|
|
Promise.reject(new Error("listen boom"));
|
|
|
|
await expect(connected.subscribeToResource(RESOURCE_URI)).rejects.toThrow(
|
|
/listen boom/,
|
|
);
|
|
expect(connected.getSubscribedResources()).toEqual([]);
|
|
expect(connected.getResourceSubscriptionStreamState().active).toBe(false);
|
|
|
|
internals(connected).client.listen = real;
|
|
});
|
|
|
|
it("reflects the subscription optimistically as 'connecting' before the ack", async () => {
|
|
// No list-change opt-ins → no connect-time stream, so the subscribe's
|
|
// `listen()` is the first one and this test's held ack is the one it waits
|
|
// on (#1920).
|
|
const started = await startServer({});
|
|
const { connected } = await connect(
|
|
started.url,
|
|
"modern",
|
|
NO_LIST_CHANGED,
|
|
);
|
|
const int = internals(connected);
|
|
|
|
// Hold the listen ack so we can observe the pre-ack (optimistic) state.
|
|
let ack: (sub: McpSubscription) => void = () => {};
|
|
int.client.listen = () =>
|
|
new Promise<McpSubscription>((resolve) => {
|
|
ack = resolve;
|
|
});
|
|
|
|
const pending = connected.subscribeToResource(RESOURCE_URI);
|
|
// The URI and a "connecting" stream state are visible immediately, without
|
|
// waiting for the round-trip.
|
|
expect(connected.getSubscribedResources()).toEqual([RESOURCE_URI]);
|
|
const connecting = connected.getResourceSubscriptionStreamState();
|
|
expect(connecting.active).toBe(true);
|
|
expect(connecting.status).toBe("connecting");
|
|
|
|
// The ack lands → acknowledged.
|
|
ack(makeFakeSub().sub);
|
|
await pending;
|
|
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
|
"acknowledged",
|
|
);
|
|
});
|
|
|
|
it("reconnects by re-listing after an unexpected 'remote' drop", async () => {
|
|
const started = await startServer({});
|
|
const { connected } = await connect(started.url, "modern");
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
|
|
const int = internals(connected);
|
|
const fake = await installFakeSubscription(int);
|
|
int.onModernSubscriptionClosed(
|
|
fake.sub,
|
|
"remote",
|
|
int.modernListenGeneration,
|
|
);
|
|
|
|
// Synchronously flips to reconnecting, then re-lists and re-acknowledges.
|
|
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
|
"reconnecting",
|
|
);
|
|
await vi.waitFor(() => {
|
|
const s = connected.getResourceSubscriptionStreamState();
|
|
expect(s.active).toBe(true);
|
|
expect(s.status).toBe("acknowledged");
|
|
});
|
|
});
|
|
|
|
it("retries a failing re-listen with backoff and gives up past the cap", async () => {
|
|
const started = await startServer({});
|
|
const { connected } = await connect(started.url, "modern");
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
const int = internals(connected);
|
|
const fake = await installFakeSubscription(int);
|
|
|
|
vi.useFakeTimers();
|
|
try {
|
|
// Every re-listen fails, so the failure count climbs each retry.
|
|
int.client.listen = () => Promise.reject(new Error("re-listen boom"));
|
|
|
|
int.onModernSubscriptionClosed(
|
|
fake.sub,
|
|
"remote",
|
|
int.modernListenGeneration,
|
|
);
|
|
// A single failure retries (not ended yet).
|
|
await vi.advanceTimersByTimeAsync(20_000);
|
|
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
|
"reconnecting",
|
|
);
|
|
|
|
// Keep failing until the consecutive-failure cap gives up.
|
|
for (let i = 0; i < 12; i++) {
|
|
if (connected.getResourceSubscriptionStreamState().status === "ended")
|
|
break;
|
|
await vi.advanceTimersByTimeAsync(20_000);
|
|
}
|
|
const state = connected.getResourceSubscriptionStreamState();
|
|
expect(state.status).toBe("ended");
|
|
// Subscriptions remain, so the ended badge stays visible.
|
|
expect(state.active).toBe(true);
|
|
} finally {
|
|
vi.useRealTimers();
|
|
}
|
|
});
|
|
|
|
it("resets the failure count after a successful reconnect", async () => {
|
|
const started = await startServer({});
|
|
const { connected } = await connect(started.url, "modern");
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
const int = internals(connected);
|
|
const fake = await installFakeSubscription(int);
|
|
|
|
vi.useFakeTimers();
|
|
try {
|
|
// The first two re-lists fail (count climbs), the third acknowledges.
|
|
let failures = 0;
|
|
int.client.listen = () => {
|
|
if (failures < 2) {
|
|
failures += 1;
|
|
return Promise.reject(new Error("re-listen boom"));
|
|
}
|
|
return Promise.resolve(makeFakeSub().sub);
|
|
};
|
|
|
|
int.onModernSubscriptionClosed(
|
|
fake.sub,
|
|
"remote",
|
|
int.modernListenGeneration,
|
|
);
|
|
// Advance through the two failures and the successful ack.
|
|
for (let i = 0; i < 4; i++) {
|
|
await vi.advanceTimersByTimeAsync(20_000);
|
|
}
|
|
// A successful ack resets the run, so a subsequent drop starts fresh.
|
|
expect(int.modernReconnectAttempts).toBe(0);
|
|
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
|
"acknowledged",
|
|
);
|
|
} finally {
|
|
vi.useRealTimers();
|
|
}
|
|
});
|
|
|
|
it("does not reconnect when the subscription set empties before the timer fires", async () => {
|
|
// No list-change opt-ins, so emptying the set empties the *filter* —
|
|
// which is what the timer's bail now tests (#1920).
|
|
const started = await startServer({});
|
|
const { connected } = await connect(
|
|
started.url,
|
|
"modern",
|
|
NO_LIST_CHANGED,
|
|
);
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
const int = internals(connected);
|
|
const fake = await installFakeSubscription(int);
|
|
|
|
vi.useFakeTimers();
|
|
try {
|
|
int.client.listen = () => {
|
|
throw new Error("re-listen should not run once the set is empty");
|
|
};
|
|
int.onModernSubscriptionClosed(
|
|
fake.sub,
|
|
"remote",
|
|
int.modernListenGeneration,
|
|
);
|
|
expect(connected.getResourceSubscriptionStreamState().status).toBe(
|
|
"reconnecting",
|
|
);
|
|
// Empty the set without going through unsubscribe (which would clear the
|
|
// timer), then fire it: the guard bails instead of re-listing.
|
|
int.subscribedResources.clear();
|
|
await vi.advanceTimersByTimeAsync(20_000);
|
|
expect(int.modernSubscription).toBeNull();
|
|
} finally {
|
|
vi.useRealTimers();
|
|
}
|
|
});
|
|
|
|
it("keeps the ended badge active on a graceful close while subscriptions remain", async () => {
|
|
const started = await startServer({});
|
|
const { connected } = await connect(started.url, "modern");
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
|
|
const int = internals(connected);
|
|
const fake = await installFakeSubscription(int);
|
|
int.onModernSubscriptionClosed(
|
|
fake.sub,
|
|
"graceful",
|
|
int.modernListenGeneration,
|
|
);
|
|
const state = connected.getResourceSubscriptionStreamState();
|
|
expect(state.status).toBe("ended");
|
|
// Subscriptions remain, so the ended badge stays visible (parity with the
|
|
// reconnect give-up state).
|
|
expect(state.active).toBe(true);
|
|
});
|
|
|
|
it("goes inactive on a graceful close once no subscriptions remain", async () => {
|
|
const started = await startServer({});
|
|
const { connected } = await connect(started.url, "modern");
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
|
|
const int = internals(connected);
|
|
const fake = await installFakeSubscription(int);
|
|
// No URIs left → the ended state is inactive (no badge).
|
|
int.subscribedResources.clear();
|
|
int.onModernSubscriptionClosed(
|
|
fake.sub,
|
|
"graceful",
|
|
int.modernListenGeneration,
|
|
);
|
|
expect(connected.getResourceSubscriptionStreamState().active).toBe(false);
|
|
});
|
|
|
|
it("ignores a close callback from a superseded generation", async () => {
|
|
const started = await startServer({});
|
|
const { connected } = await connect(started.url, "modern");
|
|
await connected.subscribeToResource(RESOURCE_URI);
|
|
|
|
const int = internals(connected);
|
|
const fake = await installFakeSubscription(int);
|
|
const before = connected.getResourceSubscriptionStreamState();
|
|
// A stale generation → the callback is a no-op (no reconnect, no change).
|
|
int.onModernSubscriptionClosed(
|
|
fake.sub,
|
|
"remote",
|
|
int.modernListenGeneration - 1,
|
|
);
|
|
expect(connected.getResourceSubscriptionStreamState()).toEqual(before);
|
|
});
|
|
});
|
|
});
|