stream: need to cleanup event listeners if last stream is readable

lib/internal/streams/pipeline.js

@@ -45,14 +45,17 @@ function destroyer(stream, reading, writing) {
     finished = true;
   });
 
-  eos(stream, { readable: reading, writable: writing }, (err) => {
+  const cleanup = eos(stream, { readable: reading, writable: writing }, (err) => {
     finished = !err;
   });
 
-  return (err) => {
-    if (finished) return;
-    finished = true;
-    destroyImpl.destroyer(stream, err || new ERR_STREAM_DESTROYED('pipe'));
+  return {
+    callback: (err) => {
+      if (finished) return;
+      finished = true;
+      destroyImpl.destroyer(stream, err || new ERR_STREAM_DESTROYED('pipe'));
+    },
+    cleanup
   };
 }
 
@@ -159,6 +162,10 @@ function pipelineImpl(streams, callback, opts) {
   const signal = ac.signal;
   const outerSignal = opts?.signal;
 
+  // Need to cleanup event listeners if last stream is readable
+  // https://github.com/nodejs/node/issues/35452
+  const lastStreamCleanup = streams[streams.length - 1].readable && [];
+
   validateAbortSignal(outerSignal, 'options.signal');
 
   function abort() {
@@ -194,6 +201,11 @@ function pipelineImpl(streams, callback, opts) {
     ac.abort();
 
     if (final) {
+      if (!error) {
+        while (lastStreamCleanup && lastStreamCleanup.length) {
+          lastStreamCleanup.shift()();
+        }
+      }
       process.nextTick(callback, error, value);
     }
   }
@@ -204,22 +216,34 @@ function pipelineImpl(streams, callback, opts) {
     const reading = i < streams.length - 1;
     const writing = i > 0;
     const end = reading || opts?.end !== false;
+    const isLastStream = i === streams.length - 1;
 
     if (isNodeStream(stream)) {
       if (end) {
-        destroys.push(destroyer(stream, reading, writing));
+        const { callback, cleanup } = destroyer(stream, reading, writing);
+        destroys.push(callback);
+
+        if (lastStreamCleanup && isLastStream) {
+          lastStreamCleanup.push(cleanup);
+        }
       }
 
       // Catch stream errors that occur after pipe/pump has completed.
-      stream.on('error', (err) => {
+      function onError(err) {
         if (
           err &&
           err.name !== 'AbortError' &&
           err.code !== 'ERR_STREAM_PREMATURE_CLOSE'
         ) {
           finish(err);
         }
-      });
+      }
+      stream.on('error', onError);
+      if (lastStreamCleanup && isLastStream) {
+        lastStreamCleanup.push(() => {
+          stream.removeListener('error', onError);
+        });
+      }
     }
 
     if (i === 0) {
@@ -285,12 +309,15 @@ function pipelineImpl(streams, callback, opts) {
 
         ret = pt;
 
-        destroys.push(destroyer(ret, false, true));
+        destroys.push(destroyer(ret, false, true).callback);
       }
     } else if (isNodeStream(stream)) {
       if (isReadableNodeStream(ret)) {
         finishCount += 2;
-        pipe(ret, stream, finish, { end });
+        const cleanup = pipe(ret, stream, finish, { end });
+        if (lastStreamCleanup && isLastStream) {
+          lastStreamCleanup.push(cleanup);
+        }
       } else if (isIterable(ret)) {
         finishCount++;
         pump(ret, stream, finish, { end });
@@ -345,7 +372,7 @@ function pipe(src, dst, finish, { end }) {
       finish(err);
     }
   });
-  eos(dst, { readable: false, writable: true }, finish);
+  return eos(dst, { readable: false, writable: true }, finish);
 }
 
 module.exports = { pipelineImpl, pipeline };

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

@@ -0,0 +1,54 @@
+'use strict';
+
+const common = require('../common');
+const { pipeline, Duplex, PassThrough, Writable } = require('stream');
+const assert = require('assert');
+
+const a = new PassThrough();
+a.end('foobar');
+
+const b = new Duplex({
+  write(chunk, encoding, callback) {
+    callback();
+  }
+});
+
+process.on('uncaughtException', common.mustCall((err) => {
+  assert.strictEqual(err.message, 'no way');
+}, 1));
+
+pipeline(a, b, common.mustCall((error) => {
+  if (error) {
+    assert.ifError(error);
+  }
+
+  // Ensure that listeners is removed if last stream is readble
+  // And other stream's listeners unchanged
+  assert(a.listenerCount('error') > 0);
+  assert.strictEqual(b.listenerCount('error'), 0);
+  setTimeout(() => {
+    assert.strictEqual(b.listenerCount('error'), 0);
+    b.destroy(new Error('no way'));
+  }, 100);
+}));
+
+// If last stream is not readable, will not throw and remove listeners
+const c = new PassThrough();
+c.end('foobar');
+const d = new Writable({
+  write(chunk, encoding, callback) {
+    callback();
+  }
+});
+pipeline(c, d, common.mustCall((error) => {
+  if (error) {
+    assert.ifError(error);
+  }
+
+  assert(c.listenerCount('error') > 0);
+  assert(d.listenerCount('error') > 0);
+  setTimeout(() => {
+    assert(d.listenerCount('error') > 0);
+    d.destroy(new Error('no way'));
+  }, 100);
+}));