stream: do not emit 'end' after 'error'

Refs: #6083

PR-URL: #31182
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

Author: ronag

lib/_stream_readable.js

@@ -1215,7 +1215,7 @@ function endReadableNT(state, stream) {
   debug('endReadableNT', state.endEmitted, state.length);
 
   // Check that we didn't get one last unshift.
-  if (!state.endEmitted && state.length === 0) {
+  if (!state.errorEmitted && !state.endEmitted && state.length === 0) {
     state.endEmitted = true;
     stream.readable = false;
     stream.emit('end');

lib/internal/streams/destroy.js

@@ -7,6 +7,8 @@ function destroy(err, cb) {
   const r = this._readableState;
   const w = this._writableState;
 
+  // TODO(ronag): readable & writable = false?
+
   if (err) {
     if (w) {
       w.errored = true;
@@ -129,6 +131,8 @@ function errorOrDestroy(stream, err, sync) {
   const r = stream._readableState;
   const w = stream._writableState;
 
+  // TODO(ronag): readable & writable = false?
+
   if ((r && r.autoDestroy) || (w && w.autoDestroy))
     stream.destroy(err);
   else if (err) {

test/parallel/test-http2-client-destroy.js

@@ -95,7 +95,7 @@ const Countdown = require('../common/countdown');
     });
 
     req.resume();
-    req.on('end', common.mustCall());
+    req.on('end', common.mustNotCall());
     req.on('close', common.mustCall(() => server.close()));
   }));
 }

test/parallel/test-http2-client-stream-destroy-before-connect.js

@@ -50,5 +50,5 @@ server.listen(0, common.mustCall(() => {
 
   req.on('response', common.mustNotCall());
   req.resume();
-  req.on('end', common.mustCall());
+  req.on('end', common.mustNotCall());
 }));

test/parallel/test-http2-head-request.js

@@ -10,7 +10,7 @@ const errCheck = common.expectsError({
   name: 'Error',
   code: 'ERR_STREAM_WRITE_AFTER_END',
   message: 'write after end'
-}, 2);
+}, 1);
 
 const {
   HTTP2_HEADER_PATH,
@@ -41,12 +41,6 @@ server.listen(0, () => {
     [HTTP2_HEADER_PATH]: '/'
   });
 
-  // Because it is a HEAD request, the payload is meaningless. The
-  // option.endStream flag is set automatically making the stream
-  // non-writable.
-  req.on('error', errCheck);
-  req.write('data');
-
   req.on('response', common.mustCall((headers, flags) => {
     assert.strictEqual(headers[HTTP2_HEADER_STATUS], 200);
     assert.strictEqual(flags, 5); // The end of stream flag is set

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

@@ -0,0 +1,15 @@
+'use strict';
+
+const common = require('../common');
+const { Readable } = require('stream');
+
+{
+  const r = new Readable({ read() {} });
+
+  r.on('end', common.mustNotCall());
+  r.on('data', common.mustCall());
+  r.on('error', common.mustCall());
+  r.push('asd');
+  r.push(null);
+  r.destroy(new Error('kaboom'));
+}

test/parallel/test-stream-unshift-read-race.js

@@ -68,6 +68,9 @@ r._read = function(n) {
 };
 
 function pushError() {
+  r.unshift(Buffer.allocUnsafe(1));
+  w.end();
+
   assert.throws(() => {
     r.push(Buffer.allocUnsafe(1));
   }, {
@@ -85,10 +88,7 @@ w._write = function(chunk, encoding, cb) {
   cb();
 };
 
-r.on('end', common.mustCall(function() {
-  r.unshift(Buffer.allocUnsafe(1));
-  w.end();
-}));
+r.on('end', common.mustNotCall());
 
 r.on('readable', function() {
   let chunk;