From 591616f37feea7ffa532cb53cb1f4415679821bb Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Fri, 25 Sep 2020 18:40:01 +0200
Subject: [PATCH 1/5] stream: writableNeedDrain

Don't write to a stream which already has a full buffer.

Fixes: https://github.com/nodejs/node/issues/35341
---
 doc/api/stream.md                |  9 ++++++
 lib/_http_outgoing.js            |  5 +++
 lib/internal/streams/duplex.js   |  2 ++
 lib/internal/streams/pipeline.js |  4 +++
 lib/internal/streams/readable.js | 55 +++++++++++++++++++-------------
 lib/internal/streams/writable.js |  8 +++++
 6 files changed, 60 insertions(+), 23 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index 97e848f56fa1fa..7f09c4d216eedf 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -573,6 +573,15 @@ This property contains the number of bytes (or objects) in the queue
 ready to be written. The value provides introspection data regarding
 the status of the `highWaterMark`.
 
+##### `writable.writableNeedDrain`
+<!-- YAML
+added: REPLACEME
+-->
+
+* {boolean}
+
+Is `true` if buffer is full.
+
 ##### `writable.writableObjectMode`
 <!-- YAML
 added: v12.3.0
diff --git a/lib/_http_outgoing.js b/lib/_http_outgoing.js
index e683b61e94a927..aa7030d7143bba 100644
--- a/lib/_http_outgoing.js
+++ b/lib/_http_outgoing.js
@@ -649,6 +649,11 @@ ObjectDefineProperty(OutgoingMessage.prototype, 'writableEnded', {
   get: function() { return this.finished; }
 });
 
+ObjectDefineProperty(OutgoingMessage.prototype, 'writableNeedDrain', {
+  get: function() {
+    return !this.destroyed && !this.finished && this[kNeedDrain];
+  }
+});
 
 const crlf_buf = Buffer.from('\r\n');
 OutgoingMessage.prototype.write = function write(chunk, encoding, callback) {
diff --git a/lib/internal/streams/duplex.js b/lib/internal/streams/duplex.js
index 3bab26d9cef934..596d30a90a0418 100644
--- a/lib/internal/streams/duplex.js
+++ b/lib/internal/streams/duplex.js
@@ -87,6 +87,8 @@ ObjectDefineProperties(Duplex.prototype, {
     ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableCorked'),
   writableEnded:
     ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableEnded'),
+  writableNeedDrain:
+    ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableNeedDrain'),
 
   destroyed: {
     get() {
diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
index c662bce8db1ab3..3a69555421abfb 100644
--- a/lib/internal/streams/pipeline.js
+++ b/lib/internal/streams/pipeline.js
@@ -126,6 +126,10 @@ async function pump(iterable, writable, finish) {
   }
   let error;
   try {
+    if (writable.writableNeedDrain === true) {
+      await EE.once(writable, 'drain');
+    }
+
     for await (const chunk of iterable) {
       if (!writable.write(chunk)) {
         if (writable.destroyed) return;
diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index abd7faf59f2723..c668bdef757629 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -727,35 +727,44 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
       ondrain();
   }
 
+  function pause() {
+    // If the user unpiped during `dest.write()`, it is possible
+    // to get stuck in a permanently paused state if that write
+    // also returned false.
+    // => Check whether `dest` is still a piping destination.
+    if (!cleanedUp) {
+      if (state.pipes.length === 1 && state.pipes[0] === dest) {
+        debug('false write response, pause', 0);
+        state.awaitDrainWriters = dest;
+        state.multiAwaitDrain = false;
+      } else if (state.pipes.length > 1 && state.pipes.includes(dest)) {
+        debug('false write response, pause', state.awaitDrainWriters.size);
+        state.awaitDrainWriters.add(dest);
+      }
+      src.pause();
+    }
+    if (!ondrain) {
+      // When the dest drains, it reduces the awaitDrain counter
+      // on the source.  This would be more elegant with a .once()
+      // handler in flow(), but adding and removing repeatedly is
+      // too slow.
+      ondrain = pipeOnDrain(src, dest);
+      dest.on('drain', ondrain);
+    }
+  }
+
   src.on('data', ondata);
+
+  if (dest.writableNeedDrain === true) {
+    pause();
+  }
+
   function ondata(chunk) {
     debug('ondata');
     const ret = dest.write(chunk);
     debug('dest.write', ret);
     if (ret === false) {
-      // If the user unpiped during `dest.write()`, it is possible
-      // to get stuck in a permanently paused state if that write
-      // also returned false.
-      // => Check whether `dest` is still a piping destination.
-      if (!cleanedUp) {
-        if (state.pipes.length === 1 && state.pipes[0] === dest) {
-          debug('false write response, pause', 0);
-          state.awaitDrainWriters = dest;
-          state.multiAwaitDrain = false;
-        } else if (state.pipes.length > 1 && state.pipes.includes(dest)) {
-          debug('false write response, pause', state.awaitDrainWriters.size);
-          state.awaitDrainWriters.add(dest);
-        }
-        src.pause();
-      }
-      if (!ondrain) {
-        // When the dest drains, it reduces the awaitDrain counter
-        // on the source.  This would be more elegant with a .once()
-        // handler in flow(), but adding and removing repeatedly is
-        // too slow.
-        ondrain = pipeOnDrain(src, dest);
-        dest.on('drain', ondrain);
-      }
+      pause();
     }
   }
 
diff --git a/lib/internal/streams/writable.js b/lib/internal/streams/writable.js
index e3d13dc1bfc675..cb36e322e7bb62 100644
--- a/lib/internal/streams/writable.js
+++ b/lib/internal/streams/writable.js
@@ -805,6 +805,14 @@ ObjectDefineProperties(Writable.prototype, {
     }
   },
 
+  writableNeedDrain: {
+    get() {
+      const wState = this._writableState;
+      if (!wState) return false;
+      return !this._writable.destroyed && !wState.ending && wState.needDrain;
+    }
+  },
+
   writableHighWaterMark: {
     get() {
       return this._writableState && this._writableState.highWaterMark;

From 08118868a71cb6a9d1e99ccabd8d3bfc39e0430b Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Fri, 9 Oct 2020 12:26:33 +0200
Subject: [PATCH 2/5] fixup: docs

---
 doc/api/stream.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index 7f09c4d216eedf..7291db8cb660f6 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -580,7 +580,7 @@ added: REPLACEME
 
 * {boolean}
 
-Is `true` if buffer is full.
+Is `true` if the stream's buffer has been full and stream will emit `'drain'`.
 
 ##### `writable.writableObjectMode`
 <!-- YAML

From d9d8e8684779eec45d7a84cfebca2176e0286d74 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Sat, 10 Oct 2020 12:59:43 +0200
Subject: [PATCH 3/5] fixup

---
 lib/internal/streams/readable.js            | 13 +++++----
 lib/internal/streams/writable.js            |  2 +-
 test/parallel/test-stream-pipe-needDrain.js | 29 +++++++++++++++++++++
 3 files changed, 36 insertions(+), 8 deletions(-)
 create mode 100644 test/parallel/test-stream-pipe-needDrain.js

diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index c668bdef757629..3099bbd6ccc161 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -753,12 +753,6 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
     }
   }
 
-  src.on('data', ondata);
-
-  if (dest.writableNeedDrain === true) {
-    pause();
-  }
-
   function ondata(chunk) {
     debug('ondata');
     const ret = dest.write(chunk);
@@ -810,7 +804,12 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
   dest.emit('pipe', src);
 
   // Start the flow if it hasn't been started already.
-  if (!state.flowing) {
+
+  if (dest.writableNeedDrain === true) {
+    if (state.flowing) {
+      src.pause();
+    }
+  } else if (!state.flowing) {
     debug('pipe resume');
     src.resume();
   }
diff --git a/lib/internal/streams/writable.js b/lib/internal/streams/writable.js
index cb36e322e7bb62..3baa9f2c9e6fa8 100644
--- a/lib/internal/streams/writable.js
+++ b/lib/internal/streams/writable.js
@@ -809,7 +809,7 @@ ObjectDefineProperties(Writable.prototype, {
     get() {
       const wState = this._writableState;
       if (!wState) return false;
-      return !this._writable.destroyed && !wState.ending && wState.needDrain;
+      return !wState.destroyed && !wState.ending && wState.needDrain;
     }
   },
 
diff --git a/test/parallel/test-stream-pipe-needDrain.js b/test/parallel/test-stream-pipe-needDrain.js
new file mode 100644
index 00000000000000..7e8f5f6f47a70a
--- /dev/null
+++ b/test/parallel/test-stream-pipe-needDrain.js
@@ -0,0 +1,29 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const Readable = require('_stream_readable');
+const Writable = require('_stream_writable');
+
+// Pipe should not continue writing if writable needs drain.
+{
+  const w = new Writable({
+    write(buf, encoding, callback) {
+
+    }
+  });
+
+  while (w.write('asd'));
+
+  assert.strictEqual(w.writableNeedDrain, true);
+
+  const r = new Readable({
+    read() {
+      this.push('asd');
+    }
+  });
+
+  w.write = common.mustNotCall();
+
+  r.pipe(w);
+}

From dfb77b495c1c5c26f17452e17556fbd1615d54ca Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Sat, 10 Oct 2020 14:25:09 +0200
Subject: [PATCH 4/5] fixup

---
 lib/internal/streams/readable.js | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index 3099bbd6ccc161..7da0c927d79c6f 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -753,6 +753,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
     }
   }
 
+  src.on('data', ondata);
   function ondata(chunk) {
     debug('ondata');
     const ret = dest.write(chunk);
@@ -807,7 +808,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
 
   if (dest.writableNeedDrain === true) {
     if (state.flowing) {
-      src.pause();
+      pause();
     }
   } else if (!state.flowing) {
     debug('pipe resume');

From 3dd02b3c3c675d86c6be17b6fdb8e8994a1d65ed Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Sat, 10 Oct 2020 14:33:20 +0200
Subject: [PATCH 5/5] fixup

---
 lib/internal/streams/readable.js | 52 +++++++++++++++-----------------
 1 file changed, 24 insertions(+), 28 deletions(-)

diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index 7da0c927d79c6f..c223f3017e5a94 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -727,39 +727,35 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
       ondrain();
   }
 
-  function pause() {
-    // If the user unpiped during `dest.write()`, it is possible
-    // to get stuck in a permanently paused state if that write
-    // also returned false.
-    // => Check whether `dest` is still a piping destination.
-    if (!cleanedUp) {
-      if (state.pipes.length === 1 && state.pipes[0] === dest) {
-        debug('false write response, pause', 0);
-        state.awaitDrainWriters = dest;
-        state.multiAwaitDrain = false;
-      } else if (state.pipes.length > 1 && state.pipes.includes(dest)) {
-        debug('false write response, pause', state.awaitDrainWriters.size);
-        state.awaitDrainWriters.add(dest);
-      }
-      src.pause();
-    }
-    if (!ondrain) {
-      // When the dest drains, it reduces the awaitDrain counter
-      // on the source.  This would be more elegant with a .once()
-      // handler in flow(), but adding and removing repeatedly is
-      // too slow.
-      ondrain = pipeOnDrain(src, dest);
-      dest.on('drain', ondrain);
-    }
-  }
-
   src.on('data', ondata);
   function ondata(chunk) {
     debug('ondata');
     const ret = dest.write(chunk);
     debug('dest.write', ret);
     if (ret === false) {
-      pause();
+      // If the user unpiped during `dest.write()`, it is possible
+      // to get stuck in a permanently paused state if that write
+      // also returned false.
+      // => Check whether `dest` is still a piping destination.
+      if (!cleanedUp) {
+        if (state.pipes.length === 1 && state.pipes[0] === dest) {
+          debug('false write response, pause', 0);
+          state.awaitDrainWriters = dest;
+          state.multiAwaitDrain = false;
+        } else if (state.pipes.length > 1 && state.pipes.includes(dest)) {
+          debug('false write response, pause', state.awaitDrainWriters.size);
+          state.awaitDrainWriters.add(dest);
+        }
+        src.pause();
+      }
+      if (!ondrain) {
+        // When the dest drains, it reduces the awaitDrain counter
+        // on the source.  This would be more elegant with a .once()
+        // handler in flow(), but adding and removing repeatedly is
+        // too slow.
+        ondrain = pipeOnDrain(src, dest);
+        dest.on('drain', ondrain);
+      }
     }
   }
 
@@ -808,7 +804,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
 
   if (dest.writableNeedDrain === true) {
     if (state.flowing) {
-      pause();
+      src.pause();
     }
   } else if (!state.flowing) {
     debug('pipe resume');