stream: avoid duplicated endReadableNT scheduling

Calling read() on an ended stream multiple times before the microtask
queue drains scheduled one endReadableNT tick per call, as the only
guard was endEmitted, which is set inside the tick itself. A hello-world
HTTP server was scheduling it four times per request while dumping the
unread request body.

Introduce a kEndScheduled flag armed when the tick is scheduled and
cleared when it runs. Clearing it unconditionally matters for reused
sockets: undestroy() resets endEmitted through the state descriptors
but cannot reach this flag, and a stale value would block the 'end'
event after a net.Socket reconnect.

Signed-off-by: Matteo Collina <hello@matteocollina.com>

Author: mcollina

lib/internal/streams/readable.js

@@ -131,6 +131,7 @@ const kFlowing = 1 << 24;
 const kHasPaused = 1 << 25;
 const kPaused = 1 << 26;
 const kDataListening = 1 << 27;
+const kEndScheduled = 1 << 28;
 
 // TODO(benjamingr) it is likely slower to do it this way than with free functions
 function makeBitMapDescriptor(bit) {
@@ -1757,15 +1758,21 @@ function endReadable(stream) {
   const state = stream._readableState;
 
   debug('endReadable');
-  if ((state[kState] & kEndEmitted) === 0) {
-    state[kState] |= kEnded;
+  if ((state[kState] & (kEndEmitted | kEndScheduled)) === 0) {
+    state[kState] |= kEnded | kEndScheduled;
     process.nextTick(endReadableNT, state, stream);
   }
 }
 
 function endReadableNT(state, stream) {
   debug('endReadableNT');
 
+  // The scheduled tick is running; allow endReadable() to schedule again.
+  // This matters both when the 'end' emission is skipped below (e.g. after
+  // an unshift()) and when the stream is later reset for reuse
+  // (see undestroy()), which clears kEndEmitted but not this flag.
+  state[kState] &= ~kEndScheduled;
+
   // Check that we didn't get one last unshift.
   if ((state[kState] & (kErrored | kCloseEmitted | kEndEmitted)) === 0 && state.length === 0) {
     state[kState] |= kEndEmitted;