Commit 3dc6564
stream: fix enqueue race condition on esm modules
stream: use nextTick on close
PR-URL: #40901
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>1 parent e27e827 commit 3dc6564
2 files changed
Lines changed: 48 additions & 7 deletions
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -1446,13 +1446,18 @@ function readableStreamTee(stream, cloneForBranch2) { | |||
| 1446 | 1446 | }); | |
| 1447 | 1447 | }, | |
| 1448 | 1448 | [kClose]() { | |
| 1449 | - reading = false; | ||
| 1450 | - if (!canceled1) | ||
| 1451 | - readableStreamDefaultControllerClose(branch1[kState].controller); | ||
| 1452 | - if (!canceled2) | ||
| 1453 | - readableStreamDefaultControllerClose(branch2[kState].controller); | ||
| 1454 | - if (!canceled1 || !canceled2) | ||
| 1455 | - cancelPromise.resolve(); | ||
| 1449 | + // The `process.nextTick()` is not part of the spec. | ||
| 1450 | + // This approach was needed to avoid a race condition working with esm | ||
| 1451 | + // Further information, see: https://github.com/nodejs/node/issues/39758 | ||
| 1452 | + process.nextTick(() => { | ||
| 1453 | + reading = false; | ||
| 1454 | + if (!canceled1) | ||
| 1455 | + readableStreamDefaultControllerClose(branch1[kState].controller); | ||
| 1456 | + if (!canceled2) | ||
| 1457 | + readableStreamDefaultControllerClose(branch2[kState].controller); | ||
| 1458 | + if (!canceled1 || !canceled2) | ||
| 1459 | + cancelPromise.resolve(); | ||
| 1460 | + }); | ||
| 1456 | 1461 | }, | |
| 1457 | 1462 | [kError]() { | |
| 1458 | 1463 | reading = false; | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -0,0 +1,36 @@ | |||
| 1 | + import { mustCall } from '../common/index.mjs'; | ||
| 2 | + import { ReadableStream } from 'stream/web'; | ||
| 3 | + import assert from 'assert'; | ||
| 4 | + | ||
| 5 | + { | ||
| 6 | + // Test tee() with close in the nextTick after enqueue | ||
| 7 | + async function read(stream) { | ||
| 8 | + const chunks = []; | ||
| 9 | + for await (const chunk of stream) | ||
| 10 | + chunks.push(chunk); | ||
| 11 | + return Buffer.concat(chunks).toString(); | ||
| 12 | + } | ||
| 13 | + | ||
| 14 | + const [r1, r2] = new ReadableStream({ | ||
| 15 | + start(controller) { | ||
| 16 | + process.nextTick(() => { | ||
| 17 | + controller.enqueue(new Uint8Array([102, 111, 111, 98, 97, 114])); | ||
| 18 | + | ||
| 19 | + process.nextTick(() => { | ||
| 20 | + controller.close(); | ||
| 21 | + }); | ||
| 22 | + }); | ||
| 23 | + } | ||
| 24 | + }).tee(); | ||
| 25 | + | ||
| 26 | + (async () => { | ||
| 27 | + const [dataReader1, dataReader2] = await Promise.all([ | ||
| 28 | + read(r1), | ||
| 29 | + read(r2), | ||
| 30 | + ]); | ||
| 31 | + | ||
| 32 | + assert.strictEqual(dataReader1, dataReader2); | ||
| 33 | + assert.strictEqual(dataReader1, 'foobar'); | ||
| 34 | + assert.strictEqual(dataReader2, 'foobar'); | ||
| 35 | + })().then(mustCall()); | ||
| 36 | + } | ||
0 commit comments