From 08c7bf7f588459b2a139b75e9dab43db9c658e0f Mon Sep 17 00:00:00 2001
From: Deepanshu Pal <40927968+DeepanshuPal@users.noreply.github.com>
Date: Wed, 7 Oct 2026 16:29:03 +0530
Subject: [PATCH 1/2] fix: settle closed request bodies and clear rejected-body
 timers

---
 src/utils/create-stream-promise.ts | 25 +++++++++++++++++++------
 1 file changed, 19 insertions(+), 6 deletions(-)

diff --git a/src/utils/create-stream-promise.ts b/src/utils/create-stream-promise.ts
index b4a58590ab6..d1439487571 100644
--- a/src/utils/create-stream-promise.ts
+++ b/src/utils/create-stream-promise.ts
@@ -16,9 +16,16 @@ const createStreamPromise = function (
     let dataLength = 0
 
     let timeoutId: NodeJS.Timeout | null = null
+    const clearTimer = () => {
+      if (timeoutId) {
+        clearTimeout(timeoutId)
+        timeoutId = null
+      }
+    }
     if (timeoutSeconds != null && Number.isFinite(timeoutSeconds)) {
       timeoutId = setTimeout(() => {
         data = null
+        clearTimer()
         reject(new Error('Request timed out waiting for body'))
       }, timeoutSeconds * SEC_TO_MILLISEC)
     }
@@ -31,6 +38,7 @@ const createStreamPromise = function (
       dataLength += chunk.length
       if (dataLength > bytesLimit) {
         data = null
+        clearTimer()
         reject(new Error('Stream body too big'))
       } else {
         data.push(chunk)
@@ -40,16 +48,21 @@ const createStreamPromise = function (
     stream.on('error', function onError(error) {
       data = null
       reject(error)
-      if (timeoutId) {
-        clearTimeout(timeoutId)
+      clearTimer()
+    })
+    stream.on('close', () => {
+      if (data) {
+        data = null
+        clearTimer()
+        reject(new Error('Stream closed before body completed'))
       }
     })
     stream.on('end', function onEnd() {
-      if (timeoutId) {
-        clearTimeout(timeoutId)
-      }
+      clearTimer()
       if (data) {
-        resolve(Buffer.concat(data))
+        const body = Buffer.concat(data)
+        data = null
+        resolve(body)
       }
     })
   })

From 991e8c52297fadb30b0594ca9ee1d28985e270e1 Mon Sep 17 00:00:00 2001
From: Deepanshu Pal <40927968+DeepanshuPal@users.noreply.github.com>
Date: Wed, 7 Oct 2026 16:29:20 +0530
Subject: [PATCH 2/2] test: cover rejected and prematurely closed body streams

---
 .../unit/utils/create-stream-promise.test.ts  | 84 +++++++++++++++++++
 1 file changed, 84 insertions(+)
 create mode 100644 tests/unit/utils/create-stream-promise.test.ts

diff --git a/tests/unit/utils/create-stream-promise.test.ts b/tests/unit/utils/create-stream-promise.test.ts
new file mode 100644
index 00000000000..683f339a4d0
--- /dev/null
+++ b/tests/unit/utils/create-stream-promise.test.ts
@@ -0,0 +1,84 @@
+import { PassThrough } from 'node:stream'
+import { afterEach, describe, expect, test, vi } from 'vitest'
+import createStreamPromise from '../../../src/utils/create-stream-promise.js'
+
+afterEach(() => vi.useRealTimers())
+
+describe('createStreamPromise', () => {
+  test('clears the request timeout when the size limit rejects a live stream', async () => {
+    vi.useFakeTimers()
+    const stream = new PassThrough()
+    const body = createStreamPromise(stream, 30, 2)
+    stream.write(Buffer.from('abc'))
+    await expect(body).rejects.toThrow('Stream body too big')
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  test('rejects a closed stream promptly instead of waiting for the request timeout', async () => {
+    vi.useFakeTimers()
+    const stream = new PassThrough()
+    const body = createStreamPromise(stream, 30)
+    const rejected = body.catch((error: unknown) => error)
+    stream.emit('close')
+    await vi.advanceTimersByTimeAsync(0)
+    expect(vi.getTimerCount()).toBe(0)
+    expect(await rejected).toBeInstanceOf(Error)
+    expect(((await rejected) as Error).message).toMatch(
+      /Stream closed before body completed|Request timed out waiting for body/,
+    )
+  })
+
+  test('rejects premature closure even without a timeout', async () => {
+    const stream = new PassThrough()
+    const body = createStreamPromise(stream, Number.POSITIVE_INFINITY)
+    const rejected = body.catch((error: unknown) => error)
+    stream.destroy()
+    expect(await rejected).toBeInstanceOf(Error)
+    expect(((await rejected) as Error).message).toMatch(
+      /Stream closed before body completed|Request timed out waiting for body/,
+    )
+  })
+
+  test('ignores later events after a size rejection', async () => {
+    vi.useFakeTimers()
+    const stream = new PassThrough()
+    const body = createStreamPromise(stream, 30, 2)
+    stream.write(Buffer.from('abc'))
+    await expect(body).rejects.toThrow('Stream body too big')
+    stream.emit('error', new Error('later error'))
+    stream.emit('close')
+    stream.end()
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  test('joins chunks when the stream ends normally', async () => {
+    const stream = new PassThrough()
+    const body = createStreamPromise(stream, 30)
+    stream.write(Buffer.from('ab'))
+    stream.end(Buffer.from('cd'))
+    await expect(body).resolves.toEqual(Buffer.from('abcd'))
+  })
+
+  test('preserves stream errors and clears the timeout', async () => {
+    vi.useFakeTimers()
+    const stream = new PassThrough()
+    const body = createStreamPromise(stream, 30)
+    const error = new Error('read failed')
+    stream.emit('error', error)
+    await expect(body).rejects.toBe(error)
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  test('still rejects a stalled stream at the timeout', async () => {
+    vi.useFakeTimers()
+    const stream = new PassThrough()
+    const body = createStreamPromise(stream, 1)
+    const rejected = body.catch((error: unknown) => error)
+    await vi.advanceTimersByTimeAsync(1000)
+    expect(await rejected).toBeInstanceOf(Error)
+    expect(((await rejected) as Error).message).toMatch(
+      /Stream closed before body completed|Request timed out waiting for body/,
+    )
+    expect(vi.getTimerCount()).toBe(0)
+  })
+})