From a2c5fcc897669cd6531393b1045f6a5f0d9ae11e Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Fri, 4 Feb 2022 10:36:58 +0100
Subject: [PATCH 1/2] stream: resume stream on drain

Previously we would just resume "flowing" the stream without
reseting the "paused" state. Fixes this by properly using
pause/resume methods for .pipe.

Fixes: https://github.com/nodejs/node/issues/41785
---
 lib/internal/streams/readable.js                |  3 +--
 .../test-stream-readable-pause-and-resume.js    | 17 +++++++++++++++++
 2 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index d0125386c8ae8e..7dc2e5346afad9 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -853,8 +853,7 @@ function pipeOnDrain(src, dest) {
 
     if ((!state.awaitDrainWriters || state.awaitDrainWriters.size === 0) &&
       EE.listenerCount(src, 'data')) {
-      state.flowing = true;
-      flow(src);
+      src.resume();
     }
   };
 }
diff --git a/test/parallel/test-stream-readable-pause-and-resume.js b/test/parallel/test-stream-readable-pause-and-resume.js
index 294ef2c35d4608..f28112da10e1bb 100644
--- a/test/parallel/test-stream-readable-pause-and-resume.js
+++ b/test/parallel/test-stream-readable-pause-and-resume.js
@@ -56,3 +56,20 @@ function readAndPause() {
     assert(readable.isPaused());
   });
 }
+
+{
+  const { PassThrough } = require('stream');
+
+  const source3 = new PassThrough();
+  const target3 = new PassThrough();
+
+  const chunk = Buffer.allocUnsafe(1000);
+  let chunks = 1;
+  while (target3.write(chunk)) chunks++;
+
+  source3.pipe(target3);
+  target3.on('drain', common.mustCall(() => {
+    assert(!source3.isPaused());
+  }));
+  target3.on('data', () => {});
+}

From 4916f0cde3fb81bf226d9759e07d84d515aeef5e Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Fri, 4 Feb 2022 10:47:25 +0100
Subject: [PATCH 2/2] fixup

---
 test/parallel/test-stream-readable-pause-and-resume.js | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/test/parallel/test-stream-readable-pause-and-resume.js b/test/parallel/test-stream-readable-pause-and-resume.js
index f28112da10e1bb..53229ec3339e5c 100644
--- a/test/parallel/test-stream-readable-pause-and-resume.js
+++ b/test/parallel/test-stream-readable-pause-and-resume.js
@@ -64,8 +64,7 @@ function readAndPause() {
   const target3 = new PassThrough();
 
   const chunk = Buffer.allocUnsafe(1000);
-  let chunks = 1;
-  while (target3.write(chunk)) chunks++;
+  while (target3.write(chunk));
 
   source3.pipe(target3);
   target3.on('drain', common.mustCall(() => {