stream: fix backpressure when multiple sync

PR-URL: https://github.com/nodejs/node/pull/19613
Fixes: https://github.com/nodejs/node/issues/19601
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>

1 files changed, 39 insertions, 0 deletions

diff --git a/test/parallel/test-stream-backpressure.js b/test/parallel/test-stream-backpressure.js
new file mode 100644
index 0000000000..03bcc233c8
--- /dev/null
+++ b/test/parallel/test-stream-backpressure.js
@@ -0,0 +1,39 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const stream = require('stream');
+
+let pushes = 0;
+const total = 65500 + 40 * 1024;
+const rs = new stream.Readable({
+  read: common.mustCall(function() {
+    if (pushes++ === 10) {
+      this.push(null);
+      return;
+    }
+
+    const length = this._readableState.length;
+
+    // We are at most doing two full runs of _reads
+    // before stopping, because Readable is greedy
+    // to keep its buffer full
+    assert(length <= total);
+
+    this.push(Buffer.alloc(65500));
+    for (let i = 0; i < 40; i++) {
+      this.push(Buffer.alloc(1024));
+    }
+
+    // We will be over highWaterMark at this point
+    // but a new call to _read is scheduled anyway.
+  }, 11)
+});
+
+const ws = stream.Writable({
+  write: common.mustCall(function(data, enc, cb) {
+    setImmediate(cb);
+  }, 41 * 10)
+});
+
+rs.pipe(ws);