From e01c0ae4817bff27d6e8cd6f8e42b9ef8845b010 Mon Sep 17 00:00:00 2001 From: ayush00git Date: Wed, 2 Sep 2026 14:10:59 +0530 Subject: [PATCH] fix(client): preserve resumption token when a resumed SSE stream drops before any id-bearing event Seed _handleSseStream's lastEventId tracker from the resumptionToken the stream was opened with, so a reconnect that saw no new events re-sends the same Last-Event-ID instead of dropping it. Fixes #2499 --- ...rve-resumption-token-on-empty-reconnect.md | 20 ++++ packages/client/src/client/streamableHttp.ts | 9 +- .../client/test/client/streamableHttp.test.ts | 113 ++++++++++++++++++ 3 files changed, 140 insertions(+), 2 deletions(-) create mode 100644 .changeset/preserve-resumption-token-on-empty-reconnect.md diff --git a/.changeset/preserve-resumption-token-on-empty-reconnect.md b/.changeset/preserve-resumption-token-on-empty-reconnect.md new file mode 100644 index 0000000000..eebbdd234c --- /dev/null +++ b/.changeset/preserve-resumption-token-on-empty-reconnect.md @@ -0,0 +1,20 @@ +--- +'@modelcontextprotocol/client': patch +--- + +Preserve the resumption token across a resumed SSE stream that drops before any +id-bearing event arrives. `StreamableHTTPClientTransport._handleSseStream()` tracked the +latest event id in a local that started as `undefined` and was never seeded from the +`resumptionToken` the stream was opened with, yet both reconnect paths forwarded that +local as the next attempt's token. A stream resumed with `Last-Event-ID: e1` that closed +again before replaying any event — a load-balancer idle timeout, a server restart — was +therefore reconnected with no `Last-Event-ID` header at all. The server treated that as a +brand-new stream, missed events were never replayed, and a long-running request hung +until its timeout. + +The tracker is now seeded from the incoming `resumptionToken`, so a reconnect that saw no +new events re-sends the same cursor. Replay from an already-seen cursor is idempotent, +and a newer event id still overrides the seed as soon as one arrives. `onresumptiontoken` +is unchanged: it fires only for ids actually received on the wire, not for the seed. + +Fixes #2499. diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index c6eaef46d8..0d45959a02 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -720,7 +720,7 @@ export class StreamableHTTPClientTransport implements Transport { options.onRequestStreamEnd?.(); return; } - const { onresumptiontoken, replayMessageId, requestSignal, onRequestStreamEnd } = options; + const { resumptionToken, onresumptiontoken, replayMessageId, requestSignal, onRequestStreamEnd } = options; // An intentional abort — transport-wide close OR a per-request abort // (McpSubscription.close() aborting its `requestSignal`) — must read as // a clean shutdown: no misleading "SSE stream disconnected" onerror, @@ -728,7 +728,12 @@ export class StreamableHTTPClientTransport implements Transport { // caller just tore down. const isIntentionalAbort = (): boolean => this._abortController?.signal.aborted === true || requestSignal?.aborted === true; - let lastEventId: string | undefined; + // Seed from the token this stream was opened with: if the stream drops + // again before any id-bearing event arrives (LB idle timeout, server + // restart), the reconnect must re-send the same Last-Event-ID rather + // than silently downgrade to a fresh, non-resumed stream. Replay from + // the same cursor is idempotent, so re-sending it is always safe. + let lastEventId: string | undefined = resumptionToken; // Track whether we've received a priming event (event with ID) // Per spec, server SHOULD send a priming event with ID before closing let hasPrimingEvent = false; diff --git a/packages/client/test/client/streamableHttp.test.ts b/packages/client/test/client/streamableHttp.test.ts index a36bbc0ad3..1494945660 100644 --- a/packages/client/test/client/streamableHttp.test.ts +++ b/packages/client/test/client/streamableHttp.test.ts @@ -1246,6 +1246,119 @@ describe('StreamableHTTPClientTransport', () => { expect(fetchMock.mock.calls[1]![1]?.method).toBe('GET'); }); + describe('resumption token preserved across an empty resumed stream', () => { + // Regression for #2499: a stream opened with `resumptionToken` that + // drops again BEFORE any id-bearing event arrives (LB idle timeout, + // server restart) must re-send the same Last-Event-ID on reconnect + // rather than silently downgrading to a fresh, non-resumed stream. + const sseResponse = (body: ReadableStream | null) => ({ + ok: true, + status: 200, + headers: new Headers({ 'content-type': 'text/event-stream' }), + body + }); + + const makeTransport = () => + new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), { + reconnectionOptions: { + initialReconnectionDelay: 10, + maxRetries: 1, + maxReconnectionDelay: 1000, + reconnectionDelayGrowFactor: 1 + } + }); + + const lastEventIdOf = (fetchMock: Mock, call: number): string | null => + (fetchMock.mock.calls[call]![1]?.headers as Headers).get('last-event-id'); + + it('re-sends the same Last-Event-ID when the resumed stream closes gracefully before any id-bearing event', async () => { + // ARRANGE + transport = makeTransport(); + const fetchMock = globalThis.fetch as Mock; + // Resumed GET: accepted, but closes cleanly with no events. + fetchMock.mockResolvedValueOnce( + sseResponse( + new ReadableStream({ + start(controller) { + controller.close(); + } + }) + ) + ); + // The reconnect GET. + fetchMock.mockResolvedValueOnce(sseResponse(new ReadableStream())); + + // ACT + await transport.start(); + await transport['_startOrAuthSse']({ resumptionToken: 'event-1' }); + await vi.advanceTimersByTimeAsync(20); + + // ASSERT + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(lastEventIdOf(fetchMock, 0)).toBe('event-1'); + expect(lastEventIdOf(fetchMock, 1)).toBe('event-1'); + }); + + it('re-sends the same Last-Event-ID when the resumed stream errors before any id-bearing event', async () => { + // ARRANGE + transport = makeTransport(); + transport.onerror = vi.fn(); + const fetchMock = globalThis.fetch as Mock; + // Resumed GET: accepted, then the socket dies with no events. + fetchMock.mockResolvedValueOnce( + sseResponse( + new ReadableStream({ + start(controller) { + controller.error(new Error('Network failure')); + } + }) + ) + ); + fetchMock.mockResolvedValueOnce(sseResponse(new ReadableStream())); + + // ACT + await transport.start(); + await transport['_startOrAuthSse']({ resumptionToken: 'event-1' }); + await vi.advanceTimersByTimeAsync(20); + + // ASSERT + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(lastEventIdOf(fetchMock, 1)).toBe('event-1'); + }); + + it('prefers a newer event id over the seeded token once one arrives', async () => { + // ARRANGE + transport = makeTransport(); + const onresumptiontoken = vi.fn(); + const fetchMock = globalThis.fetch as Mock; + const encoder = new TextEncoder(); + // Resumed GET replays one id-bearing event, then closes. + fetchMock.mockResolvedValueOnce( + sseResponse( + new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('id: event-2\ndata: \n\n')); + controller.close(); + } + }) + ) + ); + fetchMock.mockResolvedValueOnce(sseResponse(new ReadableStream())); + + // ACT + await transport.start(); + await transport['_startOrAuthSse']({ resumptionToken: 'event-1', onresumptiontoken }); + await vi.advanceTimersByTimeAsync(20); + + // ASSERT + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(lastEventIdOf(fetchMock, 1)).toBe('event-2'); + // The seed is not re-announced; only genuinely new ids surface. + expect(onresumptiontoken).toHaveBeenCalledTimes(1); + expect(onresumptiontoken).toHaveBeenCalledWith('event-2'); + }); + }); + it('should NOT reconnect a POST-initiated stream that fails', async () => { // ARRANGE transport = new StreamableHTTPClientTransport(new URL('http://localhost:1234/mcp'), {