stream: fix destroy() behavior

Ensure errorEmitted is always set. Only emit 'error' once.

PR-URL: https://github.com/nodejs/node/pull/29058
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Franziska Hinkelmann <franziska.hinkelmann@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

1 files changed, 60 insertions, 47 deletions

diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js
index 200c75459a..0d3eb6327b 100644
--- a/lib/internal/streams/destroy.js
+++ b/lib/internal/streams/destroy.js
@@ -1,22 +1,37 @@
 'use strict';
 
+function needError(stream, err) {
+  if (!err) {
+    return false;
+  }
+
+  const r = stream._readableState;
+  const w = stream._writableState;
+
+  if ((w && w.errorEmitted) || (r && r.errorEmitted)) {
+    return false;
+  }
+
+  if (w) {
+    w.errorEmitted = true;
+  }
+  if (r) {
+    r.errorEmitted = true;
+  }
+
+  return true;
+}
+
 // Undocumented cb() API, needed for core, not for public API
 function destroy(err, cb) {
-  const readableDestroyed = this._readableState &&
-    this._readableState.destroyed;
-  const writableDestroyed = this._writableState &&
-    this._writableState.destroyed;
+  const r = this._readableState;
+  const w = this._writableState;
 
-  if (readableDestroyed || writableDestroyed) {
+  if ((w && w.destroyed) || (r && r.destroyed)) {
     if (cb) {
       cb(err);
-    } else if (err) {
-      if (!this._writableState) {
-        process.nextTick(emitErrorNT, this, err);
-      } else if (!this._writableState.errorEmitted) {
-        this._writableState.errorEmitted = true;
-        process.nextTick(emitErrorNT, this, err);
-      }
+    } else if (needError(this, err)) {
+      process.nextTick(emitErrorNT, this, err);
     }
 
     return this;
@@ -25,28 +40,19 @@ function destroy(err, cb) {
   // We set destroyed to true before firing error callbacks in order
   // to make it re-entrance safe in case destroy() is called within callbacks
 
-  if (this._readableState) {
-    this._readableState.destroyed = true;
+  if (w) {
+    w.destroyed = true;
   }
-
-  // If this is a duplex stream mark the writable part as destroyed as well
-  if (this._writableState) {
-    this._writableState.destroyed = true;
+  if (r) {
+    r.destroyed = true;
   }
 
   this._destroy(err || null, (err) => {
-    if (!cb && err) {
-      if (!this._writableState) {
-        process.nextTick(emitErrorAndCloseNT, this, err);
-      } else if (!this._writableState.errorEmitted) {
-        this._writableState.errorEmitted = true;
-        process.nextTick(emitErrorAndCloseNT, this, err);
-      } else {
-        process.nextTick(emitCloseNT, this);
-      }
-    } else if (cb) {
+    if (cb) {
       process.nextTick(emitCloseNT, this);
       cb(err);
+    } else if (needError(this, err)) {
+      process.nextTick(emitErrorAndCloseNT, this, err);
     } else {
       process.nextTick(emitCloseNT, this);
     }
@@ -61,29 +67,36 @@ function emitErrorAndCloseNT(self, err) {
 }
 
 function emitCloseNT(self) {
-  if (self._writableState && !self._writableState.emitClose)
+  const r = self._readableState;
+  const w = self._writableState;
+
+  if (w && !w.emitClose)
     return;
-  if (self._readableState && !self._readableState.emitClose)
+  if (r && !r.emitClose)
     return;
   self.emit('close');
 }
 
 function undestroy() {
-  if (this._readableState) {
-    this._readableState.destroyed = false;
-    this._readableState.reading = false;
-    this._readableState.ended = false;
-    this._readableState.endEmitted = false;
+  const r = this._readableState;
+  const w = this._writableState;
+
+  if (r) {
+    r.destroyed = false;
+    r.reading = false;
+    r.ended = false;
+    r.endEmitted = false;
+    r.errorEmitted = false;
   }
 
-  if (this._writableState) {
-    this._writableState.destroyed = false;
-    this._writableState.ended = false;
-    this._writableState.ending = false;
-    this._writableState.finalCalled = false;
-    this._writableState.prefinished = false;
-    this._writableState.finished = false;
-    this._writableState.errorEmitted = false;
+  if (w) {
+    w.destroyed = false;
+    w.ended = false;
+    w.ending = false;
+    w.finalCalled = false;
+    w.prefinished = false;
+    w.finished = false;
+    w.errorEmitted = false;
   }
 }
 
@@ -98,12 +111,12 @@ function errorOrDestroy(stream, err) {
   // the error to be emitted nextTick. In a future
   // semver major update we should change the default to this.
 
-  const rState = stream._readableState;
-  const wState = stream._writableState;
+  const r = stream._readableState;
+  const w = stream._writableState;
 
-  if ((rState && rState.autoDestroy) || (wState && wState.autoDestroy))
+  if ((r && r.autoDestroy) || (w && w.autoDestroy))
     stream.destroy(err);
-  else
+  else if (needError(stream, err))
     stream.emit('error', err);
 }