http2: fix flakiness in timeout

PR-URL: https://github.com/nodejs/node/pull/14239
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

lib/internal/http2/core.js

@@ -673,15 +673,23 @@ function submitShutdown(options) {
 }
 
 function finishSessionDestroy(socket) {
+  const state = this[kState];
+  if (state.destroyed)
+    return;
+
   if (!socket.destroyed)
     socket.destroy();
 
+  state.destroying = false;
+  state.destroyed = true;
+
   // Destroy the handle
   const handle = this[kHandle];
   if (handle !== undefined) {
-    handle.destroy();
+    handle.destroy(state.skipUnconsume);
     debug(`[${sessionName(this[kType])}] nghttp2session handle destroyed`);
   }
+  this[kHandle] = undefined;
 
   process.nextTick(emit.bind(this, 'close'));
   debug(`[${sessionName(this[kType])}] nghttp2session destroyed`);
@@ -825,7 +833,7 @@ class Http2Session extends EventEmitter {
 
   // Submits a SETTINGS frame to be sent to the remote peer.
   settings(settings) {
-    if (this[kState].destroyed)
+    if (this[kState].destroyed || this[kState].destroying)
       throw new errors.Error('ERR_HTTP2_INVALID_SESSION');
 
     // Validate the input first
@@ -871,7 +879,7 @@ class Http2Session extends EventEmitter {
 
   // Submits a PRIORITY frame to be sent to the remote peer.
   priority(stream, options) {
-    if (this[kState].destroyed)
+    if (this[kState].destroyed || this[kState].destroying)
       throw new errors.Error('ERR_HTTP2_INVALID_SESSION');
 
     if (!(stream instanceof Http2Stream)) {
@@ -905,6 +913,8 @@ class Http2Session extends EventEmitter {
   // Submits an RST-STREAM frame to be sent to the remote peer. This will
   // cause the stream to be closed.
   rstStream(stream, code = NGHTTP2_NO_ERROR) {
+    // Do not check destroying here, as the rstStream may be sent while
+    // the session is in the middle of being destroyed.
     if (this[kState].destroyed)
       throw new errors.Error('ERR_HTTP2_INVALID_SESSION');
 
@@ -946,7 +956,6 @@ class Http2Session extends EventEmitter {
     const state = this[kState];
     if (state.destroyed || state.destroying)
       return;
-
     debug(`[${sessionName(this[kType])}] destroying nghttp2session`);
     state.destroying = true;
 
@@ -963,8 +972,8 @@ class Http2Session extends EventEmitter {
     delete this[kSocket];
     delete this[kServer];
 
-    state.destroyed = true;
+    state.destroyed = false;
-    state.destroying = false;
+    state.destroying = true;
 
     if (this[kHandle] !== undefined)
       this[kHandle].destroying();
@@ -975,7 +984,7 @@ class Http2Session extends EventEmitter {
   // Graceful or immediate shutdown of the Http2Session. Graceful shutdown
   // is only supported on the server-side
   shutdown(options, callback) {
-    if (this[kState].destroyed)
+    if (this[kState].destroyed || this[kState].destroying)
       throw new errors.Error('ERR_HTTP2_INVALID_SESSION');
 
     if (this[kState].shutdown || this[kState].shuttingDown)
@@ -1037,7 +1046,7 @@ class Http2Session extends EventEmitter {
   }
 
   _onTimeout() {
-    this.emit('timeout');
+    process.nextTick(emit.bind(this, 'timeout'));
   }
 }
 
@@ -1061,7 +1070,7 @@ class ClientHttp2Session extends Http2Session {
   // Submits a new HTTP2 request to the connected peer. Returns the
   // associated Http2Stream instance.
   request(headers, options) {
-    if (this[kState].destroyed)
+    if (this[kState].destroyed || this[kState].destroying)
       throw new errors.Error('ERR_HTTP2_INVALID_SESSION');
     debug(`[${sessionName(this[kType])}] initiating request`);
     _unrefActive(this);
@@ -1317,7 +1326,7 @@ class Http2Stream extends Duplex {
   }
 
   _onTimeout() {
-    this.emit('timeout');
+    process.nextTick(emit.bind(this, 'timeout'));
   }
 
   // true if the Http2Stream was aborted abornomally.
@@ -2104,7 +2113,7 @@ const onTimeout = {
   configurable: false,
   enumerable: false,
   value: function() {
-    this.emit('timeout');
+    process.nextTick(emit.bind(this, 'timeout'));
   }
 };
 
@@ -2191,6 +2200,7 @@ function socketOnError(error) {
 // of the session.
 function socketOnTimeout() {
   debug('socket timeout');
+  process.nextTick(() => {
     const server = this[kServer];
     const session = this[kSession];
     // If server or session are undefined, then we're already in the process of
@@ -2205,6 +2215,7 @@ function socketOnTimeout() {
         },
         this.destroy.bind(this));
     }
+  });
 }
 
 // Handles the on('stream') event for a session and forwards
@@ -2346,6 +2357,8 @@ function setupCompat(ev) {
 function socketOnClose(hadError) {
   const session = this[kSession];
   if (session !== undefined && !session.destroyed) {
+    // Skip unconsume because the socket is destroyed.
+    session[kState].skipUnconsume = true;
     session.destroy();
   }
 }

src/node_http2.cc

@@ -407,6 +407,12 @@ void Http2Session::Destroy(const FunctionCallbackInfo<Value>& args) {
   Http2Session* session;
   ASSIGN_OR_RETURN_UNWRAP(&session, args.Holder());
   DEBUG_HTTP2("Http2Session: destroying session %d\n", session->type());
+  Environment* env = Environment::GetCurrent(args);
+  Local<Context> context = env->context();
+
+  bool skipUnconsume = args[0]->BooleanValue(context).ToChecked();
+
+  if (!skipUnconsume)
     session->Unconsume();
   session->Free();
 }

src/node_http2_core.cc

@@ -176,7 +176,7 @@ void Nghttp2Session::GetTrailers(nghttp2_session* session,
     handle->OnTrailers(stream, &trailers);
     if (trailers.length() > 0) {
       DEBUG_HTTP2("Nghttp2Session %d: sending trailers for stream %d, "
-                  "count: %d\n", handle->session_type_, id,
+                  "count: %d\n", handle->session_type_, stream->id(),
                   trailers.length());
       *flags |= NGHTTP2_DATA_FLAG_NO_END_STREAM;
       nghttp2_submit_trailer(session,

test/parallel/test-http2-server-timeout.js

@@ -9,11 +9,10 @@ server.setTimeout(common.platformTimeout(1));
 
 const onServerTimeout = common.mustCall((session) => {
   session.destroy();
-  server.removeListener('timeout', onServerTimeout);
 });
 
 server.on('stream', common.mustNotCall());
-server.on('timeout', onServerTimeout);
+server.once('timeout', onServerTimeout);
 
 server.listen(0, common.mustCall(() => {
   const url = `http://localhost:${server.address().port}`;