stream.pipeline doesn't wait for 'close' on error

[wh0]

Version

v20.11.0

Platform

Linux cs-63018580108-ephemeral-4kum 6.1.58+ #1 SMP PREEMPT_DYNAMIC Sat Dec 30 15:31:26 UTC 2023 x86_64 GNU/Linux

Subsystem

internal/streams/pipeline

What steps will reproduce the bug?

const stream = require('stream');

const src = new stream.Readable();
const dst = new stream.Writable({
  write(chunk, encoding, callback) {
    callback();
  },
  destroy(error, callback) {
    // takes a while to destroy
    setImmediate(() => {
      callback(error);
    });
  },
});

stream.pipeline(src, dst, (err) => {
  console.log(`pipeline done, err=${err.message}, dst.closed=${dst.closed}`);
});
src.destroy(new Error('problem'));

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

always

What is the expected behavior? Why is that the expected behavior?

dst.closed=true

maybe? #32158 sounds like it should

What do you see instead?

pipeline done, err=problem, dst.closed=false

Additional information

I notice that this finishCount countdown goes from 2 down to -3.

1. (2 -> 1) src error event ("Catch stream errors that occur after pipe/pump has completed.")
2. (1 -> 0, final) src err event (same one) (pipe src eos onerror)
3. (0 -> -1) dst error event ("Catch stream errors that occur after pipe/pump has completed.")
4. (-1 -> -2) dst error event (same one) (pipe dst eos onerror)
5. (-2 -> -3) dst close event ("Finish if the destination closes before the source has completed.")

[jakecastelli]

Hi, I am trying to get a bit more context here, do you expect the close event on Writable stream happens before the callback is being called?

[wh0]

yes. from all that I've seen, I do think the intention of stream.pipeline was to wait for the Writable to finish closing before calling the callback. there's this finishCount code that's meant to keep track of everything closing, but it's getting decremented way more times than the code that sets up the initial count expects

[jakecastelli]

I do think the intention of stream.pipeline was to wait for the Writable to finish closing before calling the callback

I think you are right on this one. I have taken some time to dig, what I found it out is that because the ERR_METHOD_NOT_IMPLEMENTED error that messed up the finishCount (I used the debugger but you can just log the err here and I think you will understand what I mean.

If you implement the read{} function in the Readable stream then it is the behaviour you expected, or if you can take a look the patch I made here and let me know if it makes sense or not 👀

Thanks a lot for narrowing down the directions, if the fix is good are you happy for me to add you as a co-author on the PR?

---

EDIT: Sorry, I think my fix was incorrect!

[jakecastelli]

Hi @ronag @lpinca, the doc said the callback is called when the pipeline is fully done

How do we understand its fully done here? 👀 as we can see when the callback is called, the dst (writable stream) is not closed.

[lpinca]

IIRC the first error wins. The pipeline is "done" and all streams destroyed when the first error occurs.

[jakecastelli]

so if I understand correctly "done" here means all the streams have been destroyed but not guaranteed closed?

[lpinca]

Yes.

[wh0]

about read not implemented: that was unintentional. wasn't new stream.Readable() the way to make readable stream where you'd call .push(chunk) and .push(null)? it should use the default read implementation that waits for something to call push.

I was using the closed property to check that the stream was destroyed, done being destroyed, specifically. now that I look more into the streams API, I see that you can configure streams not to emit a close event when you destroy them. that complicates things.

for longer pipelines, it looks like the current logic would wait for about half of the streams to emit error before calling the callback, and that also feels weird.