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

[meixg]

fix: #35452

const {pipeline, Duplex, PassThrough, Writable} = require('stream');

const a = new PassThrough();
a.end('foobar');

const b = new Duplex({
    write(chunk, encoding, callback) {
        callback();
    }
});

pipeline(a, b, error => {
    if (error) {
        throw error;
    }
    console.log(b.listenerCount('error'));
    setTimeout(() => {
        console.log(b.listenerCount('error'));
        b.destroy(new Error('no way'));
    }, 100);
});

output:

0
0
node:events:505
      throw er; // Unhandled 'error' event
      ^

Error: no way

@ronag @mcollina

[nodejs-github-bot]

Review requested:

• @nodejs/streams

[mcollina]

Can you please add a unit test?

[ronag]

This has subtle problems. I think a possible solution is to just cleanup the listeners for the last stream.

[meixg]

This has subtle problems. I think a possible solution is to just cleanup the listeners for the last stream.

This change uses an isLastStream variable to only process the last stream.

[meixg]

Can you please add a unit test?

Unit test added :)

[ronag]

You need to check if the stream is readable at the end before the cleanup.

[meixg]

You need to check if the stream is readable at the end before the cleanup.

I checked at the beginning using .readable，will this change between the end？

[ronag]

I don't think this will work properly. There are a lot of edge cases that are not handled. I would recommend to not try and to only partly achieve a behaviour and cause lots of unexpected edge cases.

If this is behaviour is important than you can achieve it with something like:

const ret = pipeline(...)
const dst = new PassThrough()
ret.on('error', err => dst.destroy(err)).pipe(dst)

[ronag]

Suggested change

	callback: (err) => {
	destroy: (err) => {

[ronag]

Suggested change

	while (lastStreamCleanup && lastStreamCleanup.length) {
	lastStreamCleanup.shift()();
	}
	lastStreamCleanup.forEach(fn => fn());

[ronag]

Suggested change

	const { callback, cleanup } = destroyer(stream, reading, writing);
	destroys.push(callback);
	if (lastStreamCleanup && isLastStream) {
	lastStreamCleanup.push(cleanup);
	}
	const { destroy, cleanup } = destroyer(stream, reading, writing);
	destroys.push(destroy);
	if (isReadable(stream) && isLastStream) {
	lastStreamCleanup.push(cleanup);
	}

[ronag]

Suggested change

	if (lastStreamCleanup && isLastStream) {
	lastStreamCleanup.push(() => {
	stream.removeListener('error', onError);
	});
	}
	if (isReadable(stream) && isLastStream) {
	lastStreamCleanup.push(() => {
	stream.removeListener('error', onError);
	});
	}

[ronag]

Suggested change

	destroys.push(destroyer(ret, false, true).callback);
	const { destroy, cleanup } = (destroyer(ret, false, true)
	destroys.push(destroy);
	if (isLastStream) {
	lastStreamCleanup.push(cleanup);
	}

[ronag]

Suggested change

	if (lastStreamCleanup && isLastStream) {
	lastStreamCleanup.push(cleanup);
	}
	if (isReadable(stream) && isLastStream) {
	lastStreamCleanup.push(cleanup);
	}

[ronag]

Maybe ok. A little scary. Added some fixes.

[ronag]

Will also need a doc update.

[meixg]

Maybe ok. A little scary. Added some fixes.

Applied suggestions from code review, and added a test case for async generators.

The doc is updated too.

[mcollina]

lgtm

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/42528/

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/42543/

[meixg]

Can you help me with this CI? @ronag

[nodejs-github-bot]

Commit Queue failed

- Loading data for nodejs/node/pull/41954
✔  Done loading data for nodejs/node/pull/41954
----------------------------------- PR info ------------------------------------
Title      stream: need to cleanup event listeners if last stream is readable (#41954)
Author     Xuguang Mei  (@meixg)
Branch     meixg:stream-meixg -> nodejs:master
Labels     stream, semver-major, author ready, needs-ci
Commits    2
 - stream: need to cleanup event listeners if last stream is readable
 - stream: need to cleanup event listeners if last stream is readable
Committers 1
 - meixg 
PR-URL: https://github.com/nodejs/node/pull/41954
Fixes: https://github.com/nodejs/node/issues/35452
Reviewed-By: Robert Nagy 
Reviewed-By: Matteo Collina 
------------------------------ Generated metadata ------------------------------
PR-URL: https://github.com/nodejs/node/pull/41954
Fixes: https://github.com/nodejs/node/issues/35452
Reviewed-By: Robert Nagy 
Reviewed-By: Matteo Collina 
--------------------------------------------------------------------------------
   ℹ  This PR was created on Sun, 13 Feb 2022 08:44:23 GMT
   ✔  Approvals: 2
   ✔  - Robert Nagy (@ronag) (TSC): https://github.com/nodejs/node/pull/41954#pullrequestreview-881018526
   ✔  - Matteo Collina (@mcollina) (TSC): https://github.com/nodejs/node/pull/41954#pullrequestreview-881044648
   ✖  This PR needs to wait 25 more minutes to land
   ✔  Last GitHub CI successful
   ℹ  Last Full PR CI on 2022-02-15T07:54:47Z: https://ci.nodejs.org/job/node-test-pull-request/42575/
- Querying data for job/node-test-pull-request/42575/
   ✔  Last Jenkins CI successful
--------------------------------------------------------------------------------
   ✔  Aborted `git node land` session in /home/runner/work/node/node/.ncu

https://github.com/nodejs/node/actions/runs/1845761432

[meixg]

This PR needs to wait 25 more minutes to land

can land now :)

[nodejs-github-bot]

Commit Queue failed

- Loading data for nodejs/node/pull/41954
✔  Done loading data for nodejs/node/pull/41954
----------------------------------- PR info ------------------------------------
Title      stream: need to cleanup event listeners if last stream is readable (#41954)
Author     Xuguang Mei  (@meixg)
Branch     meixg:stream-meixg -> nodejs:master
Labels     stream, semver-major, author ready, needs-ci
Commits    2
 - stream: need to cleanup event listeners if last stream is readable
 - stream: need to cleanup event listeners if last stream is readable
Committers 1
 - meixg 
PR-URL: https://github.com/nodejs/node/pull/41954
Fixes: https://github.com/nodejs/node/issues/35452
Reviewed-By: Robert Nagy 
Reviewed-By: Matteo Collina 
------------------------------ Generated metadata ------------------------------
PR-URL: https://github.com/nodejs/node/pull/41954
Fixes: https://github.com/nodejs/node/issues/35452
Reviewed-By: Robert Nagy 
Reviewed-By: Matteo Collina 
--------------------------------------------------------------------------------
   ℹ  This PR was created on Sun, 13 Feb 2022 08:44:23 GMT
   ✔  Approvals: 2
   ✔  - Robert Nagy (@ronag) (TSC): https://github.com/nodejs/node/pull/41954#pullrequestreview-881018526
   ✔  - Matteo Collina (@mcollina) (TSC): https://github.com/nodejs/node/pull/41954#pullrequestreview-881044648
   ✔  Last GitHub CI successful
   ℹ  Last Full PR CI on 2022-02-15T08:19:08Z: https://ci.nodejs.org/job/node-test-pull-request/42575/
- Querying data for job/node-test-pull-request/42575/
   ✔  Last Jenkins CI successful
--------------------------------------------------------------------------------
   ✔  No git cherry-pick in progress
   ✔  No git am in progress
   ✔  No git rebase in progress
--------------------------------------------------------------------------------
- Bringing origin/master up to date...
From https://github.com/nodejs/node
 * branch                  master     -> FETCH_HEAD
✔  origin/master is now up-to-date
- Downloading patch for 41954
From https://github.com/nodejs/node
 * branch                  refs/pull/41954/merge -> FETCH_HEAD
✔  Fetched commits as cc505a505a16..97fb400489ed
--------------------------------------------------------------------------------
[master c8109f4cf6] stream: need to cleanup event listeners if last stream is readable
 Author: meixg 
 Date: Sun Feb 13 16:38:29 2022 +0800
 2 files changed, 92 insertions(+), 11 deletions(-)
 create mode 100644 test/parallel/test-stream-pipeline-listeners.js
[master 32dd26b122] stream: need to cleanup event listeners if last stream is readable
 Author: meixg 
 Date: Mon Feb 14 00:36:12 2022 +0800
 3 files changed, 54 insertions(+), 27 deletions(-)
   ✔  Patches applied
There are 2 commits in the PR. Attempting autorebase.
Rebasing (2/4)
Executing: git node land --amend --yes

--------------------------------- New Message ----------------------------------

stream: need to cleanup event listeners if last stream is readable
fix: #35452
PR-URL: #41954

Fixes: #35452

Reviewed-By: Robert Nagy ronagy@icloud.com

Reviewed-By: Matteo Collina matteo.collina@gmail.com
[detached HEAD 54d48f479e] stream: need to cleanup event listeners if last stream is readable

Author: meixg meixg@foxmail.com

Date: Sun Feb 13 16:38:29 2022 +0800

2 files changed, 92 insertions(+), 11 deletions(-)

create mode 100644 test/parallel/test-stream-pipeline-listeners.js

Rebasing (3/4)

Rebasing (4/4)
Executing: git node land --amend --yes

--------------------------------- New Message ----------------------------------

stream: need to cleanup event listeners if last stream is readable
Apply suggestions from code review.

Add test case for async generators.

Update doc.
PR-URL: #41954

Fixes: #35452

Reviewed-By: Robert Nagy ronagy@icloud.com

Reviewed-By: Matteo Collina matteo.collina@gmail.com
[detached HEAD a0db88ebcc] stream: need to cleanup event listeners if last stream is readable

Author: meixg meixg@foxmail.com

Date: Mon Feb 14 00:36:12 2022 +0800

3 files changed, 54 insertions(+), 27 deletions(-)
Successfully rebased and updated refs/heads/master.
ℹ  Use --fixupAll option, squash the PR manually or land the PR from the command line.

https://github.com/nodejs/node/actions/runs/1845985560

[meixg]

May need to use "commit-queue-squash"

[meixg]

Seems that the "commit-queue-squash" needs to work with "commit-queue"

[nodejs-github-bot]

Landed in 9fb7ac3

[meixg]

thanks! @ronag