stream: avoid premature close when will not emit close

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

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

test/parallel/test-stream-finished.js

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