From 1aed35c74f5751a2ef7e437d28c531a577efa39c Mon Sep 17 00:00:00 2001 From: Tavian Barnes Date: Wed, 9 Sep 2026 09:46:40 -0400 Subject: [PATCH 1/2] Prevent Output#passthrough from hanging on cancellation The `ensure` block in `passthrough` attempts to close the stream with `self.close_write(error)`, but there are two issues: - For a cancelled task, `error` will be `nil` since it won't get set by the `rescue => error` block, as `Async::Cancel` isn't a `StandardError` - More seriously, the actual `write()` can block indefinitely, stalling reactor shutdown This led to backtraces like this during Falcon worker shutdown, forcing Falcon to `SIGKILL` the worker process after the timeout: async-2.45.1 async/scheduler.rb:382 IO::Event::Selector::URing#io_write async-2.45.1 async/scheduler.rb:382 Async::Scheduler#io_write io-stream-0.11 io/stream/buffered.rb:112 IO#write io-stream-0.11 io/stream/buffered.rb:112 IO::Stream::Buffered#syswrite io-stream-0.11 io/stream/writable.rb:99 IO::Stream::Writable#drain io-stream-0.11 io/stream/writable.rb:47 block in IO::Stream::Writable#flush io-stream-0.11 io/stream/writable.rb:46 Thread::Mutex#synchronize io-stream-0.11 io/stream/writable.rb:46 IO::Stream::Writable#flush protocol-http2 protocol/http2/framer.rb:58 Protocol::HTTP2::Framer#flush protocol-http2 protocol/http2/connection.rb:253 Protocol::HTTP2::Connection#write_frame protocol-http2 protocol/http2/stream.rb:115 Protocol::HTTP2::Stream#write_frame protocol-http2 protocol/http2/stream.rb:203 Protocol::HTTP2::Stream#write_data protocol-http2 protocol/http2/stream.rb:213 Protocol::HTTP2::Stream#send_data async-http-0.94.2 http2/stream.rb:140 Async::HTTP::Protocol::HTTP2::Stream#finish_output async-http-0.94.2 http2/output.rb:68 Async::HTTP::Protocol::HTTP2::Output#close_write async-http-0.94.2 http2/output.rb:120 Async::HTTP::Protocol::HTTP2::Output#passthrough async-2.45.1 async/task.rb:225 block in Async::Task#run async-2.45.1 async/task.rb:523 block in Async::Task#schedule Fix this with `rescue Exception => error`, and add a 1-second timeout around `close_write` if `error` is non-`nil`. --- lib/async/http/protocol/http2/output.rb | 20 ++++- test/async/http/protocol/http2/output.rb | 107 +++++++++++++++++++++++ 2 files changed, 125 insertions(+), 2 deletions(-) create mode 100644 test/async/http/protocol/http2/output.rb diff --git a/lib/async/http/protocol/http2/output.rb b/lib/async/http/protocol/http2/output.rb index 5320068..c83e923 100644 --- a/lib/async/http/protocol/http2/output.rb +++ b/lib/async/http/protocol/http2/output.rb @@ -11,6 +11,9 @@ module Protocol module HTTP2 # Writes body data to an HTTP/2 stream, respecting flow control windows. class Output + CLOSE_WRITE_TIMEOUT = 1 + private_constant :CLOSE_WRITE_TIMEOUT + # Initialize the output handler. # @parameter stream [Stream] The HTTP/2 stream to write to. # @parameter body [Protocol::HTTP::Body::Readable] The body to read from. @@ -135,7 +138,7 @@ def passthrough(task) # chunk.clear unless chunk.frozen? # GC.start end - rescue => error + rescue Exception => error raise ensure # Ensure the body we are reading from is fully closed: @@ -145,7 +148,20 @@ def passthrough(task) end # Ensure the output of this body is closed: - self.close_write(error) + if error + connection = @stream&.connection + + # Don't block forever if we're shutting down + task.with_timeout(CLOSE_WRITE_TIMEOUT) do + self.close_write(error) + rescue Async::TimeoutError + # Close the socket to avoid Connection#close attempting to flush anything else, and to avoid leaving the connection in an invalid state if our write was interrupted mid-frame: + connection&.stream&.io&.close + connection&.close(error) + end + else + self.close_write(nil) + end end # Send `maximum_size` bytes of data using the specified `stream`. If the buffer has no more chunks, `END_STREAM` will be sent on the final chunk. diff --git a/test/async/http/protocol/http2/output.rb b/test/async/http/protocol/http2/output.rb new file mode 100644 index 0000000..fa816d9 --- /dev/null +++ b/test/async/http/protocol/http2/output.rb @@ -0,0 +1,107 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Tavian Barnes. + +require "async/http/protocol/http2/output" +require "async/http/body/writable" + +require "sus/fixtures/async" + +describe Async::HTTP::Protocol::HTTP2::Output do + include Sus::Fixtures::Async::ReactorContext + + let(:io) do + Class.new do + def initialize + @closed = false + end + + def close + @closed = true + end + + def closed? + @closed + end + end.new + end + + let(:connection) do + Class.new do + def initialize(io) + @stream = Struct.new(:io).new(io) + @closed = false + end + + attr :stream + + def close(error = nil) + @closed = true + end + + def closed? + @closed + end + end.new(io) + end + + let(:stream) do + Class.new do + def initialize(connection) + @connection = connection + @errors = [] + end + + attr :connection + attr :errors + + def finish_output(error = nil) + @errors << error + end + end.new(connection) + end + + let(:body) {Async::HTTP::Body::Writable.new} + + let(:output) {subject.new(stream, body)} + + it "resets the stream when the task is cancelled" do + task = output.start + + task.cancel + task.wait + + expect(stream.errors.size).to be == 1 + expect(stream.errors.first).to be_a(Async::Cancel) + end + + with "a stream which blocks while finishing output" do + let(:stream) do + Class.new do + def initialize(connection) + @connection = connection + end + + attr :connection + + def finish_output(error = nil) + sleep + end + end.new(connection) + end + + it "closes the connection when the task is cancelled" do + task = output.start + + task.cancel + + Async::Task.current.with_timeout(5) do + task.wait + end + + expect(io).to be(:closed?) + expect(connection).to be(:closed?) + end + end +end From 2580b5a9d9626bfa8892b667873342f96d285202 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Thu, 10 Sep 2026 14:09:22 +1200 Subject: [PATCH 2/2] Propagate cancellation through HTTP/2 output Signed-off-by: Samuel Williams Assisted-By: devx/2fb5639b-d636-44ad-a666-9088d8c41cc1 --- lib/async/http/protocol/http2/output.rb | 20 +------ test/async/http/protocol/http2/output.rb | 72 +----------------------- 2 files changed, 5 insertions(+), 87 deletions(-) diff --git a/lib/async/http/protocol/http2/output.rb b/lib/async/http/protocol/http2/output.rb index c83e923..ea7019a 100644 --- a/lib/async/http/protocol/http2/output.rb +++ b/lib/async/http/protocol/http2/output.rb @@ -11,9 +11,6 @@ module Protocol module HTTP2 # Writes body data to an HTTP/2 stream, respecting flow control windows. class Output - CLOSE_WRITE_TIMEOUT = 1 - private_constant :CLOSE_WRITE_TIMEOUT - # Initialize the output handler. # @parameter stream [Stream] The HTTP/2 stream to write to. # @parameter body [Protocol::HTTP::Body::Readable] The body to read from. @@ -138,7 +135,7 @@ def passthrough(task) # chunk.clear unless chunk.frozen? # GC.start end - rescue Exception => error + rescue Async::Cancel, StandardError => error raise ensure # Ensure the body we are reading from is fully closed: @@ -148,20 +145,7 @@ def passthrough(task) end # Ensure the output of this body is closed: - if error - connection = @stream&.connection - - # Don't block forever if we're shutting down - task.with_timeout(CLOSE_WRITE_TIMEOUT) do - self.close_write(error) - rescue Async::TimeoutError - # Close the socket to avoid Connection#close attempting to flush anything else, and to avoid leaving the connection in an invalid state if our write was interrupted mid-frame: - connection&.stream&.io&.close - connection&.close(error) - end - else - self.close_write(nil) - end + self.close_write(error) end # Send `maximum_size` bytes of data using the specified `stream`. If the buffer has no more chunks, `END_STREAM` will be sent on the final chunk. diff --git a/test/async/http/protocol/http2/output.rb b/test/async/http/protocol/http2/output.rb index fa816d9..676affc 100644 --- a/test/async/http/protocol/http2/output.rb +++ b/test/async/http/protocol/http2/output.rb @@ -11,62 +11,25 @@ describe Async::HTTP::Protocol::HTTP2::Output do include Sus::Fixtures::Async::ReactorContext - let(:io) do - Class.new do - def initialize - @closed = false - end - - def close - @closed = true - end - - def closed? - @closed - end - end.new - end - - let(:connection) do - Class.new do - def initialize(io) - @stream = Struct.new(:io).new(io) - @closed = false - end - - attr :stream - - def close(error = nil) - @closed = true - end - - def closed? - @closed - end - end.new(io) - end - let(:stream) do Class.new do - def initialize(connection) - @connection = connection + def initialize @errors = [] end - attr :connection attr :errors def finish_output(error = nil) @errors << error end - end.new(connection) + end.new end let(:body) {Async::HTTP::Body::Writable.new} let(:output) {subject.new(stream, body)} - it "resets the stream when the task is cancelled" do + it "propagates cancellation when the task is cancelled" do task = output.start task.cancel @@ -75,33 +38,4 @@ def finish_output(error = nil) expect(stream.errors.size).to be == 1 expect(stream.errors.first).to be_a(Async::Cancel) end - - with "a stream which blocks while finishing output" do - let(:stream) do - Class.new do - def initialize(connection) - @connection = connection - end - - attr :connection - - def finish_output(error = nil) - sleep - end - end.new(connection) - end - - it "closes the connection when the task is cancelled" do - task = output.start - - task.cancel - - Async::Task.current.with_timeout(5) do - task.wait - end - - expect(io).to be(:closed?) - expect(connection).to be(:closed?) - end - end end