stream: ensure Stream.pipeline re-throws errors without callback

Fixes an issue where Stream.pipeline wouldn't re-throw errors
on a stream if no callback was specified, thus swallowing
said errors.

Fixes: #20303

PR-URL: #20437
Reviewed-By: Anna Henningsen <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>

Author: phated

lib/internal/streams/pipeline.js

@@ -19,7 +19,10 @@ function once(callback) {
   };
 }
 
-function noop() {}
+function noop(err) {
+  // Rethrow the error if it exists to avoid swallowing it
+  if (err) throw err;
+}
 
 function isRequest(stream) {
   return stream.setHeader && typeof stream.abort === 'function';

test/parallel/test-stream-pipeline.js

@@ -142,7 +142,7 @@ common.crashOnUnhandledRejection();
       }
     });
 
-    pipeline(rs, res);
+    pipeline(rs, res, () => {});
   });
 
   server.listen(0, () => {
@@ -177,7 +177,7 @@ common.crashOnUnhandledRejection();
       })
     });
 
-    pipeline(rs, res);
+    pipeline(rs, res, () => {});
   });
 
   server.listen(0, () => {
@@ -206,7 +206,7 @@ common.crashOnUnhandledRejection();
       })
     });
 
-    pipeline(rs, res);
+    pipeline(rs, res, () => {});
   });
 
   let cnt = 10;
@@ -481,3 +481,35 @@ common.crashOnUnhandledRejection();
 
   run();
 }
+
+{
+  const read = new Readable({
+    read() {}
+  });
+
+  const transform = new Transform({
+    transform(data, enc, cb) {
+      cb(new Error('kaboom'));
+    }
+  });
+
+  const write = new Writable({
+    write(data, enc, cb) {
+      cb();
+    }
+  });
+
+  read.on('close', common.mustCall());
+  transform.on('close', common.mustCall());
+  write.on('close', common.mustCall());
+
+  process.on('uncaughtException', common.mustCall((err) => {
+    assert.deepStrictEqual(err, new Error('kaboom'));
+  }));
+
+  const dst = pipeline(read, transform, write);
+
+  assert.strictEqual(dst, write);
+
+  read.push('hello');
+}