[REG 20.9->20.10] stream: Pipe is stopped after error on another pipe from the same source

[orgads]

Version

20.13.1

Platform

Microsoft Windows NT 10.0.19045.0 x64

Subsystem

stream

What steps will reproduce the bug?

import assert from 'assert';
import stream from 'stream';

class SizeCounter extends stream.Writable {
  constructor(name, fail) {
    super();
    this.totalSize = 0;
    this.name = name;
    this.fail = fail;
  }

  _write(chunk, _encoding, callback) {
    this.totalSize += chunk.length;
    console.log(this.name, this.totalSize);
    if (this.fail)
      return callback(new Error('You asked me to fail'));
    return callback();
  }

  _final(callback) {
    console.log(`${this.name} Total size: ${this.totalSize}`);
    callback();
  }
}

const src = new stream.PassThrough();
const dst1 = new SizeCounter('1', true);
const dst2 = new SizeCounter('2', false);

src.write(Buffer.alloc(20000));
src.write(Buffer.alloc(20000));
src.end();

dst1.on('error', () => {
  // uncomment as a workaround
  // src.resume();
});

stream.finished(src, () => console.log('src ended'));
stream.finished(dst1, (err) => console.log('dst1 ended', err?.message));
stream.finished(dst2, (err) => console.log('dst2 ended', err?.message));

src.pipe(dst1, { end: true });
src.pipe(dst2, { end: true });
setImmediate(() => assert.strictEqual(dst2.totalSize, 40000));

Output with 20.9.0:

1 20000
2 20000
dst1 ended You asked me to fail
2 40000
2 Total size: 40000
src ended
dst2 ended undefined

Output with 20.10.0:

1 20000
2 20000
dst1 ended You asked me to fail
node:assert:125
  throw new AssertionError(obj);
  ^

AssertionError [ERR_ASSERTION]: Expected values to be strictly equal:

20000 !== 40000

    at Immediate._onImmediate (file:///home/orgads/test/test.mjs:45:27)
    at process.processImmediate (node:internal/timers:478:21) {
  generatedMessage: true,
  code: 'ERR_ASSERTION',
  actual: 20000,
  expected: 40000,
  operator: 'strictEqual'
}

Node.js v20.10.0

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

Always

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

I expect the stream to keep draining (at least when it still has listeners to data).

What do you see instead?

The stream is paused.

Additional information

Bisected to #50014.

Related bug:

• piping to a stream with highWaterMark: 0 and objectMode: false is broken since node v20.10.0 #51930

[orgads]

@ronag @mcollina @benjamingr

[ronag]

I'm not sure I would consider this a bug. The stream should stop IMO. However, it is a semver-major change of behavior that should not have landed on v20.0.

[orgads]

Why should the stream stop if it still has other listeners on 'data'?

[ronag]

when piping to multiple destinations the data flow should not be faster than the slowest destintation (this is the behavior in the non failing case). A failed destination is no longer consuming and should therefore stop the flow. Just like in the non-error flow.

[orgads]

I see. It's worth at least documenting this change in "Notable changes".

[mcollina]

when piping to multiple destinations the data flow should not be faster than the slowest destintation (this is the behavior in the non failing case). A failed destination is no longer consuming and should therefore stop the flow. Just like in the non-error flow.

I have a rough feeling this is not what the user would expect. If I'm piping to two streams, I don't want one of the two pipes to end, surprisingly, because the other had an unrelated failure. While we must slow down at the pace of the slowest consumer, we should definitely not break the other user. I think we should consider either reverting or fixing this.

Wdyt @ronag?

Note that this kind of double-piping is problematic in many other ways, so it might not be useful to revert. If we consider this behavior as ok, we should consider adding a clone() or bring https://github.com/mcollina/cloneable-readable into core.

[ronag]

I'll have a look if we can fix this so it behaves as it used to.