Do not close HTTP/2 connections while a graceful GOAWAY is being drained

async-http.gemspec

@@ -25,11 +25,11 @@ 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"
 	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

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

lib/async/http/protocol/http2/connection.rb

@@ -88,16 +88,27 @@ def start_connection
 
 					# Close the connection and stop the background reader.
 					def close(error = nil)
-						# Ensure the reader task is stopped.
-						if @reader
-							reader = @reader
+						if reader = @reader
 							@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 +153,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.

releases.md

@@ -1,5 +1,9 @@
 # Releases
 
+## Unreleased
+
+  - 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
 
   - Handle remote disconnects in `Async::HTTP::Protocol::HTTP1::Server#each` without reporting them as server failures.

test/async/http/protocol/http2/graceful_goaway.rb

@@ -0,0 +1,242 @@
+# 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).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).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 "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!
+
+			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(: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)
+
+			expect(connection).to be(:closed?)
+			expect(connection.framer).to be_nil
+			expect(sockets.first).to be(:closed?)
+		end
+	end
+end