http2: fix graceful session close

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);
   }
 }
 

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")                                                     \

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(

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);

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));
+}));