feat: add support for AbortSignal to abort reading the body

.github/workflows/ci.yml

@@ -41,7 +41,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest, windows-latest, macos-latest]
-        node-version: [18.0.0, 18, 19, 20, 21, 22, 23, 24, 25]
+        node-version: [18.9.0, 18, 19, 20, 21, 22, 23, 24, 25]
         # Node.js release schedule: https://nodejs.org/en/about/releases/
     name: Node.js ${{ matrix.node-version }} - ${{matrix.os}}
     runs-on: ${{ matrix.os }}

HISTORY.md

@@ -1,6 +1,11 @@
 unreleased
 ==================
 
+  * Add `signal` option to abort reading the body with an `AbortSignal`;
+    aborting produces a 408 `request.timeout` error with the signal's
+    reason in `cause`, distinct from a client abort's 400 `request.aborted`
+  * Error with a 400 `request.aborted` when a node stream closes before
+    ending, instead of never settling
   * Add support for WHATWG `ReadableStream` (web streams): `fetch`
     `Request`/`Response` bodies, `Blob.stream()`, `TransformStream`
     readables, and `Readable.toWeb()` bridges
@@ -28,7 +33,7 @@ unreleased
   * Add `decoder` option to plug in a custom decoder (e.g. `iconv-lite`'s
     `getDecoder`) for encodings outside the WHATWG Encoding Standard
   * Remove the check for a global `Promise` when no callback is provided
-  * Breaking Change: Node.js 18 is the minimum supported version
+  * Breaking Change: Node.js 18.9 is the minimum supported version
 
 3.0.2 / 2025-11-21
 ======================

README.md

@@ -85,6 +85,13 @@ getRawBody(stream, {
   If the function throws, a `415` error is returned to signal the encoding is
   unsupported.
 
+- `signal` - An [`AbortSignal`](https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal)
+  to abort reading the body, for example from `AbortSignal.timeout()` or an
+  `AbortController` tied to the request lifecycle. When the signal aborts,
+  reading stops and a `408` error with type `request.timeout` is returned,
+  carrying the signal's reason in `cause`. This is distinct from a client
+  disconnect, which returns a `400` `request.aborted`.
+
 You can also pass a string in place of options to just specify the encoding.
 
 If an error occurs, the stream will be paused, everything unpiped,
@@ -133,6 +140,12 @@ an entity that is larger.
 This error will occur when the request stream is aborted by the client before
 reading the body has finished.
 
+#### request.timeout
+
+This error will occur when the `AbortSignal` passed via the `signal` option
+aborts before reading the body has finished, for example on a server-side
+read timeout.
+
 #### request.size.invalid
 
 This error will occur when the `length` option is specified, but the stream has

index.d.ts

@@ -27,6 +27,12 @@ declare namespace getRawBody {
      * encoding is unsupported.
      */
     decoder?: (encoding: string) => Decoder;
+    /**
+     * An `AbortSignal` to abort reading the body. When it aborts, a
+     * 408 `request.timeout` error is returned with the signal's
+     * reason in `cause`.
+     */
+    signal?: AbortSignal;
   }
 
   export interface Decoder {

index.js

@@ -104,6 +104,32 @@ function abortedError (length, received, cause) {
   return err
 }
 
+/**
+ * Create a 408 request timeout error, used when the `signal`
+ * option aborts (server-initiated), as opposed to a client abort.
+ *
+ * @param {number} length
+ * @param {number} received
+ * @param {*} [cause]
+ * @private
+ */
+
+function requestTimeoutError (length, received, cause) {
+  const err = createError(408, 'request timeout', {
+    code: 'ECONNABORTED',
+    expected: length,
+    length,
+    received,
+    type: 'request.timeout'
+  })
+
+  if (cause !== undefined) {
+    err.cause = cause
+  }
+
+  return err
+}
+
 /**
  * Create a 400 size mismatch error.
  *
@@ -223,6 +249,11 @@ function getRawBody (stream, options, callback) {
     throw new TypeError('option decoder must be a function')
   }
 
+  // validate signal is an AbortSignal, if provided
+  if (opts.signal !== undefined && !(opts.signal instanceof AbortSignal)) {
+    throw new TypeError('option signal must be an AbortSignal')
+  }
+
   // convert the limit to an integer
   const limit = bytes.parse(opts.limit)
 
@@ -240,11 +271,11 @@ function getRawBody (stream, options, callback) {
 
   if (done) {
     // classic callback style
-    return read(stream, encoding, length, limit, opts.decoder, AsyncResource.bind(done, done.name || 'bound-anonymous-fn', null))
+    return read(stream, encoding, length, limit, opts.decoder, opts.signal, AsyncResource.bind(done, done.name || 'bound-anonymous-fn', null))
   }
 
   return new Promise(function executor (resolve, reject) {
-    read(stream, encoding, length, limit, opts.decoder, function onRead (err, buf) {
+    read(stream, encoding, length, limit, opts.decoder, opts.signal, function onRead (err, buf) {
       if (err) return reject(err)
       resolve(buf)
     })
@@ -276,11 +307,12 @@ function halt (stream) {
  * @param {number} length
  * @param {number} limit
  * @param {function} createDecoder
+ * @param {AbortSignal} signal
  * @param {function} callback
  * @public
  */
 
-function readStream (stream, encoding, length, limit, createDecoder, callback) {
+function readStream (stream, encoding, length, limit, createDecoder, signal, callback) {
   let buffer
   let complete = false
   let sync = true
@@ -315,9 +347,17 @@ function readStream (stream, encoding, length, limit, createDecoder, callback) {
     ? ''
     : []
 
+  if (signal) {
+    if (signal.aborted) {
+      return done(requestTimeoutError(length, received, signal.reason))
+    }
+
+    signal.addEventListener('abort', onSignalAbort, { once: true })
+  }
+
   // attach listeners
   stream.on('aborted', onAborted)
-  stream.on('close', cleanup)
+  stream.on('close', onClose)
   stream.on('data', onData)
   stream.on('end', onEnd)
   stream.on('error', onEnd)
@@ -360,6 +400,20 @@ function readStream (stream, encoding, length, limit, createDecoder, callback) {
     done(abortedError(length, received))
   }
 
+  function onClose () {
+    if (complete) return
+
+    // closed without end or error: the read can never settle
+    // on its own, so treat it as an aborted body
+    done(abortedError(length, received))
+  }
+
+  function onSignalAbort () {
+    if (complete) return
+
+    done(requestTimeoutError(length, received, signal.reason))
+  }
+
   function onData (chunk) {
     if (complete) return
 
@@ -391,11 +445,15 @@ function readStream (stream, encoding, length, limit, createDecoder, callback) {
   function cleanup () {
     buffer = null
 
+    if (signal) {
+      signal.removeEventListener('abort', onSignalAbort)
+    }
+
     stream.removeListener('aborted', onAborted)
     stream.removeListener('data', onData)
     stream.removeListener('end', onEnd)
     stream.removeListener('error', onEnd)
-    stream.removeListener('close', cleanup)
+    stream.removeListener('close', onClose)
   }
 }
 
@@ -425,13 +483,15 @@ function toBuffer (chunk) {
  * @param {number} length
  * @param {number} limit
  * @param {function} createDecoder
+ * @param {AbortSignal} signal
  * @param {function} callback
  * @private
  */
 
-function readWebStream (stream, encoding, length, limit, createDecoder, callback) {
+function readWebStream (stream, encoding, length, limit, createDecoder, signal, callback) {
   let buffer
   let reader = null
+  let settled = false
 
   // check the length and limit options.
   // note: on error the reader lock is released but the stream is
@@ -459,8 +519,16 @@ function readWebStream (stream, encoding, length, limit, createDecoder, callback
     ? ''
     : []
 
+  if (signal && signal.aborted) {
+    return fail(requestTimeoutError(length, received, signal.reason))
+  }
+
   reader = stream.getReader()
 
+  if (signal) {
+    signal.addEventListener('abort', onSignalAbort, { once: true })
+  }
+
   read()
 
   function read () {
@@ -486,16 +554,31 @@ function readWebStream (stream, encoding, length, limit, createDecoder, callback
     done(new Error('stream error', { cause: err }))
   }
 
+  function onSignalAbort () {
+    done(requestTimeoutError(length, received, signal.reason))
+  }
+
   function fail (err) {
     // defer, so the callback is never invoked synchronously
     process.nextTick(done, err)
   }
 
   function done (err, string) {
+    // a signal abort races the pending read's rejection after
+    // releaseLock: only the first settlement wins
+    if (settled) return
+    settled = true
+
+    if (signal) {
+      signal.removeEventListener('abort', onSignalAbort)
+    }
+
     buffer = null
 
     if (reader) {
-      // release the stream, so users can handle the rest themselves
+      // release the stream, so users can handle the rest
+      // themselves; a read in flight rejects, and the settled
+      // flag ignores it
       reader.releaseLock()
     }
 

package.json

@@ -23,7 +23,7 @@
     "nyc": "17.0.0"
   },
   "engines": {
-    "node": ">=18"  
+    "node": ">=18.9"
   },
   "files": [
     "LICENSE",

test/http.js

@@ -135,4 +135,38 @@ describe('using http streams', function () {
       })
     })
   })
+
+  it('should time out a slow body via the signal', function (done) {
+    let socket
+    const server = http.createServer(function onRequest (req, res) {
+      getRawBody(req, {
+        length: req.headers['content-length'],
+        signal: AbortSignal.timeout(30)
+      }, function (err) {
+        server.close()
+        socket.destroy()
+
+        // a server-side read timeout is distinct from a client
+        // abort: 408 request.timeout, not 400 request.aborted
+        assert.ok(err)
+        assert.strictEqual(err.status, 408)
+        assert.strictEqual(err.type, 'request.timeout')
+        assert.strictEqual(err.code, 'ECONNABORTED')
+        assert.strictEqual(err.expected, 50)
+        assert.strictEqual(err.received, 10)
+        assert.strictEqual(err.cause.name, 'TimeoutError')
+        done()
+      })
+    })
+
+    server.listen(function onListen () {
+      socket = net.connect(server.address().port, function () {
+        socket.write('POST / HTTP/1.0\r\n')
+        socket.write('Content-Length: 50\r\n')
+        socket.write('\r\n')
+        // send only 10 of the 50 bytes, then stall the connection
+        socket.write('testing...')
+      })
+    })
+  })
 })

test/index.js

@@ -236,6 +236,83 @@ describe('Raw Body', function () {
     })
   })
 
+  it('should validate the signal option', function () {
+    assert.throws(function () {
+      getRawBody(createStream(), { signal: 'nope' }, function () {})
+    }, /option signal must be an AbortSignal/)
+  })
+
+  it('should abort reading when the signal aborts', function (done) {
+    const controller = new AbortController()
+
+    // a stalled stream: only the signal can end this read
+    const stream = new Readable({ read () {} })
+    stream.push('partial')
+
+    getRawBody(stream, { signal: controller.signal }, function (err) {
+      assert.ok(err)
+      assert.strictEqual(err.status, 408)
+      assert.strictEqual(err.type, 'request.timeout')
+      assert.strictEqual(err.received, 7)
+      assert.strictEqual(err.cause, controller.signal.reason)
+      assert.ok(stream.isPaused())
+      done()
+    })
+
+    setTimeout(function () { controller.abort() }, 10)
+  })
+
+  it('should error immediately when the signal is already aborted', function (done) {
+    getRawBody(createStream(), { signal: AbortSignal.abort() }, function (err) {
+      assert.ok(err)
+      assert.strictEqual(err.status, 408)
+      assert.strictEqual(err.type, 'request.timeout')
+      done()
+    })
+  })
+
+  it('should ignore a signal that aborts after the body is read', function (done) {
+    const controller = new AbortController()
+    const stream = new Readable({ read () {} })
+    stream.push('hello, world!')
+    stream.push(null)
+
+    let calls = 0
+
+    getRawBody(stream, { signal: controller.signal }, function (err, buf) {
+      calls++
+      assert.ifError(err)
+      assert.strictEqual(buf.toString(), 'hello, world!')
+
+      // aborting after completion must not invoke the callback
+      // again or surface a late error
+      controller.abort()
+      setTimeout(function () {
+        assert.strictEqual(calls, 1)
+        done()
+      }, 10)
+    })
+  })
+
+  it('should error when the stream closes before ending', function (done) {
+    const stream = new Readable({ read () {} })
+
+    getRawBody(stream, function (err) {
+      assert.ok(err)
+      assert.strictEqual(err.status, 400)
+      assert.strictEqual(err.type, 'request.aborted')
+      assert.strictEqual(err.received, 7)
+      done()
+    })
+
+    stream.push('partial')
+
+    setTimeout(function () {
+      // emits only 'close': the read would otherwise never settle
+      stream.destroy()
+    }, 10)
+  })
+
   it('should prefer the node stream interface when both are present', function (done) {
     const stream = createStream(Buffer.from('hello, world!'))