stream: fix regression introduced in #26059

In #26059, we introduced a bug that caused 'readable' to be nextTicked
on EOF of a ReadableStream. This breaks the dicer module on CITGM.
That change was partially reverted to still fix the bug in #25810 and
not break dicer.

See: #26059
Fixes: #25810

PR-URL: #26643
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: mcollina

lib/_stream_readable.js

@@ -510,12 +510,24 @@ function onEofChunk(stream, state) {
     }
   }
   state.ended = true;
-  state.needReadable = false;
 
-  // We are not protecting if emittedReadable = true,
-  // so 'readable' gets scheduled anyway.
-  state.emittedReadable = true;
-  process.nextTick(emitReadable_, stream);
+  if (state.sync) {
+    // If we are sync, wait until next tick to emit the data.
+    // Otherwise we risk emitting data in the flow()
+    // the readable code triggers during a read() call
+    emitReadable(stream);
+  } else {
+    // Emit 'readable' now to make sure it gets picked up.
+    state.needReadable = false;
+    state.emittedReadable = true;
+    // We have to emit readable now that we are EOF. Modules
+    // in the ecosystem (e.g. dicer) rely on this event being sync.
+    if (state.ended) {
+      emitReadable_(stream);
+    } else {
+      process.nextTick(emitReadable_, stream);
+    }
+  }
 }
 
 // Don't emit readable right away in sync mode, because this can trigger

test/parallel/test-stream-readable-emit-readable-short-stream.js

@@ -54,7 +54,7 @@ const assert = require('assert');
         break;
       assert.strictEqual(chunk.toString(), 'content');
     }
-  }, 2));
+  }));
 }
 
 {
@@ -78,7 +78,7 @@ const assert = require('assert');
         break;
       assert.strictEqual(chunk.toString(), 'content');
     }
-  }, 2));
+  }));
 }
 
 {
@@ -94,7 +94,7 @@ const assert = require('assert');
         break;
       assert.strictEqual(chunk.toString(), 'content');
     }
-  }, 2));
+  }));
 
   t.push('content');
   t.push(null);

test/parallel/test-stream-readable-emittedReadable.js

@@ -43,23 +43,12 @@ const noRead = new Readable({
   read: () => {}
 });
 
-noRead.once('readable', common.mustCall(() => {
+noRead.on('readable', common.mustCall(() => {
   // emittedReadable should be true when the readable event is emitted
   assert.strictEqual(noRead._readableState.emittedReadable, true);
   noRead.read(0);
   // emittedReadable is not reset during read(0)
   assert.strictEqual(noRead._readableState.emittedReadable, true);
-
-  noRead.on('readable', common.mustCall(() => {
-    // The second 'readable' is emitted because we are ending
-
-    // emittedReadable should be true when the readable event is emitted
-    assert.strictEqual(noRead._readableState.emittedReadable, false);
-    noRead.read(0);
-    // emittedReadable is not reset during read(0)
-    assert.strictEqual(noRead._readableState.emittedReadable, false);
-
-  }));
 }));
 
 noRead.push('foo');

test/parallel/test-stream-readable-needReadable.js

@@ -14,7 +14,7 @@ readable.on('readable', common.mustCall(() => {
   // When the readable event fires, needReadable is reset.
   assert.strictEqual(readable._readableState.needReadable, false);
   readable.read();
-}, 2));
+}));
 
 // If a readable listener is attached, then a readable event is needed.
 assert.strictEqual(readable._readableState.needReadable, true);

test/parallel/test-stream-readable-reading-readingMore.js

@@ -31,7 +31,7 @@ const Readable = require('stream').Readable;
     assert.strictEqual(state.reading, false);
   }
 
-  const expectedReadingMore = [true, false, false];
+  const expectedReadingMore = [true, true, false];
   readable.on('readable', common.mustCall(() => {
     // There is only one readingMore scheduled from on('data'),
     // after which everything is governed by the .read() call

test/parallel/test-stream2-httpclient-response-end.js

@@ -15,7 +15,7 @@ const server = http.createServer(function(req, res) {
       while ((chunk = res.read()) !== null) {
         data += chunk;
       }
-    }, 2));
+    }));
     res.on('end', common.mustCall(function() {
       console.log('end event');
       assert.strictEqual(msg, data);

test/parallel/test-stream2-transform.js

@@ -321,16 +321,10 @@ const Transform = require('_stream_transform');
 
   pt.end();
 
-  // The next readable is emitted on the next tick.
-  assert.strictEqual(emits, 0);
-
-  process.on('nextTick', function() {
-    assert.strictEqual(emits, 1);
-    assert.strictEqual(pt.read(5).toString(), 'l');
-    assert.strictEqual(pt.read(5), null);
-
-    assert.strictEqual(emits, 1);
-  });
+  assert.strictEqual(emits, 1);
+  assert.strictEqual(pt.read(5).toString(), 'l');
+  assert.strictEqual(pt.read(5), null);
+  assert.strictEqual(emits, 1);
 }
 
 {