Revert "stream: prevent 'end' to be emitted after 'error'"

This reverts commit 0857790. PR-URL: #20449 Fixes: #20334 Reviewed-By: Luigi Pinca <luigipinca@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Anna Henningsen <anna@addaleax.net>

Author: mscdex

lib/_stream_readable.js

@@ -99,9 +99,6 @@ function ReadableState(options, stream, isDuplex) {
  this.endEmitted = false;
  this.reading = false;
 
- // Flipped if an 'error' is emitted.
- this.errorEmitted = false;
-
  // a flag to be able to tell if the event 'readable'/'data' is emitted
  // immediately, or on a later tick. We set this to true at first, because
  // any actions that shouldn't happen until "later" should generally also
@@ -1072,23 +1069,20 @@ function fromList(n, state) {
 function endReadable(stream) {
  var state = stream._readableState;
 
- debug('endReadable', state.endEmitted, state.errorEmitted);
- if (!state.endEmitted && !state.errorEmitted) {
+ debug('endReadable', state.endEmitted);
+ if (!state.endEmitted) {
  state.ended = true;
  process.nextTick(endReadableNT, state, stream);
  }
 }
 
 function endReadableNT(state, stream) {
- debug('endReadableNT', state.endEmitted, state.length, state.errorEmitted);
+ debug('endReadableNT', state.endEmitted, state.length);
 
  // Check that we didn't get one last unshift.
  if (!state.endEmitted && state.length === 0) {
+ state.endEmitted = true;
  stream.readable = false;
-
- if (!state.errorEmitted) {
- state.endEmitted = true;
- stream.emit('end');
- }
+ stream.emit('end');
  }
 }

lib/_stream_writable.js

@@ -424,22 +424,12 @@ function onwriteError(stream, state, sync, er, cb) {
  // this can emit finish, and it will always happen
  // after error
  process.nextTick(finishMaybe, stream, state);
-
- // needed for duplex, fixes https://github.com/nodejs/node/issues/6083
- if (stream._readableState) {
- stream._readableState.errorEmitted = true;
- }
  stream._writableState.errorEmitted = true;
  stream.emit('error', er);
  } else {
  // the caller expect this to happen before if
  // it is async
  cb(er);
-
- // needed for duplex, fixes https://github.com/nodejs/node/issues/6083
- if (stream._readableState) {
- stream._readableState.errorEmitted = true;
- }
  stream._writableState.errorEmitted = true;
  stream.emit('error', er);
  // this can emit finish, but finish must

lib/internal/streams/destroy.js

@@ -8,14 +8,10 @@ function destroy(err, cb) {
  this._writableState.destroyed;
 
  if (readableDestroyed || writableDestroyed) {
- const readableErrored = this._readableState &&
- this._readableState.errorEmitted;
- const writableErrored = this._writableState &&
- this._writableState.errorEmitted;
-
  if (cb) {
  cb(err);
- } else if (err && !readableErrored && !writableErrored) {
+ } else if (err &&
+ (!this._writableState || !this._writableState.errorEmitted)) {
  process.nextTick(emitErrorNT, this, err);
  }
  return this;
@@ -36,11 +32,6 @@ function destroy(err, cb) {
  this._destroy(err || null, (err) => {
  if (!cb && err) {
  process.nextTick(emitErrorAndCloseNT, this, err);
-
- if (this._readableState) {
- this._readableState.errorEmitted = true;
- }
-
  if (this._writableState) {
  this._writableState.errorEmitted = true;
  }
@@ -74,7 +65,6 @@ function undestroy() {
  this._readableState.reading = false;
  this._readableState.ended = false;
  this._readableState.endEmitted = false;
- this._readableState.errorEmitted = false;
  }
 
  if (this._writableState) {

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

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

test/parallel/test-http2-client-onconnect-errors.js

@@ -101,6 +101,7 @@ function runTest(test) {
  });
  }
 
+ req.on('end', common.mustCall());
  req.on('close', common.mustCall(() => {
  client.destroy();
 

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

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

test/parallel/test-http2-compat-serverresponse-destroy.js

@@ -63,6 +63,7 @@ server.listen(0, common.mustCall(() => {
  req.on('close', common.mustCall(() => countdown.dec()));
 
  req.resume();
+ req.on('end', common.mustCall());
  }
 
  {
@@ -77,5 +78,6 @@ server.listen(0, common.mustCall(() => {
  req.on('close', common.mustCall(() => countdown.dec()));
 
  req.resume();
+ req.on('end', common.mustCall());
  }
 }));

test/parallel/test-http2-max-concurrent-streams.js

@@ -45,6 +45,7 @@ server.listen(0, common.mustCall(() => {
  req.on('aborted', common.mustCall());
  req.on('response', common.mustNotCall());
  req.resume();
+ req.on('end', common.mustCall());
  req.on('close', common.mustCall(() => countdown.dec()));
  req.on('error', common.expectsError({
  code: 'ERR_HTTP2_STREAM_ERROR',

test/parallel/test-http2-misused-pseudoheaders.js

@@ -41,6 +41,7 @@ server.listen(0, common.mustCall(() => {
 
  req.on('response', common.mustCall());
  req.resume();
+ req.on('end', common.mustCall());
  req.on('close', common.mustCall(() => {
  server.close();
  client.close();

test/parallel/test-http2-multi-content-length.js

@@ -53,6 +53,7 @@ server.listen(0, common.mustCall(() => {
  // header to be set for non-payload bearing requests...
  const req = client.request({ 'content-length': 1 });
  req.resume();
+ req.on('end', common.mustCall());
  req.on('close', common.mustCall(() => countdown.dec()));
  req.on('error', common.expectsError({
  code: 'ERR_HTTP2_STREAM_ERROR',