Repository: node
Issue: Race condition when ending a stream
Labels: stream

### Version

v14.17.5 and v16.11.1

### Platform

Linux torri 5.13.14 #29 SMP Mon Sep 6 12:28:17 CEST 2021 x86_64 GNU/Linux

### Subsystem

stream

### What steps will reproduce the bug?

Clone https://github.com/julienw/streams-problems
Run the script `node-streams.js` with node v14 or node v16.

Here is the content of this script:
```js
import Stream, { Writable, PassThrough } from "stream";
import util from "util";

const pipeline = util.promisify(Stream.pipeline);
const nextTick = util.promisify(process.nextTick);

// This Transform cheaply checks that a gzipped stream looks like a json.
class MyStream extends Writable {
  _write(chunk, encoding, callback) {
    console.log("_write", chunk);
    callback();
    // Calling end will also stop any piping gracefully.
    // Using nextTick allows some bufferred write to finish.
    process.nextTick(() => this.end());
  }

  // This is called when all the data has been given to _write and the
  // stream is ended.
  _final(callback) {
    console.log("_final()");
    callback();
  }
}

async function run() {
  console.log("START TEST");
  const fixture = "WRITE SOMETHING";
  const checker = new MyStream();
  const input = new PassThrough();
  const pipelinePromise = pipeline(input, checker);
  console.log("WRITE 1");
  input.write(fixture.slice(0, 3));
  await nextTick();
  console.log("WRITE 2");
  input.end(fixture.slice(3));
  console.log("WAIT FOR PIPELINE END");
  await pipelinePromise;
  console.log("FINISHED");
}

// Keeps the node process running
const intervalId = setInterval(() => {}, 1000);

console.log("Starting");
run()
  .then(() => console.log("End!"))
  .catch((e) => console.error(e))
  .then(() => clearInterval(intervalId));
```
(This needs to be run with `type:"module"` so it's probably easier to clone the repository)

In v12, we get this output:
```
Starting
START TEST
WRITE 1
_write <Buffer 57 52 49>
_final()
WRITE 2
WAIT FOR PIPELINE END
FINISHED
End!
```
But in v14 and v16 we get this:
```
Starting
START TEST
WRITE 1
_write <Buffer 57 52 49>
_final()
WRITE 2
WAIT FOR PIPELINE END
```
And this never ends.

As you see, we're calling `this.end()` in `_write()` after a nextTick. The original idea was that this stream's purpose was finished (checking something) and calling `end` would unpipe, and we wouldn't get more writes. The goal is to stop doing more work.

### How often does it reproduce? Is there a required condition?

Always

### What is the expected behavior?

I think we should get one of these results:
1. like v12, the pipeline is unpiped and therefore it should end because Passthrough will just finish consuming the data. (I think)
2. otherwise, we should get an error because we're trying to write after `end` (`[ERR_STREAM_WRITE_AFTER_END]: write after end`).


### What do you see instead?

Here it looks like that the pipeline isn't unpiped AND we don't get an error. So the written data isn't consumed by the stream that's been ended, and we're waiting forever.


### Additional information

Replacing `process.nextTick(() => this.end());` with something like `await nextTick(); this.end();` gets an error (behavior 2 above), ~~which is different than node v12~~ but at least we're not waiting forever and we get some clue.  __Update Oct 18__: we get the same error in v12 and v14, but in v16 we get the behavior of waiting forever.

Any insight will be very much appreciated. Especially can you point to what changed in v14 in this topic compared to v12 (I do know that a lot of the streams code changed in-between)? Also it seems to be than waiting forever isn't the right behavior.

Thanks!