stream: forward errored to callback

Refs: #39356
PR-URL: #39364
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>

Author: ronag

lib/internal/streams/writable.js

@@ -515,12 +515,12 @@ function errorBuffer(state) {
 const{ chunk, callback }=state.buffered[n];
 constlen=state.objectMode ? 1 : chunk.length;
 state.length-=len;
-callback(newERR_STREAM_DESTROYED('write'));
+callback(state.errored??newERR_STREAM_DESTROYED('write'));
 }
 
 constonfinishCallbacks=state[kOnFinished].splice(0);
 for(leti=0;i<onfinishCallbacks.length;i++){
-onfinishCallbacks[i](newERR_STREAM_DESTROYED('end'));
+onfinishCallbacks[i](state.errored??newERR_STREAM_DESTROYED('end'));
 }
 
 resetBuffer(state);

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

@@ -351,33 +351,35 @@ const assert = require('assert');
 constwrite=newWritable({
 write(chunk,enc,cb){process.nextTick(cb);}
 });
+const_err=newError('asd');
 write.once('error',common.mustCall((err)=>{
 assert.strictEqual(err.message,'asd');
 }));
 write.end('asd',common.mustCall((err)=>{
-assert.strictEqual(err.code,'ERR_STREAM_DESTROYED');
+assert.strictEqual(err,_err);
 }));
-write.destroy(newError('asd'));
+write.destroy(_err);
 }
 
 {
 // Call buffered write callback with error
 
+const_err=newError('asd');
 constwrite=newWritable({
 write(chunk,enc,cb){
-process.nextTick(cb,newError('asd'));
+process.nextTick(cb,_err);
 },
 autoDestroy: false
 });
 write.cork();
 write.write('asd',common.mustCall((err)=>{
-assert.strictEqual(err.message,'asd');
+assert.strictEqual(err,_err);
 }));
 write.write('asd',common.mustCall((err)=>{
-assert.strictEqual(err.code,'ERR_STREAM_DESTROYED');
+assert.strictEqual(err,_err);
 }));
 write.on('error',common.mustCall((err)=>{
-assert.strictEqual(err.message,'asd');
+assert.strictEqual(err,_err);
 }));
 write.uncork();
 }
@@ -471,3 +473,19 @@ const assert = require('assert');
 write.destroy();
 write.destroy();
 }
+
+{
+// https://github.com/nodejs/node/issues/39356
+consts=newWritable({
+final(){}
+});
+const_err=newError('oh no');
+// Remove `callback` and it works
+s.end(common.mustCall((err)=>{
+assert.strictEqual(err,_err);
+}));
+s.on('error',common.mustCall((err)=>{
+assert.strictEqual(err,_err);
+}));
+s.destroy(_err);
+}

test/parallel/test-stream-writable-end-cb-error.js

@@ -8,19 +8,20 @@ const stream = require('stream');
 // Invoke end callback on failure.
 constwritable=newstream.Writable();
 
+const_err=newError('kaboom');
 writable._write=(chunk,encoding,cb)=>{
-process.nextTick(cb,newError('kaboom'));
+process.nextTick(cb,_err);
 };
 
 writable.on('error',common.mustCall((err)=>{
-assert.strictEqual(err.message,'kaboom');
+assert.strictEqual(err,_err);
 }));
 writable.write('asd');
 writable.end(common.mustCall((err)=>{
-assert.strictEqual(err.code,'ERR_STREAM_DESTROYED');
+assert.strictEqual(err,_err);
 }));
 writable.end(common.mustCall((err)=>{
-assert.strictEqual(err.code,'ERR_STREAM_DESTROYED');
+assert.strictEqual(err,_err);
 }));
 }
 
@@ -57,18 +58,12 @@ const stream = require('stream');
 }
 });
 w.end('testing ended state',common.mustCall((err)=>{
-// This errors since .destroy(err), which is invoked by errors
-// in same tick below, will error all pending callbacks.
-// Does this make sense? Not sure.
-assert.strictEqual(err.code,'ERR_STREAM_DESTROYED');
+assert.strictEqual(err.code,'ERR_STREAM_WRITE_AFTER_END');
 }));
 assert.strictEqual(w.destroyed,false);
 assert.strictEqual(w.writableEnded,true);
 w.end(common.mustCall((err)=>{
-// This errors since .destroy(err), which is invoked by errors
-// in same tick below, will error all pending callbacks.
-// Does this make sense? Not sure.
-assert.strictEqual(err.code,'ERR_STREAM_DESTROYED');
+assert.strictEqual(err.code,'ERR_STREAM_WRITE_AFTER_END');
 }));
 assert.strictEqual(w.destroyed,false);
 assert.strictEqual(w.writableEnded,true);

test/parallel/test-stream-writable-end-cb-uncaught.js

@@ -9,15 +9,16 @@ process.on('uncaughtException', common.mustCall((err) => {
 }));
 
 constwritable=newstream.Writable();
+const_err=newError('kaboom');
 
 writable._write=(chunk,encoding,cb)=>{
 cb();
 };
 writable._final=(cb)=>{
-cb(newError('kaboom'));
+cb(_err);
 };
 
 writable.write('asd');
 writable.end(common.mustCall((err)=>{
-assert.strictEqual(err.code,'ERR_STREAM_DESTROYED');
+assert.strictEqual(err,_err);
 }));