stream: avoid premature close when will not emit close

Fixes: #45281
PR-URL: #45301
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Minwoo Jung <nodecorelab@gmail.com>

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);
+  }));
+}