diff --git a/src/node_http2.cc b/src/node_http2.cc index 86b09c1d4efb..02860722927c 100644 --- a/src/node_http2.cc +++ b/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 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(); diff --git a/src/node_http2.h b/src/node_http2.h index c042c46ab595..70d00b065f04 100644 --- a/src/node_http2.h +++ b/src/node_http2.h @@ -1001,8 +1001,8 @@ class Http2Session : public AsyncWrap, std::vector outgoing_storage_; size_t outgoing_length_ = 0; std::vector 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(); diff --git a/test/parallel/test-http2-session-destroy-pending-write.js b/test/parallel/test-http2-session-destroy-pending-write.js new file mode 100644 index 000000000000..7ff3941a0724 --- /dev/null +++ b/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])); +}