stream: fix readable state `awaitDrain` increase in recursion

PR-URL: https://github.com/nodejs/node/pull/27572
Reviewed-By: Anna Henningsen <anna@addaleax.net>

5 files changed, 50 insertions, 22 deletions

diff --git a/test/parallel/test-stream-await-drain-writers-in-synchronously-recursion-write.js b/test/parallel/test-stream-await-drain-writers-in-synchronously-recursion-write.js
new file mode 100644
index 0000000000..110d46bb9f
--- /dev/null
+++ b/test/parallel/test-stream-await-drain-writers-in-synchronously-recursion-write.js
@@ -0,0 +1,28 @@
+'use strict';
+const common = require('../common');
+const { PassThrough } = require('stream');
+
+const encode = new PassThrough({
+  highWaterMark: 1
+});
+
+const decode = new PassThrough({
+  highWaterMark: 1
+});
+
+const send = common.mustCall((buf) => {
+  encode.write(buf);
+}, 4);
+
+let i = 0;
+const onData = common.mustCall(() => {
+  if (++i === 2) {
+    send(Buffer.from([0x3]));
+    send(Buffer.from([0x4]));
+  }
+}, 4);
+
+encode.pipe(decode).on('data', onData);
+
+send(Buffer.from([0x1]));
+send(Buffer.from([0x2]));
diff --git a/test/parallel/test-stream-pipe-await-drain-manual-resume.js b/test/parallel/test-stream-pipe-await-drain-manual-resume.js
index 37acead996..a95a5e05ae 100644
--- a/test/parallel/test-stream-pipe-await-drain-manual-resume.js
+++ b/test/parallel/test-stream-pipe-await-drain-manual-resume.js
@@ -28,10 +28,10 @@ readable.pipe(writable);
 
 readable.once('pause', common.mustCall(() => {
   assert.strictEqual(
-    readable._readableState.awaitDrain,
-    1,
-    'Expected awaitDrain to equal 1 but instead got ' +
-    `${readable._readableState.awaitDrain}`
+    readable._readableState.awaitDrainWriters,
+    writable,
+    'Expected awaitDrainWriters to be a Writable but instead got ' +
+    `${readable._readableState.awaitDrainWriters}`
   );
   // First pause, resume manually. The next write() to writable will still
   // return false, because chunks are still being buffered, so it will increase
@@ -43,10 +43,10 @@ readable.once('pause', common.mustCall(() => {
 
   readable.once('pause', common.mustCall(() => {
     assert.strictEqual(
-      readable._readableState.awaitDrain,
-      1,
-      '.resume() should not reset the counter but instead got ' +
-      `${readable._readableState.awaitDrain}`
+      readable._readableState.awaitDrainWriters,
+      writable,
+      '.resume() should not reset the awaitDrainWriters, but instead got ' +
+      `${readable._readableState.awaitDrainWriters}`
     );
     // Second pause, handle all chunks from now on. Once all callbacks that
     // are currently queued up are handled, the awaitDrain drain counter should
@@ -65,10 +65,11 @@ readable.push(null);
 
 writable.on('finish', common.mustCall(() => {
   assert.strictEqual(
-    readable._readableState.awaitDrain,
-    0,
-    'awaitDrain should equal 0 after all chunks are written but instead got' +
-    `${readable._readableState.awaitDrain}`
+    readable._readableState.awaitDrainWriters,
+    null,
+    `awaitDrainWriters should be reset to null
+    after all chunks are written but instead got
+    ${readable._readableState.awaitDrainWriters}`
   );
   // Everything okay, all chunks were written.
 }));
diff --git a/test/parallel/test-stream-pipe-await-drain-push-while-write.js b/test/parallel/test-stream-pipe-await-drain-push-while-write.js
index d14ad46cb0..6dbf3c669b 100644
--- a/test/parallel/test-stream-pipe-await-drain-push-while-write.js
+++ b/test/parallel/test-stream-pipe-await-drain-push-while-write.js
@@ -6,8 +6,8 @@ const assert = require('assert');
 const writable = new stream.Writable({
   write: common.mustCall(function(chunk, encoding, cb) {
     assert.strictEqual(
-      readable._readableState.awaitDrain,
-      0
+      readable._readableState.awaitDrainWriters,
+      null,
     );
 
     if (chunk.length === 32 * 1024) { // first chunk
@@ -15,7 +15,7 @@ const writable = new stream.Writable({
       // We should check if awaitDrain counter is increased in the next
       // tick, because awaitDrain is incremented after this method finished
       process.nextTick(() => {
-        assert.strictEqual(readable._readableState.awaitDrain, 1);
+        assert.strictEqual(readable._readableState.awaitDrainWriters, writable);
       });
     }
 
diff --git a/test/parallel/test-stream-pipe-await-drain.js b/test/parallel/test-stream-pipe-await-drain.js
index 9286ceb791..3ae248e08b 100644
--- a/test/parallel/test-stream-pipe-await-drain.js
+++ b/test/parallel/test-stream-pipe-await-drain.js
@@ -24,10 +24,10 @@ writer1._write = common.mustCall(function(chunk, encoding, cb) {
 
 writer1.once('chunk-received', () => {
   assert.strictEqual(
-    reader._readableState.awaitDrain,
+    reader._readableState.awaitDrainWriters.size,
     0,
     'awaitDrain initial value should be 0, actual is ' +
-    reader._readableState.awaitDrain
+    reader._readableState.awaitDrainWriters
   );
   setImmediate(() => {
     // This one should *not* get through to writer1 because writer2 is not
@@ -39,10 +39,10 @@ writer1.once('chunk-received', () => {
 // A "slow" consumer:
 writer2._write = common.mustCall((chunk, encoding, cb) => {
   assert.strictEqual(
-    reader._readableState.awaitDrain,
+    reader._readableState.awaitDrainWriters.size,
     1,
     'awaitDrain should be 1 after first push, actual is ' +
-    reader._readableState.awaitDrain
+    reader._readableState.awaitDrainWriters
   );
   // Not calling cb here to "simulate" slow stream.
   // This should be called exactly once, since the first .write() call
@@ -51,10 +51,10 @@ writer2._write = common.mustCall((chunk, encoding, cb) => {
 
 writer3._write = common.mustCall((chunk, encoding, cb) => {
   assert.strictEqual(
-    reader._readableState.awaitDrain,
+    reader._readableState.awaitDrainWriters.size,
     2,
     'awaitDrain should be 2 after second push, actual is ' +
-    reader._readableState.awaitDrain
+    reader._readableState.awaitDrainWriters
   );
   // Not calling cb here to "simulate" slow stream.
   // This should be called exactly once, since the first .write() call
diff --git a/test/parallel/test-stream2-basic.js b/test/parallel/test-stream2-basic.js
index 5e0f9c6e91..7121f7bda7 100644
--- a/test/parallel/test-stream2-basic.js
+++ b/test/parallel/test-stream2-basic.js
@@ -355,7 +355,6 @@ class TestWriter extends EE {
   assert.strictEqual(v, null);
 
   const w = new R();
-
   w.write = function(buffer) {
     written = true;
     assert.strictEqual(ended, false);