Merge remote-tracking branch 'upstream/main'

* upstream/main: fix: use explicit flag for when use has interacted with stream (nodejs#3361) refactor: simplify signal handling (nodejs#3362) fix: consider bytes read when dumping (nodejs#3360) websocket: don't use pooled buffer in mask pool (nodejs#3357) Revert "fix: post request signal (nodejs#3354)" (nodejs#3359) fix: post request signal (nodejs#3354) build(deps): bump node from `075a5cc` to `9af472b` in /build (nodejs#3355)
nxtedition · Jun 23, 2024 · 1aa10e3 · 1aa10e3
2 parents 2e522b4 + 59fc6d5
commit 1aa10e3
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 20 deletions.
diff --git a/build/Dockerfile b/build/Dockerfile
@@ -1,4 +1,4 @@
-FROM node:22-alpine3.19@sha256:075a5cc188c3c9a49acacd481a9e8a3c9abf4223f02c658e37fdb8e9fe2c4664
+FROM node:22-alpine3.19@sha256:9af472b2578996eb3d6affbcb82fdee6f086da2c43121e75038a4a70317f784f
 
 ARG UID=1000
 ARG GID=1000

diff --git a/lib/api/api-request.js b/lib/api/api-request.js
@@ -64,12 +64,6 @@ class RequestHandler {
           } else if (this.abort) {
             this.abort(this.reason)
           }
-
-          if (this.removeAbortListener) {
-            this.res?.off('close', this.removeAbortListener)
-            this.removeAbortListener()
-            this.removeAbortListener = null
-          }
         })
       }
     }
@@ -110,6 +104,7 @@ class RequestHandler {
 
     if (this.removeAbortListener) {
       res.on('close', this.removeAbortListener)
+      this.removeAbortListener = null
     }
 
     this.callback = null
@@ -152,7 +147,6 @@ class RequestHandler {
     }
 
     if (this.removeAbortListener) {
-      res?.off('close', this.removeAbortListener)
       this.removeAbortListener()
       this.removeAbortListener = null
     }

diff --git a/lib/api/readable.js b/lib/api/readable.js
@@ -16,6 +16,8 @@ const kContentLength = Symbol('kContentLength')
 const kMethod = Symbol('kMethod')
 const kStatusCode = Symbol('kStatusCode')
 const kHeaders = Symbol('kHeaders')
+const kUsed = Symbol('kUsed')
+const kBytesRead = Symbol('kBytesRead')
 
 const noop = () => {}
 
@@ -40,9 +42,11 @@ class BodyReadable extends Readable {
 
     this[kAbort] = abort
     this[kConsume] = null
+    this[kBytesRead] = 0
     this[kBody] = null
+    this[kUsed] = false
     this[kContentType] = contentType
-    this[kContentLength] = contentLength
+    this[kContentLength] = Number.isFinite(contentLength) ? contentLength : null
     this[kMethod] = method
     this[kStatusCode] = statusCode
     this[kHeaders] = headers
@@ -70,7 +74,7 @@ class BodyReadable extends Readable {
     return this
   }
 
-  destroy (err) {
+  _destroy (err, callback) {
     if (!err && !this._readableState.endEmitted) {
       err = new RequestAbortedError()
     }
@@ -79,15 +83,11 @@ class BodyReadable extends Readable {
       this[kAbort]()
     }
 
-    return super.destroy(err)
-  }
-
-  _destroy (err, callback) {
     // Workaround for Node "bug". If the stream is destroyed in same
     // tick as it is created, then a user who is waiting for a
     // promise (i.e micro tick) for installing a 'error' listener will
     // never get a chance and will always encounter an unhandled exception.
-    if (!this[kReading]) {
+    if (!this[kUsed]) {
       setImmediate(() => {
         callback(err)
       })
@@ -99,6 +99,7 @@ class BodyReadable extends Readable {
   on (ev, ...args) {
     if (ev === 'data' || ev === 'readable') {
       this[kReading] = true
+      this[kUsed] = true
     }
     return super.on(ev, ...args)
   }
@@ -123,6 +124,8 @@ class BodyReadable extends Readable {
   }
 
   push (chunk) {
+    this[kBytesRead] += chunk ? chunk.length : 0
+
     if (this[kConsume] && chunk !== null) {
       consumePush(this[kConsume], chunk)
       return this[kReading] ? super.push(chunk) : true
@@ -156,7 +159,7 @@ class BodyReadable extends Readable {
   }
 
   async dump (opts) {
-    let limit = Number.isFinite(opts?.limit) ? opts.limit : 128 * 1024
+    const limit = Number.isFinite(opts?.limit) ? opts.limit : 128 * 1024
     const signal = opts?.signal
 
     if (signal != null && (typeof signal !== 'object' || !('aborted' in signal))) {
@@ -170,7 +173,7 @@ class BodyReadable extends Readable {
     }
 
     return await new Promise((resolve, reject) => {
-      if (this[kContentLength] > limit) {
+      if (this[kContentLength] > limit || this[kBytesRead] > limit) {
         this.destroy(new AbortError())
       }
 
@@ -190,8 +193,7 @@ class BodyReadable extends Readable {
         })
         .on('error', noop)
         .on('data', function (chunk) {
-          limit -= chunk.length
-          if (limit <= 0) {
+          if (this[kBytesRead] > limit) {
             this.destroy()
           }
         })

diff --git a/lib/core/util.js b/lib/core/util.js
@@ -534,7 +534,7 @@ function addAbortListener (signal, listener) {
     signal.addEventListener('abort', listener, { once: true })
     return () => signal.removeEventListener('abort', listener)
   }
-  signal.addListener('abort', listener)
+  signal.once('abort', listener)
   return () => signal.removeListener('abort', listener)
 }