From 17a41c60698e0ccdf0146e96cd3def3e6408dbe4 Mon Sep 17 00:00:00 2001
From: David Halls <dahalls@gmail.com>
Date: Tue, 14 Jun 2016 08:47:04 +0100
Subject: [PATCH] stream: ensure awaitDrain is increased once

Guard against the call to write() inside pipe's ondata pushing more data
back onto the Readable, thus causing ondata to be called again.

This is fine but results in awaitDrain being increased more than once.
The problem with that is when the destination does drain, only a single
'drain' event is emitted, so awaitDrain in this case will never reach
zero and we end up with a permanently paused stream.

Fixes: https://github.com/nodejs/node/issues/7278
PR-URL: https://github.com/nodejs/node/pull/7292
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
---
 lib/_stream_readable.js                       |  9 +++++-
 ...tream-pipe-await-drain-push-while-write.js | 28 +++++++++++++++++++
 2 files changed, 36 insertions(+), 1 deletion(-)
 create mode 100644 test/parallel/test-stream-pipe-await-drain-push-while-write.js

diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index 6fe9d198533cb9..c9beafa9bee8e7 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -530,11 +530,17 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
       ondrain();
   }
 
+  // If the user pushes more data while we're writing to dest then we'll end up
+  // in ondata again. However, we only want to increase awaitDrain once because
+  // dest will only emit one 'drain' event for the multiple writes.
+  // => Introduce a guard on increasing awaitDrain.
+  var increasedAwaitDrain = false;
   src.on('data', ondata);
   function ondata(chunk) {
     debug('ondata');
+    increasedAwaitDrain = false;
     var ret = dest.write(chunk);
-    if (false === ret) {
+    if (false === ret && !increasedAwaitDrain) {
       // If the user unpiped during `dest.write()`, it is possible
       // to get stuck in a permanently paused state if that write
       // also returned false.
@@ -544,6 +550,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
           !cleanedUp) {
         debug('false write response, pause', src._readableState.awaitDrain);
         src._readableState.awaitDrain++;
+        increasedAwaitDrain = true;
       }
       src.pause();
     }
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
new file mode 100644
index 00000000000000..1dfdfdb80c8d71
--- /dev/null
+++ b/test/parallel/test-stream-pipe-await-drain-push-while-write.js
@@ -0,0 +1,28 @@
+'use strict';
+const common = require('../common');
+const stream = require('stream');
+
+// A writable stream which pushes data onto the stream which pipes into it,
+// but only the first time it's written to. Since it's not paused at this time,
+// a second write will occur. If the pipe increases awaitDrain twice, we'll
+// never get subsequent chunks because 'drain' is only emitted once.
+const writable = new stream.Writable({
+  write: common.mustCall((chunk, encoding, cb) => {
+    if (chunk.length === 32 * 1024) { // first chunk
+      readable.push(new Buffer(33 * 1024)); // above hwm
+    }
+    cb();
+  }, 3)
+});
+
+// A readable stream which produces two buffers.
+const bufs = [new Buffer(32 * 1024), new Buffer(33 * 1024)]; // above hwm
+const readable = new stream.Readable({
+  read: function() {
+    while (bufs.length > 0) {
+      this.push(bufs.shift());
+    }
+  }
+});
+
+readable.pipe(writable);