diff --git a/lib/async/http/body/pipe.rb b/lib/async/http/body/pipe.rb index 05b8d23f..da221778 100644 --- a/lib/async/http/body/pipe.rb +++ b/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 diff --git a/lib/async/http/protocol/http2/input.rb b/lib/async/http/protocol/http2/input.rb index 34e5ff61..7a50ba82 100644 --- a/lib/async/http/protocol/http2/input.rb +++ b/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 diff --git a/lib/async/http/protocol/http2/output.rb b/lib/async/http/protocol/http2/output.rb index 64301896..53200685 100644 --- a/lib/async/http/protocol/http2/output.rb +++ b/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) diff --git a/lib/async/http/protocol/http2/response.rb b/lib/async/http/protocol/http2/response.rb index 1bff4108..83fcf2b8 100644 --- a/lib/async/http/protocol/http2/response.rb +++ b/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? diff --git a/lib/async/http/protocol/http2/stream.rb b/lib/async/http/protocol/http2/stream.rb index 414d90dc..3ab171aa 100644 --- a/lib/async/http/protocol/http2/stream.rb +++ b/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 diff --git a/test/async/http/protocol/http2.rb b/test/async/http/protocol/http2.rb index b33cbd8a..31b7cbc7 100644 --- a/test/async/http/protocol/http2.rb +++ b/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,7 +81,7 @@ 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| @@ -88,7 +89,7 @@ def make_client(endpoint, **options) reactor.async do |task| begin - 100.times do |i| + 1000.times do |i| body.write("Chunk #{i}") sleep (0.01) end @@ -96,7 +97,7 @@ def make_client(endpoint, **options) # 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 diff --git a/test/async/http/protocol/http2/input.rb b/test/async/http/protocol/http2/input.rb new file mode 100644 index 00000000..e6b3d0c2 --- /dev/null +++ b/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 diff --git a/test/async/http/protocol/http2/input_close.rb b/test/async/http/protocol/http2/input_close.rb new file mode 100644 index 00000000..e5892805 --- /dev/null +++ b/test/async/http/protocol/http2/input_close.rb @@ -0,0 +1,57 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "async/http/protocol/http2" +require "async/http/body/hijack" +require "async/promise" +require "sus/fixtures/async/http" + +describe Async::HTTP::Protocol::HTTP2 do + with "closed input and active output" do + include Sus::Fixtures::Async::HTTP::ServerContext + let(:protocol) {subject} + + let(:data) {"Hello World!"} + let(:request_body) {Async::Promise.new} + + let(:app) do + Protocol::HTTP::Middleware.for do |request| + Async::HTTP::Body::Hijack.response(request, 200, {}) do |stream| + stream.write("x" * 128 * 1024) + stream.flush + + request_body.resolve(stream.read(data.bytesize)) + ensure + stream.close + end + end + end + + it "discards incoming data while continuing to write" do + input = Async::HTTP::Body::Writable.new + response = client.connect(authority: "localhost:1", body: input) + + response.body.close + input.write(data) + + current_task = Async::Task.current + received = current_task.with_timeout(1) do + request_body.wait + end + + expect(received).to be == data + input.close_write + + current_task.with_timeout(1) do + current_task.yield while client.pool.busy? + end + + expect(client.pool).not.to be(:busy?) + ensure + input&.close + response&.close + end + end +end diff --git a/test/async/http/proxy.rb b/test/async/http/proxy.rb index e532a0cf..34b99df3 100644 --- a/test/async/http/proxy.rb +++ b/test/async/http/proxy.rb @@ -116,6 +116,59 @@ end end + with "independent tunnel directions" do + let(:request_closed) {Async::Promise.new} + let(:write_response) {Async::Promise.new} + + let(:app) do + Protocol::HTTP::Middleware.for do |request| + Async::HTTP::Body::Hijack.response(request, 200, {}) do |stream| + begin + while stream.read_partial(1024) + end + ensure + request_closed.resolve(true) + end + + write_response.wait + stream.write("Hello World!") + stream.flush + ensure + stream.close + end + end + end + + it "closes the response when forwarding to the closed peer fails" do + proxy = Async::HTTP::Proxy.tcp(client, "localhost", 1) + peer = proxy.connect + + expect(proxy.client.pool).to be(:busy?) + + peer.close + peer = nil + + current_task = Async::Task.current + current_task.with_timeout(1) do + request_closed.wait + end + + expect(proxy.client.pool).to be(:busy?) + + write_response.resolve(true) + + current_task.with_timeout(1) do + current_task.yield while proxy.client.pool.busy? + end + + expect(proxy.client.pool).not.to be(:busy?) + ensure + write_response.resolve(true) unless write_response.resolved? + peer&.close + proxy&.close + end + end + with "proxied client" do let(:app) do Protocol::HTTP::Middleware.for do |request| diff --git a/test/protocol/http/body/streamable.rb b/test/protocol/http/body/streamable.rb index 17f2cef6..424123a7 100644 --- a/test/protocol/http/body/streamable.rb +++ b/test/protocol/http/body/streamable.rb @@ -5,6 +5,7 @@ require "async/http/protocol/http" require "protocol/http/body/streamable" +require "async/promise" require "sus/fixtures/async/http" AnEchoServer = Sus::Shared("an echo server") do @@ -122,6 +123,50 @@ end end +AClosingServer = Sus::Shared("a closing server") do + let(:data) {"Hello World!"} + let(:finished) {Async::Promise.new} + + let(:app) do + ::Protocol::HTTP::Middleware.for do |request| + streamable = ::Protocol::HTTP::Body::Streamable.response(request) do |stream| + stream.write(stream.read(data.bytesize)) + stream.flush + + while stream.read_partial(1024) + end + + finished.resolve(true) + ensure + stream.close + end + + ::Protocol::HTTP::Response[200, {}, streamable] + end + end + + it "should finish normally when the peer closes" do + output = ::Protocol::HTTP::Body::Writable.new + response = client.post("/", body: output) + stream = ::Protocol::HTTP::Body::Stream.new(response.body, output) + + stream.write(data) + expect(stream.read(data.bytesize)).to be == data + + stream.close + stream = nil + + result = Async::Task.current.with_timeout(1) do + finished.wait + end + + expect(result).to be == true + ensure + stream&.close + response&.close + end +end + [Async::HTTP::Protocol::HTTP1, Async::HTTP::Protocol::HTTP2].each do |protocol| describe protocol, unique: protocol.name do include Sus::Fixtures::Async::HTTP::ServerContext @@ -130,5 +175,6 @@ it_behaves_like AnEchoServer it_behaves_like AnEchoClient + it_behaves_like AClosingServer end end