http2: fix graceful session close

Fix issue where session.close() prematurely destroys the session
when response.end() was called with an empty payload while active
http2 streams still existed. This change ensures that sessions are
closed gracefully only after all http2 streams complete and clients
properly receive the GOAWAY frame as per the HTTP/2 spec.

Refs: https://nodejs.org/api/http2.html\#http2sessionclosecallback

Author: pandeykushagra51

Diff for: lib/internal/http2/core.js

@@ -1068,6 +1068,7 @@ function setupHandle(socket, type, options) {
   if (typeof options.selectPadding === 'function')
     this[kSelectPadding] = options.selectPadding;
   handle.consume(socket._handle);
+  handle.onstreamafterwrite = this[kMaybeDestroy].bind(this, null);
 
   this[kHandle] = handle;
   if (this[kNativeFields]) {
@@ -1609,11 +1610,13 @@ class Http2Session extends EventEmitter {
   // * session is closed and there are no more pending or open streams
   [kMaybeDestroy](error) {
     if (error == null) {
+      const handle = this[kHandle];
+      const hasPendingData = !!handle && handle.hasPendingData();
       const state = this[kState];
       // Do not destroy if we're not closed and there are pending/open streams
       if (!this.closed ||
           state.streams.size > 0 ||
-          state.pendingStreams.size > 0) {
+          state.pendingStreams.size > 0 || hasPendingData) {
         return;
       }
     }
@@ -3300,7 +3303,7 @@ function socketOnClose() {
     state.streams.forEach((stream) => stream.close(NGHTTP2_CANCEL));
     state.pendingStreams.forEach((stream) => stream.close(NGHTTP2_CANCEL));
     session.close();
-    session[kMaybeDestroy](err);
+    closeSession(session, NGHTTP2_NO_ERROR, err);
   }
 }
 

Diff for: src/env_properties.h

@@ -285,6 +285,7 @@
   V(onsignal_string, "onsignal")                                               \
   V(onunpipe_string, "onunpipe")                                               \
   V(onwrite_string, "onwrite")                                                 \
+  V(onstreamafterwrite_string, "onstreamafterwrite")                           \
   V(openssl_error_stack, "opensslErrorStack")                                  \
   V(options_string, "options")                                                 \
   V(order_string, "order")                                                     \

Diff for: src/node_http2.cc

@@ -765,6 +765,22 @@ void Http2Stream::EmitStatistics() {
   });
 }
 
+void Http2Session::HasPendingData(const FunctionCallbackInfo<Value>& args) {
+  Http2Session* session;
+  ASSIGN_OR_RETURN_UNWRAP(&session, args.Holder());
+  args.GetReturnValue().Set(session->HasPendingData());
+}
+
+bool Http2Session::HasPendingData() const {
+  nghttp2_session* session = session_.get();
+  int want_write = nghttp2_session_want_write(session);
+  int want_read = nghttp2_session_want_read(session);
+  if (want_write == 0 && want_read == 0) {
+    return false;
+  }
+  return true;
+}
+
 void Http2Session::EmitStatistics() {
   if (!HasHttp2Observer(env())) [[likely]] {
     return;
@@ -1743,6 +1759,8 @@ void Http2Session::HandleSettingsFrame(const nghttp2_frame* frame) {
 void Http2Session::OnStreamAfterWrite(WriteWrap* w, int status) {
   Debug(this, "write finished with status %d", status);
 
+  HandleScope scope(env()->isolate());
+  MakeCallback(env()->onstreamafterwrite_string(), 0, nullptr);
   CHECK(is_write_in_progress());
   set_write_in_progress(false);
 
@@ -1965,6 +1983,8 @@ uint8_t Http2Session::SendPendingData() {
   if (!res.async) {
     set_write_in_progress(false);
     ClearOutgoing(res.err);
+    HandleScope scope(env()->isolate());
+    MakeCallback(env()->onstreamafterwrite_string(), 0, nullptr);
   }
 
   MaybeStopReading();
@@ -3478,6 +3498,8 @@ void Initialize(Local<Object> target,
   SetProtoMethod(isolate, session, "receive", Http2Session::Receive);
   SetProtoMethod(isolate, session, "destroy", Http2Session::Destroy);
   SetProtoMethod(isolate, session, "goaway", Http2Session::Goaway);
+  SetProtoMethod(
+      isolate, session, "hasPendingData", Http2Session::HasPendingData);
   SetProtoMethod(isolate, session, "settings", Http2Session::Settings);
   SetProtoMethod(isolate, session, "request", Http2Session::Request);
   SetProtoMethod(

Diff for: src/node_http2.h

@@ -712,6 +712,7 @@ class Http2Session : public AsyncWrap,
   static void Consume(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Receive(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Destroy(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void HasPendingData(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Settings(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Request(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void SetNextStreamID(const v8::FunctionCallbackInfo<v8::Value>& args);
@@ -735,6 +736,7 @@ class Http2Session : public AsyncWrap,
 
   BaseObjectPtr<Http2Ping> PopPing();
   bool AddPing(const uint8_t* data, v8::Local<v8::Function> callback);
+  bool HasPendingData() const;
 
   BaseObjectPtr<Http2Settings> PopSettings();
   bool AddSettings(v8::Local<v8::Function> callback);

Diff for: test/parallel/test-http2-session-graceful-close.js

@@ -0,0 +1,47 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const assert = require('assert');
+const h2 = require('http2');
+
+const server = h2.createServer();
+let response;
+
+server.on('session', common.mustCall(function(session) {
+  session.on('stream', common.mustCall(function (stream) {
+    response.end();
+    session.close();
+  }));
+  session.on('close', common.mustCall(function() {
+    server.close();
+  }));
+}));
+
+server.listen(0, common.mustCall(function() {
+  const port = server.address().port;
+  server.once('request', common.mustCall(function (request, resp) {
+    response = resp;
+  }));
+
+  const url = `http://localhost:${port}`;
+  const client = h2.connect(url, common.mustCall(function() {
+    const headers = {
+      ':path': '/',
+      ':method': 'GET',
+      ':scheme': 'http',
+      ':authority': `localhost:${port}`
+    };
+    const request = client.request(headers);
+    request.on('response', common.mustCall(function(headers) {
+      assert.strictEqual(headers[':status'], 200);
+    }, 1));
+    request.on('end', common.mustCall(function() {
+      client.close();
+    }));
+    request.end();
+    request.resume();
+  }));
+  client.on('goaway', common.mustCallAtLeast(1));
+}));