stream: finished should error on errored stream

Calling finished before or after a stream has errored or closed
should end up with the same behavior.
PR-URL: #39235
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

Author: ronag

lib/_http_client.js

@@ -806,6 +806,9 @@ function onSocketNT(req, socket, err) {
 socket.emit('free');
 }else{
 finished(socket.destroy(err||req[kError]),(er)=>{
+if(er?.code==='ERR_STREAM_PREMATURE_CLOSE'){
+er=null;
+}
 _destroy(req,er||err);
 });
 return;

lib/_http_incoming.js

@@ -188,6 +188,9 @@ IncomingMessage.prototype._destroy = function _destroy(err, cb) {
 if(this.socket&&!this.socket.destroyed&&this.aborted){
 this.socket.destroy(err);
 constcleanup=finished(this.socket,(e)=>{
+if(e?.code==='ERR_STREAM_PREMATURE_CLOSE'){
+e=null;
+}
 cleanup();
 process.nextTick(onError,this,e||err,cb);
 });

lib/internal/streams/end-of-stream.js

@@ -98,8 +98,7 @@ function eos(stream, options, callback) {
 isWritable(stream)===writable
 );
 
-letwritableFinished=stream.writableFinished||
-(wState&&wState.finished);
+letwritableFinished=stream.writableFinished||wState?.finished;
 constonfinish=()=>{
 writableFinished=true;
 // Stream should not be destroyed here. If it is that
@@ -111,8 +110,7 @@ function eos(stream, options, callback) {
 if(!readable||readableEnded)callback.call(stream);
 };
 
-letreadableEnded=stream.readableEnded||
-(rState&&rState.endEmitted);
+letreadableEnded=stream.readableEnded||rState?.endEmitted;
 constonend=()=>{
 readableEnded=true;
 // Stream should not be destroyed here. If it is that
@@ -128,7 +126,17 @@ function eos(stream, options, callback) {
 callback.call(stream,err);
 };
 
+letclosed=wState?.closed||rState?.closed;
+
 constonclose=()=>{
+closed=true;
+
+consterrored=wState?.errored||rState?.errored;
+
+if(errored&&typeoferrored!=='boolean'){
+returncallback.call(stream,errored);
+}
+
 if(readable&&!readableEnded){
 if(!isReadableEnded(stream))
 returncallback.call(stream,
@@ -139,6 +147,7 @@ function eos(stream, options, callback) {
 returncallback.call(stream,
 newERR_STREAM_PREMATURE_CLOSE());
 }
+
 callback.call(stream);
 };
 
@@ -168,29 +177,29 @@ function eos(stream, options, callback) {
 if(options.error!==false)stream.on('error',onerror);
 stream.on('close',onclose);
 
-// _closed is for OutgoingMessage which is not a proper Writable.
-constclosed=(!wState&&!rState&&stream._closed===true)||(
-(wState&&wState.closed)||
-(rState&&rState.closed)||
-(wState&&wState.errorEmitted)||
-(rState&&rState.errorEmitted)||
-(rState&&stream.req&&stream.aborted)||
-(
-(!writable||(wState&&wState.finished))&&
-(!readable||(rState&&rState.endEmitted))
-)
-);
-
 if(closed){
-// TODO(ronag): Re-throw error if errorEmitted?
-// TODO(ronag): Throw premature close as if finished was called?
-// before being closed? i.e. if closed but not errored, ended or finished.
-// TODO(ronag): Throw some kind of error? Does it make sense
-// to call finished() on a "finished" stream?
-// TODO(ronag): willEmitClose?
-process.nextTick(()=>{
-callback();
-});
+process.nextTick(onclose);
+}elseif(wState?.errorEmitted||rState?.errorEmitted){
+if(!willEmitClose){
+process.nextTick(onclose);
+}
+}elseif(
+!readable&&
+(!willEmitClose||stream.readable)&&
+writableFinished
+){
+process.nextTick(onclose);
+}elseif(
+!writable&&
+(!willEmitClose||stream.writable)&&
+readableEnded
+){
+process.nextTick(onclose);
+}elseif(!wState&&!rState&&stream._closed===true){
+// _closed is for OutgoingMessage which is not a proper Writable.
+process.nextTick(onclose);
+}elseif((rState&&stream.req&&stream.aborted)){
+process.nextTick(onclose);
 }
 
 constcleanup=()=>{

test/parallel/test-stream-finished.js

@@ -608,3 +608,26 @@ testClosed((opts) => new Writable({ write() {}, ...opts }));
 assert.strictEqual(closed,true);
 }));
 }
+
+{
+constw=newWritable();
+const_err=newError();
+w.destroy(_err);
+finished(w,common.mustCall((err)=>{
+assert.strictEqual(_err,err);
+finished(w,common.mustCall((err)=>{
+assert.strictEqual(_err,err);
+}));
+}));
+}
+
+{
+constw=newWritable();
+w.destroy();
+finished(w,common.mustCall((err)=>{
+assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
+finished(w,common.mustCall((err)=>{
+assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
+}));
+}));
+}