pipeline leaves a hanging error handler

[szmarczak]

• Version: 14.12.0
• Platform: Linux SZM-DESKTOP 4.19.104-microsoft-standard #1 SMP Wed Feb 19 06:37:35 UTC 2020 x86_64 x86_64 x86_64 GNU/Linux
• Subsystem: stream

What steps will reproduce the bug?

const {pipeline, Duplex, PassThrough} = 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);
});

How often does it reproduce? Is there a required condition?

Always.

What is the expected behavior?

0
0
[Uncaught] Error: no way

What do you see instead?

2
1

/cc @ronag

[ronag]

This is by design.

[ronag]

The docs say:

stream.pipeline() leaves dangling event listeners on the streams after the callback has been invoked. In the case of reuse of streams after failure, this can cause event listener leaks and swallowed errors.

[szmarczak]

This is by design.

Can this design be improved? Or is this a no?

[possum1102]

Hmmm

[targos]

/cc @nodejs/streams

[ronag]

I don’t think there is anything more to do here. Unless someone has suggestion?

[vweevers]

@ronag I agree. Using pipeline() means letting pipeline() manage the lifetime of the streams, after which the streams are destroyed and should not be touched anymore.

[szmarczak]

Using pipeline() means letting pipeline() manage the lifetime of the streams

It does not. b.destroyed is false. So the readable side is still open. The fix is either to remove the hanging handler or destroy the end stream.

[vweevers]

I missed that the last stream in the example is a duplex stream. Not sure how that should behave, because the readable side can't be consumed.

[szmarczak]

the readable side can't be consumed.

What do you mean?

[vweevers]

If it were pipeline(readableStream, duplexStream, writableStream) then the readable side of duplexStream would be piped into writableStream. But if duplexStream is the last stream in the pipeline, then no one is reading from it.

If we swap the duplex stream in your example for a writable stream, then it does get destroyed:

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

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

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

pipeline(a, b, function (err) {
  if (err) {
    throw err
  }

  console.log(a.destroyed) // true
  console.log(b.destroyed) // true
})

[szmarczak]

But if duplexStream is the last stream in the pipeline, then no one is reading from it.

You don't know. It's readable so it's not destroyed. Something can be reading it a tick later.

[ronag]

Removing the error handler on the last (returned) stream might make sense if it is readable. @nodejs/streams wdyt?

[mcollina]

I'm ok in removing it when we do not call stream.destroy().

[ronag]

@szmarczak would you like to prepare a PR?

[szmarczak]

Sure, would love to.

[mcollina]

@szmarczak any updates here?

[szmarczak]

Sorry, totally forgot about this one. Will sketch something in a few hours.

[joaoofreitas]

Sorry, totally forgot about this one. Will sketch something in a few hours.

Count on me to help! @szmarczak

[szmarczak]

@joaoofreitas awesome! Feel free to send a PR first and I'll chime in and let you know my thoughts :D

[lvndry]

If this is still work in progress I'd be happy to help

[mcollina]

Sure thing!

[Heikrana]

Hey @mcollina. Can I help in this? (If @lvndry isn't working on it anymore)

I'm new to NodeJS and to Open Source. Might be hard for me, but I'd like to try.

[mcollina]

@Heikrana this might end up a bit too hard. Maybe try something easier first? It's one of the hardest part of Node.

[Heikrana]

@Heikrana this might end up a bit too hard. Maybe try something easier first? It's one of the hardest part of Node.

Okay, I see your point. I've been trying to understand the stream API and yeah, it's a bit confusing.
I'd still try to solve this issue (since no one else is working on it).

But in the meantime, can you suggest me some resources (which would help me contribute to NodeJS) or another issue which might be more of my level. I really want to be part of this community :)