stream: don't emit end after close

Readable stream could emit 'end' after 'close'.

PR-URL: #33076
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>

lib/_stream_readable.js

@@ -150,6 +150,10 @@ function ReadableState(options, stream, isDuplex) {
   // Indicates whether the stream has finished destroying.
   this.closed = false;
 
+  // True if close has been emitted or would have been emitted
+  // depending on emitClose.
+  this.closeEmitted = false;
+
   // Crypto is kind of old and crusty.  Historically, its default string
   // encoding is 'binary' so we have to make this configurable.
   // Everything else in the universe uses 'utf8', though.
@@ -1213,7 +1217,8 @@ function endReadableNT(state, stream) {
   debug('endReadableNT', state.endEmitted, state.length);
 
   // Check that we didn't get one last unshift.
-  if (!state.errorEmitted && !state.endEmitted && state.length === 0) {
+  if (!state.errorEmitted && !state.closeEmitted &&
+      !state.endEmitted && state.length === 0) {
     state.endEmitted = true;
     stream.emit('end');
 

lib/internal/streams/destroy.js

@@ -76,6 +76,9 @@ function emitCloseNT(self) {
   if (w) {
     w.closeEmitted = true;
   }
+  if (r) {
+    r.closeEmitted = true;
+  }
 
   if ((w && w.emitClose) || (r && r.emitClose)) {
     self.emit('close');
@@ -106,12 +109,13 @@ function undestroy() {
 
   if (r) {
     r.closed = false;
+    r.closeEmitted = false;
     r.destroyed = false;
     r.errored = false;
+    r.errorEmitted = false;
     r.reading = false;
     r.ended = false;
     r.endEmitted = false;
-    r.errorEmitted = false;
   }
 
   if (w) {

test/parallel/test-stream-duplex-destroy.js

@@ -124,7 +124,7 @@ const assert = require('assert');
 
   duplex.removeListener('end', fail);
   duplex.removeListener('finish', fail);
-  duplex.on('end', common.mustCall());
+  duplex.on('end', common.mustNotCall());
   duplex.on('finish', common.mustCall());
   assert.strictEqual(duplex.destroyed, true);
 }

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

@@ -113,7 +113,7 @@ const assert = require('assert');
   read.destroy();
 
   read.removeListener('end', fail);
-  read.on('end', common.mustCall());
+  read.on('end', common.mustNotCall());
   assert.strictEqual(read.destroyed, true);
 }
 

test/parallel/test-stream-readable-end-destroyed.js

@@ -0,0 +1,17 @@
+'use strict';
+
+const common = require('../common');
+const { Readable } = require('stream');
+
+{
+  // Don't emit 'end' after 'close'.
+
+  const r = new Readable();
+
+  r.on('end', common.mustNotCall());
+  r.resume();
+  r.destroy();
+  r.on('close', common.mustCall(() => {
+    r.push(null);
+  }));
+}