stream: regression in v14, this.push(null) in Transform doesn't emit end

[rlidwka]

• Version: v14.5.0
• Platform: Ubuntu/Linux
• Subsystem: stream

let stream = require('stream');

let src = new stream.Readable({
  read() {
    console.log('push')
    this.push(Buffer.alloc(20000));
  }
});

let dst = new stream.Transform({
  transform(chunk, output, fn) {
    this.push(null);
    fn();
  }
});

src.pipe(dst);

function parser_end(error) {
  console.log('parser ended', error);
  dst.removeAllListeners();
}

dst.once('data', data => console.log(data));
dst.once('end', () => parser_end());
dst.on('error', error => parser_end(error));

• Expected behavior (tested on node v10, 12): end event is emitted by dst, transform stops, source pauses.
• Actual behavior (tested on node v14): end event does not get emitted by dst, transform runs indefinitely.

[lpinca]

@nodejs/streams @ronag

[mcollina]

Confirmed, this is a bug, just tested in the latest v14 and on master.

[ronag]

In this case... shouldn't it actually fail with ERR_STREAM_WRITE_AFTER_END since pipe will continue writing?

If find this case quite problematic.

[puzrin]

In this case... shouldn't it actually fail with ERR_STREAM_WRITE_AFTER_END since pipe will continue writing?

Problem is data not processed in consumer at all. Handlers, which have to call end/destroy are not reached. Looks like data buffered somewhere until stream end.

https://github.com/nodeca/probe-image-size/blob/master/stream.js - full source if anyone interested.

The same with generator-based source:

async function * generate() {
  for (;;) {
    yield Buffer.alloc(20000);
  }
}

let src = Readable.from(generate());

[mcollina]

In this case... shouldn't it actually fail with ERR_STREAM_WRITE_AFTER_END since pipe will continue writing?

No. .pipe() historically automatically unpiped from the source if the destination ended, i.e. the source is not paused while it should be.

I think we'd need to bisect this if you do not have a suspect.

[ronag]

I think we'd need to bisect this if you do not have a suspect.

Haven't started digging yet. Trying to understand how it should work first.

No. .pipe() historically automatically unpiped from the source if the destination ended, i.e. the source is not paused while it should be.

I'm not sure i entirely agree with this. .pipe() will unpipe on 'end', however 'end' is emitted sometime after the destination has been .end():ed so there is still a possibility (which is quite probable) that pipe will call .write() after .end() but before 'end'.

I suspect we would need to update .pipe() to check for writableEnded before .write() to correctly achieve the behavior you are assuming.

[mcollina]

let's get the fix in and then do a follow-up refactor.