stream: do not emit end on readable error

PR-URL: https://github.com/nodejs/node/pull/39607
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

lib/internal/streams/readable.js

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

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

@@ -163,7 +163,7 @@ const { getEventListeners } = require('events');
 
     client.close();
     req.resume();
-    req.on('end', common.mustCall());
+    req.on('end', common.mustNotCall());
     req.on('close', common.mustCall(() => server.close()));
   }));
 }

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

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

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

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

test/parallel/test-http2-empty-frame-without-eof.js

@@ -34,7 +34,9 @@ async function main() {
       client.on('error', common.expectsError(expected));
     }
     stream.resume();
-    await once(stream, 'end');
+    await new Promise((resolve) => {
+      stream.once('close', resolve);
+    });
     client.close();
   }
   server.close();

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

@@ -45,7 +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('end', common.mustNotCall());
     req.on('close', common.mustCall(() => countdown.dec()));
     req.on('error', common.expectsError({
       code: 'ERR_HTTP2_STREAM_ERROR',

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

@@ -54,7 +54,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('end', common.mustNotCall());
     req.on('close', common.mustCall(() => countdown.dec()));
     req.on('error', common.expectsError({
       code: 'ERR_HTTP2_STREAM_ERROR',

test/parallel/test-http2-respond-file-fd-invalid.js

@@ -40,7 +40,8 @@ server.listen(0, () => {
   req.on('response', common.mustCall());
   req.on('error', errorCheck);
   req.on('data', common.mustNotCall());
-  req.on('end', common.mustCall(() => {
+  req.on('end', common.mustNotCall());
+  req.on('close', common.mustCall(() => {
     assert.strictEqual(req.rstCode, NGHTTP2_INTERNAL_ERROR);
     client.close();
     server.close();

test/parallel/test-http2-respond-nghttperrors.js

@@ -83,12 +83,13 @@ function runTest(test) {
     name: 'Error',
     message: 'Stream closed with error code NGHTTP2_INTERNAL_ERROR'
   }));
+  req.on('end', common.mustNotCall());
 
   currentError = test;
   req.resume();
   req.end();
 
-  req.on('end', common.mustCall(() => {
+  req.on('close', common.mustCall(() => {
     client.close();
 
     if (!tests.length) {

test/parallel/test-http2-respond-with-fd-errors.js

@@ -91,12 +91,13 @@ function runTest(test) {
     name: 'Error',
     message: 'Stream closed with error code NGHTTP2_INTERNAL_ERROR'
   }));
+  req.on('end', common.mustNotCall());
 
   currentError = test;
   req.resume();
   req.end();
 
-  req.on('end', common.mustCall(() => {
+  req.on('close', common.mustCall(() => {
     client.close();
 
     if (!tests.length) {

test/parallel/test-http2-server-shutdown-before-respond.js

@@ -32,5 +32,6 @@ server.on('listening', common.mustCall(() => {
   }));
   req.resume();
   req.on('data', common.mustNotCall());
-  req.on('end', common.mustCall(() => server.close()));
+  req.on('end', common.mustNotCall());
+  req.on('close', common.mustCall(() => server.close()));
 }));

test/parallel/test-http2-server-socket-destroy.js

@@ -8,6 +8,7 @@ if (!common.hasCrypto)
 const assert = require('assert');
 const h2 = require('http2');
 const { kSocket } = require('internal/http2/util');
+const { once } = require('events');
 
 const server = h2.createServer();
 
@@ -36,7 +37,7 @@ function onStream(stream) {
 
 server.listen(0);
 
-server.on('listening', common.mustCall(() => {
+server.on('listening', common.mustCall(async () => {
   const client = h2.connect(`http://localhost:${server.address().port}`);
   // The client may have an ECONNRESET error here depending on the operating
   // system, due mainly to differences in the timing of socket closing. Do
@@ -58,5 +59,8 @@ server.on('listening', common.mustCall(() => {
 
   req.on('aborted', common.mustCall());
   req.resume();
-  req.on('end', common.mustCall());
+
+  try {
+    await once(req, 'end');
+  } catch {}
 }));