nodejs · ronag · Jul 15, 2019 · Aug 6, 2019
diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
@@ -117,6 +117,9 @@ function ReadableState(options, stream, isDuplex) {
  this.resumeScheduled = false;
  this.paused = true;

+  // True if the error was already emitted and should not be thrown again
+  this.errorEmitted = false;
+
  // Should close be emitted on destroy. Defaults to true.
  this.emitClose = options.emitClose !== false;


diff --git a/lib/internal/streams/async_iterator.js b/lib/internal/streams/async_iterator.js
@@ -132,7 +132,8 @@ const createReadableStreamAsyncIterator = (stream) => {
    [kLastReject]: { value: null, writable: true },
    [kError]: { value: null, writable: true },
    [kEnded]: {
-      value: stream._readableState.endEmitted,
+      value: stream._readableState.endEmitted ||
+        stream._readableState.errorEmitted,
      writable: true
    },
    // The function passed to new Promise is cached so we avoid allocating a new

diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js
@@ -2,21 +2,32 @@

 // 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 rState = this._readableState;
+  const wState = this._writableState;
+
+  const readableDestroyed = rState && rState.destroyed;
+  const writableDestroyed = wState && wState.destroyed;

  if (readableDestroyed || writableDestroyed) {
    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);
+      const errorEmitted = (rState && rState.errorEmitted) ||
+        (wState && wState.errorEmitted);
+
+      if (errorEmitted) {
+        return this;
+      }
+
+      if (rState) {
+        rState.errorEmitted = true;
      }
+
+      if (wState) {
+        wState.errorEmitted = true;
+      }
+
+      process.nextTick(emitErrorNT, this, err);
    }

    return this;
@@ -25,13 +36,13 @@ 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 (rState) {
+    rState.destroyed = true;
  }

  // If this is a duplex stream mark the writable part as destroyed as well
-  if (this._writableState) {
-    this._writableState.destroyed = true;
+  if (wState) {
+    wState.destroyed = true;
  }

  this._destroy(err || null, (err) => {
@@ -74,6 +85,7 @@ function undestroy() {
    this._readableState.reading = false;
    this._readableState.ended = false;
    this._readableState.endEmitted = false;
+    this._readableState.errorEmitted = false;
  }

  if (this._writableState) {
@@ -101,10 +113,17 @@ function errorOrDestroy(stream, err) {
  const rState = stream._readableState;
  const wState = stream._writableState;

-  if ((rState && rState.autoDestroy) || (wState && wState.autoDestroy))
+  if ((rState && rState.autoDestroy) || (wState && wState.autoDestroy)) {
    stream.destroy(err);
-  else
+  } else {
+    if (wState) {
+      wState.errorEmitted = true;
+    }
+    if (rState) {
+      rState.errorEmitted = true;
+    }
    stream.emit('error', err);
+  }
 }



diff --git a/test/parallel/test-stream2-writable.js b/test/parallel/test-stream2-writable.js
@@ -402,3 +402,45 @@ const helloWorldBuffer = Buffer.from('hello world');
  w.write(Buffer.allocUnsafe(1));
  w.end(Buffer.allocUnsafe(0));
 }
+
+{
+  // Verify that error is only emitted once when failing in _finish.
+  const w = new W();
+
+  w._final = common.mustCall(function(cb) {
+    cb(new Error('test'));
+  });
+  w._write = function(chunk, e, cb) {
+    process.nextTick(cb);
+  };
+  w.once('error', common.mustCall((err) => {
+    assert.strictEqual(w._writableState.errorEmitted, true);
+    assert.strictEqual(err.message, 'test');
+    w.on('error', common.mustNotCall());
+    w.destroy(new Error());
+  }));
+  w.end();
+}
+
+{
+  // Verify that error is only emitted once when failing in write.
+  const w = new W();
+  w.on('error', common.mustCall((err) => {
+    assert.strictEqual(w._writableState.errorEmitted, true);
+    assert.strictEqual(err.code, 'ERR_STREAM_NULL_VALUES');
+  }));
+  w.write(null);
+  w.destroy(new Error());
+}
+
+{
+  // Verify that error is only emitted once when failing in write after end.
+  const w = new W();
+  w.on('error', common.mustCall((err) => {
+    assert.strictEqual(w._writableState.errorEmitted, true);
+    assert.strictEqual(err.code, 'ERR_STREAM_WRITE_AFTER_END');
+  }));
+  w.end();
+  w.write('hello');
+  w.destroy(new Error());
+}