http2: flush final GOAWAY after pending writes

Keep the native close pending when an active socket write prevents the
terminal GOAWAY from being serialized. Finish the close after the
outgoing buffers become available, preserving the requested error code.

Handle synchronous completion and transport errors without retrying
writes to a failed transport or losing the socket close notification.

Add a deterministic regression for default and custom GOAWAY codes, both
inside and outside the receive callback.

Fixes: #66832
Assisted-by: Codex
Signed-off-by: Emerson Gomes <emerson.gomes@gmail.com>

Author: emerzon

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