stream: fix destroy() behavior

Ensure errorEmitted is always set. Only emit 'error' once.
PR-URL: #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>

Author: ronag

doc/api/stream.md

@@ -281,6 +281,9 @@ The stream is not closed when the `'error'` event is emitted unless the
 [`autoDestroy`][writable-new] option was set to `true` when creating the
 stream.
 
+After `'error'`, no further events other than `'close'`*should* be emitted
+(including `'error'` events).
+
 ##### Event: 'finish'
 <!-- YAML
 added: v0.9.4

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/_stream_writable.js

@@ -429,13 +429,11 @@ function onwriteError(stream, state, sync, er, cb) {
 // This can emit finish, and it will always happen
 // after error
 process.nextTick(finishMaybe,stream,state);
-stream._writableState.errorEmitted=true;
 errorOrDestroy(stream,er);
 }else{
 // The caller expect this to happen before if
 // it is async
 cb(er);
-stream._writableState.errorEmitted=true;
 errorOrDestroy(stream,er);
 // This can emit finish, but finish must
 // always follow error

lib/internal/streams/destroy.js

@@ -1,22 +1,37 @@
 'use strict';
 
+functionneedError(stream,err){
+if(!err){
+returnfalse;
+}
+
+constr=stream._readableState;
+constw=stream._writableState;
+
+if((w&&w.errorEmitted)||(r&&r.errorEmitted)){
+returnfalse;
+}
+
+if(w){
+w.errorEmitted=true;
+}
+if(r){
+r.errorEmitted=true;
+}
+
+returntrue;
+}
+
 // Undocumented cb() API, needed for core, not for public API
 functiondestroy(err,cb){
-constreadableDestroyed=this._readableState&&
-this._readableState.destroyed;
-constwritableDestroyed=this._writableState&&
-this._writableState.destroyed;
+constr=this._readableState;
+constw=this._writableState;
 
-if(readableDestroyed||writableDestroyed){
+if((w&&w.destroyed)||(r&&r.destroyed)){
 if(cb){
 cb(err);
-}elseif(err){
-if(!this._writableState){
-process.nextTick(emitErrorNT,this,err);
-}elseif(!this._writableState.errorEmitted){
-this._writableState.errorEmitted=true;
-process.nextTick(emitErrorNT,this,err);
-}
+}elseif(needError(this,err)){
+process.nextTick(emitErrorNT,this,err);
 }
 
 returnthis;
@@ -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);
-}elseif(!this._writableState.errorEmitted){
-this._writableState.errorEmitted=true;
-process.nextTick(emitErrorAndCloseNT,this,err);
-}else{
-process.nextTick(emitCloseNT,this);
-}
-}elseif(cb){
+if(cb){
 process.nextTick(emitCloseNT,this);
 cb(err);
+}elseif(needError(this,err)){
+process.nextTick(emitErrorAndCloseNT,this,err);
 }else{
 process.nextTick(emitCloseNT,this);
 }
@@ -61,29 +67,36 @@ function emitErrorAndCloseNT(self, err) {
 }
 
 functionemitCloseNT(self){
-if(self._writableState&&!self._writableState.emitClose)
+constr=self._readableState;
+constw=self._writableState;
+
+if(w&&!w.emitClose)
 return;
-if(self._readableState&&!self._readableState.emitClose)
+if(r&&!r.emitClose)
 return;
 self.emit('close');
 }
 
 functionundestroy(){
-if(this._readableState){
-this._readableState.destroyed=false;
-this._readableState.reading=false;
-this._readableState.ended=false;
-this._readableState.endEmitted=false;
+constr=this._readableState;
+constw=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.
 
-constrState=stream._readableState;
-constwState=stream._writableState;
+constr=stream._readableState;
+constw=stream._writableState;
 
-if((rState&&rState.autoDestroy)||(wState&&wState.autoDestroy))
+if((r&&r.autoDestroy)||(w&&w.autoDestroy))
 stream.destroy(err);
-else
+elseif(needError(stream,err))
 stream.emit('error',err);
 }
 

test/parallel/test-net-connect-buffer.js

@@ -69,12 +69,14 @@ tcp.listen(0, common.mustCall(function() {
 [],
 {}
 ].forEach((value)=>{
-common.expectsError(()=>socket.write(value),{
+// We need to check the callback since 'error' will only
+// be emitted once per instance.
+socket.write(value,common.expectsError({
 code: 'ERR_INVALID_ARG_TYPE',
 type: TypeError,
 message: 'The "chunk" argument must be one of type string or Buffer. '+
 `Received type ${typeofvalue}`
-});
+}));
 });
 
 // Write a string that contains a multi-byte character sequence to test that

test/parallel/test-stream-error-once.js

@@ -0,0 +1,19 @@
+'use strict';
+constcommon=require('../common');
+const{ Writable, Readable }=require('stream');
+
+{
+constwritable=newWritable();
+writable.on('error',common.mustCall());
+writable.end();
+writable.write('h');
+writable.write('h');
+}
+
+{
+constreadable=newReadable();
+readable.on('error',common.mustCall());
+readable.push(null);
+readable.push('h');
+readable.push('h');
+}

test/parallel/test-stream-readable-invalid-chunk.js

@@ -3,17 +3,32 @@
 constcommon=require('../common');
 conststream=require('stream');
 
-constreadable=newstream.Readable({
-read: ()=>{}
-});
-
-functioncheckError(fn){
-common.expectsError(fn,{
+functiontestPushArg(val){
+constreadable=newstream.Readable({
+read: ()=>{}
+});
+readable.on('error',common.expectsError({
 code: 'ERR_INVALID_ARG_TYPE',
 type: TypeError
+}));
+readable.push(val);
+}
+
+testPushArg([]);
+testPushArg({});
+testPushArg(0);
+
+functiontestUnshiftArg(val){
+constreadable=newstream.Readable({
+read: ()=>{}
 });
+readable.on('error',common.expectsError({
+code: 'ERR_INVALID_ARG_TYPE',
+type: TypeError
+}));
+readable.unshift(val);
 }
 
-checkError(()=>readable.push([]));
-checkError(()=>readable.push({}));
-checkError(()=>readable.push(0));
+testUnshiftArg([]);
+testUnshiftArg({});
+testUnshiftArg(0);

test/parallel/test-stream-readable-unshift.js

@@ -112,23 +112,6 @@ const { Readable } = require('stream');
 
 }
 
-{
-// Check that error is thrown for invalid chunks
-
-constreadable=newReadable({read(){}});
-functioncheckError(fn){
-common.expectsError(fn,{
-code: 'ERR_INVALID_ARG_TYPE',
-type: TypeError
-});
-}
-
-checkError(()=>readable.unshift([]));
-checkError(()=>readable.unshift({}));
-checkError(()=>readable.unshift(0));
-
-}
-
 {
 // Check that ObjectMode works
 constreadable=newReadable({objectMode: true,read(){}});

test/parallel/test-stream-unshift-read-race.js

@@ -86,13 +86,7 @@ w._write = function(chunk, encoding, cb) {
 };
 
 r.on('end',common.mustCall(function(){
-common.expectsError(function(){
-r.unshift(Buffer.allocUnsafe(1));
-},{
-code: 'ERR_STREAM_UNSHIFT_AFTER_END_EVENT',
-type: Error,
-message: 'stream.unshift() after end event'
-});
+r.unshift(Buffer.allocUnsafe(1));
 w.end();
 }));
 

test/parallel/test-stream2-writable.js

@@ -402,3 +402,42 @@ 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.
+constw=newW();
+
+w._final=common.mustCall(function(cb){
+cb(newError('test'));
+});
+w.on('error',common.mustCall((err)=>{
+assert.strictEqual(w._writableState.errorEmitted,true);
+assert.strictEqual(err.message,'test');
+w.on('error',common.mustNotCall());
+w.destroy(newError());
+}));
+w.end();
+}
+
+{
+// Verify that error is only emitted once when failing in write.
+constw=newW();
+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(newError());
+}
+
+{
+// Verify that error is only emitted once when failing in write after end.
+constw=newW();
+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(newError());
+}