stream: readable error emitted

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;
 

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

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);
+  }
 }
 
 

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());
+}