stream: save error in state

Useful for future PR's to resolve situations where e.g. finished() is invoked on an already errored streams. PR-URL: #34103 Backport-PR-URL: #34887 Refs: #34680 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Denys Otrishko <[email protected]> Reviewed-By: Anna Henningsen <[email protected]> Reviewed-By: Luigi Pinca <[email protected]>

Author: ronag

lib/_stream_readable.js

@@ -145,8 +145,11 @@ function ReadableState(options, stream, isDuplex){
 // Has it been destroyed.
 this.destroyed=false;
 
-// Indicates whether the stream has errored.
-this.errored=false;
+// Indicates whether the stream has errored. When true no further
+// _read calls, 'data' or 'readable' events should occur. This is needed
+// since when autoDestroy is disabled we need a way to tell whether the
+// stream has failed.
+this.errored=null;
 
 // Indicates whether the stream has finished destroying.
 this.closed=false;

lib/_stream_writable.js

@@ -171,7 +171,7 @@ function WritableState(options, stream, isDuplex){
 // Indicates whether the stream has errored. When true all write() calls
 // should return false. This is needed since when autoDestroy
 // is disabled we need a way to tell whether the stream has failed.
-this.errored=false;
+this.errored=null;
 
 // Indicates whether the stream has finished destroying.
 this.closed=false;
@@ -407,7 +407,19 @@ function onwrite(stream, er){
 state.writelen=0;
 
 if(er){
-state.errored=true;
+// Avoid V8 leak, https://github.com/nodejs/node/pull/34103#issuecomment-652002364
+er.stack;
+
+if(!state.errored){
+state.errored=er;
+}
+
+// In case of duplex streams we need to notify the readable side of the
+// error.
+if(stream._readableState&&!stream._readableState.errored){
+stream._readableState.errored=er;
+}
+
 if(sync){
 process.nextTick(onwriteError,stream,state,er,cb);
 }else{

lib/internal/streams/destroy.js

@@ -15,11 +15,14 @@ function destroy(err, cb){
 }
 
 if(err){
-if(w){
-w.errored=true;
+// Avoid V8 leak, https://github.com/nodejs/node/pull/34103#issuecomment-652002364
+err.stack;
+
+if(w&&!w.errored){
+w.errored=err;
 }
-if(r){
-r.errored=true;
+if(r&&!r.errored){
+r.errored=err;
 }
 }
 
@@ -35,11 +38,14 @@ function destroy(err, cb){
 
 this._destroy(err||null,(err)=>{
 if(err){
-if(w){
-w.errored=true;
+// Avoid V8 leak, https://github.com/nodejs/node/pull/34103#issuecomment-652002364
+err.stack;
+
+if(w&&!w.errored){
+w.errored=err;
 }
-if(r){
-r.errored=true;
+if(r&&!r.errored){
+r.errored=err;
 }
 }
 
@@ -108,7 +114,7 @@ function undestroy(){
 r.closed=false;
 r.closeEmitted=false;
 r.destroyed=false;
-r.errored=false;
+r.errored=null;
 r.errorEmitted=false;
 r.reading=false;
 r.ended=false;
@@ -118,7 +124,7 @@ function undestroy(){
 if(w){
 w.closed=false;
 w.destroyed=false;
-w.errored=false;
+w.errored=null;
 w.ended=false;
 w.ending=false;
 w.finalCalled=false;
@@ -145,11 +151,14 @@ function errorOrDestroy(stream, err, sync){
 if((r&&r.autoDestroy)||(w&&w.autoDestroy))
 stream.destroy(err);
 elseif(err){
-if(w){
-w.errored=true;
+// Avoid V8 leak, https://github.com/nodejs/node/pull/34103#issuecomment-652002364
+err.stack;
+
+if(w&&!w.errored){
+w.errored=err;
 }
-if(r){
-r.errored=true;
+if(r&&!r.errored){
+r.errored=err;
 }
 
 if(sync){

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

@@ -134,13 +134,13 @@ const assert = require('assert');
 read.on('error',common.mustCall((err)=>{
 assert.strictEqual(ticked,true);
 assert.strictEqual(read._readableState.errorEmitted,true);
-assert.strictEqual(read._readableState.errored,true);
+assert.strictEqual(read._readableState.errored,expected);
 assert.strictEqual(err,expected);
 }));
 
 read.destroy();
 assert.strictEqual(read._readableState.errorEmitted,false);
-assert.strictEqual(read._readableState.errored,true);
+assert.strictEqual(read._readableState.errored,expected);
 assert.strictEqual(read.destroyed,true);
 ticked=true;
 }
@@ -190,15 +190,15 @@ const assert = require('assert');
 assert.strictEqual(err,expected);
 }));
 
-assert.strictEqual(read._readableState.errored,false);
+assert.strictEqual(read._readableState.errored,null);
 assert.strictEqual(read._readableState.errorEmitted,false);
 
 read.destroy(expected,common.mustCall(function(err){
-assert.strictEqual(read._readableState.errored,true);
+assert.strictEqual(read._readableState.errored,expected);
 assert.strictEqual(err,expected);
 }));
 assert.strictEqual(read._readableState.errorEmitted,false);
-assert.strictEqual(read._readableState.errored,true);
+assert.strictEqual(read._readableState.errored,expected);
 ticked=true;
 }
 
@@ -223,14 +223,14 @@ const assert = require('assert');
 
 readable.destroy();
 assert.strictEqual(readable.destroyed,true);
-assert.strictEqual(readable._readableState.errored,false);
+assert.strictEqual(readable._readableState.errored,null);
 assert.strictEqual(readable._readableState.errorEmitted,false);
 
 // Test case where `readable.destroy()` is called again with an error before
 // the `_destroy()` callback is called.
 readable.destroy(newError('kaboom 2'));
 assert.strictEqual(readable._readableState.errorEmitted,false);
-assert.strictEqual(readable._readableState.errored,false);
+assert.strictEqual(readable._readableState.errored,null);
 
 ticked=true;
 }
@@ -253,3 +253,18 @@ const assert = require('assert');
 assert.strictEqual(read.destroyed,true);
 read.read();
 }
+
+{
+constread=newReadable({
+autoDestroy: false,
+read(){
+this.push(null);
+this.push('asd');
+}
+});
+
+read.on('error',common.mustCall(()=>{
+assert(read._readableState.errored);
+}));
+read.resume();
+}

test/parallel/test-stream-writable-destroy.js

@@ -167,9 +167,10 @@ const assert = require('assert');
 assert.strictEqual(write._writableState.errorEmitted,true);
 }));
 
-write.destroy(newError('kaboom 1'));
+constexpected=newError('kaboom 1');
+write.destroy(expected);
 write.destroy(newError('kaboom 2'));
-assert.strictEqual(write._writableState.errored,true);
+assert.strictEqual(write._writableState.errored,expected);
 assert.strictEqual(write._writableState.errorEmitted,false);
 assert.strictEqual(write.destroyed,true);
 ticked=true;
@@ -200,14 +201,14 @@ const assert = require('assert');
 
 writable.destroy();
 assert.strictEqual(writable.destroyed,true);
-assert.strictEqual(writable._writableState.errored,false);
+assert.strictEqual(writable._writableState.errored,null);
 assert.strictEqual(writable._writableState.errorEmitted,false);
 
 // Test case where `writable.destroy()` is called again with an error before
 // the `_destroy()` callback is called.
 writable.destroy(newError('kaboom 2'));
 assert.strictEqual(writable._writableState.errorEmitted,false);
-assert.strictEqual(writable._writableState.errored,false);
+assert.strictEqual(writable._writableState.errored,null);
 
 ticked=true;
 }
@@ -401,3 +402,18 @@ const assert = require('assert');
 }));
 write.destroy();
 }
+
+{
+constwrite=newWritable({
+autoDestroy: false,
+write(chunk,enc,cb){
+cb();
+cb();
+}
+});
+
+write.on('error',common.mustCall(()=>{
+assert(write._writableState.errored);
+}));
+write.write('asd');
+}

test/parallel/test-stream2-readable-wrap-error.js

@@ -10,23 +10,25 @@ oldStream.pause = () =>{};
 oldStream.resume=()=>{};
 
 {
+consterr=newError();
 constr=newReadable({autoDestroy: true})
 .wrap(oldStream)
 .on('error',common.mustCall(()=>{
 assert.strictEqual(r._readableState.errorEmitted,true);
-assert.strictEqual(r._readableState.errored,true);
+assert.strictEqual(r._readableState.errored,err);
 assert.strictEqual(r.destroyed,true);
 }));
-oldStream.emit('error',newError());
+oldStream.emit('error',err);
 }
 
 {
+consterr=newError();
 constr=newReadable({autoDestroy: false})
 .wrap(oldStream)
 .on('error',common.mustCall(()=>{
 assert.strictEqual(r._readableState.errorEmitted,true);
-assert.strictEqual(r._readableState.errored,true);
+assert.strictEqual(r._readableState.errored,err);
 assert.strictEqual(r.destroyed,false);
 }));
-oldStream.emit('error',newError());
+oldStream.emit('error',err);
 }