From 945773a9fe73675b39edb814275541b76f711ec6 Mon Sep 17 00:00:00 2001 From: Felix Weinberger <3823880+felixweinberger@users.noreply.github.com> Date: Fri, 23 Oct 2026 15:25:57 +0000 Subject: [PATCH 1/2] fix(client): do not wait for the end of an unread POST reply body **Before:** after a successful POST whose reply body the client does not read (a 202, a 2xx reply to a message with no requests, a 2xx reply with an unexpected content type, any OK reply on the SSE transport), `send()` awaited `response.text()`, which settles only when the body ends. A server or proxy that sends the headers and keeps the response open stalled `send()`, and with it `Client.connect()`. **After:** the body is cancelled without awaiting. A fully received reply keeps its connection for reuse; a partly received one has it closed. **History of these lines:** #1214 added `await response.body?.cancel()`. #1320 replaced it with `await response.text()`, because `cancel()` could hang. Both forms wait, and both waits can hang: `cancel()` on a response that a custom fetch has cloned does not settle until the other branch is consumed. Not awaiting avoids both; a test covers the cloned case. A custom fetch may return a body that is not a WHATWG stream (node-fetch returns a Node stream, which has no `cancel()`). Such a body is still read with `response.text()`, also without awaiting. The four sites share one internal helper, `releaseBody()`, which never throws or rejects. The non-2xx drains, the GET stream and the DELETE are left as they are. --- .changeset/release-post-reply-body.md | 5 + packages/client/src/client/sse.ts | 3 +- packages/client/src/client/streamableHttp.ts | 7 +- .../core-internal/src/shared/transport.ts | 6 + .../test/client/heldOpenPostReply.test.ts | 195 ++++++++++++++++++ 5 files changed, 212 insertions(+), 4 deletions(-) create mode 100644 .changeset/release-post-reply-body.md create mode 100644 test/integration/test/client/heldOpenPostReply.test.ts diff --git a/.changeset/release-post-reply-body.md b/.changeset/release-post-reply-body.md new file mode 100644 index 0000000000..29073d9d07 --- /dev/null +++ b/.changeset/release-post-reply-body.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +`StreamableHTTPClientTransport.send()` and `SSEClientTransport.send()` no longer wait for the end of a successful POST reply whose body they do not read (a `202`, a reply to a message with no requests, a reply with an unexpected content type). A server or proxy that sends the headers and keeps the response open no longer stalls `send()` or `Client.connect()`. The unread body is now cancelled without waiting, instead of being read to its end; when part of it has not arrived yet, the connection is closed instead of reused. A body with no `cancel()`, as a custom `fetch` may return, is still read, but no longer waited for. diff --git a/packages/client/src/client/sse.ts b/packages/client/src/client/sse.ts index 55c346ac85..8cadfcd5bf 100644 --- a/packages/client/src/client/sse.ts +++ b/packages/client/src/client/sse.ts @@ -5,6 +5,7 @@ import { fetchLeavingRedirects, fetchWithinOrigin, JSONRPCMessageSchema, + releaseBody, SdkError, SdkErrorCode, SdkHttpError, @@ -491,7 +492,7 @@ export class SSEClientTransport implements Transport { } // Release connection - POST responses don't have content we need - await response.text?.().catch(() => {}); + releaseBody(response); } catch (error) { this.onerror?.(error as Error); throw error; diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 091674e101..9f49b4ecb5 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -16,6 +16,7 @@ import { mcpNameSource, mediaTypeEssence, PROTOCOL_VERSION_META_KEY, + releaseBody, SdkError, SdkErrorCode, SdkHttpError, @@ -1164,7 +1165,7 @@ export class StreamableHTTPClientTransport implements Transport { // If the response is 202 Accepted, there's no body to process if (response.status === 202) { - await response.text?.().catch(() => {}); + releaseBody(response); // if the accepted notification is initialized, we start the SSE stream // if it's supported by the server if (isInitializedNotification(message)) { @@ -1208,14 +1209,14 @@ export class StreamableHTTPClientTransport implements Transport { this.onmessage?.(msg); } } else { - await response.text?.().catch(() => {}); + releaseBody(response); throw new SdkError(SdkErrorCode.ClientHttpUnexpectedContent, `Unexpected content type: ${contentType}`, { contentType }); } } else { // No requests in message but got 200 OK - still need to release connection - await response.text?.().catch(() => {}); + releaseBody(response); } } catch (error) { // Intentional per-request abort BEFORE response headers (the diff --git a/packages/core-internal/src/shared/transport.ts b/packages/core-internal/src/shared/transport.ts index 1679995081..89194fd444 100644 --- a/packages/core-internal/src/shared/transport.ts +++ b/packages/core-internal/src/shared/transport.ts @@ -111,6 +111,12 @@ export function unfollowedRedirect(url: string | URL, response: Response): strin return `Redirect to ${target.href} not followed; use that URL as the endpoint if it is the intended server`; } +/** Releases the unread body of `response` without waiting for it; reads it to its end when it has no `cancel()`. Never throws. */ +export function releaseBody(response: Response): void { + const release = async () => (typeof response.body?.cancel === 'function' ? response.body.cancel() : response.text?.()); + release().catch(() => {}); +} + /** * Options for sending a JSON-RPC message. */ diff --git a/test/integration/test/client/heldOpenPostReply.test.ts b/test/integration/test/client/heldOpenPostReply.test.ts new file mode 100644 index 0000000000..fac63c36d6 --- /dev/null +++ b/test/integration/test/client/heldOpenPostReply.test.ts @@ -0,0 +1,195 @@ +import type { IncomingMessage, Server, ServerResponse } from 'node:http'; +import { createServer } from 'node:http'; +import type { AddressInfo } from 'node:net'; +import { Readable } from 'node:stream'; + +import { Client, SSEClientTransport, StreamableHTTPClientTransport } from '@modelcontextprotocol/client'; +import { afterEach, describe, expect, it } from 'vitest'; + +const initializeResult = (id: unknown) => ({ + jsonrpc: '2.0', + id, + result: { protocolVersion: '2025-06-18', capabilities: {}, serverInfo: { name: 'held-open', version: '1.0.0' } } +}); + +const readJson = async (req: IncomingMessage) => { + let raw = ''; + for await (const chunk of req) raw += chunk; + return JSON.parse(raw); +}; + +// Sends the headers of a bodiless reply and never ends the response. +const holdOpen = (res: ServerResponse, status = 202, headers?: Record) => { + res.writeHead(status, headers); + res.flushHeaders(); +}; + +// A fetch whose POST replies carry `body` in place of their own. +const fetchWithBody = + (body: unknown, replies: Response[] = []): typeof fetch => + async (input, init) => { + const response = await fetch(input, init); + if (init?.method !== 'POST') return response; + replies.push(response); + return Object.defineProperty(response, 'body', { value: body }); + }; + +describe('POST reply that is never ended', () => { + let server: Server; + + const listen = async (handler: (req: IncomingMessage, res: ServerResponse) => void) => { + server = createServer(handler); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + return new URL(`http://127.0.0.1:${(server.address() as AddressInfo).port}/mcp`); + }; + + afterEach(async () => { + server.closeAllConnections(); + await new Promise(resolve => server.close(resolve)); + }); + + const streamableHttpServer = () => + listen(async (req, res) => { + if (req.method !== 'POST') { + res.writeHead(405).end(); + return; + } + const message = await readJson(req); + if (message.method === 'initialize') { + res.writeHead(200, { 'content-type': 'application/json' }).end(JSON.stringify(initializeResult(message.id))); + return; + } + holdOpen(res); + }); + + const sseServer = () => { + let stream: ServerResponse | undefined; + return listen(async (req, res) => { + if (req.method === 'GET') { + res.writeHead(200, { 'content-type': 'text/event-stream' }); + res.write('event: endpoint\ndata: /messages\n\n'); + stream = res; + return; + } + const message = await readJson(req); + holdOpen(res); + if (message.method === 'initialize') { + stream?.write(`event: message\ndata: ${JSON.stringify(initializeResult(message.id))}\n\n`); + } + }); + }; + + describe.each([ + ['the global fetch', fetch], + ['a fetch whose reply body is a Node stream, which has no cancel()', fetchWithBody(Readable.from([]))], + [ + 'a fetch whose reply body has a cancel() that throws', + fetchWithBody({ + cancel() { + throw new Error('cancel failed'); + } + }) + ], + ['a fetch whose reply body has a cancel() that returns undefined', fetchWithBody({ cancel() {} })] + ])('with %s', (_name, fetch) => { + it('does not stall send() on the streamable HTTP transport', async () => { + const transport = new StreamableHTTPClientTransport(await streamableHttpServer(), { fetch }); + await transport.start(); + + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + + await transport.close(); + }); + + it('does not stall Client.connect() over streamable HTTP', async () => { + const client = new Client({ name: 'test', version: '1.0.0' }); + + await client.connect(new StreamableHTTPClientTransport(await streamableHttpServer(), { fetch })); + + expect(client.getServerVersion()?.name).toBe('held-open'); + await client.close(); + }); + + it('does not stall send() of a notification on a 200 reply', async () => { + const url = await listen((_req, res) => holdOpen(res, 200)); + const transport = new StreamableHTTPClientTransport(url, { fetch }); + await transport.start(); + + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + + await transport.close(); + }); + + it('does not stall the rejection of a reply with an unexpected content type', async () => { + const url = await listen((_req, res) => holdOpen(res, 200, { 'content-type': 'text/plain' })); + const transport = new StreamableHTTPClientTransport(url, { fetch }); + await transport.start(); + + await expect(transport.send({ jsonrpc: '2.0', id: 1, method: 'ping' })).rejects.toThrow('Unexpected content type: text/plain'); + + await transport.close(); + }); + + it('does not stall send() on the SSE transport', async () => { + const transport = new SSEClientTransport(await sseServer(), { fetch }); + await transport.start(); + + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + + await transport.close(); + }); + + it('does not stall Client.connect() over SSE', async () => { + const client = new Client({ name: 'test', version: '1.0.0' }); + + await client.connect(new SSEClientTransport(await sseServer(), { fetch })); + + expect(client.getServerVersion()?.name).toBe('held-open'); + await client.close(); + }); + }); + + it('does not stall send() when a custom fetch has cloned the response', async () => { + const url = await listen((req, res) => { + req.resume(); + res.writeHead(202).end(); + }); + const fetchWithClone: typeof fetch = async (input, init) => { + const response = await fetch(input, init); + response.clone(); + return response; + }; + const transport = new StreamableHTTPClientTransport(url, { fetch: fetchWithClone }); + await transport.start(); + + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + + await transport.close(); + }); + + it('closes the connection of the reply', async () => { + let closed: Promise | undefined; + const url = await listen((_req, res) => { + closed = new Promise(resolve => res.on('close', resolve)); + holdOpen(res); + }); + const transport = new StreamableHTTPClientTransport(url); + await transport.start(); + + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + + await closed; + await transport.close(); + }); + + it('reads a reply whose body has no cancel()', async () => { + const replies: Response[] = []; + const transport = new StreamableHTTPClientTransport(await streamableHttpServer(), { fetch: fetchWithBody({}, replies) }); + await transport.start(); + + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + + expect(replies[0]?.bodyUsed).toBe(true); + await transport.close(); + }); +}); From c140325c1d1f01ca9e4d1f97627fd540d06897e1 Mon Sep 17 00:00:00 2001 From: Felix Weinberger <3823880+felixweinberger@users.noreply.github.com> Date: Tue, 27 Oct 2026 19:46:30 +0000 Subject: [PATCH 2/2] fix(client): let an unread POST reply end before releasing it **Before:** `releaseBody()` cancelled the unread body of a 2xx POST reply at once. That ended the stall on a reply that is never ended, but it also cut replies of normal servers whose end had not arrived yet. Measured against `main`, with nothing held open: about one new connection per 202 through a TLS reverse proxy built on Go's `httputil.ReverseProxy` (85 to 107 connections against 3 to 8, for 100 notifications and 100 calls), from a workerd client to this SDK's servers (50 to 52 against 0 to 2, for 50 notifications) and on the SSE transport to a Python server over TLS (206 against 3); Chromium logged each 202 POST as `net::ERR_ABORTED`; and with a custom `fetch` that returns `response.clone()`, `close()` was followed by an unhandled `AbortError` on Node 20 and 22. **After:** the body is read to its end, still without `send()` waiting for it, and is cancelled only if it has not ended after 100 ms. The same runs now give the results of `main`. A reply that is never ended still does not stall `send()` or `Client.connect()`; its connection closes after 100 ms instead of at once. The timer is cleared when the body ends or fails. A body with no `getReader()` (node-fetch) is read with `response.text()`, also without waiting, and is never cancelled. **Still open:** only POST replies with a 2xx status are covered. Replies with another status, the redirect drain and `terminateSession()` wait as before. When the reply has not ended after 100 ms and the custom `fetch` returns `response.clone()`, a later `close()` is still followed by the unhandled `AbortError` on Node 20 and 22. Each of the four call sites now has tests that fail when its call is deleted. --- .changeset/release-post-reply-body.md | 2 +- .../core-internal/src/shared/transport.ts | 18 +- .../test/client/heldOpenPostReply.test.ts | 173 ++++++++++++++++-- 3 files changed, 175 insertions(+), 18 deletions(-) diff --git a/.changeset/release-post-reply-body.md b/.changeset/release-post-reply-body.md index 29073d9d07..4142859d57 100644 --- a/.changeset/release-post-reply-body.md +++ b/.changeset/release-post-reply-body.md @@ -2,4 +2,4 @@ '@modelcontextprotocol/client': patch --- -`StreamableHTTPClientTransport.send()` and `SSEClientTransport.send()` no longer wait for the end of a successful POST reply whose body they do not read (a `202`, a reply to a message with no requests, a reply with an unexpected content type). A server or proxy that sends the headers and keeps the response open no longer stalls `send()` or `Client.connect()`. The unread body is now cancelled without waiting, instead of being read to its end; when part of it has not arrived yet, the connection is closed instead of reused. A body with no `cancel()`, as a custom `fetch` may return, is still read, but no longer waited for. +`StreamableHTTPClientTransport.send()` and `SSEClientTransport.send()` no longer wait for the end of a POST reply with a 2xx status whose body they do not read (a `202`, a reply to a message with no requests, a reply with an unexpected content type). A server or proxy that sends the headers of such a reply and keeps it open no longer stalls `send()` or `Client.connect()`. The unread body is still read to its end, but `send()` does not wait for that; a body that has not ended after 100 ms is cancelled, which closes its HTTP/1.1 connection unless a custom `fetch` holds a clone of the reply. A body with no `getReader()`, as a custom `fetch` may return, is read in the same way but never cancelled. POST replies with any other status and `terminateSession()` still wait for the end of the body. diff --git a/packages/core-internal/src/shared/transport.ts b/packages/core-internal/src/shared/transport.ts index 89194fd444..3bb485df92 100644 --- a/packages/core-internal/src/shared/transport.ts +++ b/packages/core-internal/src/shared/transport.ts @@ -111,9 +111,23 @@ export function unfollowedRedirect(url: string | URL, response: Response): strin return `Redirect to ${target.href} not followed; use that URL as the endpoint if it is the intended server`; } -/** Releases the unread body of `response` without waiting for it; reads it to its end when it has no `cancel()`. Never throws. */ +/** How long `releaseBody()` lets a body end by itself. */ +const RELEASE_BODY_DEADLINE_MS = 100; + +/** Reads the unread body of `response` to its end without waiting for it, and cancels what has not ended at the deadline. Never throws. */ export function releaseBody(response: Response): void { - const release = async () => (typeof response.body?.cancel === 'function' ? response.body.cancel() : response.text?.()); + const release = async () => { + if (typeof response.body?.getReader !== 'function') return void (await response.text()); + const reader = response.body.getReader(); + const cancel = async () => reader.cancel(); + const timer = setTimeout(() => void cancel().catch(() => {}), RELEASE_BODY_DEADLINE_MS); + try { + let done = false; + while (!done) ({ done } = await reader.read()); + } finally { + clearTimeout(timer); + } + }; release().catch(() => {}); } diff --git a/test/integration/test/client/heldOpenPostReply.test.ts b/test/integration/test/client/heldOpenPostReply.test.ts index fac63c36d6..b3a79af5d7 100644 --- a/test/integration/test/client/heldOpenPostReply.test.ts +++ b/test/integration/test/client/heldOpenPostReply.test.ts @@ -4,7 +4,7 @@ import type { AddressInfo } from 'node:net'; import { Readable } from 'node:stream'; import { Client, SSEClientTransport, StreamableHTTPClientTransport } from '@modelcontextprotocol/client'; -import { afterEach, describe, expect, it } from 'vitest'; +import { afterEach, describe, expect, it, onTestFinished, vi } from 'vitest'; const initializeResult = (id: unknown) => ({ jsonrpc: '2.0', @@ -18,8 +18,12 @@ const readJson = async (req: IncomingMessage) => { return JSON.parse(raw); }; +// Settles when the connection of the last held-open reply closes. +let closed: Promise; + // Sends the headers of a bodiless reply and never ends the response. const holdOpen = (res: ServerResponse, status = 202, headers?: Record) => { + closed = new Promise(resolve => res.on('close', resolve)); res.writeHead(status, headers); res.flushHeaders(); }; @@ -34,7 +38,29 @@ const fetchWithBody = return Object.defineProperty(response, 'body', { value: body }); }; -describe('POST reply that is never ended', () => { +// A mock that is called with each rejection that nothing handles, until the test ends. +const watchUnhandledRejections = () => { + const unhandled = vi.fn(); + process.on('unhandledRejection', unhandled); + onTestFinished(() => void process.off('unhandledRejection', unhandled)); + return unhandled; +}; + +const notification = { jsonrpc: '2.0', method: 'notifications/progress', params: { progressToken: 1, progress: 1 } } as const; + +// The four kinds of 2xx POST reply whose body a transport does not read; send() rejects on the third. +const unreadReplies = [ + ['a 202 reply', { status: 202, message: notification }], + ['a 200 reply to a notification', { status: 200, message: notification }], + [ + 'a 200 reply with an unexpected content type', + { status: 200, headers: { 'content-type': 'text/plain' }, message: { jsonrpc: '2.0', id: 1, method: 'ping' } } + ], + ['a reply on the SSE transport', { status: 202, message: notification, sse: true }] +] as const; +const accepted = unreadReplies[0][1]; + +describe('POST reply whose body the client does not read', () => { let server: Server; const listen = async (handler: (req: IncomingMessage, res: ServerResponse) => void) => { @@ -44,6 +70,8 @@ describe('POST reply that is never ended', () => { }; afterEach(async () => { + vi.useRealTimers(); + vi.restoreAllMocks(); server.closeAllConnections(); await new Promise(resolve => server.close(resolve)); }); @@ -79,18 +107,34 @@ describe('POST reply that is never ended', () => { }); }; + // Starts a transport whose next POST gets `reply`: held open by the server, or with `body` and without the network. + const start = async ({ status, headers, sse }: { status: number; headers?: Record; sse?: boolean }, body?: unknown) => { + const url = sse ? await sseServer() : await listen((_req, res) => holdOpen(res, status, headers)); + const fetchWithoutPost: typeof fetch = async (input, init) => + init?.method === 'POST' + ? Object.defineProperty(new Response(null, { status, headers }), 'body', { value: body }) + : fetch(input, init); + const options = body === undefined ? {} : { fetch: fetchWithoutPost }; + const transport = sse ? new SSEClientTransport(url, options) : new StreamableHTTPClientTransport(url, options); + await transport.start(); + return transport; + }; + describe.each([ ['the global fetch', fetch], - ['a fetch whose reply body is a Node stream, which has no cancel()', fetchWithBody(Readable.from([]))], + ['a fetch whose reply body is a Node stream, which has no getReader()', fetchWithBody(Readable.from([]))], [ - 'a fetch whose reply body has a cancel() that throws', + 'a fetch whose reply body has a getReader() that throws', fetchWithBody({ - cancel() { - throw new Error('cancel failed'); + getReader() { + throw new Error('locked'); } }) ], - ['a fetch whose reply body has a cancel() that returns undefined', fetchWithBody({ cancel() {} })] + [ + 'a fetch whose reply body has a read() that rejects', + fetchWithBody({ getReader: () => ({ read: () => Promise.reject(new Error('read failed')) }) }) + ] ])('with %s', (_name, fetch) => { it('does not stall send() on the streamable HTTP transport', async () => { const transport = new StreamableHTTPClientTransport(await streamableHttpServer(), { fetch }); @@ -167,22 +211,121 @@ describe('POST reply that is never ended', () => { await transport.close(); }); - it('closes the connection of the reply', async () => { - let closed: Promise | undefined; - const url = await listen((_req, res) => { - closed = new Promise(resolve => res.on('close', resolve)); - holdOpen(res); + it('leaves no unhandled rejection when a custom fetch returns a clone of the response', async () => { + const url = await listen((req, res) => { + req.resume(); + res.writeHead(202).end(); + }); + const originals: Response[] = []; + const fetchReturningClone: typeof fetch = async (input, init) => { + const response = await fetch(input, init); + originals.push(response); + return response.clone(); + }; + const unhandled = watchUnhandledRejections(); + const transport = new StreamableHTTPClientTransport(url, { fetch: fetchReturningClone }); + await transport.start(); + + await transport.send(notification); + await transport.close(); + + await new Promise(resolve => setTimeout(resolve, 10)); + expect(unhandled).not.toHaveBeenCalled(); + }); + + it('does not cut a reply that ends after its headers', async () => { + let ended: Promise | undefined; + const url = await listen((req, res) => { + ended = new Promise(resolve => res.on('close', () => resolve(res.writableFinished && !req.socket.destroyed))); + res.writeHead(202); + res.flushHeaders(); + setTimeout(() => res.end(), 10); }); const transport = new StreamableHTTPClientTransport(url); await transport.start(); - await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + await transport.send(notification); + + expect(await ended).toBe(true); + await transport.close(); + }); - await closed; + describe.each(unreadReplies)('%s', (_name, reply) => { + it('has its connection closed when it never ends', async () => { + const transport = await start(reply); + + await transport.send(reply.message).catch(() => {}); + + await closed; + await transport.close(); + }); + + it('is cancelled when it has not ended after 100 ms, and send() does not wait for that', async () => { + const cancel = vi.fn(); + const transport = await start(reply, new ReadableStream({ cancel })); + vi.useFakeTimers(); + + await transport.send(reply.message).catch(() => {}); + + await vi.advanceTimersByTimeAsync(99); + expect(cancel).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + expect(cancel).toHaveBeenCalledOnce(); + await transport.close(); + }); + + it('is read to its end and not cancelled when it ends', async () => { + const cancel = vi.fn(); + const pull = vi.fn((controller: ReadableStreamDefaultController) => + pull.mock.calls.length < 3 ? controller.enqueue(new Uint8Array(1)) : controller.close() + ); + const transport = await start(reply, new ReadableStream({ pull, cancel }, { highWaterMark: 0 })); + vi.useFakeTimers(); + + await transport.send(reply.message).catch(() => {}); + + await vi.advanceTimersByTimeAsync(0); + expect(pull).toHaveBeenCalledTimes(3); + expect(vi.getTimerCount()).toBe(0); + expect(cancel).not.toHaveBeenCalled(); + await transport.close(); + }); + }); + + it('leaves no timer when the body fails', async () => { + const transport = await start(accepted, new ReadableStream({ start: controller => controller.error(new Error('failed')) })); + vi.useFakeTimers(); + + await transport.send(notification); + + await vi.advanceTimersByTimeAsync(0); + expect(vi.getTimerCount()).toBe(0); + await transport.close(); + }); + + it.each([ + [ + 'throws', + () => { + throw new Error('cancel failed'); + } + ], + ['rejects', () => Promise.reject(new Error('cancel failed'))], + ['returns undefined', () => {}], + ['never settles', () => new Promise(() => {})] + ])('tolerates a cancel() that %s', async (_name, cancel) => { + const unhandled = watchUnhandledRejections(); + const transport = await start(accepted, { getReader: () => ({ read: () => new Promise(() => {}), cancel }) }); + vi.useFakeTimers(); + + await transport.send(notification); + + await vi.advanceTimersByTimeAsync(100); + expect(unhandled).not.toHaveBeenCalled(); await transport.close(); }); - it('reads a reply whose body has no cancel()', async () => { + it('reads a reply whose body has no getReader()', async () => { const replies: Response[] = []; const transport = new StreamableHTTPClientTransport(await streamableHttpServer(), { fetch: fetchWithBody({}, replies) }); await transport.start();