diff --git a/.changeset/abortable-auth-awaits.md b/.changeset/abortable-auth-awaits.md new file mode 100644 index 0000000000..f59c5ac053 --- /dev/null +++ b/.changeset/abortable-auth-awaits.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +Make the streamable HTTP transport's auth awaits abortable. `AuthProvider.token()`, `onUnauthorized()` 401 recovery, and insufficient-scope step-up authorization were awaited with no way for `TransportSendOptions.requestSignal` (or the transport's own lifetime signal) to reach them, so a hung token refresh or recovery flow parked `send()` forever past its abort. These awaits are now raced against the combined request/transport signal, and the signal is offered to `onUnauthorized` via the new optional `UnauthorizedContext.signal` field so cooperative providers can cancel their own recovery work. An abort during the auth chain rejects the send with the abort reason (unstamped, treated as an intentional teardown, no spurious `onerror`). Also fixes resume-via-`send()`: the resumed GET now preserves the caller's `onresumptiontoken`/`onRequestStreamEnd` observers, so resumed streams keep the token-persistence chain, report their terminal end, and an outright resume failure no longer dead-ends silently. Reconnect-attempt failures now reach `onerror` exactly once, as the underlying error — the `Failed to reconnect SSE stream: …` wrapper that duplicated each per-attempt report is gone (the `Maximum reconnection attempts (N) exceeded.` exhaustion message is unchanged). A reconnect scheduled before the resumed stream delivered any ID-bearing event now falls back to the stream's prior resumption token instead of silently dropping `Last-Event-ID`. diff --git a/docs/clients/machine-auth.md b/docs/clients/machine-auth.md index 1b3c6614bb..ec8447617c 100644 --- a/docs/clients/machine-auth.md +++ b/docs/clients/machine-auth.md @@ -40,7 +40,7 @@ const authProvider: AuthProvider = { token: async () => getStoredToken() }; const transport = new StreamableHTTPClientTransport(new URL('https://api.example.com/mcp'), { authProvider }); ``` -The transport calls `token()` before every request and sets the `Authorization` header from whatever it returns. Without `onUnauthorized`, a 401 throws `UnauthorizedError`. Add `onUnauthorized(ctx)` to refresh the credential and the transport retries the request once. +The transport calls `token()` before every request and sets the `Authorization` header from whatever it returns. Without `onUnauthorized`, a 401 throws `UnauthorizedError`. Add `onUnauthorized(ctx)` to refresh the credential and the transport retries the request once. `ctx.signal` (when present) aborts once the caller or transport gives up on the request — forward it to your own fetches so the recovery work stops with it. ## Sign with a private key instead of a secret diff --git a/docs/migration/upgrade-to-v2.md b/docs/migration/upgrade-to-v2.md index 19f4127733..8d52afd762 100644 --- a/docs/migration/upgrade-to-v2.md +++ b/docs/migration/upgrade-to-v2.md @@ -1091,7 +1091,8 @@ The transport `authProvider` option is widened to `AuthProvider | OAuthClientPro **`AuthProvider`** is a new minimal interface — `{ token(): Promise; onUnauthorized?(ctx): Promise }` — for static-token / non-OAuth bearer auth. Transports call `token()` before every request and `onUnauthorized()` on 401 (then retry -once). Existing `OAuthClientProvider` implementations need no changes — transports adapt +once); `ctx.signal` (when present) aborts once the caller or transport gives up — forward +it to your own fetches. Existing `OAuthClientProvider` implementations need no changes — transports adapt them internally via the new `adaptOAuthProvider()` export. Also exported: `isOAuthClientProvider()` (type guard) and `handleOAuthUnauthorized()` (the standard OAuth `onUnauthorized` behavior, for composing your own adapter). @@ -1511,7 +1512,12 @@ rewrite required unless noted. standalone GET-stream reconnection behavior and its exhaustion signal carry over from v1: when retries run out, the transport emits `onerror` with a plain `Error` whose message is `Maximum reconnection attempts (N) exceeded.` — there is no typed error - class for this condition, so monitors that match the message text keep working. + class for this condition, so monitors that match the message text keep working. The + message-text guarantee is scoped to that exhaustion message: each failed attempt + before exhaustion now reaches `onerror` exactly once, as the underlying error itself — + the `Failed to reconnect SSE stream: …` wrapper that previously accompanied every + per-attempt report is gone, so monitors matching that wrapper text should match the + underlying error (or the exhaustion message) instead. - **Also unchanged: elicitation response validation.** `elicitInput`'s local validation of elicitation responses against `requestedSchema`, the resulting `-32602` error message wording (`Elicitation response content does not match requested schema: …`), diff --git a/packages/client/src/client/auth.ts b/packages/client/src/client/auth.ts index 9ebc6fd251..941b162db7 100644 --- a/packages/client/src/client/auth.ts +++ b/packages/client/src/client/auth.ts @@ -55,6 +55,14 @@ export interface UnauthorizedContext { serverUrl: URL; /** Fetch function configured with the transport's `requestInit`, for making auth requests. */ fetchFn: FetchLike; + /** + * Abort signal for the request (or transport) whose 401 triggered this + * recovery. The transport stops waiting for `onUnauthorized` when it + * aborts; cooperative implementations should pass it to their own fetches + * so the recovery work stops too. Optional — absent when the transport + * has no lifetime signal to offer. + */ + signal?: AbortSignal; } /** diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index ace0663158..a24e080022 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -303,6 +303,89 @@ function anySignal(a: AbortSignal, b: AbortSignal): AbortSignal { return controller.signal; } +/** + * `signal?.aborted === true` behind a function boundary: `aborted` flips + * asynchronously (across awaits), so an inline check after an earlier guard + * gets unsoundly narrowed to `false` by the type checker. + */ +function isAborted(signal: AbortSignal | undefined): boolean { + return signal?.aborted === true; +} + +/** + * Normalize an aborted signal's `reason` to an `Error` (matching how the + * listen driver and `fetch` surface aborts) so a raced auth await rejects + * with something callers can inspect. + */ +function abortReasonError(signal: AbortSignal): Error { + const reason: unknown = signal.reason; + return reason instanceof Error ? reason : new Error(String(reason ?? 'Aborted')); +} + +/** + * Race a pending auth-chain await (`AuthProvider.token()`, `onUnauthorized`, + * step-up authorization) against the request/transport abort signal so an + * abort settles `send()` even when the underlying promise never does. The + * loser keeps running — the `AuthProvider` contract has no cancellation + * channel of its own beyond the optional `ctx.signal` — but the transport + * stops waiting on it, which is what `TransportSendOptions.requestSignal` + * promises. The abort listener is removed as soon as the promise settles so + * a long-lived transport signal does not accumulate closures per request. + */ +function raceWithSignal(produce: () => Promise, signal: AbortSignal | undefined): Promise { + // Aborted fast-path BEFORE the producer is invoked: an already-dead + // request must not start side-effectful auth work (a token refresh, an + // OAuth step-up that mutates discovery state or redirects the user) + // just to have its result discarded. + if (signal !== undefined && isAborted(signal)) { + return Promise.reject(abortReasonError(signal)); + } + // Normalize: a plain-JS provider may return a bare value (or a foreign + // thenable) where the types say Promise — the plain `await` these call + // sites used before the race tolerated that, so the race must too + // (`.then` on a bare string would TypeError and get misstamped as an + // auth failure). Identity-preserving for native promises. A synchronous + // throw from the producer becomes a rejection, matching how the plain + // `await` call sites surfaced it. + let promise: Promise; + try { + promise = Promise.resolve(produce()); + } catch (error) { + return Promise.reject(error); + } + if (signal === undefined) { + return promise; + } + if (signal.aborted) { + // Attach a no-op handler so the loser's eventual rejection (if any) + // does not escape as an unhandledRejection. + promise.catch(() => {}); + return Promise.reject(abortReasonError(signal)); + } + return new Promise((resolve, reject) => { + const onAbort = (): void => { + promise.catch(() => {}); + reject(abortReasonError(signal)); + }; + signal.addEventListener('abort', onAbort, { once: true }); + promise.then( + value => { + signal.removeEventListener('abort', onAbort); + resolve(value); + }, + (error: unknown) => { + signal.removeEventListener('abort', onAbort); + // Verbatim: the loser's rejection must flow through + // identity-preserved — the auth-seam stamp is a property on + // the thrown object (Error or not), and rewrapping would strip + // it (and any `.cause` chain) on paths with no re-stamping + // catch, e.g. the step-up awaits. + reject(error); + } + ); + }); +} + /** * Client transport for Streamable HTTP: this implements the MCP Streamable HTTP transport specification. * It will connect to a server using HTTP `POST` for sending messages and HTTP `GET` with Server-Sent Events @@ -432,12 +515,46 @@ export class StreamableHTTPClientTransport implements Transport { }); } - private async _commonHeaders(): Promise { + /** + * The abort signal governing one request: the transport-lifetime signal + * combined with the caller's per-request signal when both exist, + * whichever one exists otherwise. Computed BEFORE header acquisition so + * the auth chain (token(), 401 recovery, step-up) is raced against the + * same abort as the request itself. + */ + private _combinedSignal(requestSignal?: AbortSignal): AbortSignal | undefined { + const transportSignal = this._abortController?.signal; + return requestSignal !== undefined && transportSignal !== undefined + ? anySignal(transportSignal, requestSignal) + : (requestSignal ?? transportSignal); + } + + /** + * Whether teardown has been signalled for the transport (`close()`) or + * for the observed request (its `requestSignal`). Errors surfacing after + * either are intentional-abort fallout, not reportable failures — the + * guard-site mirror of {@linkcode _combinedSignal}. + */ + private _isIntentionalAbort(requestSignal?: AbortSignal): boolean { + return isAborted(this._abortController?.signal) || isAborted(requestSignal); + } + + private async _commonHeaders(signal?: AbortSignal): Promise { const headers: RequestInit['headers'] & Record = {}; + const provider = this._authProvider; let token: string | undefined; try { - token = await this._authProvider?.token(); + // Raced against the per-request/transport abort so a hung + // `token()` (wedged refresh, slow broker) cannot park the send + // past its `requestSignal` (#2643). + token = provider === undefined ? undefined : await raceWithSignal(() => provider.token(), signal); } catch (error) { + // An abort is an intentional teardown, never an auth failure — + // leave it unstamped so the send-catch and the negotiation + // probe's classifier treat it as a plain abort. + if (isAborted(signal)) { + throw error; + } // Auth-seam stamp: a throwing token() is an auth failure, never a // network failure. throw markAuthSeamEscape(error); @@ -519,17 +636,13 @@ export class StreamableHTTPClientTransport implements Transport { private async _startOrAuthSse(options: StartSSEOptions, isAuthRetry = false, stepUpRetries = 0): Promise { const { resumptionToken, requestSignal } = options; - // Same guard as `_handleSseStream`: a resurrected listen stream (the - // POST-SSE → GET reconnect path threads `requestSignal` through - // `StartSSEOptions`) must honour the per-request abort exactly as the - // original POST did — both as a fetch signal and as a "do not surface - // onerror" gate. - const isIntentionalAbort = (): boolean => this._abortController?.signal.aborted === true || requestSignal?.aborted === true; try { + const signal = this._combinedSignal(requestSignal); + // Try to open an initial SSE stream with GET to listen for server messages // This is optional according to the spec - server may not support it - const headers = await this._commonHeaders(); + const headers = await this._commonHeaders(signal); const userAccept = headers.get('accept'); const types = [...(userAccept?.split(',').map(s => s.trim().toLowerCase()) ?? []), 'text/event-stream']; headers.set('accept', [...new Set(types)].join(', ')); @@ -539,11 +652,6 @@ export class StreamableHTTPClientTransport implements Transport { headers.set('last-event-id', resumptionToken); } - const transportSignal = this._abortController?.signal; - const signal = - requestSignal !== undefined && transportSignal !== undefined - ? anySignal(transportSignal, requestSignal) - : (requestSignal ?? transportSignal); const response = await (this._fetch ?? fetch)(this._url, { ...this._requestInit, method: 'GET', @@ -561,14 +669,29 @@ export class StreamableHTTPClientTransport implements Transport { this._scope = computeScopeUnion(this._scope, scope); } - if (this._authProvider.onUnauthorized && !isAuthRetry) { + const onUnauthorized = this._authProvider.onUnauthorized?.bind(this._authProvider); + if (onUnauthorized && !isAuthRetry) { try { - await this._authProvider.onUnauthorized({ - response, - serverUrl: this._url, - fetchFn: this._fetchWithInit - }); + // Raced against the per-request/transport abort so + // a hung 401 recovery cannot park the GET (#2643); + // the signal is also handed to the provider so a + // cooperative implementation can cancel its own work. + await raceWithSignal( + () => + onUnauthorized({ + response, + serverUrl: this._url, + fetchFn: this._fetchWithInit, + signal + }), + signal + ); } catch (error) { + // An abort is an intentional teardown, not an auth + // failure — leave it unstamped. + if (isAborted(signal)) { + throw error; + } // Auth-seam stamp: covers the SDK's OAuth flow and // custom onUnauthorized callbacks alike. throw markAuthSeamEscape(error); @@ -593,9 +716,15 @@ export class StreamableHTTPClientTransport implements Transport { const { resourceMetadataUrl, scope, error, errorDescription } = extractWWWAuthenticateParams(response); if (error === 'insufficient_scope') { const text = await response.text?.().catch(() => null); - const result = await this._stepUpAuthorize( - { scope, resourceMetadataUrl, errorDescription, statusText: response.statusText, text }, - stepUpRetries + // Raced against the per-request/transport abort so a + // hung step-up authorization cannot park the GET (#2643). + const result = await raceWithSignal( + () => + this._stepUpAuthorize( + { scope, resourceMetadataUrl, errorDescription, statusText: response.statusText, text }, + stepUpRetries + ), + signal ); if (result !== 'AUTHORIZED') { throw markAuthSeamEscape(new UnauthorizedError()); @@ -629,7 +758,11 @@ export class StreamableHTTPClientTransport implements Transport { this._handleSseStream(response.body, options, true); } catch (error) { - if (!isIntentionalAbort()) { + // A resurrected listen stream (the POST-SSE → GET reconnect path + // threads `requestSignal` through `StartSSEOptions`) must honour + // the per-request abort exactly as the original POST did — both + // as a fetch signal and as a "do not surface onerror" gate. + if (!this._isIntentionalAbort(requestSignal)) { this.onerror?.(error as Error); } throw error; @@ -683,14 +816,22 @@ export class StreamableHTTPClientTransport implements Transport { // Honour BOTH the transport-wide abort and the per-request abort // (a listen subscription closed during the backoff delay): do not // resurrect a stream the caller already tore down. - if (this._abortController?.signal.aborted || options.requestSignal?.aborted) return; - this._startOrAuthSse(options).catch(error => { - if (this._abortController?.signal.aborted || options.requestSignal?.aborted) return; - this.onerror?.(new Error(`Failed to reconnect SSE stream: ${error instanceof Error ? error.message : String(error)}`)); + if (this._isIntentionalAbort(options.requestSignal)) return; + this._startOrAuthSse(options).catch(() => { + if (this._isIntentionalAbort(options.requestSignal)) return; + // `_startOrAuthSse`'s own catch already routed the failure to + // onerror before rethrowing — re-emitting here would report + // every failed attempt twice. Just schedule the next try. try { this._scheduleReconnection(options, attemptCount + 1); } catch (scheduleError) { this.onerror?.(scheduleError instanceof Error ? scheduleError : new Error(String(scheduleError))); + // A scheduler that throws while re-arming ends the retry + // chain — that is TERMINAL for an observed per-request + // stream, exactly like retry-budget exhaustion above, so + // the caller must be told or a listen() observer waits + // forever. + options.onRequestStreamEnd?.(); } }); }; @@ -719,7 +860,7 @@ export class StreamableHTTPClientTransport implements Transport { // a clean shutdown: no misleading "SSE stream disconnected" onerror, // and no GET+Last-Event-ID reconnect that would resurrect a stream the // caller just tore down. - const isIntentionalAbort = (): boolean => this._abortController?.signal.aborted === true || requestSignal?.aborted === true; + const isIntentionalAbort = (): boolean => this._isIntentionalAbort(requestSignal); let lastEventId: string | undefined; // Track whether we've received a priming event (event with ID) @@ -791,7 +932,12 @@ export class StreamableHTTPClientTransport implements Transport { if (needsReconnect && this._abortController && !isIntentionalAbort()) { this._scheduleReconnection( { - resumptionToken: lastEventId, + // Fall back to the token this stream was RESUMED + // from: a resumed stream that drops before its + // first ID-bearing event must not lose its + // Last-Event-ID (a token-less GET starts a fresh + // stream and the pending response never replays). + resumptionToken: lastEventId ?? options.resumptionToken, onresumptiontoken, replayMessageId, requestSignal, @@ -824,7 +970,10 @@ export class StreamableHTTPClientTransport implements Transport { try { this._scheduleReconnection( { - resumptionToken: lastEventId, + // Same fallback as the graceful-close path + // above: keep the token this stream was + // resumed from when it dies pre-first-event. + resumptionToken: lastEventId ?? options.resumptionToken, onresumptiontoken, replayMessageId, requestSignal, @@ -957,12 +1106,40 @@ export class StreamableHTTPClientTransport implements Transport { this._startOrAuthSse({ resumptionToken, replayMessageId: isJSONRPCRequest(message) ? message.id : undefined, - requestSignal: options?.requestSignal - }).catch(error => this.onerror?.(error)); + requestSignal: options?.requestSignal, + // Keep the caller's stream observers across the resume, + // matching the fresh-POST path below: without them a + // resume-via-send() dropped later resumption tokens (the + // persistence chain) and never reported the resumed + // stream's terminal end. + onresumptiontoken, + onRequestStreamEnd: options?.onRequestStreamEnd + }).catch(() => { + // `_startOrAuthSse`'s own catch already routed the + // failure to onerror (suppressed for intentional aborts) + // before rethrowing — no second emission here. Same + // abort guard as `_scheduleReconnection`'s reconnect(): + // an abort of either signal during the resume (now + // reachable mid-auth-chain too) is intentional teardown. + if (this._isIntentionalAbort(options?.requestSignal)) { + return; + } + // An outright resume failure (network, non-401/403/405 + // HTTP error) is TERMINAL for the per-request stream the + // caller is observing — this is the only channel that + // reports it (the 405 case fires inside _startOrAuthSse). + options?.onRequestStreamEnd?.(); + }); return; } - const headers = await this._commonHeaders(); + // Per-request abort: when the caller supplies a request-scoped + // signal (the `subscriptions/listen` driver), aborting it cancels + // this POST and its SSE response stream without closing the + // transport. + const signal = this._combinedSignal(options?.requestSignal); + + const headers = await this._commonHeaders(signal); this._applyBodyDerivedHeaders(headers, message); // A new session starts "without a session ID attached" (2025-11-25 transports §Session Management). const isHandshake = Array.isArray(message) ? message.some(m => isInitializeRequest(m)) : isInitializeRequest(message); @@ -987,15 +1164,6 @@ export class StreamableHTTPClientTransport implements Transport { const types = [...(userAccept?.split(',').map(s => s.trim().toLowerCase()) ?? []), 'application/json', 'text/event-stream']; headers.set('accept', [...new Set(types)].join(', ')); - // Per-request abort: when the caller supplies a request-scoped - // signal (the `subscriptions/listen` driver), aborting it cancels - // this POST and its SSE response stream without closing the - // transport. - const transportSignal = this._abortController?.signal; - const signal = - options?.requestSignal !== undefined && transportSignal !== undefined - ? anySignal(transportSignal, options.requestSignal) - : (options?.requestSignal ?? transportSignal); const init = { ...this._requestInit, method: 'POST', @@ -1023,14 +1191,29 @@ export class StreamableHTTPClientTransport implements Transport { this._scope = computeScopeUnion(this._scope, scope); } - if (this._authProvider.onUnauthorized && !isAuthRetry) { + const onUnauthorized = this._authProvider.onUnauthorized?.bind(this._authProvider); + if (onUnauthorized && !isAuthRetry) { try { - await this._authProvider.onUnauthorized({ - response, - serverUrl: this._url, - fetchFn: this._fetchWithInit - }); + // Raced against the per-request/transport abort so + // a hung 401 recovery cannot park the send (#2643); + // the signal is also handed to the provider so a + // cooperative implementation can cancel its own work. + await raceWithSignal( + () => + onUnauthorized({ + response, + serverUrl: this._url, + fetchFn: this._fetchWithInit, + signal + }), + signal + ); } catch (error) { + // An abort is an intentional teardown, not an auth + // failure — leave it unstamped. + if (isAborted(signal)) { + throw error; + } // Auth-seam stamp: covers the SDK's OAuth flow and // custom onUnauthorized callbacks alike. throw markAuthSeamEscape(error); @@ -1057,9 +1240,16 @@ export class StreamableHTTPClientTransport implements Transport { const { resourceMetadataUrl, scope, error, errorDescription } = extractWWWAuthenticateParams(response); if (error === 'insufficient_scope') { - const result = await this._stepUpAuthorize( - { scope, resourceMetadataUrl, errorDescription, statusText: response.statusText, text }, - stepUpRetries + // Raced against the per-request/transport abort so a + // hung step-up authorization (metadata discovery, token + // exchange) cannot park the send (#2643). + const result = await raceWithSignal( + () => + this._stepUpAuthorize( + { scope, resourceMetadataUrl, errorDescription, statusText: response.statusText, text }, + stepUpRetries + ), + signal ); if (result !== 'AUTHORIZED') { throw markAuthSeamEscape(new UnauthorizedError()); @@ -1111,8 +1301,13 @@ export class StreamableHTTPClientTransport implements Transport { // if the accepted notification is initialized, we start the SSE stream // if it's supported by the server if (isInitializedNotification(message)) { - // Start without a lastEventId since this is a fresh connection - this._startOrAuthSse({ resumptionToken: undefined }).catch(error => this.onerror?.(error)); + // Start without a lastEventId since this is a fresh + // connection. `_startOrAuthSse`'s own catch already + // routes failures to onerror (suppressed for intentional + // aborts) before rethrowing — emitting here again would + // double-report. Standalone GET: no per-request stream, + // so there is no terminal callback to fire either. + this._startOrAuthSse({ resumptionToken: undefined }).catch(() => {}); } return; } @@ -1161,13 +1356,13 @@ export class StreamableHTTPClientTransport implements Transport { await response.text?.().catch(() => {}); } } catch (error) { - // Intentional per-request abort BEFORE response headers (the - // `subscriptions/listen` driver aborting its `requestSignal`): - // fetch rejects with AbortError. Same guard as - // `_handleSseStream`'s `isIntentionalAbort` — do not surface a - // misleading onerror; still rethrow so `listen()`'s send-catch - // settles the per-subscription state machine. - if (options?.requestSignal?.aborted !== true) { + // Intentional abort BEFORE response headers — the + // `subscriptions/listen` driver aborting its `requestSignal`, or + // `close()` aborting the transport signal while the send is in + // the auth chain or the fetch. Do not surface a misleading + // onerror; still rethrow so `listen()`'s send-catch settles the + // per-subscription state machine. + if (!this._isIntentionalAbort(options?.requestSignal)) { this.onerror?.(error as Error); } throw error; @@ -1195,13 +1390,14 @@ export class StreamableHTTPClientTransport implements Transport { } try { - const headers = await this._commonHeaders(); + const signal = this._combinedSignal(); + const headers = await this._commonHeaders(signal); const init = { ...this._requestInit, method: 'DELETE', headers, - signal: this._abortController?.signal + signal }; const response = await (this._fetch ?? fetch)(this._url, init); @@ -1222,7 +1418,13 @@ export class StreamableHTTPClientTransport implements Transport { this._sessionId = undefined; } catch (error) { - this.onerror?.(error as Error); + // A transport-lifetime abort (close() during a DELETE parked in + // the auth chain or the fetch) is intentional teardown — same + // discipline as `_send`'s catch. No per-request signal on this + // path. Still rethrow so the caller observes the failure. + if (!this._isIntentionalAbort()) { + this.onerror?.(error as Error); + } throw error; } } diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index cfe9f88389..0d6f3c0e39 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -15,7 +15,8 @@ export type { OAuthClientInformationContext, OAuthClientProvider, OAuthDiscoveryState, - OAuthServerInfo + OAuthServerInfo, + UnauthorizedContext } from './client/auth'; export { assertSecureTokenEndpoint, diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index a36bbc0ad3..caa7436ca9 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -1246,6 +1246,84 @@ describe('StreamableHTTPClientTransport', () => { expect(fetchMock.mock.calls[1]![1]?.method).toBe('GET'); }); + it('a resumed stream dropping before its first ID-bearing event keeps the prior resumption token', async () => { + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 1, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + const fetchMock = globalThis.fetch as Mock; + // The resumed GET's stream errors before delivering any event… + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream({ + start(controller) { + controller.error(new Error('dropped before first event')); + } + }) + }); + // …and the retry GET succeeds. + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: new ReadableStream() + }); + + await transport.start(); + await transport['_startOrAuthSse']({ resumptionToken: 'tok-1' }); + await vi.advanceTimersByTimeAsync(20); + + expect(fetchMock).toHaveBeenCalledTimes(2); + // The retry must NOT lose Last-Event-ID: no event arrived to + // supersede the token the dropped stream was resumed from. + const retryHeaders = (fetchMock.mock.calls[1]![1] as { headers: Headers }).headers; + expect(retryHeaders.get('last-event-id')).toBe('tok-1'); + }); + + it('reports a failed reconnect attempt via onerror exactly once', async () => { + // _startOrAuthSse's own catch routes the failure to onerror before + // rethrowing; the reconnect scheduling must not report it again. + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 1, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + const errorSpy = vi.fn(); + transport.onerror = errorSpy; + + const failingStream = new ReadableStream({ + start(controller) { + controller.error(new Error('Network failure')); + } + }); + const fetchMock = globalThis.fetch as Mock; + // Initial GET stream drops, scheduling a reconnect… + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body: failingStream + }); + // …whose GET then fails outright. + fetchMock.mockRejectedValueOnce(new Error('connection refused')); + + await transport.start(); + await transport['_startOrAuthSse']({}); + await vi.advanceTimersByTimeAsync(20); + + const refusedReports = errorSpy.mock.calls.filter(([e]) => String((e as Error).message).includes('connection refused')); + expect(refusedReports).toHaveLength(1); + }); + it('should NOT reconnect a POST-initiated stream that fails', async () => { // ARRANGE transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { @@ -2621,6 +2699,37 @@ describe('StreamableHTTPClientTransport', () => { vi.useRealTimers(); }); + it('a scheduler that throws while re-arming fires onRequestStreamEnd (terminal for observed streams)', async () => { + let armed: (() => void) | undefined; + transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions, + reconnectionScheduler: (cb, _delay, attemptCount) => { + if (attemptCount > 0) throw new Error('scheduler exploded'); + armed = cb; + return () => {}; + } + }); + const onerror = vi.fn(); + transport.onerror = onerror; + const onRequestStreamEnd = vi.fn(); + (globalThis.fetch as Mock).mockRejectedValue(new Error('connection refused')); + await transport.start(); + + (transport as unknown as { _scheduleReconnection(opts: StartSSEOptions, attempt?: number): void })._scheduleReconnection( + { onRequestStreamEnd }, + 0 + ); + expect(armed).toBeDefined(); + // Attempt 0 runs, its GET fails, and re-arming attempt 1 throws. + armed!(); + await vi.advanceTimersByTimeAsync(0); + + // The retry chain is dead — that is terminal for an observed + // per-request stream, same as budget exhaustion. + expect(onRequestStreamEnd).toHaveBeenCalledTimes(1); + expect(onerror.mock.calls.some(([e]) => (e as Error).message.includes('scheduler exploded'))).toBe(true); + }); + it('invokes the custom scheduler with reconnect, delay, and attemptCount', () => { const scheduler = vi.fn(); transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { @@ -2736,4 +2845,186 @@ describe('StreamableHTTPClientTransport', () => { expect(onclose).toHaveBeenCalledTimes(1); }); }); + + describe('abortable auth awaits (requestSignal reaches the auth chain)', () => { + const sleep = (ms: number) => new Promise(r => setTimeout(r, ms)); + const request: JSONRPCMessage = { jsonrpc: '2.0', method: 'tools/list', params: {}, id: 1 }; + + const settledOrHung = async (p: Promise): Promise => { + const settled = p.then( + () => 'resolved', + e => e + ); + return Promise.race([settled, sleep(500).then(() => 'send() hung')]); + }; + + it('requestSignal abort settles a send parked in a hung token()', async () => { + const hungTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { token: () => new Promise(() => {}) } + }); + const onerror = vi.fn(); + hungTransport.onerror = onerror; + await hungTransport.start(); + + const ac = new AbortController(); + const pending = hungTransport.send(request, { requestSignal: ac.signal }); + setTimeout(() => ac.abort(new Error('caller abort')), 20); + + const outcome = await settledOrHung(pending); + expect(outcome).toBeInstanceOf(Error); + expect((outcome as Error).message).toContain('caller abort'); + // A per-request abort is intentional — no misleading onerror. + expect(onerror).not.toHaveBeenCalled(); + // The token() fetch never ran. + expect(globalThis.fetch).not.toHaveBeenCalled(); + await hungTransport.close(); + }); + + it('transport close() settles a send parked in a hung token()', async () => { + const hungTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { token: () => new Promise(() => {}) } + }); + const onerror = vi.fn(); + hungTransport.onerror = onerror; + await hungTransport.start(); + + const pending = hungTransport.send(request); + setTimeout(() => void hungTransport.close(), 20); + + const outcome = await settledOrHung(pending); + expect(outcome).toBeInstanceOf(Error); + expect(outcome).not.toBe('send() hung'); + // A transport-lifetime abort is intentional teardown — same + // isIntentionalAbort discipline as the per-request abort. + expect(onerror).not.toHaveBeenCalled(); + }); + + it('transport close() settles terminateSession() parked in a hung token(), without onerror', async () => { + const hungTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { token: () => new Promise(() => {}) } + }); + const onerror = vi.fn(); + hungTransport.onerror = onerror; + await hungTransport.start(); + // terminateSession early-returns without a session id. + hungTransport['_sessionId'] = 'session-1'; + + const pending = hungTransport.terminateSession(); + setTimeout(() => void hungTransport.close(), 20); + + const outcome = await settledOrHung(pending); + expect(outcome).toBeInstanceOf(Error); + expect(outcome).not.toBe('send() hung'); + // Transport-lifetime abort is intentional teardown — no onerror. + expect(onerror).not.toHaveBeenCalled(); + }); + + it('an already-aborted requestSignal rejects send() before the provider is invoked', async () => { + // The aborted fast-path must run BEFORE the producer: a dead + // request must not start side-effectful auth work. + const token = vi.fn(() => Promise.resolve('t')); + const gatedTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { token } + }); + await gatedTransport.start(); + + const ac = new AbortController(); + ac.abort(new Error('already dead')); + await expect(gatedTransport.send(request, { requestSignal: ac.signal })).rejects.toThrow('already dead'); + expect(token).not.toHaveBeenCalled(); + expect(globalThis.fetch).not.toHaveBeenCalled(); + await gatedTransport.close(); + }); + + it('a plain-JS provider returning a bare (non-thenable) token still works through the race', async () => { + const bareTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + // What a plain-JS caller can hand over despite the types: a + // synchronous return. The pre-race `await` tolerated it. + authProvider: { token: (() => 'bare-token') as unknown as () => Promise } + }); + await bareTransport.start(); + + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: true, + status: 202, + headers: new Headers() + }); + + await bareTransport.send({ jsonrpc: '2.0', method: 'notifications/x', params: {} }); + const [, init] = (globalThis.fetch as Mock).mock.calls[0] as [unknown, { headers: Headers }]; + expect(init.headers.get('authorization')).toBe('Bearer bare-token'); + await bareTransport.close(); + }); + + it('a non-Error rejection from the raced auth await passes through identity-preserved (auth-seam stamp intact)', async () => { + // The auth-seam stamp is a property set on the thrown VALUE + // (Error or not); a rewrap inside the race would strip it on + // paths with no re-stamping catch (the step-up awaits). + const sentinel = { code: 'custom-auth-failure' }; + const failingTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { token: () => Promise.reject(sentinel) } + }); + await failingTransport.start(); + // The transport-lifetime signal engages the race; the rejection + // must surface as the very same object (markAuthSeamEscape stamps + // in place, preserving identity). + const outcome = await failingTransport.send(request).then( + () => 'resolved (unexpected)', + (e: unknown) => e + ); + expect(outcome).toBe(sentinel); + await failingTransport.close(); + }); + + it('requestSignal abort settles a send parked in a hung onUnauthorized(); the signal is offered to the provider', async () => { + let seenSignal: AbortSignal | undefined; + const hungTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { + token: async () => 'stale-token', + onUnauthorized: ctx => { + seenSignal = ctx.signal; + return new Promise(() => {}); + } + } + }); + const onerror = vi.fn(); + hungTransport.onerror = onerror; + await hungTransport.start(); + + (globalThis.fetch as Mock).mockResolvedValueOnce({ + ok: false, + status: 401, + statusText: 'Unauthorized', + headers: new Headers(), + text: async () => '' + }); + + const ac = new AbortController(); + const pending = hungTransport.send(request, { requestSignal: ac.signal }); + setTimeout(() => ac.abort(new Error('caller abort')), 20); + + const outcome = await settledOrHung(pending); + expect(outcome).toBeInstanceOf(Error); + expect((outcome as Error).message).toContain('caller abort'); + expect(onerror).not.toHaveBeenCalled(); + // The provider was handed the combined signal so cooperative + // implementations can cancel their own recovery work. + expect(seenSignal).toBeDefined(); + expect(seenSignal?.aborted).toBe(true); + await hungTransport.close(); + }); + + it('a token() rejection (without an abort) is still surfaced as an auth failure, not swallowed by the race', async () => { + const failingTransport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + authProvider: { + token: async () => { + throw new UnauthorizedError('no credentials'); + } + } + }); + await failingTransport.start(); + await expect(failingTransport.send(request)).rejects.toThrow('no credentials'); + await failingTransport.close(); + }); + }); }); diff --git a/test/e2e/scenarios/transport-http.test.ts b/test/e2e/scenarios/transport-http.test.ts index 7cda0f28aa..4a03d59941 100644 --- a/test/e2e/scenarios/transport-http.test.ts +++ b/test/e2e/scenarios/transport-http.test.ts @@ -1482,7 +1482,11 @@ verifies('client-transport:http:reconnect-failure-onerror', async (_args: TestAr await vi.waitFor(() => expect(transportErrors.filter(e => e.message === 'Maximum reconnection attempts (2) exceeded.')).toHaveLength(1) ); - expect(transportErrors.filter(e => e.message.startsWith('Failed to reconnect SSE stream:'))).toHaveLength(2); + // EXACTLY once per failed attempt: the underlying failure itself. + // The 'Failed to reconnect SSE stream:' re-wrap that used to + // accompany each report was a duplicate of the same failure. + expect(transportErrors.filter(e => e.message.startsWith('Failed to open SSE stream:'))).toHaveLength(2); + expect(transportErrors.filter(e => e.message.startsWith('Failed to reconnect SSE stream:'))).toHaveLength(0); expect(records.filter(r => r.method === 'GET')).toHaveLength(3); // The reconnection failure stays on onerror: an unrelated request issued afterwards still succeeds