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

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

Fixes: https://github.com/nodejs/node/issues/35341

PR-URL: https://github.com/nodejs/node/pull/35348
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
---
 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            |  7 ++++-
 lib/internal/streams/writable.js            |  8 ++++++
 test/parallel/test-stream-pipe-needDrain.js | 29 +++++++++++++++++++++
 7 files changed, 63 insertions(+), 1 deletion(-)
 create mode 100644 test/parallel/test-stream-pipe-needDrain.js

diff --git a/doc/api/stream.md b/doc/api/stream.md
index e73624739a6b62..22a2deee83fba9 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -580,6 +580,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 the stream's buffer has been full and stream will emit `'drain'`.
+
 ##### `writable.writableObjectMode`
 <!-- YAML
 added: v12.3.0
diff --git a/lib/_http_outgoing.js b/lib/_http_outgoing.js
index 48d113bb5cf232..14219857e5f899 100644
--- a/lib/_http_outgoing.js
+++ b/lib/_http_outgoing.js
@@ -660,6 +660,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 a73e17f1f29eed..5bd655aa1e7a9e 100644
--- a/lib/internal/streams/pipeline.js
+++ b/lib/internal/streams/pipeline.js
@@ -123,6 +123,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 43b53447d8734e..93153908fef4e1 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -783,7 +783,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 a3fc3dba50a93f..104a3c43452931 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 !wState.destroyed && !wState.ending && wState.needDrain;
+    }
+  },
+
   writableHighWaterMark: {
     get() {
       return this._writableState && this._writableState.highWaterMark;
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);
+}