`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
})