stream: fix broken pipeline error propagation

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

lib/internal/streams/pipeline.js

@@ -132,9 +132,10 @@ function pipeline(...streams) {
  }
 
  let error;
+ let value;
  const destroys = [];
 
- function finish(err, val, final) {
+ function finish(err, final) {
  if (!error && err) {
  error = err;
  }
@@ -146,13 +147,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);
  }));
  }
 
@@ -197,12 +198,12 @@ function pipeline(...streams) {
  const pt = new PassThrough();
  if (isPromise(ret)) {
  ret
+ .catch((err) => {
+ pt.destroy(err);
+ })
  .then((val) => {
+ value = val;
  pt.end(val);
- finish(null, val, true);
- })
- .catch((err) => {
- finish(err, null, true);
  });
  } else if (isIterable(ret, true)) {
  pump(ret, pt, finish);
@@ -212,7 +213,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, 'asd');
+}));
+
+// Ensure that pipeline that ends with Promise
+// still propagates error to uncaughtException.
+const s = new PassThrough();
+s.end('asd');
+pipeline(s, async function(source) {
+ for await (const chunk of source) {
+ chunk;
+ }
+}, common.mustCall((err) => {
+ assert.ok(!err);
+ throw new Error('asd');
+}));

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