stream: pipeline should drain empty readable

This simplifies some cases where the last stream is a Duplex
without any expected output.

await pipeline(readable, duplex)

PR-URL: #40654
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>

lib/internal/streams/pipeline.js

@@ -33,6 +33,7 @@ const {
   isIterable,
   isReadableNodeStream,
   isNodeStream,
+  isReadableFinished,
 } = require('internal/streams/utils');
 const { AbortController } = require('internal/abort_controller');
 
@@ -229,7 +230,14 @@ function pipelineImpl(streams, callback, opts) {
 
     if (isNodeStream(stream)) {
       finishCount++;
-      destroys.push(destroyer(stream, reading, writing, finish));
+      destroys.push(destroyer(stream, reading, writing, (err) => {
+        if (!err && !reading && isReadableFinished(stream, false)) {
+          stream.read(0);
+          destroyer(stream, true, writing, finish);
+        } else {
+          finish(err);
+        }
+      }));
     }
 
     if (i === 0) {

test/parallel/test-stream-pipeline.js

@@ -1027,7 +1027,7 @@ const tsp = require('timers/promises');
   const src = new PassThrough();
   const dst = new PassThrough();
   pipeline(src, dst, common.mustSucceed(() => {
-    assert.strictEqual(dst.destroyed, false);
+    assert.strictEqual(dst.destroyed, true);
   }));
   src.end();
 }
@@ -1447,3 +1447,23 @@ const tsp = require('timers/promises');
     assert.strictEqual(text, 'Hello World!');
   }));
 }
+
+{
+  const pipelinePromise = promisify(pipeline);
+
+  async function run() {
+    const read = new Readable({
+      read() {}
+    });
+
+    const duplex = new PassThrough();
+
+    read.push(null);
+
+    await pipelinePromise(read, duplex);
+
+    assert.strictEqual(duplex.destroyed, true);
+  }
+
+  run();
+}