Preserve orderly HTTP/2 duplex shutdown.

lib/async/http/body/pipe.rb

@@ -63,7 +63,7 @@ def reader(task)
 				end
 
 				# Read from the head of the pipe and write to the @output stream.
-				# If the @tail is closed, this will cause chunk to be nil, which in turn will call `@output.close` and `@head.close`
+				# A write-side close on @tail produces EOF and closes @output independently of the input direction.
 				def writer(task)
 					@writer = task
 

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

@@ -25,8 +25,8 @@ def initialize(stream, length)
 					# @returns [String | Nil] The next chunk, or `nil` if the body is complete.
 					def read
 						if chunk = super
-							# If we read a chunk fron the stream, we want to extend the window if required so more data will be provided.
-							@stream.request_window_update
+							# If we read a chunk from the stream, we want to extend the window if required so more data will be provided.
+							@stream&.request_window_update
 						end
 
 						# We track the expected length and check we got what we were expecting.
@@ -42,6 +42,17 @@ def read
 
 						return chunk
 					end
+
+					# Close the application-facing input body and notify the stream that incoming data is no longer being consumed. While local output is active, the HTTP/2 stream remains open. Once output also closes, the remaining wire stream is terminated without an error.
+					# @parameter error [Exception | Nil] The error that caused the input to be closed, if any.
+					def close(error = nil)
+						super
+
+						if stream = @stream
+							@stream = nil
+							stream.finish_input(self, error)
+						end
+					end
 				end
 			end
 		end

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

@@ -54,22 +54,24 @@ def window_updated(size)
 					# @parameter chunk [String] The data to write.
 					def write(chunk)
 						until chunk.empty?
-							maximum_size = @stream.available_frame_size
+							stream = @stream or raise IOError, "HTTP/2 stream is closed!"
+							maximum_size = stream.available_frame_size
 
 							# We try to avoid synchronization if possible:
 							if maximum_size <= 0
 								@guard.synchronize do
-									maximum_size = @stream.available_frame_size
+									maximum_size = stream.available_frame_size
 
 									while maximum_size <= 0
 										@window_updated.wait(@guard)
 
-										maximum_size = @stream.available_frame_size
+										stream = @stream or raise IOError, "HTTP/2 stream is closed!"
+										maximum_size = stream.available_frame_size
 									end
 								end
 							end
 
-							break unless chunk = send_data(chunk, maximum_size)
+							break unless chunk = send_data(stream, chunk, maximum_size)
 						end
 					end
 
@@ -96,6 +98,19 @@ def stop(error)
 						end
 					end
 
+					# Close the wire output without cancelling a streamable body. This allows bidirectional bodies to observe an orderly input closure and finish normally. A non-streaming producer has no input side through which closure can propagate, so it is stopped directly.
+					def close_stream
+						if @body.stream?
+							@stream = nil
+
+							@guard.synchronize do
+								@window_updated.broadcast
+							end
+						else
+							stop(nil)
+						end
+					end
+
 					private
 
 					def stream(task)
@@ -137,11 +152,11 @@ def passthrough(task)
 					# @param maximum_size [Integer] send up to this many bytes of data.
 					# @param stream [Stream] the stream to use for sending data frames.
 					# @return [String, nil] any data that could not be written.
-					def send_data(chunk, maximum_size)
+					def send_data(stream, chunk, maximum_size)
 						if chunk.bytesize <= maximum_size
-							@stream.send_data(chunk, maximum_size: maximum_size)
+							stream.send_data(chunk, maximum_size: maximum_size)
 						else
-							@stream.send_data(chunk.byteslice(0, maximum_size), maximum_size: maximum_size)
+							stream.send_data(chunk.byteslice(0, maximum_size), maximum_size: maximum_size)
 
 							# The window was not big enough to send all the data, so we save it for next time:
 							return chunk.byteslice(maximum_size, chunk.bytesize - maximum_size)

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

@@ -30,11 +30,13 @@ def initialize(*)
 						# Wait for the response headers and return the response body.
 						# @returns [Protocol::HTTP::Body::Readable | Nil] The response body.
 						def wait_for_input
+							response = @response
+
 							# The input isn't ready until the response headers have been received:
-							@response.wait
+							response.wait
 
 							# There is a possible race condition if you try to access @input - it might already be closed and nil.
-							return @response.body
+							return response.body
 						end
 
 						# Handle a push promise stream from the server.
@@ -169,6 +171,17 @@ def wait
 						@stream.wait
 					end
 
+					# Close this response as quickly as possible. If the response body is still active, cancel the HTTP/2 exchange rather than draining it.
+					# @parameter error [Exception | Nil] The error which closed the response.
+					def close(error = nil)
+						if @body && !@stream.closed?
+							code = error ? ::Protocol::HTTP2::Error::INTERNAL_ERROR : ::Protocol::HTTP2::Error::CANCEL
+							@stream.send_reset_stream(code)
+						end
+
+						super
+					end
+
 					# @returns [Boolean] Whether the original request was a HEAD request.
 					def head?
 						@request&.head?

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

@@ -28,6 +28,9 @@ def initialize(*)
 						@length = nil
 						@input = nil
 
+						# The application can close its input before the peer finishes sending. HTTP/2 cannot close only the receiving side of a stream, so incoming data is discarded until local output also finishes. At that point, a no-error reset terminates the remaining wire stream.
+						@input_closed = false
+
 						# Output buffer, writing request body or response body (window_updated):
 						@output = nil
 					end
@@ -114,14 +117,17 @@ def update_local_window(frame)
 					def process_data(frame)
 						data = frame.unpack
 
-						if @input
+						if input = @input
 							unless data.empty?
-								@input.write(data)
+								input.write(data)
 							end
 
 							if frame.end_stream?
-								@input.close_write
+								input.close_write
 							end
+						else
+							# The application has closed the input, so discard incoming data while maintaining flow control for the stream.
+							request_window_update
 						end
 
 						return data
@@ -131,6 +137,22 @@ def process_data(frame)
 						send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR)
 					end
 
+					# Close the application-facing receiving side of the stream. While local output remains active, incoming data is discarded with flow-control updates. Once local output is also closed, the remaining wire stream is terminated without an error.
+					# @parameter input [Input] The input body being closed.
+					# @parameter error [Exception | Nil] The error which closed the input.
+					def finish_input(input, error = nil)
+						if @input.equal?(input)
+							@input = nil
+							@input_closed = true
+
+							if error
+								send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR)
+							else
+								close_if_finished
+							end
+						end
+					end
+
 					# Set the body and begin sending it.
 					def send_body(body, trailer = nil)
 						@output = Output.new(self, body, trailer)
@@ -169,11 +191,32 @@ def window_updated(size)
 						return true
 					end
 
+					# Send headers and apply any pending application-side closure.
+					def send_headers(...)
+						result = super
+						close_if_finished
+						return result
+					end
+
+					# Send data and apply any pending application-side closure.
+					def send_data(...)
+						result = super
+						close_if_finished
+						return result
+					end
+
 					# When the stream transitions to the closed state, this method is called. There are roughly two ways this can happen:
 					# - A frame is received which causes this stream to enter the closed state. This method will be invoked from the background reader task.
 					# - A frame is sent which causes this stream to enter the closed state. This method will be invoked from that task.
 					# While the input stream is relatively straight forward, the output stream can trigger the second case above
 					def closed(error)
+						orderly_reset = error.is_a?(::Protocol::HTTP2::StreamError) &&
+							error.code == ::Protocol::HTTP2::Error::NO_ERROR
+
+						if orderly_reset
+							error = nil
+						end
+
 						super
 
 						if input = @input
@@ -183,7 +226,12 @@ def closed(error)
 
 						if output = @output
 							@output = nil
-							output.stop(error)
+
+							if orderly_reset
+								output.close_stream
+							else
+								output.stop(error)
+							end
 						end
 
 						if pool = @pool and @connection
@@ -192,6 +240,15 @@ def closed(error)
 
 						return self
 					end
+
+					private
+
+					# If both application-facing directions are closed but the peer has not finished, terminate the remaining wire stream without an error.
+					def close_if_finished
+						if @input_closed && @state == :half_closed_local
+							send_reset_stream(::Protocol::HTTP2::Error::NO_ERROR)
+						end
+					end
 				end
 			end
 		end

test/async/http/protocol/http2.rb

@@ -5,6 +5,7 @@
 
 require "async/http/protocol/http2"
 require "async/http/a_protocol"
+require "async/promise"
 
 describe Async::HTTP::Protocol::HTTP2 do
 	it_behaves_like Async::HTTP::AProtocol
@@ -80,23 +81,23 @@ def make_client(endpoint, **options)
 		end
 
 		with "stopping requests" do
-			let(:notification) {Async::Notification.new}
+			let(:finished) {Async::Promise.new}
 
 			let(:app) do
 				Protocol::HTTP::Middleware.for do |request|
 					body = Async::HTTP::Body::Writable.new
 
 					reactor.async do |task|
 						begin
-							100.times do |i|
+							1000.times do |i|
 								body.write("Chunk #{i}")
 								sleep (0.01)
 							end
 						rescue
 							# puts "Response generation failed: #{$!}"
 						ensure
 							body.close
-							notification.signal
+							finished.resolve(true)
 						end
 					end
 
@@ -115,7 +116,7 @@ def make_client(endpoint, **options)
 
 				response.close
 
-				notification.wait
+				finished.wait(timeout: 1)
 
 				expect(response.stream.connection).to be(:reusable?)
 			end

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

@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "async/http/protocol/http2/input"
+
+describe Async::HTTP::Protocol::HTTP2::Input do
+	let(:stream) do
+		Class.new do
+			attr_reader :window_updates
+			attr_reader :finished_inputs
+
+			def initialize
+				@window_updates = 0
+				@finished_inputs = []
+			end
+
+			def request_window_update
+				@window_updates += 1
+			end
+
+			def finish_input(input, error = nil)
+				@finished_inputs << [input, error]
+			end
+		end.new
+	end
+
+	let(:input) {subject.new(stream, nil)}
+
+	it "requests a window update when data is consumed" do
+		input.write("Hello World")
+
+		expect(input.read).to be == "Hello World"
+		expect(stream.window_updates).to be == 1
+	end
+
+	it "notifies the stream when closed" do
+		error = RuntimeError.new("Input closed")
+
+		input.close(error)
+		input.close
+
+		expect(stream.finished_inputs).to be == [[input, error]]
+	end
+end