http2: flush final GOAWAY after pending writes

src/node_http2.cc

@@ -829,19 +829,22 @@ void Http2Session::Close(uint32_t code, bool socket_closed) {
     return;
   set_closing();
 
+  pending_close_code_ = code;
+  pending_close_socket_closed_ = socket_closed;
+
   // Do not flush GOAWAY from inside nghttp2_session_mem_recv() callbacks.
   // ConsumeHTTP2Data() finishes the close once mem_recv returns.
   if (is_receiving()) {
     set_close_pending();
-    pending_close_code_ = code;
-    pending_close_socket_closed_ = socket_closed;
     return;
   }
 
   FinishClose(code, socket_closed);
 }
 
-void Http2Session::FinishClose(uint32_t code, bool socket_closed) {
+void Http2Session::FinishClose(uint32_t code,
+                              bool socket_closed,
+                              int write_status) {
   CHECK(is_closing());
 
   // Stop reading on the i/o stream
@@ -850,15 +853,20 @@ void Http2Session::FinishClose(uint32_t code, bool socket_closed) {
     stream_->ReadStop();
   }
 
-  // If the socket is not closed, then attempt to send a closing GOAWAY
+  // If the socket is still writable, attempt to send a closing GOAWAY
   // frame. There is no guarantee that this GOAWAY will be received by
   // the peer but the HTTP/2 spec recommends sending it anyway. We'll
   // make a best effort.
-  if (!socket_closed) {
+  if (!socket_closed && write_status == 0) {
     Debug(this, "terminating session with code %d", code);
     CHECK_EQ(nghttp2_session_terminate_session(session_.get(), code), 0);
-    SendPendingData();
-  } else if (stream_ != nullptr) {
+    if (SendPendingData() != 0) {
+      // An earlier write still owns the outgoing buffers. Keep the session
+      // alive until its completion lets us flush the terminal GOAWAY.
+      set_close_pending();
+      return;
+    }
+  } else if (socket_closed && stream_ != nullptr) {
     // so that the previous listener of the socket, typically, JS code of a
     // (tls) socket will be notified of any activity later
     stream_->RemoveStreamListener(this);
@@ -893,10 +901,12 @@ void Http2Session::FinishClose(uint32_t code, bool socket_closed) {
   EmitStatistics();
 }
 
-void Http2Session::MaybeFinishPendingClose() {
-  if (!is_close_pending() || is_destroyed()) return;
+void Http2Session::MaybeFinishPendingClose(int status) {
+  if (!is_close_pending() || is_destroyed() || is_receiving() || is_sending()) {
+    return;
+  }
   set_close_pending(false);
-  FinishClose(pending_close_code_, pending_close_socket_closed_);
+  FinishClose(pending_close_code_, pending_close_socket_closed_, status);
 }
 
 // Locates an existing known stream by ID. nghttp2 has a similar method
@@ -1844,9 +1854,16 @@ void Http2Session::OnStreamAfterWrite(WriteWrap* w, int status) {
   CHECK(is_write_in_progress());
   set_write_in_progress(false);
 
+  // ClearOutgoing() may synchronously finish a pending close.
+  const bool close_pending = is_close_pending();
   // Inform all pending writes about their completion.
   ClearOutgoing(status);
 
+  if (close_pending || is_close_pending()) {
+    MaybeFinishPendingClose(status);
+    return;
+  }
+
   if (is_reading_stopped() &&
       !is_write_in_progress() &&
       nghttp2_session_want_read(session_.get())) {
@@ -1943,7 +1960,7 @@ void Http2Session::ClearOutgoing(int status) {
 
   // Now that we've finished sending queued data, if there are any pending
   // RstStreams we should try sending again and then flush them one by one.
-  if (!pending_rst_streams_.empty()) {
+  if (status == 0 && !pending_rst_streams_.empty()) {
     std::vector<int32_t> current_pending_rst_streams;
     pending_rst_streams_.swap(current_pending_rst_streams);
 
@@ -2065,6 +2082,7 @@ uint8_t Http2Session::SendPendingData() {
     set_write_in_progress(false);
     ClearOutgoing(res.err);
     MaybeNotifyGracefulCloseComplete();
+    MaybeFinishPendingClose(res.err);
   }
 
   MaybeStopReading();

src/node_http2.h

@@ -1001,8 +1001,8 @@ class Http2Session : public AsyncWrap,
   std::vector<uint8_t> outgoing_storage_;
   size_t outgoing_length_ = 0;
   std::vector<int32_t> pending_rst_streams_;
-  // Saved arguments for Close() deferred while nghttp2_session_mem_recv()
-  // callbacks are active.
+  // Saved arguments for Close() deferred while receiving data or waiting for
+  // an outgoing write to finish.
   uint32_t pending_close_code_ = NGHTTP2_NO_ERROR;
   bool pending_close_socket_closed_ = false;
   // Count streams that have been rejected while being opened. Exceeding a fixed
@@ -1019,8 +1019,8 @@ class Http2Session : public AsyncWrap,
 
   void CopyDataIntoOutgoing(const uint8_t* src, size_t src_length);
   void ClearOutgoing(int status);
-  void FinishClose(uint32_t code, bool socket_closed);
-  void MaybeFinishPendingClose();
+  void FinishClose(uint32_t code, bool socket_closed, int write_status = 0);
+  void MaybeFinishPendingClose(int status = 0);
 
   void MaybeNotifyGracefulCloseComplete();
 

test/parallel/test-http2-session-destroy-pending-write.js

@@ -0,0 +1,91 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+
+// Destroying a session while a socket write is pending must flush the final
+// GOAWAY before ending the socket, including when called from a stream handler.
+const assert = require('assert');
+const http2 = require('http2');
+const { Duplex } = require('stream');
+const {
+  HeadersFrame,
+  SettingsFrame,
+  kClientMagic,
+  kFakeRequestHeaders,
+} = require('../common/http2');
+
+for (const [code, deferDestroy] of [
+  [0, false],
+  [http2.constants.NGHTTP2_REFUSED_STREAM, false],
+  [0, true],
+  [http2.constants.NGHTTP2_REFUSED_STREAM, true],
+]) {
+  const output = [];
+  let pendingWrite;
+  const socket = new Duplex({
+    read() {},
+    write(chunk, encoding, callback) {
+      output.push(Buffer.from(chunk));
+      if (pendingWrite === undefined) {
+        // Hold the initial SETTINGS write until the stream handler destroys
+        // the session. No timers or large response bodies are needed.
+        pendingWrite = callback;
+        setImmediate(() => {
+          socket.push(new HeadersFrame(1, kFakeRequestHeaders, 0, true).data);
+        });
+      } else {
+        callback();
+      }
+    },
+  });
+
+  const session = http2.performServerHandshake(socket);
+  session.on('stream', common.mustCall((stream) => {
+    assert.strictEqual(typeof pendingWrite, 'function');
+    stream.on('close', common.mustCall());
+    if (code !== 0) {
+      stream.on('error', common.mustCall((err) => {
+        assert.strictEqual(err.code, 'ERR_HTTP2_SESSION_ERROR');
+      }));
+    }
+    const destroy = common.mustCall(() => {
+      session.destroy(code);
+      assert.strictEqual(session.destroyed, true);
+      setImmediate(common.mustCall(() => {
+        const callback = pendingWrite;
+        pendingWrite = null;
+        callback();
+      }));
+    });
+    if (deferDestroy)
+      setImmediate(destroy);
+    else
+      destroy();
+  }));
+  session.on('close', common.mustCall());
+  if (code !== 0) {
+    session.on('error', common.mustCall((err) => {
+      assert.strictEqual(err.code, 'ERR_HTTP2_SESSION_ERROR');
+    }));
+  }
+
+  socket.on('finish', common.mustCall(() => {
+    const data = Buffer.concat(output);
+    let goaways = 0;
+    for (let offset = 0; offset < data.length;) {
+      const length = data.readUIntBE(offset, 3);
+      assert(data.length >= offset + 9 + length);
+      if (data[offset + 3] === 7) {
+        goaways++;
+        assert.strictEqual(data.readUInt32BE(offset + 9), 1);
+        assert.strictEqual(data.readUInt32BE(offset + 13), code);
+      }
+      offset += 9 + length;
+    }
+    assert.strictEqual(goaways, 1);
+  }));
+
+  socket.push(Buffer.concat([kClientMagic, new SettingsFrame().data]));
+}