From 071f980e2b85eb367c22093b59e6615b870b131b Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 1 Sep 2026 09:50:51 +1200 Subject: [PATCH 01/13] Close HTTP/2 streams when tunnel peers disconnect Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- lib/async/http/body/pipe.rb | 6 +- lib/async/http/protocol/http2/input.rb | 17 +++++- lib/async/http/protocol/http2/stream.rb | 44 +++++++++++++- test/async/http/protocol/http2/input.rb | 46 +++++++++++++++ test/async/http/protocol/http2/input_close.rb | 57 +++++++++++++++++++ test/async/http/proxy.rb | 38 +++++++++++++ 6 files changed, 202 insertions(+), 6 deletions(-) create mode 100644 test/async/http/protocol/http2/input.rb create mode 100644 test/async/http/protocol/http2/input_close.rb diff --git a/lib/async/http/body/pipe.rb b/lib/async/http/body/pipe.rb index 05b8d23f..9aa1f204 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 here. A full close also stops the reader so both sides of the pipe can finish. def writer(task) @writer = task @@ -75,6 +75,10 @@ def writer(task) rescue => error ensure @output.close_write(error) + + if @tail.closed? + @reader&.stop + end close_head if @reader&.finished? end diff --git a/lib/async/http/protocol/http2/input.rb b/lib/async/http/protocol/http2/input.rb index 34e5ff61..7df4e682 100644 --- a/lib/async/http/protocol/http2/input.rb +++ b/lib/async/http/protocol/http2/input.rb @@ -25,8 +25,10 @@ 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. + if stream = @stream + stream.request_window_update + end end # We track the expected length and check we got what we were expecting. @@ -42,6 +44,17 @@ def read return chunk end + + # Close the input body and notify the stream that incoming data is no longer being consumed. + # @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) + end + end end end end diff --git a/lib/async/http/protocol/http2/stream.rb b/lib/async/http/protocol/http2/stream.rb index 414d90dc..0475e929 100644 --- a/lib/async/http/protocol/http2/stream.rb +++ b/lib/async/http/protocol/http2/stream.rb @@ -27,6 +27,7 @@ def initialize(*) # Input buffer, reading request body, or response body (receive_data): @length = nil @input = nil + @input_closed = false # Output buffer, writing request body or response body (window_updated): @output = nil @@ -114,14 +115,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 +135,17 @@ def process_data(frame) send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR) end + # Called when the application is no longer consuming incoming data. + # @parameter input [Input] The input body being closed. + def finish_input(input) + if @input.equal?(input) + @input = nil + @input_closed = true + + close_if_finished + end + end + # Set the body and begin sending it. def send_body(body, trailer = nil) @output = Output.new(self, body, trailer) @@ -169,6 +184,20 @@ def window_updated(size) return true end + # Send headers and check whether they completed the local side of the stream. + def send_headers(...) + result = super + close_if_finished + return result + end + + # Send data and check whether it completed the local side of the stream. + 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. @@ -192,6 +221,15 @@ def closed(error) return self end + + private + + # Once both application-facing directions are closed, cancel a stream whose remote side remains open. + def close_if_finished + if @input_closed && @state == :half_closed_local + send_reset_stream(::Protocol::HTTP2::Error::CANCEL) + end + end end end 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..b5f88cff --- /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) + @finished_inputs << input + 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] + 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..094bd930 100644 --- a/test/async/http/proxy.rb +++ b/test/async/http/proxy.rb @@ -116,6 +116,44 @@ end end + with "idle tunnel" do + let(:finish_response) {Async::Notification.new} + + let(:app) do + Protocol::HTTP::Middleware.for do |request| + Async::HTTP::Body::Hijack.response(request, 200, {}) do |stream| + while stream.read_partial(1024) + end + + finish_response.wait + ensure + stream.close + end + end + end + + it "releases the proxy connection when the peer is closed" 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 + current_task.yield while proxy.client.pool.busy? + end + + expect(proxy.client.pool).not.to be(:busy?) + ensure + finish_response.signal + peer&.close + proxy&.close + end + end + with "proxied client" do let(:app) do Protocol::HTTP::Middleware.for do |request| From 6c8fcf0c070e6c30d4d1ed6883ad1e08147d22b5 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 1 Sep 2026 10:15:38 +1200 Subject: [PATCH 02/13] Simplify HTTP/2 input window updates Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- lib/async/http/protocol/http2/input.rb | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/lib/async/http/protocol/http2/input.rb b/lib/async/http/protocol/http2/input.rb index 7df4e682..c5236bbb 100644 --- a/lib/async/http/protocol/http2/input.rb +++ b/lib/async/http/protocol/http2/input.rb @@ -26,9 +26,7 @@ def initialize(stream, length) def read if chunk = super # If we read a chunk from the stream, we want to extend the window if required so more data will be provided. - if stream = @stream - stream.request_window_update - end + @stream&.request_window_update end # We track the expected length and check we got what we were expecting. From 26ee0c12e4e3956948ff0c4ad1513681aa86b738 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 1 Sep 2026 10:21:40 +1200 Subject: [PATCH 03/13] Document HTTP/2 input closure state Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- lib/async/http/protocol/http2/stream.rb | 1 + 1 file changed, 1 insertion(+) diff --git a/lib/async/http/protocol/http2/stream.rb b/lib/async/http/protocol/http2/stream.rb index 0475e929..856b38c2 100644 --- a/lib/async/http/protocol/http2/stream.rb +++ b/lib/async/http/protocol/http2/stream.rb @@ -27,6 +27,7 @@ def initialize(*) # Input buffer, reading request body, or response body (receive_data): @length = nil @input = nil + # `@input.nil?` can mean the input has not been prepared, the peer ended the stream without a body, or the stream has closed. Track explicit application closure separately. @input_closed = false # Output buffer, writing request body or response body (window_updated): From 75baf31bef2940415462fdd0eb34e183304dd5f1 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 1 Sep 2026 10:22:26 +1200 Subject: [PATCH 04/13] Expand HTTP/2 input state documentation Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- lib/async/http/protocol/http2/stream.rb | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/lib/async/http/protocol/http2/stream.rb b/lib/async/http/protocol/http2/stream.rb index 856b38c2..29630b9e 100644 --- a/lib/async/http/protocol/http2/stream.rb +++ b/lib/async/http/protocol/http2/stream.rb @@ -27,7 +27,14 @@ def initialize(*) # Input buffer, reading request body, or response body (receive_data): @length = nil @input = nil - # `@input.nil?` can mean the input has not been prepared, the peer ended the stream without a body, or the stream has closed. Track explicit application closure separately. + + # `@input.nil?` does not tell us why input is absent: + # + # - The input has not been prepared yet. + # - The peer ended its sending side without a body. + # - The stream has already closed. + # + # `@input_closed` specifically records that the application explicitly closed the input while the peer may still be sending. This prevents a bodyless request from being cancelled before its response arrives. @input_closed = false # Output buffer, writing request body or response body (window_updated): From 16dc9c958a13bc7ab91a6d43a67d0a76262e502e Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 1 Sep 2026 10:23:53 +1200 Subject: [PATCH 05/13] Document HTTP/2 input closure lifecycle Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- lib/async/http/protocol/http2/input.rb | 2 +- lib/async/http/protocol/http2/stream.rb | 14 ++++++-------- 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/lib/async/http/protocol/http2/input.rb b/lib/async/http/protocol/http2/input.rb index c5236bbb..33eee510 100644 --- a/lib/async/http/protocol/http2/input.rb +++ b/lib/async/http/protocol/http2/input.rb @@ -43,7 +43,7 @@ def read return chunk end - # Close the input body and notify the stream that incoming data is no longer being consumed. + # Close the application-facing input body and notify the stream that incoming data is no longer being consumed. The HTTP/2 stream may remain open until local output has finished. # @parameter error [Exception | Nil] The error that caused the input to be closed, if any. def close(error = nil) super diff --git a/lib/async/http/protocol/http2/stream.rb b/lib/async/http/protocol/http2/stream.rb index 29630b9e..8809ff94 100644 --- a/lib/async/http/protocol/http2/stream.rb +++ b/lib/async/http/protocol/http2/stream.rb @@ -28,13 +28,11 @@ def initialize(*) @length = nil @input = nil - # `@input.nil?` does not tell us why input is absent: + # Application input closure is independent of the HTTP/2 wire state: # - # - The input has not been prepared yet. - # - The peer ended its sending side without a body. - # - The stream has already closed. - # - # `@input_closed` specifically records that the application explicitly closed the input while the peer may still be sending. This prevents a bodyless request from being cancelled before its response arrives. + # - Closing the input means the application will no longer consume incoming data, but it cannot immediately reset the stream because `RST_STREAM` would also close active local output. + # - Incoming data is discarded and flow-control credit is restored to preserve progress for bidirectional applications while local output remains active. + # - Once local output sends `END_STREAM`, the remaining remote half is cancelled with `RST_STREAM(CANCEL)` because neither application-facing direction remains in use. @input_closed = false # Output buffer, writing request body or response body (window_updated): @@ -143,7 +141,7 @@ def process_data(frame) send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR) end - # Called when the application is no longer consuming incoming data. + # Record that the application is no longer consuming incoming data and cancel the stream if local output has already finished. # @parameter input [Input] The input body being closed. def finish_input(input) if @input.equal?(input) @@ -232,7 +230,7 @@ def closed(error) private - # Once both application-facing directions are closed, cancel a stream whose remote side remains open. + # Once both application-facing directions are closed, cancel the remaining remote half of the stream. def close_if_finished if @input_closed && @state == :half_closed_local send_reset_stream(::Protocol::HTTP2::Error::CANCEL) From 7c83f81974848a6d34807a693ffb273085314c8f Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 1 Sep 2026 10:28:22 +1200 Subject: [PATCH 06/13] Fix blank line indentation Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- lib/async/http/body/pipe.rb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/async/http/body/pipe.rb b/lib/async/http/body/pipe.rb index 9aa1f204..becacb29 100644 --- a/lib/async/http/body/pipe.rb +++ b/lib/async/http/body/pipe.rb @@ -75,7 +75,7 @@ def writer(task) rescue => error ensure @output.close_write(error) - + if @tail.closed? @reader&.stop end From 32334b3ae2297703a89dd852e1b0d86476599ba4 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 1 Sep 2026 10:50:50 +1200 Subject: [PATCH 07/13] Preserve orderly HTTP/2 duplex shutdown Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- lib/async/http/body/pipe.rb | 5 ++- lib/async/http/protocol/http2/input.rb | 6 ++-- lib/async/http/protocol/http2/stream.rb | 34 +++++++++++++------ test/async/http/protocol/http2/input.rb | 6 ++-- test/async/http/protocol/http2/input_close.rb | 31 +++++++++++++++++ 5 files changed, 66 insertions(+), 16 deletions(-) diff --git a/lib/async/http/body/pipe.rb b/lib/async/http/body/pipe.rb index becacb29..1e4516e9 100644 --- a/lib/async/http/body/pipe.rb +++ b/lib/async/http/body/pipe.rb @@ -55,6 +55,8 @@ def reader(task) end @head.close_write + rescue Async::Stop => error + raise rescue => error ensure @input.close(error) @@ -63,7 +65,8 @@ def reader(task) end # Read from the head of the pipe and write to the @output stream. - # A write-side close on @tail produces EOF here. A full close also stops the reader so both sides of the pipe can finish. + # A write-side close on @tail produces EOF here. A full close also + # stops the reader so both sides of the pipe can finish. def writer(task) @writer = task diff --git a/lib/async/http/protocol/http2/input.rb b/lib/async/http/protocol/http2/input.rb index 33eee510..2a84d9a0 100644 --- a/lib/async/http/protocol/http2/input.rb +++ b/lib/async/http/protocol/http2/input.rb @@ -43,14 +43,16 @@ def read return chunk end - # Close the application-facing input body and notify the stream that incoming data is no longer being consumed. The HTTP/2 stream may remain open until local output has finished. + # Close the application-facing input body and notify the stream that + # incoming data is no longer being consumed. The HTTP/2 stream may + # remain open until local output has finished. # @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) + stream.finish_input(self, error) end end end diff --git a/lib/async/http/protocol/http2/stream.rb b/lib/async/http/protocol/http2/stream.rb index 8809ff94..7c888c5d 100644 --- a/lib/async/http/protocol/http2/stream.rb +++ b/lib/async/http/protocol/http2/stream.rb @@ -30,10 +30,16 @@ def initialize(*) # Application input closure is independent of the HTTP/2 wire state: # - # - Closing the input means the application will no longer consume incoming data, but it cannot immediately reset the stream because `RST_STREAM` would also close active local output. - # - Incoming data is discarded and flow-control credit is restored to preserve progress for bidirectional applications while local output remains active. - # - Once local output sends `END_STREAM`, the remaining remote half is cancelled with `RST_STREAM(CANCEL)` because neither application-facing direction remains in use. - @input_closed = false + # - Closing the input while local output remains active allows an + # orderly bidirectional shutdown. A reset would also discard that + # active output. + # - Incoming data is discarded and flow-control credit is restored to + # preserve progress for bidirectional applications while local output + # remains active. + # - If input is closed due to an error, or after local output has already + # finished, the remaining remote half is abandoned and cancelled with + # `RST_STREAM(CANCEL)`. + @input_abandoned = false # Output buffer, writing request body or response body (window_updated): @output = nil @@ -130,7 +136,8 @@ def process_data(frame) input.close_write end else - # The application has closed the input, so discard incoming data while maintaining flow control for the stream. + # The application has closed the input, so discard incoming data while + # maintaining flow control for the stream. request_window_update end @@ -141,12 +148,18 @@ def process_data(frame) send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR) end - # Record that the application is no longer consuming incoming data and cancel the stream if local output has already finished. + # Record that the application is no longer consuming incoming data. + # Preserve active local output during an orderly close, but arrange to + # cancel the remote half if the input was closed due to an error. # @parameter input [Input] The input body being closed. - def finish_input(input) + # @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 || @state == :half_closed_local + @input_abandoned = true + end close_if_finished end @@ -230,9 +243,10 @@ def closed(error) private - # Once both application-facing directions are closed, cancel the remaining remote half of the stream. + # Once local output has finished, cancel an abandoned remote half of + # the stream. def close_if_finished - if @input_closed && @state == :half_closed_local + if @input_abandoned && @state == :half_closed_local send_reset_stream(::Protocol::HTTP2::Error::CANCEL) end end diff --git a/test/async/http/protocol/http2/input.rb b/test/async/http/protocol/http2/input.rb index b5f88cff..e6b3d0c2 100644 --- a/test/async/http/protocol/http2/input.rb +++ b/test/async/http/protocol/http2/input.rb @@ -20,8 +20,8 @@ def request_window_update @window_updates += 1 end - def finish_input(input) - @finished_inputs << input + def finish_input(input, error = nil) + @finished_inputs << [input, error] end end.new end @@ -41,6 +41,6 @@ def finish_input(input) input.close(error) input.close - expect(stream.finished_inputs).to be == [input] + 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 index e5892805..11464702 100644 --- a/test/async/http/protocol/http2/input_close.rb +++ b/test/async/http/protocol/http2/input_close.rb @@ -9,6 +9,37 @@ require "sus/fixtures/async/http" describe Async::HTTP::Protocol::HTTP2 do + with "orderly bidirectional shutdown" do + include Sus::Fixtures::Async::HTTP::ServerContext + let(:protocol) {subject} + + let(:data) {"Hello World!"} + + let(:app) do + Protocol::HTTP::Middleware.for do |request| + Async::HTTP::Body::Hijack.response(request, 200, {}) do |stream| + stream.read(data.bytesize) + stream.write(data) + ensure + stream.close + end + end + end + + it "allows the peer to finish normally" do + input = Async::HTTP::Body::Writable.new + response = client.connect(authority: "localhost:1", body: input) + + input.write(data) + + expect(response.body.read).to be == data + expect(response.body.read).to be_nil + ensure + input&.close + response&.close + end + end + with "closed input and active output" do include Sus::Fixtures::Async::HTTP::ServerContext let(:protocol) {subject} From 801850a5bc3f3d5b13bea7c8b2a6337ec0e27961 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 1 Sep 2026 12:04:43 +1200 Subject: [PATCH 08/13] Clarify HTTP/2 stream close semantics Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- lib/async/http/body/pipe.rb | 2 - lib/async/http/protocol/http2/input.rb | 4 +- lib/async/http/protocol/http2/response.rb | 13 +++++ lib/async/http/protocol/http2/stream.rb | 50 +++---------------- test/async/http/protocol/http2.rb | 6 ++- test/async/http/protocol/http2/input_close.rb | 33 +++++++++--- test/async/http/proxy.rb | 8 +++ 7 files changed, 58 insertions(+), 58 deletions(-) diff --git a/lib/async/http/body/pipe.rb b/lib/async/http/body/pipe.rb index 1e4516e9..df5bf2ca 100644 --- a/lib/async/http/body/pipe.rb +++ b/lib/async/http/body/pipe.rb @@ -55,8 +55,6 @@ def reader(task) end @head.close_write - rescue Async::Stop => error - raise rescue => error ensure @input.close(error) diff --git a/lib/async/http/protocol/http2/input.rb b/lib/async/http/protocol/http2/input.rb index 2a84d9a0..c4841c8b 100644 --- a/lib/async/http/protocol/http2/input.rb +++ b/lib/async/http/protocol/http2/input.rb @@ -44,8 +44,8 @@ def read end # Close the application-facing input body and notify the stream that - # incoming data is no longer being consumed. The HTTP/2 stream may - # remain open until local output has finished. + # incoming data is no longer being consumed. The HTTP/2 stream remains + # open until the peer finishes unless its owner explicitly cancels it. # @parameter error [Exception | Nil] The error that caused the input to be closed, if any. def close(error = nil) super diff --git a/lib/async/http/protocol/http2/response.rb b/lib/async/http/protocol/http2/response.rb index 1bff4108..2b7adda8 100644 --- a/lib/async/http/protocol/http2/response.rb +++ b/lib/async/http/protocol/http2/response.rb @@ -169,6 +169,19 @@ 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) + body = @body + super + + if body && !@stream.closed? + code = error ? ::Protocol::HTTP2::Error::INTERNAL_ERROR : ::Protocol::HTTP2::Error::CANCEL + @stream.send_reset_stream(code) + end + 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 7c888c5d..679678c5 100644 --- a/lib/async/http/protocol/http2/stream.rb +++ b/lib/async/http/protocol/http2/stream.rb @@ -28,19 +28,6 @@ def initialize(*) @length = nil @input = nil - # Application input closure is independent of the HTTP/2 wire state: - # - # - Closing the input while local output remains active allows an - # orderly bidirectional shutdown. A reset would also discard that - # active output. - # - Incoming data is discarded and flow-control credit is restored to - # preserve progress for bidirectional applications while local output - # remains active. - # - If input is closed due to an error, or after local output has already - # finished, the remaining remote half is abandoned and cancelled with - # `RST_STREAM(CANCEL)`. - @input_abandoned = false - # Output buffer, writing request body or response body (window_updated): @output = nil end @@ -148,20 +135,19 @@ def process_data(frame) send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR) end - # Record that the application is no longer consuming incoming data. - # Preserve active local output during an orderly close, but arrange to - # cancel the remote half if the input was closed due to an error. + # Detach the application-facing input body from the wire stream. + # Incoming data will be discarded while maintaining flow control until + # the peer sends `END_STREAM`. Cancellation is an explicit operation + # performed by the request or response which owns the exchange. # @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 - if error || @state == :half_closed_local - @input_abandoned = true + if error + send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR) end - - close_if_finished end end @@ -203,20 +189,6 @@ def window_updated(size) return true end - # Send headers and check whether they completed the local side of the stream. - def send_headers(...) - result = super - close_if_finished - return result - end - - # Send data and check whether it completed the local side of the stream. - 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. @@ -240,16 +212,6 @@ def closed(error) return self end - - private - - # Once local output has finished, cancel an abandoned remote half of - # the stream. - def close_if_finished - if @input_abandoned && @state == :half_closed_local - send_reset_stream(::Protocol::HTTP2::Error::CANCEL) - end - end end end end diff --git a/test/async/http/protocol/http2.rb b/test/async/http/protocol/http2.rb index b33cbd8a..cf44a275 100644 --- a/test/async/http/protocol/http2.rb +++ b/test/async/http/protocol/http2.rb @@ -88,7 +88,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 @@ -115,7 +115,9 @@ def make_client(endpoint, **options) response.close - notification.wait + Async::Task.current.with_timeout(1) do + notification.wait + end expect(response.stream.connection).to be(:reusable?) end diff --git a/test/async/http/protocol/http2/input_close.rb b/test/async/http/protocol/http2/input_close.rb index 11464702..e41a1cab 100644 --- a/test/async/http/protocol/http2/input_close.rb +++ b/test/async/http/protocol/http2/input_close.rb @@ -13,29 +13,46 @@ include Sus::Fixtures::Async::HTTP::ServerContext let(:protocol) {subject} - let(:data) {"Hello World!"} + let(:request_closed) {Async::Notification.new} + let(:finish_response) {Async::Notification.new} let(:app) do Protocol::HTTP::Middleware.for do |request| Async::HTTP::Body::Hijack.response(request, 200, {}) do |stream| - stream.read(data.bytesize) - stream.write(data) + while stream.read_partial(1024) + end + + request_closed.signal + finish_response.wait ensure stream.close end end end - it "allows the peer to finish normally" do + it "retains the stream until the peer finishes" do input = Async::HTTP::Body::Writable.new response = client.connect(authority: "localhost:1", body: input) + stream = Protocol::HTTP::Body::Stream.new(response.body, input) - input.write(data) + stream.close + + current_task = Async::Task.current + current_task.with_timeout(1) do + request_closed.wait + end - expect(response.body.read).to be == data - expect(response.body.read).to be_nil + expect(client.pool).to be(:busy?) + finish_response.signal + + current_task.with_timeout(1) do + current_task.yield while client.pool.busy? + end + + expect(client.pool).not.to be(:busy?) ensure - input&.close + finish_response.signal + stream&.close response&.close end end diff --git a/test/async/http/proxy.rb b/test/async/http/proxy.rb index 094bd930..faf7fef7 100644 --- a/test/async/http/proxy.rb +++ b/test/async/http/proxy.rb @@ -117,6 +117,7 @@ end with "idle tunnel" do + let(:request_closed) {Async::Notification.new} let(:finish_response) {Async::Notification.new} let(:app) do @@ -125,6 +126,7 @@ while stream.read_partial(1024) end + request_closed.signal finish_response.wait ensure stream.close @@ -142,6 +144,12 @@ peer = nil current_task = Async::Task.current + current_task.with_timeout(1) do + request_closed.wait + end + + finish_response.signal + current_task.with_timeout(1) do current_task.yield while proxy.client.pool.busy? end From 6c65456009697a7cec09df0f950a1017bf7a138d Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 1 Sep 2026 13:44:42 +1200 Subject: [PATCH 09/13] Complete orderly HTTP/2 stream closure Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- lib/async/http/protocol/http2/input.rb | 5 +- lib/async/http/protocol/http2/output.rb | 32 ++++++++--- lib/async/http/protocol/http2/response.rb | 13 +++-- lib/async/http/protocol/http2/stream.rb | 55 +++++++++++++++++-- test/async/http/protocol/http2/input_close.rb | 48 ---------------- test/async/http/proxy.rb | 10 ++-- test/protocol/http/body/streamable.rb | 46 ++++++++++++++++ 7 files changed, 137 insertions(+), 72 deletions(-) diff --git a/lib/async/http/protocol/http2/input.rb b/lib/async/http/protocol/http2/input.rb index c4841c8b..566847aa 100644 --- a/lib/async/http/protocol/http2/input.rb +++ b/lib/async/http/protocol/http2/input.rb @@ -44,8 +44,9 @@ def read end # Close the application-facing input body and notify the stream that - # incoming data is no longer being consumed. The HTTP/2 stream remains - # open until the peer finishes unless its owner explicitly cancels it. + # 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 diff --git a/lib/async/http/protocol/http2/output.rb b/lib/async/http/protocol/http2/output.rb index 64301896..2a4fab99 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,22 @@ 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 +155,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 2b7adda8..0583ceab 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. @@ -173,13 +175,12 @@ def wait # 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) - body = @body - super - - if body && !@stream.closed? + 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. diff --git a/lib/async/http/protocol/http2/stream.rb b/lib/async/http/protocol/http2/stream.rb index 679678c5..bbd03f36 100644 --- a/lib/async/http/protocol/http2/stream.rb +++ b/lib/async/http/protocol/http2/stream.rb @@ -28,6 +28,12 @@ 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 @@ -135,18 +141,21 @@ def process_data(frame) send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR) end - # Detach the application-facing input body from the wire stream. - # Incoming data will be discarded while maintaining flow control until - # the peer sends `END_STREAM`. Cancellation is an explicit operation - # performed by the request or response which owns the exchange. + # 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 @@ -189,11 +198,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 @@ -203,7 +233,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 @@ -212,6 +247,16 @@ 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/input_close.rb b/test/async/http/protocol/http2/input_close.rb index e41a1cab..e5892805 100644 --- a/test/async/http/protocol/http2/input_close.rb +++ b/test/async/http/protocol/http2/input_close.rb @@ -9,54 +9,6 @@ require "sus/fixtures/async/http" describe Async::HTTP::Protocol::HTTP2 do - with "orderly bidirectional shutdown" do - include Sus::Fixtures::Async::HTTP::ServerContext - let(:protocol) {subject} - - let(:request_closed) {Async::Notification.new} - let(:finish_response) {Async::Notification.new} - - let(:app) do - Protocol::HTTP::Middleware.for do |request| - Async::HTTP::Body::Hijack.response(request, 200, {}) do |stream| - while stream.read_partial(1024) - end - - request_closed.signal - finish_response.wait - ensure - stream.close - end - end - end - - it "retains the stream until the peer finishes" do - input = Async::HTTP::Body::Writable.new - response = client.connect(authority: "localhost:1", body: input) - stream = Protocol::HTTP::Body::Stream.new(response.body, input) - - stream.close - - current_task = Async::Task.current - current_task.with_timeout(1) do - request_closed.wait - end - - expect(client.pool).to be(:busy?) - finish_response.signal - - current_task.with_timeout(1) do - current_task.yield while client.pool.busy? - end - - expect(client.pool).not.to be(:busy?) - ensure - finish_response.signal - stream&.close - response&.close - end - end - with "closed input and active output" do include Sus::Fixtures::Async::HTTP::ServerContext let(:protocol) {subject} diff --git a/test/async/http/proxy.rb b/test/async/http/proxy.rb index faf7fef7..3cb5bfa6 100644 --- a/test/async/http/proxy.rb +++ b/test/async/http/proxy.rb @@ -123,10 +123,13 @@ let(:app) do Protocol::HTTP::Middleware.for do |request| Async::HTTP::Body::Hijack.response(request, 200, {}) do |stream| - while stream.read_partial(1024) + begin + while stream.read_partial(1024) + end + ensure + request_closed.signal end - request_closed.signal finish_response.wait ensure stream.close @@ -148,13 +151,12 @@ request_closed.wait end - finish_response.signal - current_task.with_timeout(1) do current_task.yield while proxy.client.pool.busy? end expect(proxy.client.pool).not.to be(:busy?) + finish_response.signal ensure finish_response.signal peer&.close 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 From ac30c582a064a0433801534abefb6a27933c83a4 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 1 Sep 2026 16:00:30 +1200 Subject: [PATCH 10/13] Use a promise for response completion Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- test/async/http/protocol/http2.rb | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/test/async/http/protocol/http2.rb b/test/async/http/protocol/http2.rb index cf44a275..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| @@ -96,7 +97,7 @@ def make_client(endpoint, **options) # puts "Response generation failed: #{$!}" ensure body.close - notification.signal + finished.resolve(true) end end @@ -115,9 +116,7 @@ def make_client(endpoint, **options) response.close - Async::Task.current.with_timeout(1) do - notification.wait - end + finished.wait(timeout: 1) expect(response.stream.connection).to be(:reusable?) end From 2d0e397631b2b0a2688edc436c7acc541b470e28 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 1 Sep 2026 21:40:57 +1200 Subject: [PATCH 11/13] Preserve independent pipe directions Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- lib/async/http/body/pipe.rb | 8 ++------ test/async/http/proxy.rb | 19 ++++++++++++------- 2 files changed, 14 insertions(+), 13 deletions(-) diff --git a/lib/async/http/body/pipe.rb b/lib/async/http/body/pipe.rb index df5bf2ca..bef5ead6 100644 --- a/lib/async/http/body/pipe.rb +++ b/lib/async/http/body/pipe.rb @@ -63,8 +63,8 @@ def reader(task) end # Read from the head of the pipe and write to the @output stream. - # A write-side close on @tail produces EOF here. A full close also - # stops the reader so both sides of the pipe can finish. + # A write-side close on @tail produces EOF and closes @output + # independently of the input direction. def writer(task) @writer = task @@ -77,10 +77,6 @@ def writer(task) ensure @output.close_write(error) - if @tail.closed? - @reader&.stop - end - close_head if @reader&.finished? end diff --git a/test/async/http/proxy.rb b/test/async/http/proxy.rb index 3cb5bfa6..a4e4a0fa 100644 --- a/test/async/http/proxy.rb +++ b/test/async/http/proxy.rb @@ -117,8 +117,8 @@ end with "idle tunnel" do - let(:request_closed) {Async::Notification.new} - let(:finish_response) {Async::Notification.new} + let(:request_closed) {Async::Promise.new} + let(:write_response) {Async::Promise.new} let(:app) do Protocol::HTTP::Middleware.for do |request| @@ -127,17 +127,19 @@ while stream.read_partial(1024) end ensure - request_closed.signal + request_closed.resolve(true) end - finish_response.wait + write_response.wait + stream.write("Hello World!") + stream.flush ensure stream.close end end end - it "releases the proxy connection when the peer is closed" do + it "releases the proxy connection when response data arrives after the peer is closed" do proxy = Async::HTTP::Proxy.tcp(client, "localhost", 1) peer = proxy.connect @@ -151,14 +153,17 @@ 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?) - finish_response.signal ensure - finish_response.signal + write_response.resolve(true) unless write_response.resolved? peer&.close proxy&.close end From 7368584988961b34dd7ce948594fb63f46c036f4 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 1 Sep 2026 22:06:44 +1200 Subject: [PATCH 12/13] Clarify independent tunnel direction test Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- test/async/http/proxy.rb | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/async/http/proxy.rb b/test/async/http/proxy.rb index a4e4a0fa..34b99df3 100644 --- a/test/async/http/proxy.rb +++ b/test/async/http/proxy.rb @@ -116,7 +116,7 @@ end end - with "idle tunnel" do + with "independent tunnel directions" do let(:request_closed) {Async::Promise.new} let(:write_response) {Async::Promise.new} @@ -139,7 +139,7 @@ end end - it "releases the proxy connection when response data arrives after the peer is closed" do + it "closes the response when forwarding to the closed peer fails" do proxy = Async::HTTP::Proxy.tcp(client, "localhost", 1) peer = proxy.connect From 042356369618f1bce77a50d8592d98ca07ae5ce6 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 1 Sep 2026 22:09:18 +1200 Subject: [PATCH 13/13] Soft wrap HTTP/2 lifecycle comments Assisted-By: devx/618580b0-d55f-4c2e-95b6-87e648f60543 --- lib/async/http/body/pipe.rb | 3 +-- lib/async/http/protocol/http2/input.rb | 5 +---- lib/async/http/protocol/http2/output.rb | 5 +---- lib/async/http/protocol/http2/response.rb | 3 +-- lib/async/http/protocol/http2/stream.rb | 16 ++++------------ 5 files changed, 8 insertions(+), 24 deletions(-) diff --git a/lib/async/http/body/pipe.rb b/lib/async/http/body/pipe.rb index bef5ead6..da221778 100644 --- a/lib/async/http/body/pipe.rb +++ b/lib/async/http/body/pipe.rb @@ -63,8 +63,7 @@ def reader(task) end # Read from the head of the pipe and write to the @output stream. - # A write-side close on @tail produces EOF and closes @output - # independently of the input direction. + # 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 566847aa..7a50ba82 100644 --- a/lib/async/http/protocol/http2/input.rb +++ b/lib/async/http/protocol/http2/input.rb @@ -43,10 +43,7 @@ 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. + # 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 diff --git a/lib/async/http/protocol/http2/output.rb b/lib/async/http/protocol/http2/output.rb index 2a4fab99..53200685 100644 --- a/lib/async/http/protocol/http2/output.rb +++ b/lib/async/http/protocol/http2/output.rb @@ -98,10 +98,7 @@ 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. + # 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 diff --git a/lib/async/http/protocol/http2/response.rb b/lib/async/http/protocol/http2/response.rb index 0583ceab..83fcf2b8 100644 --- a/lib/async/http/protocol/http2/response.rb +++ b/lib/async/http/protocol/http2/response.rb @@ -171,8 +171,7 @@ 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. + # 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? diff --git a/lib/async/http/protocol/http2/stream.rb b/lib/async/http/protocol/http2/stream.rb index bbd03f36..3ab171aa 100644 --- a/lib/async/http/protocol/http2/stream.rb +++ b/lib/async/http/protocol/http2/stream.rb @@ -28,10 +28,7 @@ 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. + # 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): @@ -129,8 +126,7 @@ def process_data(frame) input.close_write end else - # The application has closed the input, so discard incoming data while - # maintaining flow control for the stream. + # The application has closed the input, so discard incoming data while maintaining flow control for the stream. request_window_update end @@ -141,10 +137,7 @@ 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. + # 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) @@ -250,8 +243,7 @@ def closed(error) private - # If both application-facing directions are closed but the peer has not - # finished, terminate the remaining wire stream without an error. + # 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)