stream: preserve error over AbortError in pipeline

lib/internal/streams/pipeline.js

@@ -227,7 +227,7 @@ function pipelineImpl(streams, callback, opts) {
   }
 
   function finishImpl(err, final) {
-    if (err && (!error || error.code === 'ERR_STREAM_PREMATURE_CLOSE')) {
+    if (err && (!error || error.code === 'ERR_STREAM_PREMATURE_CLOSE' || error.name === 'AbortError')) {
       error = err;
     }
 

test/parallel/test-stream-pipeline.js

@@ -1749,3 +1749,28 @@ tmpdir.refresh();
     assert.deepStrictEqual(err, new Error('booom'));
   }));
 }
+
+{
+  // Errors thrown in Readable.map inside pipeline should not be
+  // swallowed by AbortError when the source is an infinite stream.
+  async function run() {
+    await assert.rejects(
+      pipelinep(
+        new Readable({ read() { this.push('data'); } }),
+        new Transform({
+          readableObjectMode: true,
+          transform(chunk, encoding, callback) {
+            this.push({});
+            callback();
+          },
+        }),
+        (readable) => readable.map(async () => {
+          throw new Error('Boom!');
+        }),
+      ),
+      { message: 'Boom!' },
+    );
+  }
+
+  run().then(common.mustCall());
+}