From 30bb6fa994bc8967804e85ee276c5671b602d154 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Thu, 22 Oct 2026 12:14:01 +0200 Subject: [PATCH 1/2] feat(ai-client): back off reconnects and retry temporary HTTP statuses ReconnectOptions gets initialDelayMs, maxDelayMs, and budgetMs. The reconnect wait doubles with jitter, and a reconnect answered 408, 429, 502, 503, or 504 is retried, with Retry-After as the wait. The first wait stays 250 ms and the limit stays 5 attempts unless a budget is set. --- .changeset/client-reconnect-backoff.md | 5 + docs/resumable-streams/advanced.md | 22 +- docs/resumable-streams/websockets.md | 2 +- packages/ai-client/src/connection-adapters.ts | 177 ++++++++++++--- ...nection-adapters-reconnect-backoff.test.ts | 213 ++++++++++++++++++ ...tion-adapters-resumable-transports.test.ts | 42 ++++ .../connection-adapters-resumable.test.ts | 6 +- .../tests/connection-adapters-xhr.test.ts | 4 + packages/ai-client/tests/fetch-json.test.ts | 13 ++ 9 files changed, 451 insertions(+), 33 deletions(-) create mode 100644 .changeset/client-reconnect-backoff.md create mode 100644 packages/ai-client/tests/connection-adapters-reconnect-backoff.test.ts diff --git a/.changeset/client-reconnect-backoff.md b/.changeset/client-reconnect-backoff.md new file mode 100644 index 0000000000..9b0c100c76 --- /dev/null +++ b/.changeset/client-reconnect-backoff.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-client': minor +--- + +Reconnects of a durable run can ride out a server restart. The wait between reconnects now doubles from `initialDelayMs` (default 250) up to `maxDelayMs` (default 5000), with jitter. Set `budgetMs` to keep reconnecting for a total time instead of a fixed number of attempts. A reconnect answered 408, 429, 502, 503, or 504 is retried, and a `Retry-After` header sets the wait. `delayMs` still works as the old name of `initialDelayMs`. diff --git a/docs/resumable-streams/advanced.md b/docs/resumable-streams/advanced.md index 8032010e43..cb8cfc0717 100644 --- a/docs/resumable-streams/advanced.md +++ b/docs/resumable-streams/advanced.md @@ -197,11 +197,31 @@ import { fetchServerSentEvents } from '@tanstack/ai-client' function makeConnection() { return fetchServerSentEvents('/api/chat', { - reconnect: { maxAttempts: 5, delayMs: 250 }, // defaults shown + reconnect: { maxAttempts: 5, initialDelayMs: 250, maxDelayMs: 5000 }, // defaults shown }) } ``` +The wait doubles after each reconnect that delivers nothing: 250 ms, 500 ms, 1 s, and up to `maxDelayMs`. Jitter takes each wait from half of that value up to the full value, so many clients do not reconnect at the same moment. + +A reconnect also retries a response with a temporary status: 408, 429, 502, 503, or 504. A proxy sends these while the server behind it restarts. If the response has a `Retry-After` header, the client waits that long, up to `maxDelayMs`. Any other non-2xx status, for example 401 or 404, fails at once. + +#### Ride out a server restart + +A restart or a rolling deploy can take tens of seconds. With the defaults, the client gives up after about 8 seconds. To keep open chats alive through a deploy, set `budgetMs`: + +```ts +import { fetchServerSentEvents } from '@tanstack/ai-client' + +function makeConnection() { + return fetchServerSentEvents('/api/chat', { + reconnect: { budgetMs: 120_000 }, + }) +} +``` + +With `budgetMs`, the client reconnects with no attempt limit until the time is used. The budget starts at the first failed reconnect, and any new event resets it. If the next wait would go past the budget, the run fails with `StreamReconnectLimitError`. To also limit the attempts, set `maxAttempts` too. + `durableStream` limits its own read loop the same way. After a body read fails in the middle of a window, it retries from the last valid position, and it limits consecutive failures (`reconnect: { maxReadFailures: 10, delayMs: 250 }`). Normal long-poll progress is never throttled. ### WebSocket transport diff --git a/docs/resumable-streams/websockets.md b/docs/resumable-streams/websockets.md index 276c6215b4..2301dcf7ef 100644 --- a/docs/resumable-streams/websockets.md +++ b/docs/resumable-streams/websockets.md @@ -196,7 +196,7 @@ If a socket does not need to survive a drop, leave out `durability`. `toWebSocke | --- | --- | | `protocols` | WebSocket subprotocols, passed to the `WebSocket` constructor as they are. | | `body` | Static extra fields added to every outgoing `RunAgentInput` frame. Same as `body` on the HTTP connections. | -| `reconnect` | Reconnect limits (`maxAttempts`, `delayMs`). They work the same as on `fetchServerSentEvents`. See [Advanced](./advanced). | +| `reconnect` | Reconnect limits (`maxAttempts`, `initialDelayMs`, `maxDelayMs`, `budgetMs`). They work the same as on `fetchServerSentEvents`. See [Advanced](./advanced#reconnection-bounding). | | `WebSocketImpl` | Your own `WebSocket` implementation, for tests and non-browser runtimes. | [Transports](../transports/overview) shows where `webSocket()` fits next to the other connections. diff --git a/packages/ai-client/src/connection-adapters.ts b/packages/ai-client/src/connection-adapters.ts index b5a2db6086..2f317e312f 100644 --- a/packages/ai-client/src/connection-adapters.ts +++ b/packages/ai-client/src/connection-adapters.ts @@ -87,18 +87,60 @@ export class DurableStreamIncompleteError extends Error { * reconnecting without end. */ export class StreamReconnectLimitError extends Error { - constructor(attempts: number) { + constructor(attempts: number, budgetMs?: number) { super( - `Durable run exceeded its reconnect ceiling of ${attempts} attempts — giving up.`, + budgetMs === undefined + ? `Durable run exceeded its reconnect ceiling of ${attempts} attempts — giving up.` + : `Durable run did not reconnect within its budget of ${budgetMs} ms (${attempts} attempts) — giving up.`, ) this.name = 'StreamReconnectLimitError' } } +/** HTTP statuses that a proxy or server sends while it restarts or sheds load. */ +const TEMPORARY_HTTP_STATUSES = new Set([408, 429, 502, 503, 504]) + /** - * Reconnect bounding for resumable streams. A constant throttle delay prevents a - * hot loop against the origin, and the ceiling bounds a pathologically failing - * run — but only counts CONSECUTIVE reconnects that made no forward progress. + * A non-2xx response. A reconnect retries it when the status is temporary + * (408, 429, 502, 503, 504), and waits for its `Retry-After` when present. + */ +class StreamHttpError extends Error { + readonly retryAfterMs: number | undefined + constructor( + message: string, + readonly status: number, + retryAfter: string | null, + ) { + super(message) + this.name = 'StreamHttpError' + this.retryAfterMs = parseRetryAfterMs(retryAfter) + } +} + +/** `Retry-After` in seconds or as an HTTP date, in ms. */ +function parseRetryAfterMs(value: string | null): number | undefined { + if (!value) return undefined + const seconds = Number(value) + if (Number.isFinite(seconds)) return seconds >= 0 ? seconds * 1000 : undefined + const date = Date.parse(value) + return Number.isNaN(date) ? undefined : Math.max(0, date - Date.now()) +} + +function isTemporaryHttpError(error: unknown): error is StreamHttpError { + return ( + error instanceof StreamHttpError && + TEMPORARY_HTTP_STATUSES.has(error.status) + ) +} + +/** + * Reconnect bounding for resumable streams. An exponential backoff with jitter + * prevents a hot loop against the origin, and the ceiling (or the time budget) + * bounds a pathologically failing run — but only counts CONSECUTIVE reconnects + * that made no forward progress. + * + * A reconnect retries a dropped connection and a response with a temporary + * status: 408, 429, 502, 503 or 504. Any other non-2xx status fails at once. */ export interface ReconnectOptions { /** @@ -107,37 +149,72 @@ export interface ReconnectOptions { * zero whenever a reconnect makes forward progress, so a healthy long run — * even one behind a proxy that rolls the socket after every event — never * approaches it; the ceiling only fires when the run is genuinely stuck - * (reconnecting repeatedly without receiving anything new). Default 5. + * (reconnecting repeatedly without receiving anything new). Default 5, or no + * ceiling when `budgetMs` is set. */ maxAttempts?: number - /** Delay between reconnect attempts, in ms, to avoid hammering. Default 250. */ + /** + * Delay before the first reconnect, in ms. Each consecutive reconnect that + * delivers nothing doubles it, up to `maxDelayMs`. Jitter takes each delay + * from half of this value up to the full value. Default 250. + */ + initialDelayMs?: number + /** + * Longest delay between reconnects, in ms. Also caps a `Retry-After` wait. + * Default 5000. + */ + maxDelayMs?: number + /** + * Total time, in ms, to keep reconnecting without new events, counted from + * the first failed reconnect. When the next wait would go past it, the run + * fails with {@link StreamReconnectLimitError}. Default: no budget. + */ + budgetMs?: number + /** @deprecated Use `initialDelayMs`. */ delayMs?: number } interface ResolvedReconnectOptions { maxAttempts: number - delayMs: number + initialDelayMs: number + maxDelayMs: number + budgetMs: number | undefined } function resolveReconnectOptions( options: ReconnectOptions | undefined, ): ResolvedReconnectOptions { - const maxAttempts = options?.maxAttempts ?? 5 - const delayMs = options?.delayMs ?? 250 + const budgetMs = options?.budgetMs + const maxAttempts = + options?.maxAttempts ?? (budgetMs === undefined ? 5 : Infinity) + const initialDelayMs = options?.initialDelayMs ?? options?.delayMs ?? 250 + const maxDelayMs = options?.maxDelayMs ?? Math.max(5000, initialDelayMs) // Reject non-finite / negative bounds up front: a NaN or Infinity maxAttempts // would make the ceiling ineffective (unbounded reconnects), and a non-finite - // delayMs would remove throttling. Fail loudly on misconfiguration. - if (!Number.isInteger(maxAttempts) || maxAttempts < 0) { - throw new Error( - `Invalid reconnect.maxAttempts: ${maxAttempts}. Must be a non-negative integer.`, - ) + // delay would remove throttling. Fail loudly on misconfiguration. + if (options?.maxAttempts !== undefined) { + if (!Number.isInteger(maxAttempts) || maxAttempts < 0) { + throw new Error( + `Invalid reconnect.maxAttempts: ${maxAttempts}. Must be a non-negative integer.`, + ) + } + } + for (const [name, value] of [ + ['initialDelayMs', initialDelayMs], + ['maxDelayMs', maxDelayMs], + ] as const) { + if (!Number.isFinite(value) || value < 0) { + throw new Error( + `Invalid reconnect.${name}: ${value}. Must be a non-negative finite number.`, + ) + } } - if (!Number.isFinite(delayMs) || delayMs < 0) { + if (budgetMs !== undefined && (!Number.isFinite(budgetMs) || budgetMs <= 0)) { throw new Error( - `Invalid reconnect.delayMs: ${delayMs}. Must be a non-negative finite number.`, + `Invalid reconnect.budgetMs: ${budgetMs}. Must be a positive finite number.`, ) } - return { maxAttempts, delayMs } + return { maxAttempts, initialDelayMs, maxDelayMs, budgetMs } } /** @@ -160,11 +237,14 @@ export interface ReconnectTracker { /** * Throttle before a reconnect attempt. Resets the no-progress counter when * `madeProgress` is true; otherwise increments it and throws - * {@link StreamReconnectLimitError} once it exceeds the configured ceiling. + * {@link StreamReconnectLimitError} once it exceeds the configured ceiling + * or the next wait would go past the budget. `retryAfterMs` (from a + * `Retry-After` header) replaces the backoff delay, capped at `maxDelayMs`. */ waitBeforeReconnect: ( madeProgress: boolean, signal?: AbortSignal, + retryAfterMs?: number, ) => Promise } @@ -181,6 +261,8 @@ export function createReconnectTracker( const seen = new Set() let lastEventId: string | undefined let reconnectAttempts = 0 + // When the current run of no-progress reconnects started, for `budgetMs`. + let failingSince: number | undefined return { get lastEventId() { return lastEventId @@ -204,16 +286,38 @@ export function createReconnectTracker( // whose socket rolls after every event) never approaches the ceiling; it // fires only when the run is genuinely stuck — reconnecting repeatedly // with nothing new. - async waitBeforeReconnect(madeProgress, signal) { + async waitBeforeReconnect(madeProgress, signal, retryAfterMs) { if (madeProgress) { reconnectAttempts = 0 + failingSince = undefined } else { reconnectAttempts += 1 + failingSince ??= Date.now() if (reconnectAttempts > reconnect.maxAttempts) { throw new StreamReconnectLimitError(reconnect.maxAttempts) } } - await abortableDelay(reconnect.delayMs, signal) + // Exponential backoff with jitter: a delay from half of the backoff up + // to the full backoff, so a jittered delay is never longer. + const backoff = Math.min( + reconnect.maxDelayMs, + reconnect.initialDelayMs * 2 ** Math.max(0, reconnectAttempts - 1), + ) + const delayMs = + retryAfterMs === undefined + ? backoff * (0.5 + Math.random() / 2) + : Math.min(retryAfterMs, reconnect.maxDelayMs) + if ( + reconnect.budgetMs !== undefined && + failingSince !== undefined && + Date.now() - failingSince + delayMs > reconnect.budgetMs + ) { + throw new StreamReconnectLimitError( + reconnectAttempts - 1, + reconnect.budgetMs, + ) + } + await abortableDelay(delayMs, signal) }, } } @@ -502,8 +606,10 @@ async function assertResponseOk(response: Response): Promise { throw new ByokMissingError(body.error.provider) } } - throw new Error( + throw new StreamHttpError( `HTTP error! status: ${response.status} ${response.statusText}`, + response.status, + response.headers.get('retry-after'), ) } @@ -519,7 +625,11 @@ function errorFromXhrStatus(xhr: XMLHttpRequest): Error { return new ByokMissingError(parsed.error.provider) } } - return new Error(`XHR error! status: ${xhr.status} ${xhr.statusText}`) + return new StreamHttpError( + `XHR error! status: ${xhr.status} ${xhr.statusText}`, + xhr.status, + xhr.getResponseHeader('retry-after'), + ) } function chatHydrationSearchParams( @@ -798,13 +908,19 @@ async function* resumableStream( // ceiling in waitBeforeReconnect already bounds a genuinely stuck flapper, // so a per-attempt progress requirement here would only convert // recoverable drops into hard failures on flaky (mobile/edge) networks. - // Without an offset (a non-durable stream), surface the failure. + // Without an offset (a non-durable stream), surface the failure. A + // temporary status (a proxy while the server restarts) is a drop too. if ( (error instanceof StreamTruncatedError || - error instanceof StreamReadError) && + error instanceof StreamReadError || + isTemporaryHttpError(error)) && tracker.lastEventId !== undefined ) { - await tracker.waitBeforeReconnect(progressed, abortSignal) + await tracker.waitBeforeReconnect( + progressed, + abortSignal, + isTemporaryHttpError(error) ? error.retryAfterMs : undefined, + ) continue } throw error @@ -1730,9 +1846,14 @@ async function getJsonRunBody(options: JsonRunPollOptions, offset: string) { return await requestJsonRunBody(options.fetchClient, url, options.getInit) } catch (error) { const isTransportDrop = - error instanceof StreamReadError && !options.signal?.aborted + (error instanceof StreamReadError || isTemporaryHttpError(error)) && + !options.signal?.aborted if (!isTransportDrop) throw error - await tracker.waitBeforeReconnect(false, options.signal) + await tracker.waitBeforeReconnect( + false, + options.signal, + isTemporaryHttpError(error) ? error.retryAfterMs : undefined, + ) if (options.signal?.aborted) throw error } } diff --git a/packages/ai-client/tests/connection-adapters-reconnect-backoff.test.ts b/packages/ai-client/tests/connection-adapters-reconnect-backoff.test.ts new file mode 100644 index 0000000000..5681016097 --- /dev/null +++ b/packages/ai-client/tests/connection-adapters-reconnect-backoff.test.ts @@ -0,0 +1,213 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { EventType } from '@tanstack/ai/client' +import { + StreamReconnectLimitError, + fetchServerSentEvents, +} from '../src/connection-adapters' +import type { ReconnectOptions } from '../src/connection-adapters' + +/** A durable SSE body: one id-tagged event, then the server goes away. */ +function firstResponse() { + const frame = `id: 1\ndata: ${JSON.stringify({ + type: EventType.RUN_STARTED, + threadId: 'thread-1', + runId: 'run-1', + timestamp: 0, + })}\n\n` + return new Response(frame, { + headers: { 'content-type': 'text/event-stream' }, + }) +} + +/** The server is back: the run finishes. */ +function finishedResponse() { + const frame = `id: 2\ndata: ${JSON.stringify({ + type: EventType.RUN_FINISHED, + threadId: 'thread-1', + runId: 'run-1', + timestamp: 0, + })}\n\n` + return new Response(frame, { + headers: { 'content-type': 'text/event-stream' }, + }) +} + +/** + * Runs one durable run and records each request. `afterDrop` answers each + * reconnect, with its 1-based index. + */ +async function run( + afterDrop: (reconnect: number) => Promise, + reconnect?: ReconnectOptions, +) { + const requests: Array<{ at: number; lastEventId: string | null }> = [] + const start = Date.now() + const connection = fetchServerSentEvents('http://chat.test/run', { + fetchClient: async (_input, init) => { + const headers = new Headers(init?.headers) + requests.push({ + at: Date.now() - start, + lastEventId: headers.get('Last-Event-ID'), + }) + return requests.length === 1 + ? firstResponse() + : afterDrop(requests.length - 1) + }, + ...(reconnect && { reconnect }), + }) + const chunks: Array = [] + let error: unknown + const done = (async () => { + try { + for await (const chunk of connection.connect( + [{ role: 'user', content: 'Hi.' }], + undefined, + undefined, + { threadId: 'thread-1', runId: 'run-1' }, + )) { + chunks.push(chunk.type) + } + } catch (caught) { + error = caught + } + })() + await vi.runAllTimersAsync() + await done + const gaps = requests + .slice(1) + .map((request, index) => request.at - (requests[index]?.at ?? 0)) + return { requests, gaps, chunks, error } +} + +const serverDown = () => Promise.reject(new TypeError('fetch failed')) + +beforeEach(() => { + vi.useFakeTimers() + vi.setSystemTime(0) + // Jitter takes a delay from half of the backoff up to the full backoff. + // `1` gives the full backoff. + vi.spyOn(Math, 'random').mockReturnValue(1) +}) + +afterEach(() => { + vi.useRealTimers() + vi.restoreAllMocks() +}) + +describe('reconnect backoff', () => { + it('doubles the delay from 250 ms and still gives up after 5 reconnects', async () => { + const { requests, gaps, chunks, error } = await run(serverDown) + + expect(chunks).toEqual([EventType.RUN_STARTED]) + expect(error).toBeInstanceOf(StreamReconnectLimitError) + expect(requests).toHaveLength(7) + expect(requests.slice(1).every((r) => r.lastEventId === '1')).toBe(true) + // The first reconnect follows a delivered event, so it waits 250 ms. + // Then each failed reconnect doubles the wait. + expect(gaps).toEqual([250, 250, 500, 1000, 2000, 4000]) + }) + + it('caps the delay at maxDelayMs', async () => { + const { gaps } = await run(serverDown, { maxDelayMs: 600 }) + expect(gaps).toEqual([250, 250, 500, 600, 600, 600]) + }) + + it('adds jitter that never makes a delay longer', async () => { + vi.spyOn(Math, 'random').mockReturnValue(0) + const { gaps } = await run(serverDown) + expect(gaps).toEqual([125, 125, 250, 500, 1000, 2000]) + }) + + it('keeps reconnecting for budgetMs when a budget is set', async () => { + const { requests, error } = await run(serverDown, { budgetMs: 30_000 }) + + expect(error).toBeInstanceOf(StreamReconnectLimitError) + expect(String(error)).toContain('30000 ms') + // Far more than 5 reconnects, all inside the budget. + expect(requests.length).toBeGreaterThan(7) + const firstFailure = requests[1]?.at ?? 0 + expect((requests.at(-1)?.at ?? 0) - firstFailure).toBeLessThanOrEqual( + 30_000, + ) + }) + + it('rides out a restart with a budget', async () => { + // The server is down for 20 s, then it comes back. + const { chunks, error } = await run( + () => + Date.now() < 20_000 + ? serverDown() + : Promise.resolve(finishedResponse()), + { budgetMs: 60_000 }, + ) + expect(error).toBeUndefined() + expect(chunks).toEqual([EventType.RUN_STARTED, EventType.RUN_FINISHED]) + }) + + it('rejects invalid backoff options', async () => { + for (const reconnect of [ + { initialDelayMs: -1 }, + { maxDelayMs: Number.NaN }, + { budgetMs: 0 }, + { budgetMs: Number.POSITIVE_INFINITY }, + ]) { + const { error } = await run(serverDown, reconnect) + expect(String(error)).toMatch(/Invalid reconnect\./) + } + }) +}) + +describe('reconnect on a temporary HTTP status', () => { + it.each([408, 429, 502, 503, 504])( + 'retries a reconnect answered %i', + async (status) => { + const { requests, chunks, error } = await run(async (reconnect) => + reconnect < 3 + ? new Response('upstream unavailable', { status }) + : finishedResponse(), + ) + expect(error).toBeUndefined() + expect(chunks).toEqual([EventType.RUN_STARTED, EventType.RUN_FINISHED]) + expect(requests).toHaveLength(4) + }, + ) + + it('waits for Retry-After, capped at maxDelayMs', async () => { + const { gaps } = await run( + async (reconnect) => { + if (reconnect === 1) { + return new Response('', { + status: 503, + headers: { 'retry-after': '2' }, + }) + } + if (reconnect === 2) { + return new Response('', { + status: 429, + headers: { 'retry-after': '120' }, + }) + } + return finishedResponse() + }, + { maxDelayMs: 10_000 }, + ) + expect(gaps).toEqual([250, 2000, 10_000]) + }) + + it('fails at once on a status that is not temporary', async () => { + const { requests, error } = await run( + async () => new Response('no', { status: 401 }), + ) + expect(requests).toHaveLength(2) + expect(error).not.toBeInstanceOf(StreamReconnectLimitError) + expect(String(error)).toContain('HTTP error! status: 401') + }) + + it('still gives up after maxAttempts when every reconnect is a 503', async () => { + const { requests, error } = await run( + async () => new Response('', { status: 503 }), + ) + expect(error).toBeInstanceOf(StreamReconnectLimitError) + expect(requests).toHaveLength(7) + }) +}) diff --git a/packages/ai-client/tests/connection-adapters-resumable-transports.test.ts b/packages/ai-client/tests/connection-adapters-resumable-transports.test.ts index 28c9324f6b..aae331c6f9 100644 --- a/packages/ai-client/tests/connection-adapters-resumable-transports.test.ts +++ b/packages/ai-client/tests/connection-adapters-resumable-transports.test.ts @@ -176,6 +176,10 @@ class FakeXhr { this.requestHeaders[name] = value } + getResponseHeader(_name: string): string | null { + return null + } + progress(text: string): void { this.responseText += text this.onprogress?.({ type: 'progress' } as ProgressEvent) @@ -304,6 +308,44 @@ describe('resumable XHR transports', () => { expect(queue.xhrs).toHaveLength(2) }) + it('xhrHttpStream retries a reconnect answered 503', async () => { + const queue = createXhrQueue() + const adapter = xhrHttpStream('/api/chat', { + xhrFactory: queue.xhrFactory, + reconnect: { delayMs: 0 }, + }) + + const chunks: Array = [] + const done = (async () => { + for await (const chunk of adapter.connect( + [{ role: 'user', content: 'hi' }], + undefined, + undefined, + { threadId: 't', runId: 'r' }, + )) { + chunks.push(chunk) + } + })() + + await flush() + queue.xhrs[0]!.progress(envelopeLine('run@1', contentChunk('1'))) + queue.xhrs[0]!.error() + + await flush() + queue.xhrs[1]!.status = 503 + queue.xhrs[1]!.statusText = 'Service Unavailable' + queue.xhrs[1]!.end() + + await flush() + expect(queue.xhrs[2]!.requestHeaders['Last-Event-ID']).toBe('run@1') + queue.xhrs[2]!.progress(envelopeLine('run@2', finishedChunk())) + queue.xhrs[2]!.end() + + await done + expect(chunks.at(-1)?.type).toBe(EventType.RUN_FINISHED) + expect(queue.xhrs).toHaveLength(3) + }) + it('xhrServerSentEvents is resumable too (durable SSE over XHR)', async () => { const queue = createXhrQueue() const adapter = xhrServerSentEvents('/api/chat', { diff --git a/packages/ai-client/tests/connection-adapters-resumable.test.ts b/packages/ai-client/tests/connection-adapters-resumable.test.ts index 662a1ccc12..b645cfe119 100644 --- a/packages/ai-client/tests/connection-adapters-resumable.test.ts +++ b/packages/ai-client/tests/connection-adapters-resumable.test.ts @@ -558,12 +558,12 @@ describe('resumable SSE connection adapter', () => { expect(fetchClient).toHaveBeenCalledOnce() }) - it('does not retry HTTP setup failures after earlier progress', async () => { + it('does not retry a non-temporary HTTP status after earlier progress', async () => { const fetchClient = vi.fn(async () => { if (fetchClient.mock.calls.length === 1) { return sseResponse(contentEvent('run@1', '1')) } - return new Response(null, { status: 503, statusText: 'Unavailable' }) + return new Response(null, { status: 401, statusText: 'Unauthorized' }) }) const adapter = fetchServerSentEvents('/api/chat', { fetchClient, @@ -579,7 +579,7 @@ describe('resumable SSE connection adapter', () => { )) { // drain } - }).rejects.toThrow(/503 Unavailable/) + }).rejects.toThrow(/401 Unauthorized/) expect(fetchClient).toHaveBeenCalledTimes(2) }) }) diff --git a/packages/ai-client/tests/connection-adapters-xhr.test.ts b/packages/ai-client/tests/connection-adapters-xhr.test.ts index 84c920c1de..128c8a8f4f 100644 --- a/packages/ai-client/tests/connection-adapters-xhr.test.ts +++ b/packages/ai-client/tests/connection-adapters-xhr.test.ts @@ -42,6 +42,10 @@ class FakeXMLHttpRequest { this.requestHeaders[name] = value } + getResponseHeader(_name: string): string | null { + return null + } + readonly send = vi.fn((body?: string): void => { this.requestBody = body }) diff --git a/packages/ai-client/tests/fetch-json.test.ts b/packages/ai-client/tests/fetch-json.test.ts index 6603d714db..615d328b90 100644 --- a/packages/ai-client/tests/fetch-json.test.ts +++ b/packages/ai-client/tests/fetch-json.test.ts @@ -259,6 +259,19 @@ describe('fetchJson', () => { ]) }) + it('retries a poll answered 503', async () => { + const fetchClient = fetchSequence( + jsonResponse({ chunks: [contentChunk('a')], offset: '1', done: false }), + new Response('restarting', { status: 503 }), + jsonResponse({ chunks: [contentChunk('b')], offset: '2', done: true }), + ) + + const chunks = await collect(connectWith(fetchClient)) + + expect(deltasOf(chunks)).toEqual(['a', 'b']) + expect(fetchClient).toHaveBeenCalledTimes(3) + }) + it('gives up on a poll after the reconnect ceiling', async () => { const fetchClient = fetchSequence( jsonResponse({ chunks: [], offset: '1', done: false }), From 737cc72ec83fe807102d3de998d26a78fe3dfec7 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Thu, 22 Oct 2026 12:37:35 +0200 Subject: [PATCH 2/2] fix(ai-client): keep the reconnect budget with a 0 start delay, fix test mocks - With initialDelayMs 0, try 1025 computed 0 * 2 ** 1024, which is NaN. A NaN delay never reached budgetMs, so the retries did not stop. A 0 start delay now stays 0. A new test runs 1100 tries, then checks that the budget stops the next one. - Three HTTP error tests mocked a response with no headers. The error path now reads Retry-After, so the mocks get an empty Headers. --- packages/ai-client/src/connection-adapters.ts | 14 ++++++++++---- ...connection-adapters-reconnect-backoff.test.ts | 16 ++++++++++++++++ .../ai-client/tests/connection-adapters.test.ts | 3 +++ 3 files changed, 29 insertions(+), 4 deletions(-) diff --git a/packages/ai-client/src/connection-adapters.ts b/packages/ai-client/src/connection-adapters.ts index 2f317e312f..1eb30cdb78 100644 --- a/packages/ai-client/src/connection-adapters.ts +++ b/packages/ai-client/src/connection-adapters.ts @@ -299,10 +299,16 @@ export function createReconnectTracker( } // Exponential backoff with jitter: a delay from half of the backoff up // to the full backoff, so a jittered delay is never longer. - const backoff = Math.min( - reconnect.maxDelayMs, - reconnect.initialDelayMs * 2 ** Math.max(0, reconnectAttempts - 1), - ) + // A 0 start stays 0: after 1024 tries `0 * 2 ** 1024` is NaN, and a NaN + // delay never reaches the budget. + const backoff = + reconnect.initialDelayMs === 0 + ? 0 + : Math.min( + reconnect.maxDelayMs, + reconnect.initialDelayMs * + 2 ** Math.max(0, reconnectAttempts - 1), + ) const delayMs = retryAfterMs === undefined ? backoff * (0.5 + Math.random() / 2) diff --git a/packages/ai-client/tests/connection-adapters-reconnect-backoff.test.ts b/packages/ai-client/tests/connection-adapters-reconnect-backoff.test.ts index 5681016097..bafe2e25b4 100644 --- a/packages/ai-client/tests/connection-adapters-reconnect-backoff.test.ts +++ b/packages/ai-client/tests/connection-adapters-reconnect-backoff.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { EventType } from '@tanstack/ai/client' import { StreamReconnectLimitError, + createReconnectTracker, fetchServerSentEvents, } from '../src/connection-adapters' import type { ReconnectOptions } from '../src/connection-adapters' @@ -144,6 +145,21 @@ describe('reconnect backoff', () => { expect(chunks).toEqual([EventType.RUN_STARTED, EventType.RUN_FINISHED]) }) + it('stops at the budget with initialDelayMs 0, after many tries', async () => { + const tracker = createReconnectTracker({ + initialDelayMs: 0, + budgetMs: 1000, + }) + // Past 1024 tries, where `0 * 2 ** 1024` would make the delay NaN. + for (let attempt = 0; attempt < 1100; attempt++) { + await tracker.waitBeforeReconnect(false) + } + vi.setSystemTime(1001) + await expect(tracker.waitBeforeReconnect(false)).rejects.toBeInstanceOf( + StreamReconnectLimitError, + ) + }) + it('rejects invalid backoff options', async () => { for (const reconnect of [ { initialDelayMs: -1 }, diff --git a/packages/ai-client/tests/connection-adapters.test.ts b/packages/ai-client/tests/connection-adapters.test.ts index 526a222a3a..c70c077014 100644 --- a/packages/ai-client/tests/connection-adapters.test.ts +++ b/packages/ai-client/tests/connection-adapters.test.ts @@ -354,6 +354,7 @@ describe('connection-adapters', () => { ok: false, status: 500, statusText: 'Internal Server Error', + headers: new Headers(), } fetchMock.mockResolvedValue(mockResponse as any) @@ -434,6 +435,7 @@ describe('connection-adapters', () => { ok: false, status: 500, statusText: 'Internal Server Error', + headers: new Headers(), body: null, } @@ -929,6 +931,7 @@ describe('connection-adapters', () => { ok: false, status: 404, statusText: 'Not Found', + headers: new Headers(), } fetchMock.mockResolvedValue(mockResponse as any)