feat(ai-client): back off reconnects and retry temporary HTTP statuses

.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`.

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

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.

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<void>
 }
 
@@ -181,6 +261,8 @@ export function createReconnectTracker(
   const seen = new Set<string>()
   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,44 @@ 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.
+      // 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)
+          : 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 +612,10 @@ async function assertResponseOk(response: Response): Promise<void> {
       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 +631,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 +914,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 +1852,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
     }
   }