stream: fix broken pipeline error propagation

If the destination was an async function any
error thrown from that function would be swallowed.

Backport-PR-URL: #31975
PR-URL: #31835
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Denys Otrishko <shishugi@gmail.com>

lib/internal/streams/pipeline.js

@@ -163,9 +163,10 @@ function pipeline(...streams) {
   }
 
   let error;
+  let value;
   const destroys = [];
 
-  function finish(err, val, final) {
+  function finish(err, final) {
     if (!error && err) {
       error = err;
     }
@@ -177,13 +178,13 @@ function pipeline(...streams) {
     }
 
     if (final) {
-      callback(error, val);
+      callback(error, value);
     }
   }
 
   function wrap(stream, reading, writing, final) {
     destroys.push(destroyer(stream, reading, writing, (err) => {
-      finish(err, null, final);
+      finish(err, final);
     }));
   }
 
@@ -229,11 +230,10 @@ function pipeline(...streams) {
         if (isPromise(ret)) {
           ret
             .then((val) => {
+              value = val;
               pt.end(val);
-              finish(null, val, true);
-            })
-            .catch((err) => {
-              finish(err, null, true);
+            }, (err) => {
+              pt.destroy(err);
             });
         } else if (isIterable(ret, true)) {
           pump(ret, pt, finish);
@@ -243,7 +243,7 @@ function pipeline(...streams) {
         }
 
         ret = pt;
-        wrap(ret, true, false, true);
+        wrap(ret, false, true, true);
       }
     } else if (isStream(stream)) {
       if (isReadable(ret)) {

test/parallel/test-stream-pipeline-uncaught.js

@@ -0,0 +1,25 @@
+'use strict';
+
+const common = require('../common');
+const {
+  pipeline,
+  PassThrough
+} = require('stream');
+const assert = require('assert');
+
+process.on('uncaughtException', common.mustCall((err) => {
+  assert.strictEqual(err.message, 'error');
+}));
+
+// Ensure that pipeline that ends with Promise
+// still propagates error to uncaughtException.
+const s = new PassThrough();
+s.end('data');
+pipeline(s, async function(source) {
+  for await (const chunk of source) {
+    chunk;
+  }
+}, common.mustCall((err) => {
+  assert.ifError(err);
+  throw new Error('error');
+}));

test/parallel/test-stream-pipeline.js

@@ -613,11 +613,9 @@ const { promisify } = require('util');
     yield 'hello';
     yield 'world';
   }, async function*(source) {
-    const ret = [];
     for await (const chunk of source) {
-      ret.push(chunk.toUpperCase());
+      yield chunk.toUpperCase();
     }
-    yield ret;
   }, async function(source) {
     let ret = '';
     for await (const chunk of source) {
@@ -754,7 +752,6 @@ const { promisify } = require('util');
   }, common.mustCall((err) => {
     assert.strictEqual(err, undefined);
     assert.strictEqual(ret, 'asd');
-    assert.strictEqual(s.destroyed, true);
   }));
 }
 
@@ -775,7 +772,6 @@ const { promisify } = require('util');
   }, common.mustCall((err) => {
     assert.strictEqual(err, undefined);
     assert.strictEqual(ret, 'asd');
-    assert.strictEqual(s.destroyed, true);
   }));
 }