stream: fix pipeline function tail deadlock

Signed-off-by: Dayun <dlekdbs6530@gmail.com>
PR-URL: #65559
Fixes: #40685
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: dayun6530

lib/stream/promises.js

@@ -1,6 +1,7 @@
 'use strict';
 
 const {
+  ArrayIsArray,
   ArrayPrototypePop,
   Promise,
 } = primordials;
@@ -28,13 +29,22 @@ function pipeline(...streams) {
       end = options.end;
     }
 
-    pl(streams, (err, value) => {
+    let lastStream = streams[streams.length - 1];
+    if (streams.length === 1 && ArrayIsArray(streams[0])) {
+      lastStream = streams[0][streams[0].length - 1];
+    }
+
+    const stream = pl(streams, (err, value) => {
       if (err) {
         reject(err);
       } else {
         resolve(value);
       }
     }, { signal, end });
+
+    if (typeof lastStream === 'function' && stream.readable) {
+      stream.resume();
+    }
   });
 }
 

test/parallel/test-stream-pipeline.js

@@ -1657,6 +1657,22 @@ tmpdir.refresh();
   }));
 }
 
+{
+  async function* passThrough(source) {
+    for await (const chunk of source) {
+      yield chunk;
+    }
+  }
+
+  const streams = () => [
+    Readable.from(Array.from({ length: 100 }, (_, i) => i)),
+    passThrough,
+  ];
+
+  pipelinep(...streams()).then(common.mustCall());
+  pipelinep(streams()).then(common.mustCall());
+}
+
 {
   const r = new Readable();
   for (let i = 0; i < 4000; i++) {