stream: don't emit 'end' after 'error'

lib/_stream_readable.js

@@ -1214,7 +1214,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

@@ -16,6 +16,8 @@ function destroy(err, cb) {
     }
   }
 
+  // TODO(ronag): readable & writable = false?
+
   if ((w && w.destroyed) || (r && r.destroyed)) {
     if (cb) {
       cb(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,26 @@
+'use strict';
+
+const common = require('../common');
+const { Readable } = require('stream');
+const assert = require('assert');
+
+{
+  // Ensure 'end' is not emitted after 'error'.
+  // This test is slightly more complicated than
+  // needed in order to better illustrate the invariant.
+
+  const r = new Readable({ read() {} });
+
+  let errorEmitted = false;
+
+  r.on('data', common.mustCall());
+  r.on('end', () => {
+    assert.strictEqual(!errorEmitted);
+  });
+  r.on('error', () => {
+    errorEmitted = true;
+  });
+  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;