stream: avoid premature close when will not emit close

Fixes: #45281 PR-URL: #45301 Reviewed-By: Luigi Pinca <[email protected]> Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Benjamin Gruenbaum <[email protected]> Reviewed-By: Minwoo Jung <[email protected]>

Author: ronag

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

@@ -151,6 +151,18 @@ function eos(stream, options, callback){
 callback.call(stream);
 };
 
+constonclosed=()=>{
+closed=true;
+
+consterrored=isWritableErrored(stream)||isReadableErrored(stream);
+
+if(errored&&typeoferrored!=='boolean'){
+returncallback.call(stream,errored);
+}
+
+callback.call(stream);
+};
+
 constonrequest=()=>{
 stream.req.on('finish',onfinish);
 };
@@ -186,22 +198,22 @@ function eos(stream, options, callback){
 process.nextTick(onclose);
 }elseif(wState?.errorEmitted||rState?.errorEmitted){
 if(!willEmitClose){
-process.nextTick(onclose);
+process.nextTick(onclosed);
 }
 }elseif(
 !readable&&
 (!willEmitClose||isReadable(stream))&&
 (writableFinished||isWritable(stream)===false)
 ){
-process.nextTick(onclose);
+process.nextTick(onclosed);
 }elseif(
 !writable&&
 (!willEmitClose||isWritable(stream))&&
 (readableFinished||isReadable(stream)===false)
 ){
-process.nextTick(onclose);
+process.nextTick(onclosed);
 }elseif((rState&&stream.req&&stream.aborted)){
-process.nextTick(onclose);
+process.nextTick(onclosed);
 }
 
 constcleanup=()=>{

test/parallel/test-stream-finished.js

@@ -667,3 +667,17 @@ testClosed((opts) => new Writable({write(){}, ...opts }));
 ).end();
 });
 }
+
+{
+conststream=newDuplex({
+write(chunk,enc,cb){
+setImmediate(cb);
+}
+});
+
+stream.end('foo');
+
+finished(stream,{readable: false},common.mustCall((err)=>{
+assert(!err);
+}));
+}