feat(http): add bodyTimeout to BodyReaderOptions

Bounds total body read time using a deadline-based approach so a slow
or malicious peer cannot stall a connection indefinitely.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

packages/http/src/client/utils.ts

@@ -1,6 +1,7 @@
 import type { ClientConnectionOptions } from "./connection.js"
 
 const clientConnectionOptionsKeys = Object.keys({
+  bodyTimeout: 0,
   headersTimeout: 0,
   maxBodyLength: 0,
   maxHeaderCount: 0,

packages/http/src/common/objects.ts

@@ -1,6 +1,7 @@
 import type { Body, ReadableHttp, Reader, WritableHttp } from "./types.js"
 import { Headers, MutableHeaders } from "./headers.js"
 import type { BodyReaderOptions } from "./reader.js"
+import { withTimeout } from "./utils.js"
 
 export class ReadableHttpImpl implements ReadableHttp {
   #headers: Headers
@@ -28,15 +29,18 @@ export class ReadableHttpImpl implements ReadableHttp {
   stream(encoding: "utf8", options?: BodyReaderOptions): AsyncIterable<string>
   async *stream(
     encoding?: "binary" | "utf8",
-    { maxBodyLength = Infinity }: BodyReaderOptions = {},
+    { maxBodyLength = Infinity, bodyTimeout }: BodyReaderOptions = {},
   ): AsyncIterable<string | Uint8Array> {
     const body = this.#assertBody()
     this.#bodyStream = null
     const decoder = encoding === "utf8" ? new TextDecoder() : null
     let bodyLength = 0
+    const deadline =
+      bodyTimeout !== undefined && isFinite(bodyTimeout) ? Date.now() + bodyTimeout : undefined
     while (!body.closed) {
       try {
-        const chunk = await body.read()
+        const remaining = deadline !== undefined ? Math.max(0, deadline - Date.now()) : undefined
+        const chunk = await withTimeout(body.read(), remaining, "Body timeout")
         if (!chunk.length) continue
         bodyLength += chunk.length
         if (bodyLength > maxBodyLength)

packages/http/src/common/reader.ts

@@ -16,6 +16,11 @@ export type BodyReaderOptions = {
    * @defaultValue Infinity
    */
   maxBodyLength?: number
+  /**
+   * Maximum total time in ms to read the entire body.
+   * @defaultValue Infinity
+   */
+  bodyTimeout?: number
 }
 
 abstract class BaseBodyReader implements Pick<BodyReader, "onFinish"> {