From 842fc6b163a5c3e22ab1bc7ab849a4872aa40021 Mon Sep 17 00:00:00 2001 From: Denis Talakevich Date: Sun, 30 Aug 2026 15:52:14 +0300 Subject: [PATCH 1/3] Do not close HTTP/2 connections while draining a graceful GOAWAY. When a server sends a graceful GOAWAY it will not accept new streams, but it is still processing the streams at or below last_stream_id and will send their responses (RFC 9113 6.8). The client only honoured the first half: the connection was closed at once, the background reader stopped reading, and every stream still waiting was failed with "Connection closed with N active stream(s)!" -- exactly the streams the server accepted and was about to answer. For a POST that is unrecoverable, because a non-idempotent request is not retried. nginx sends a graceful GOAWAY on the keepalive_requests-th request of every HTTP/2 connection, on keepalive_time, and on every reload, so a client with requests in flight loses a burst of them each time. Reproduced against nginx 1.22.1: 20 of 100 requests failed while the backend completed all 100. With the companion change in protocol-http2, such a connection is now retired from the pool immediately but stays open until the accepted streams complete: - viable?/reusable? are false once a GOAWAY has been received, so the pool retires the connection instead of handing it to another request. - Client#call refuses a request which reaches a connection that is going away, so it is retried on a new connection, non-idempotent ones included. This is checked before closed?, because a GOAWAY which finds no streams to drain closes the connection in the same step. - close(nil) defers while the connection is draining, so the pool retiring it does not fail the streams we are waiting for. An explicit error, or the absence of a background reader, still closes immediately. - close_if_drained! stops the background reader when the drain finishes in another task: the last stream can complete on the sending side, leaving the reader parked in a blocking read, leaking the task and the socket. - close no longer stops the reader task when it is the reader task, which raised Async::Cancel in the middle of close and left the socket open. --- lib/async/http/protocol/http2/client.rb | 3 + lib/async/http/protocol/http2/connection.rb | 30 ++- releases.md | 4 + .../http/protocol/http2/graceful_goaway.rb | 227 ++++++++++++++++++ 4 files changed, 258 insertions(+), 6 deletions(-) create mode 100644 test/async/http/protocol/http2/graceful_goaway.rb diff --git a/lib/async/http/protocol/http2/client.rb b/lib/async/http/protocol/http2/client.rb index 95624bfa..b1c1dd4a 100644 --- a/lib/async/http/protocol/http2/client.rb +++ b/lib/async/http/protocol/http2/client.rb @@ -34,6 +34,9 @@ def create_response # Used by the client to send requests to the remote server. def call(request) + # The remote peer has sent a GOAWAY frame, so it will not process any new streams on this connection. The request has not been sent yet, so it is safe to retry it on a new connection, even if it is not idempotent. This is checked first, because a connection with no streams left to drain is closed by the GOAWAY itself. + raise ::Protocol::HTTP::RefusedError, "Connection is going away!" if self.goaway_received? + raise ::Protocol::HTTP2::Error, "Connection closed!" if self.closed? response = create_response diff --git a/lib/async/http/protocol/http2/connection.rb b/lib/async/http/protocol/http2/connection.rb index 0b7aa01b..dc57c54c 100644 --- a/lib/async/http/protocol/http2/connection.rb +++ b/lib/async/http/protocol/http2/connection.rb @@ -87,17 +87,33 @@ def start_connection end # Close the connection and stop the background reader. + # + # If the remote peer sent a graceful GOAWAY frame, the streams it accepted are still being processed and their responses are still on the way. Closing the connection now would fail those requests, even though the remote peer has already processed them. Instead we defer: the background reader closes the connection once the last stream completes. An explicit error still closes the connection immediately. def close(error = nil) - # Ensure the reader task is stopped. - if @reader - reader = @reader + if reader = @reader + # Only the background reader can drain the connection, so if it is not running, there is nothing to wait for: + return self if error.nil? and self.draining? + @reader = nil - reader.stop + + # The reader task can close the connection itself, e.g. when the last stream completes and the connection is released back to the pool. Stopping it here would cancel the current task in the middle of this method, leaving the underlying stream open, so we let it unwind by itself: `closed?` is now true, so the read loop exits. + reader.stop unless reader.current? end super end + # The connection has finished draining the streams which the remote peer accepted before its graceful GOAWAY. + # + # The last of those streams can complete on the sending side, in a task other than the background reader - the response arrived first and the request body was still being written. The reader is then parked in a blocking read and will never notice that the connection is closed, so we stop it and let its `ensure` close the connection. + def close_if_drained! + super + + if self.closed? and (reader = @reader) and !reader.current? + reader.stop + end + end + # Start a transient background task that reads frames from the connection. def read_in_background(parent: Task.current) raise RuntimeError, "Connection is closed!" if closed? @@ -142,12 +158,14 @@ def concurrency # Can we use this connection to make requests? def viable? - @stream&.readable? + !self.goaway_received? && @stream&.readable? end # @returns [Boolean] Whether the connection can be reused. + # + # Once the remote peer has sent a GOAWAY frame, it will not process any new streams on this connection, so it must not be handed out for another request, even while the streams it accepted are still being drained. def reusable? - !self.closed? + !self.closed? && !self.goaway_received? end # @returns [String] The HTTP version string. diff --git a/releases.md b/releases.md index 19eacd19..f79f25c8 100644 --- a/releases.md +++ b/releases.md @@ -1,5 +1,9 @@ # Releases +## Unreleased + + - HTTP/2 connections which received a graceful `GOAWAY` are no longer closed while the server is still answering the streams it accepted. Such a connection is retired from the pool, but stays open until the last accepted stream completes, so those requests no longer fail with `EOFError: Connection closed with N active stream(s)!`. + ## v0.101.0 - Handle remote disconnects in `Async::HTTP::Protocol::HTTP1::Server#each` without reporting them as server failures. diff --git a/test/async/http/protocol/http2/graceful_goaway.rb b/test/async/http/protocol/http2/graceful_goaway.rb new file mode 100644 index 00000000..f9be5b48 --- /dev/null +++ b/test/async/http/protocol/http2/graceful_goaway.rb @@ -0,0 +1,227 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Denis Talakevich. + +require "async/http/client" +require "async/http/mock" +require "async/http/protocol/http2" +require "protocol/http2/server" +require "protocol/http2/stream" +require "sus/fixtures/async/scheduler_context" + +require "async/queue" +require "io/stream" +require "socket" + +describe Async::HTTP::Protocol::HTTP2 do + include Sus::Fixtures::Async::SchedulerContext + + let(:response_headers) {[[":status", "200"]]} + + # A graceful GOAWAY is sent by the server *while it is still processing* the streams it + # accepted, which {Async::HTTP::Server} does not do, so these tests drive a raw HTTP/2 server + # over the sockets the mock endpoint hands out. + let(:peers) {Async::Queue.new} + let(:endpoint) {Async::HTTP::Mock::Endpoint.new(subject, "http", "localhost", queue: peers)} + let(:client) {Async::HTTP::Client.new(endpoint, limit: 1)} + + def goaway_frame(last_stream_id) + frame = Protocol::HTTP2::GoawayFrame.new + frame.pack(last_stream_id, 0, "") + + return frame + end + + def accept_connection(socket) + server = Protocol::HTTP2::Server.new(Protocol::HTTP2::Framer.new(socket)) + server.read_connection_preface([]) + + return server + end + + # Reads frames until `count` requests have arrived, and returns their streams in the order + # they were accepted. + def accept_requests(server, count) + streams = [] + + while streams.size < count + frame = server.read_frame + + if frame.is_a?(Protocol::HTTP2::HeadersFrame) + streams << server.streams[frame.stream_id] + end + end + + return streams + end + + # The first connection accepts three requests but tells the client it only processed the + # first one; every later connection simply answers what it is given. + def handle_connection(socket, index, processed) + server = accept_connection(socket) + + if index == 1 + streams = accept_requests(server, 3) + server.write_frame(goaway_frame(streams.first.id)) + + # The response for the accepted stream arrives *after* the GOAWAY, exactly like nginx: + processed << streams.first.id + streams.first.send_headers(response_headers, Protocol::HTTP2::END_STREAM) + else + accept_requests(server, 2).each do |stream| + stream.send_headers(response_headers, Protocol::HTTP2::END_STREAM) + end + end + + # Keep the connection alive until the client closes it: + while true + server.read_frame + end + rescue EOFError, Errno::EPIPE + # The client closed the connection. + ensure + socket.close + end + + with "a server which sends a graceful GOAWAY" do + it "completes the requests the server accepted, and retries the rest" do + connections = 0 + processed = [] + + acceptor = Async(transient: true) do |task| + while socket = peers.dequeue + index = (connections += 1) + + task.async{handle_connection(socket, index, processed)} + end + end + + responses = 3.times.map do |index| + Async do + client.post("/#{index}", {}, ["body-#{index}"]) + end + end.map(&:wait) + + # Every request completes, including the one which was in flight when the GOAWAY arrived: + expect(responses.map(&:status)).to be == [200, 200, 200] + + # The first request was answered on the connection which was going away: + expect(processed).to be == [1] + + # The two requests which the server did not process were retried on a new connection: + expect(connections).to be == 2 + ensure + client.close + acceptor&.stop + end + end + + with "a connection which received a GOAWAY" do + let(:sockets) {::Socket.pair(::Socket::AF_UNIX, ::Socket::SOCK_STREAM)} + let(:connection) {Async::HTTP::Protocol::HTTP2::Client.new(IO::Stream(sockets.first))} + + def after(error = nil) + sockets.each{|socket| socket.close unless socket.closed?} + + super + end + + with "no streams left to drain" do + def before + super + + connection.open! + connection.receive_goaway(goaway_frame(0)) + end + + it "is closed" do + expect(connection).to be(:goaway_received?) + expect(connection).not.to be(:draining?) + expect(connection).to be(:closed?) + end + + it "refuses new requests so that they are retried on another connection" do + # The connection is already closed, but the request was still not processed, so it must be refused rather than failed: + expect do + connection.call(::Protocol::HTTP::Request["POST", "/", {}, ["Hello World"]]) + end.to raise_exception(::Protocol::HTTP::RefusedError) + end + end + + with "a stream still draining" do + def before + super + + connection.open! + + # One request is in flight, and the server tells us it is the last one it will process: + response = connection.create_response + + connection.receive_goaway(goaway_frame(response.stream.id)) + end + + it "is not offered to new requests" do + expect(connection).to be(:goaway_received?) + expect(connection).to be(:draining?) + expect(connection).not.to be(:closed?) + + expect(connection).not.to be(:reusable?) + expect(connection).not.to be(:viable?) + end + + it "refuses new requests so that they are retried on another connection" do + expect do + connection.call(::Protocol::HTTP::Request["POST", "/", {}, ["Hello World"]]) + end.to raise_exception(::Protocol::HTTP::RefusedError) + end + + it "is not closed while the accepted stream is still in flight" do + Async do + connection.read_in_background + + # The pool retires a connection which is no longer reusable, but the stream the server accepted is still being processed, so the connection must stay open: + connection.close + + expect(connection).not.to be(:closed?) + expect(connection.streams).not.to be(:empty?) + ensure + connection.close(EOFError.new("Test finished!")) + end.wait + end + + it "is closed if there is no reader to drain it" do + connection.close + + expect(connection).to be(:closed?) + end + end + + it "closes itself when the last drained stream completes on the sending side" do + server = Protocol::HTTP2::Server.new(Protocol::HTTP2::Framer.new(sockets.last)) + server.open! + + connection.open! + + # The request body is still being written when the response arrives, so the stream is completed by this task rather than by the background reader: + stream = connection.create_stream + stream.send_headers([[":method", "POST"], [":path", "/"], [":authority", "localhost"]]) + + connection.read_in_background + + server.read_frame + server.write_frame(goaway_frame(stream.id)) + server.streams[stream.id].send_headers(response_headers, Protocol::HTTP2::END_STREAM) + + Async::Task.current.sleep(0.1) + expect(connection).to be(:draining?) + + stream.send_data("body", Protocol::HTTP2::END_STREAM) + Async::Task.current.sleep(0.1) + + expect(connection).to be(:closed?) + expect(connection.framer).to be_nil + expect(sockets.first).to be(:closed?) + end + end +end From cefdbc530f88fa7905ed6fe80674dee00bd4e4ca Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Mon, 31 Aug 2026 14:40:21 +1200 Subject: [PATCH 2/3] Use pool ownership for graceful GOAWAY connections Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- async-http.gemspec | 2 +- lib/async/http/protocol/http2/connection.rb | 5 -- releases.md | 2 +- .../http/protocol/http2/graceful_goaway.rb | 51 ++++++++++++------- 4 files changed, 35 insertions(+), 25 deletions(-) diff --git a/async-http.gemspec b/async-http.gemspec index 1a7eba75..02108628 100644 --- a/async-http.gemspec +++ b/async-http.gemspec @@ -25,7 +25,7 @@ Gem::Specification.new do |spec| spec.required_ruby_version = ">= 3.3" spec.add_dependency "async", ">= 2.35.1" - spec.add_dependency "async-pool", "~> 0.11" + spec.add_dependency "async-pool", "~> 0.12" spec.add_dependency "io-endpoint", "~> 0.18" spec.add_dependency "io-stream", "~> 0.14" spec.add_dependency "protocol-http", "~> 0.66" diff --git a/lib/async/http/protocol/http2/connection.rb b/lib/async/http/protocol/http2/connection.rb index dc57c54c..a33a1389 100644 --- a/lib/async/http/protocol/http2/connection.rb +++ b/lib/async/http/protocol/http2/connection.rb @@ -87,13 +87,8 @@ def start_connection end # Close the connection and stop the background reader. - # - # If the remote peer sent a graceful GOAWAY frame, the streams it accepted are still being processed and their responses are still on the way. Closing the connection now would fail those requests, even though the remote peer has already processed them. Instead we defer: the background reader closes the connection once the last stream completes. An explicit error still closes the connection immediately. def close(error = nil) if reader = @reader - # Only the background reader can drain the connection, so if it is not running, there is nothing to wait for: - return self if error.nil? and self.draining? - @reader = nil # The reader task can close the connection itself, e.g. when the last stream completes and the connection is released back to the pool. Stopping it here would cancel the current task in the middle of this method, leaving the underlying stream open, so we let it unwind by itself: `closed?` is now true, so the read loop exits. diff --git a/releases.md b/releases.md index f79f25c8..65737862 100644 --- a/releases.md +++ b/releases.md @@ -2,7 +2,7 @@ ## Unreleased - - HTTP/2 connections which received a graceful `GOAWAY` are no longer closed while the server is still answering the streams it accepted. Such a connection is retired from the pool, but stays open until the last accepted stream completes, so those requests no longer fail with `EOFError: Connection closed with N active stream(s)!`. + - HTTP/2 connections which received a graceful `GOAWAY` are removed from availability immediately, but remain in the pool until the server has finished answering the streams it accepted. The connection is closed after its final user releases it, so those requests no longer fail with `EOFError: Connection closed with N active stream(s)!`. ## v0.101.0 diff --git a/test/async/http/protocol/http2/graceful_goaway.rb b/test/async/http/protocol/http2/graceful_goaway.rb index f9be5b48..c3d76b00 100644 --- a/test/async/http/protocol/http2/graceful_goaway.rb +++ b/test/async/http/protocol/http2/graceful_goaway.rb @@ -137,7 +137,6 @@ def before it "is closed" do expect(connection).to be(:goaway_received?) - expect(connection).not.to be(:draining?) expect(connection).to be(:closed?) end @@ -163,7 +162,6 @@ def before it "is not offered to new requests" do expect(connection).to be(:goaway_received?) - expect(connection).to be(:draining?) expect(connection).not.to be(:closed?) expect(connection).not.to be(:reusable?) @@ -176,27 +174,42 @@ def before end.to raise_exception(::Protocol::HTTP::RefusedError) end - it "is not closed while the accepted stream is still in flight" do - Async do - connection.read_in_background - - # The pool retires a connection which is no longer reusable, but the stream the server accepted is still being processed, so the connection must stay open: - connection.close - - expect(connection).not.to be(:closed?) - expect(connection.streams).not.to be(:empty?) - ensure - connection.close(EOFError.new("Test finished!")) - end.wait - end - - it "is closed if there is no reader to drain it" do + it "closes synchronously" do connection.close expect(connection).to be(:closed?) end end + it "is retained by the pool until all users release it" do + connection.open! + pool = Async::Pool::Controller.wrap(limit: 1){connection} + + resource1 = pool.acquire + resource2 = pool.acquire + + response1 = connection.create_response + response2 = connection.create_response + connection.receive_goaway(goaway_frame(response2.stream.id)) + + response1.stream.close! + pool.release(resource1) + + expect(pool.resources[connection]).to be == 1 + expect(pool).not.to be(:available?) + expect(connection).not.to be(:closed?) + expect(connection.streams.keys).to be == [response2.stream.id] + + response2.stream.close! + expect(connection.framer).not.to be_nil + + pool.release(resource2) + + expect(pool.resources).not.to be(:key?, connection) + expect(connection.framer).to be_nil + expect(sockets.first).to be(:closed?) + end + it "closes itself when the last drained stream completes on the sending side" do server = Protocol::HTTP2::Server.new(Protocol::HTTP2::Framer.new(sockets.last)) server.open! @@ -214,7 +227,9 @@ def before server.streams[stream.id].send_headers(response_headers, Protocol::HTTP2::END_STREAM) Async::Task.current.sleep(0.1) - expect(connection).to be(:draining?) + expect(connection).to be(:goaway_received?) + expect(connection).not.to be(:closed?) + expect(connection.streams.keys).to be == [stream.id] stream.send_data("body", Protocol::HTTP2::END_STREAM) Async::Task.current.sleep(0.1) From 7519a41590c3d03f80d808e6847036567fea7315 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Mon, 31 Aug 2026 15:08:41 +1200 Subject: [PATCH 3/3] Require protocol-http2 0.27 Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- async-http.gemspec | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/async-http.gemspec b/async-http.gemspec index 02108628..a904b25e 100644 --- a/async-http.gemspec +++ b/async-http.gemspec @@ -30,6 +30,6 @@ Gem::Specification.new do |spec| spec.add_dependency "io-stream", "~> 0.14" spec.add_dependency "protocol-http", "~> 0.66" spec.add_dependency "protocol-http1", "~> 0.41" - spec.add_dependency "protocol-http2", "~> 0.26" + spec.add_dependency "protocol-http2", "~> 0.27" spec.add_dependency "protocol-url", "~> 0.2" end