From c4d6323bddd5452c8723d0d9d001c5060b7e64a1 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Thu, 6 Aug 2026 21:41:54 +1200 Subject: [PATCH] Preserve streaming body terminal errors --- lib/protocol/http/body/readable.rb | 25 +++++--- lib/protocol/http/body/reader.rb | 19 +++--- lib/protocol/http/body/stream.rb | 9 +++ lib/protocol/http/body/streamable.rb | 7 ++- lib/protocol/http/body/writable.rb | 53 ++++++++--------- test/protocol/http/body/readable.rb | 80 +++++++++++++++++++++++++ test/protocol/http/body/reader.rb | 32 ++++++++++ test/protocol/http/body/stream.rb | 14 +++++ test/protocol/http/body/streamable.rb | 25 ++++++++ test/protocol/http/body/writable.rb | 86 +++++++++++++++++++++++++++ 10 files changed, 302 insertions(+), 48 deletions(-) diff --git a/lib/protocol/http/body/readable.rb b/lib/protocol/http/body/readable.rb index aa03731b..c2150b71 100644 --- a/lib/protocol/http/body/readable.rb +++ b/lib/protocol/http/body/readable.rb @@ -138,17 +138,24 @@ def stream? # @parameter stream [IO | Object] An `IO`-like object that responds to `#read`, `#write` and `#flush`. # @returns [Boolean] Whether the ownership of the stream was transferred. def call(stream) - self.each do |chunk| - stream.write(chunk) - - # Flush the stream unless we are immediately expecting more data: - unless self.ready? - stream.flush + begin + self.each do |chunk| + stream.write(chunk) + + # Flush the stream unless we are immediately expecting more data: + unless self.ready? + stream.flush + end + end + rescue => error + raise + ensure + if error and stream.respond_to?(:close_with_error) + stream.close_with_error(error) + else + stream.close end end - ensure - # TODO Should this invoke close_write(error) instead? - stream.close end # Read all remaining chunks into a buffered body and close the underlying input. diff --git a/lib/protocol/http/body/reader.rb b/lib/protocol/http/body/reader.rb index c04fbc39..105814cb 100644 --- a/lib/protocol/http/body/reader.rb +++ b/lib/protocol/http/body/reader.rb @@ -15,9 +15,9 @@ module Reader # # @yields {|chunk| ...} chunks from the body. def each(&block) - if @body - @body.each(&block) + if body = @body @body = nil + body.each(&block) end end @@ -25,11 +25,10 @@ def each(&block) # # @returns [String] the entire body as a string. def read - if @body - buffer = @body.join + if body = @body @body = nil - return buffer + return body.join end end @@ -37,11 +36,10 @@ def read # # @returns [Buffered] buffers the entire body. def finish - if @body - body = @body.finish + if body = @body @body = nil - return body + return body.finish end end @@ -59,8 +57,9 @@ def discard # # @returns [Reader] itself. def buffered! - if @body - @body = @body.finish + if body = @body + @body = nil + @body = body.finish end # TODO Should this return @body instead? It seems more useful. diff --git a/lib/protocol/http/body/stream.rb b/lib/protocol/http/body/stream.rb index 72ba02e9..3025f661 100644 --- a/lib/protocol/http/body/stream.rb +++ b/lib/protocol/http/body/stream.rb @@ -375,6 +375,15 @@ def close_write(error = nil) end end + # Close the stream because an error occurred. + # + # This explicit operation allows IO-compatible consumers to propagate an error without passing an argument to {close}. + # + # @parameter error [Exception] The error that caused this stream to be closed. + def close_with_error(error) + self.close(error) + end + # Close the input and output bodies. # # @parameter error [Exception | Nil] The error that caused this stream to be closed, if any. diff --git a/lib/protocol/http/body/streamable.rb b/lib/protocol/http/body/streamable.rb index 20e95af3..11c75bcf 100644 --- a/lib/protocol/http/body/streamable.rb +++ b/lib/protocol/http/body/streamable.rb @@ -129,7 +129,12 @@ def call(stream) block.call(stream) rescue => error # If, for some reason, the block raises an error, we assume it may not have closed the stream, so we close it here: - stream.close + if stream.respond_to?(:close_with_error) + stream.close_with_error(error) + else + stream.close + end + raise end diff --git a/lib/protocol/http/body/writable.rb b/lib/protocol/http/body/writable.rb index f19823fa..2028d8f1 100644 --- a/lib/protocol/http/body/writable.rb +++ b/lib/protocol/http/body/writable.rb @@ -21,6 +21,7 @@ class Closed < StandardError def initialize(length = nil, queue: Thread::Queue.new) @length = length @queue = queue + @mutex = Thread::Mutex.new @count = 0 @error = nil end @@ -31,19 +32,23 @@ def initialize(length = nil, queue: Thread::Queue.new) # @attribute [Integer] The number of chunks written to the body. attr :count - # Stop generating output; cause the next call to write to fail with the given error. Does not prevent existing chunks from being read. In other words, this indicates both that no more data will be or should be written to the body. + # Stop consuming the body and discard any unread chunks. Future writes will fail with the given error, or {Closed} if no error is given. # - # @parameter error [Exception] The error that caused this body to be closed, if any. Will be raised on the next call to {read}. + # @parameter error [Exception | Nil] The error that caused this body to be closed, if any. def close(error = nil) - @error ||= error - - @queue.clear - @queue.close + @mutex.synchronize do + unless @queue.closed? + @error = error + @queue.close + end + + @queue.clear + end super end - # Whether the body is closed. A closed body can not be written to or read from. + # Whether the body is closed for writing. Buffered chunks may still be read. # # @returns [Boolean] Whether the body is closed. def closed? @@ -59,7 +64,7 @@ def ready? # # @returns [Boolean] Whether the body is empty. def empty? - @queue.empty? && @queue.closed? + @error.nil? && @queue.empty? && @queue.closed? end # Read the next available chunk. @@ -67,14 +72,9 @@ def empty? # @returns [String | Nil] The next chunk, or `nil` if the body is finished. # @raises [Exception] If the body was closed due to an error. def read - if @error - raise @error - end - - # This operation may result in @error being set. chunk = @queue.pop - if @error + if chunk.nil? and @error raise @error end @@ -87,20 +87,22 @@ def read # @raises [Closed] If the body has been closed without error. # @raises [Exception] If the body has been closed due to an error. def write(chunk) - if @queue.closed? - raise(@error || Closed) - end - @queue.push(chunk) @count += 1 + rescue ClosedQueueError + raise(@error || Closed) end # Signal that no more data will be written to the body. # # @parameter error [Exception] The error that caused this body to be closed, if any. def close_write(error = nil) - @error ||= error - @queue.close + @mutex.synchronize do + unless @queue.closed? + @error = error + @queue.close + end + end end # The output interface for writing chunks to the body. @@ -127,23 +129,18 @@ def write(chunk) # Close the output stream. # - # If an error is given, the error will be used to close the body by invoking {close} with the error. Otherwise, only the write side of the body will be closed. + # If an error is given, it will be raised by the reader after all buffered chunks have been consumed. # # @parameter error [Exception | Nil] The error that caused this stream to be closed, if any. def close(error = nil) @closed = true - - if error - @writable.close(error) - else - @writable.close_write - end + @writable.close_write(error) end end # Create an output wrapper which can be used to write chunks to the body. # - # If a block is given, and the block raises an error, the error will used to close the body by invoking {close} with the error. + # If a block is given, and the block raises an error, the reader will receive all buffered chunks followed by that error. # # @yields {|output| ...} if a block is given. # @parameter output [Output] The output wrapper. diff --git a/test/protocol/http/body/readable.rb b/test/protocol/http/body/readable.rb index f5cb9e6f..215b1b68 100644 --- a/test/protocol/http/body/readable.rb +++ b/test/protocol/http/body/readable.rb @@ -45,6 +45,47 @@ end end + with "#each" do + it "passes a read error to close" do + error = RuntimeError.new("Could not read the body!") + closed_error = nil + + mock(body) do |mock| + mock.replace(:read){raise error} + mock.replace(:close){|argument = nil| closed_error = argument} + end + + raised_error = begin + body.each{} + rescue => exception + exception + end + + expect(raised_error).to be_equal(error) + expect(closed_error).to be_equal(error) + end + + it "passes a consumer error to close" do + error = RuntimeError.new("Could not consume the body!") + closed_error = nil + chunks = ["Hello", nil] + + mock(body) do |mock| + mock.replace(:read){chunks.shift} + mock.replace(:close){|argument = nil| closed_error = argument} + end + + raised_error = begin + body.each{raise error} + rescue => exception + exception + end + + expect(raised_error).to be_equal(error) + expect(closed_error).to be_equal(error) + end + end + with "#call" do let(:output) {Protocol::HTTP::Body::Buffered.new} let(:stream) {Protocol::HTTP::Body::Stream.new(nil, output)} @@ -72,6 +113,45 @@ body.call(stream) end + + it "closes a plain IO normally when reading fails" do + error = RuntimeError.new("Could not read the body!") + stream = StringIO.new + + mock(body) do |mock| + mock.replace(:read){raise error} + end + + raised_error = begin + body.call(stream) + rescue => exception + exception + end + + expect(raised_error).to be_equal(error) + expect(stream).to be(:closed?) + end + + it "passes a read error to a stream with explicit error closure" do + error = RuntimeError.new("Could not read the body!") + closed_error = nil + stream = Object.new + stream.define_singleton_method(:close_with_error){|argument| closed_error = argument} + + mock(body) do |mock| + mock.replace(:read){raise error} + end + + raised_error = begin + body.call(stream) + rescue => exception + exception + end + + expect(raised_error).to be_equal(error) + expect(closed_error).to be_equal(error) + end + end with "#join" do diff --git a/test/protocol/http/body/reader.rb b/test/protocol/http/body/reader.rb index ff8086d1..b3b8c8d2 100644 --- a/test/protocol/http/body/reader.rb +++ b/test/protocol/http/body/reader.rb @@ -23,10 +23,35 @@ def initialize(body) let(:body) {Protocol::HTTP::Body::Buffered.wrap("thequickbrownfox")} let(:reader) {TestReader.new(body)} + with "#each" do + it "detaches the body after failure" do + mock(body) {|mock| mock.replace(:each){raise "Could not read!"}} + + expect{reader.each{}}.to raise_exception(RuntimeError, message: be =~ /Could not read!/) + expect(reader.body).to be_nil + end + end + + with "#read" do + it "detaches the body after failure" do + mock(body) {|mock| mock.replace(:join){raise "Could not read!"}} + + expect{reader.read}.to raise_exception(RuntimeError, message: be =~ /Could not read!/) + expect(reader.body).to be_nil + end + end + with "#finish" do it "returns a buffered representation" do expect(reader.finish).to be == body end + + it "detaches the body after failure" do + mock(body) {|mock| mock.replace(:finish){raise "Could not read!"}} + + expect{reader.finish}.to raise_exception(RuntimeError, message: be =~ /Could not read!/) + expect(reader.body).to be_nil + end end with "#discard" do @@ -41,6 +66,13 @@ def initialize(body) expect(reader.buffered!).to be_equal(reader) expect(reader.body).to be == body end + + it "detaches the body after failure" do + mock(body) {|mock| mock.replace(:finish){raise "Could not read!"}} + + expect{reader.buffered!}.to raise_exception(RuntimeError, message: be =~ /Could not read!/) + expect(reader.body).to be_nil + end end with "#close" do diff --git a/test/protocol/http/body/stream.rb b/test/protocol/http/body/stream.rb index 1e7af599..cf0e650c 100644 --- a/test/protocol/http/body/stream.rb +++ b/test/protocol/http/body/stream.rb @@ -341,6 +341,20 @@ stream.close expect(stream).to be(:closed?) end + + it "can be closed with an explicit error" do + error = RuntimeError.new("Oh no!") + closed_error = nil + + mock(output) do |mock| + mock.replace(:close_write){|argument| closed_error = argument} + end + + stream.close_with_error(error) + + expect(closed_error).to be_equal(error) + expect(stream).to be(:closed?) + end end with "IO.copy_stream" do diff --git a/test/protocol/http/body/streamable.rb b/test/protocol/http/body/streamable.rb index 40deffc5..d4ab72dc 100644 --- a/test/protocol/http/body/streamable.rb +++ b/test/protocol/http/body/streamable.rb @@ -4,6 +4,8 @@ # Copyright, 2024-2025, by Samuel Williams. require "protocol/http/body/streamable" +require "protocol/http/request" +require "protocol/http/response" require "sus/fixtures/async" describe Protocol::HTTP::Body::Streamable do @@ -134,6 +136,27 @@ expect(stream.string).to be == "Hello" end end + + with "a stream that supports explicit error closure" do + let(:error) {RuntimeError.new("Oh no!")} + let(:block) {proc{|stream| raise error}} + + it "passes the exact error when closing the stream" do + closed_error = nil + stream = Object.new + stream.define_singleton_method(:close_with_error){|argument| closed_error = argument} + + raised_error = begin + body.call(stream) + rescue => exception + exception + end + + expect(raised_error).to be_equal(error) + expect(closed_error).to be_equal(error) + end + end + end with "#close" do @@ -301,6 +324,8 @@ input.write("Hello") input.close_write(RuntimeError.new("Oh no!")) + expect(output.read).to be == "Hello" + expect do output.read end.to raise_exception(RuntimeError, message: be =~ /Oh no!/) diff --git a/test/protocol/http/body/writable.rb b/test/protocol/http/body/writable.rb index d07e7a8a..7a533373 100644 --- a/test/protocol/http/body/writable.rb +++ b/test/protocol/http/body/writable.rb @@ -102,6 +102,71 @@ body.write("Hello") expect(body).not.to be(:empty?) end + + it "should not be empty if a terminal error is pending" do + body.close_write(RuntimeError.new("Oh no!")) + expect(body).not.to be(:empty?) + end + end + + with "#close" do + it "discards buffered chunks" do + body.write("Hello") + body.close + + expect(body.read).to be_nil + expect{body.write("World")}.to raise_exception(Protocol::HTTP::Body::Writable::Closed) + end + end + + with "#close_write" do + it "drains buffered chunks before raising the terminal error" do + error = RuntimeError.new("The body was truncated!") + + body.write("Hello") + body.write("World") + body.close_write(error) + + expect(body.read).to be == "Hello" + expect(body.read).to be == "World" + + raised_error = begin + body.read + rescue => exception + exception + end + + expect(raised_error).to be_equal(error) + end + + it "does not replace clean completion with a later error" do + body.write("Hello") + body.close_write + body.close_write(RuntimeError.new("Too late!")) + + expect(body.read).to be == "Hello" + expect(body.read).to be_nil + end + + it "wakes a blocked reader with the exact error" do + error = RuntimeError.new("The body was truncated!") + started = Thread::Queue.new + + reader = Thread.new do + started << true + + begin + body.read + rescue => exception + exception + end + end + + started.pop + body.close_write(error) + + expect(reader.value).to be_equal(error) + end end with "#write" do @@ -234,5 +299,26 @@ body.read end.to raise_exception(RuntimeError, message: be =~ /Oops/) end + + it "drains buffered chunks before propagating errors" do + error = RuntimeError.new("Oops!") + + expect do + body.output do |output| + output.write("Hello") + raise error + end + end.to raise_exception(RuntimeError, message: be =~ /Oops/) + + expect(body.read).to be == "Hello" + + raised_error = begin + body.read + rescue => exception + exception + end + + expect(raised_error).to be_equal(error) + end end end