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

@@ -544,11 +544,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.
@@ -558,6 +564,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
           !cleanedUp) {
         debug('false write response, pause', src._readableState.awaitDrain);
         src._readableState.awaitDrain++;
+        increasedAwaitDrain = true;
       }
       src.pause();
     }

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);