lib: ensure readable stream flows to end

If a readable stream was set up with `highWaterMark 0`, the while-loop
in `maybeReadMore_` function would never execute.

The while loop now has an extra or-condition for the case where the
stream is flowing and there are no items. The or-condition is adapted
from the emit-condition of the `addChunk` function.

The `addChunk` also contains a check for `state.sync`. However that part
of the check was omitted here because the `maybeReadMore_` is executed
using `process.nextTick`. `state.sync` is set and then unset  within the
`read()` function so it should never be in effect in `maybeReadMore_`.

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

PR-URL: https://github.com/nodejs/node/pull/24918
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

1 files changed, 26 insertions, 4 deletions

diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index 24f58bb18a..fbab236203 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -568,16 +568,38 @@ function maybeReadMore(stream, state) {
 }
 
 function maybeReadMore_(stream, state) {
-  var len = state.length;
+  // Attempt to read more data if we should.
+  //
+  // The conditions for reading more data are (one of):
+  // - Not enough data buffered (state.length < state.highWaterMark). The loop
+  //   is responsible for filling the buffer with enough data if such data
+  //   is available. If highWaterMark is 0 and we are not in the flowing mode
+  //   we should _not_ attempt to buffer any extra data. We'll get more data
+  //   when the stream consumer calls read() instead.
+  // - No data in the buffer, and the stream is in flowing mode. In this mode
+  //   the loop below is responsible for ensuring read() is called. Failing to
+  //   call read here would abort the flow and there's no other mechanism for
+  //   continuing the flow if the stream consumer has just subscribed to the
+  //   'data' event.
+  //
+  // In addition to the above conditions to keep reading data, the following
+  // conditions prevent the data from being read:
+  // - The stream has ended (state.ended).
+  // - There is already a pending 'read' operation (state.reading). This is a
+  //   case where the the stream has called the implementation defined _read()
+  //   method, but they are processing the call asynchronously and have _not_
+  //   called push() with new data. In this case we skip performing more
+  //   read()s. The execution ends in this method again after the _read() ends
+  //   up calling push() with more data.
   while (!state.reading && !state.ended &&
-         state.length < state.highWaterMark) {
+         (state.length < state.highWaterMark ||
+          (state.flowing && state.length === 0))) {
+    const len = state.length;
     debug('maybeReadMore read 0');
     stream.read(0);
     if (len === state.length)
       // didn't get any data, stop spinning.
       break;
-    else
-      len = state.length;
   }
   state.readingMore = false;
 }