René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Race condition when ending a stream","articleBody":"### Version\r\n\r\nv14.17.5 and v16.11.1\r\n\r\n### Platform\r\n\r\nLinux torri 5.13.14 #29 SMP Mon Sep 6 12:28:17 CEST 2021 x86_64 GNU/Linux\r\n\r\n### Subsystem\r\n\r\nstream\r\n\r\n### What steps will reproduce the bug?\r\n\r\nClone https://github.com/julienw/streams-problems\r\nRun the script `node-streams.js` with node v14 or node v16.\r\n\r\nHere is the content of this script:\r\n```js\r\nimport Stream, { Writable, PassThrough } from \"stream\";\r\nimport util from \"util\";\r\n\r\nconst pipeline = util.promisify(Stream.pipeline);\r\nconst nextTick = util.promisify(process.nextTick);\r\n\r\n// This Transform cheaply checks that a gzipped stream looks like a json.\r\nclass MyStream extends Writable {\r\n  _write(chunk, encoding, callback) {\r\n    console.log(\"_write\", chunk);\r\n    callback();\r\n    // Calling end will also stop any piping gracefully.\r\n    // Using nextTick allows some bufferred write to finish.\r\n    process.nextTick(() => this.end());\r\n  }\r\n\r\n  // This is called when all the data has been given to _write and the\r\n  // stream is ended.\r\n  _final(callback) {\r\n    console.log(\"_final()\");\r\n    callback();\r\n  }\r\n}\r\n\r\nasync function run() {\r\n  console.log(\"START TEST\");\r\n  const fixture = \"WRITE SOMETHING\";\r\n  const checker = new MyStream();\r\n  const input = new PassThrough();\r\n  const pipelinePromise = pipeline(input, checker);\r\n  console.log(\"WRITE 1\");\r\n  input.write(fixture.slice(0, 3));\r\n  await nextTick();\r\n  console.log(\"WRITE 2\");\r\n  input.end(fixture.slice(3));\r\n  console.log(\"WAIT FOR PIPELINE END\");\r\n  await pipelinePromise;\r\n  console.log(\"FINISHED\");\r\n}\r\n\r\n// Keeps the node process running\r\nconst intervalId = setInterval(() => {}, 1000);\r\n\r\nconsole.log(\"Starting\");\r\nrun()\r\n  .then(() => console.log(\"End!\"))\r\n  .catch((e) => console.error(e))\r\n  .then(() => clearInterval(intervalId));\r\n```\r\n(This needs to be run with `type:\"module\"` so it's probably easier to clone the repository)\r\n\r\nIn v12, we get this output:\r\n```\r\nStarting\r\nSTART TEST\r\nWRITE 1\r\n_write \u003cBuffer 57 52 49>\r\n_final()\r\nWRITE 2\r\nWAIT FOR PIPELINE END\r\nFINISHED\r\nEnd!\r\n```\r\nBut in v14 and v16 we get this:\r\n```\r\nStarting\r\nSTART TEST\r\nWRITE 1\r\n_write \u003cBuffer 57 52 49>\r\n_final()\r\nWRITE 2\r\nWAIT FOR PIPELINE END\r\n```\r\nAnd this never ends.\r\n\r\nAs 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.\r\n\r\n### How often does it reproduce? Is there a required condition?\r\n\r\nAlways\r\n\r\n### What is the expected behavior?\r\n\r\nI think we should get one of these results:\r\n1. like v12, the pipeline is unpiped and therefore it should end because Passthrough will just finish consuming the data. (I think)\r\n2. otherwise, we should get an error because we're trying to write after `end` (`[ERR_STREAM_WRITE_AFTER_END]: write after end`).\r\n\r\n\r\n### What do you see instead?\r\n\r\nHere 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.\r\n\r\n\r\n### Additional information\r\n\r\nReplacing `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.\r\n\r\nAny 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.\r\n\r\nThanks!","author":{"url":"https://github.com/julienw","@type":"Person","name":"julienw"},"datePublished":"2021-10-15T18:01:52Z","url":"https://github.com/nodejs/node/issues/40471"}