Surface HTTP/2 response resets after buffered data

src/http2_client.jl

@@ -977,12 +977,13 @@ function _process_incoming_frame!(conn::H2Connection, frame::AbstractFrame)
         state === nothing && return nothing
         # RFC 9113 §8.1: a server MAY send RST_STREAM with NO_ERROR after a
         # complete response to ask the client to abort the request body;
-        # "clients MUST NOT discard responses as a result". Only surface the
-        # reset as a stream error when it does not follow a complete response.
+        # clients must not discard the response. A reset with another code can
+        # also race with the final response frame. Once END_STREAM completed
+        # the response normally, keep it regardless of the later reset code.
         benign = false
         lock((state::H2StreamState).lock)
         try
-            benign = rst.error_code == UInt32(0) && (state::H2StreamState).stream_done
+            benign = (state::H2StreamState).stream_done && !_stream_failed(state::H2StreamState)
         finally
             unlock((state::H2StreamState).lock)
         end
@@ -1598,6 +1599,7 @@ function body_read!(body::H2Body, dst::Vector{UInt8})::Int
         nread = 0
         done = false
         too_many = false
+        terminal_error::Union{Nothing,Exception} = nothing
         wait_deadline_ns = Int64(0)
         lock(body.state.lock)
         try
@@ -1618,8 +1620,16 @@ function body_read!(body::H2Body, dst::Vector{UInt8})::Int
                     notify(body.state.condition)
                 end
             elseif body.state.stream_done
-                _publish_h2_response_trailers!(body.state)
-                done = true
+                # DATA buffered before a stream reset is still readable, but
+                # once it is drained the terminal stream error must surface.
+                # Otherwise a partial response without Content-Length looks
+                # like a successful clean EOF.
+                if body.state.stream_error !== nothing
+                    terminal_error = body.state.stream_error
+                else
+                    _publish_h2_response_trailers!(body.state)
+                    done = true
+                end
             else
                 _throw_stream_error(body.conn, body.state)
                 deadline_ns = _request_read_deadline_ns(body.request)
@@ -1633,6 +1643,12 @@ function body_read!(body::H2Body, dst::Vector{UInt8})::Int
             unlock(body.state.lock)
         end
         wait_deadline_ns == 0 || (_wait_h2_body_progress!(body.state, wait_deadline_ns); continue)
+        if terminal_error !== nothing
+            @atomic :release body.closed = true
+            _clear_h2_cancel_callback!(body)
+            _unregister_stream!(body.conn, body.stream_id)
+            throw(terminal_error::Exception)
+        end
         if too_many
             @atomic :release body.closed = true
             @try_ignore _write_frame_h2_threadsafe!(body.conn, RSTStreamFrame(body.stream_id, UInt32(0x1)))

test/http2_client_tests.jl

@@ -269,6 +269,210 @@ end
     end
 end
 
+@testset "HTTP/2 client surfaces a reset after buffered response DATA" begin
+    for payload in ("x", "partial", repeat("z", 4097))
+        listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8)
+        laddr = NC.addr(listener)::NC.SocketAddrV4
+        address = ND.join_host_port("127.0.0.1", Int(laddr.port))
+        allow_reset = Channel{Nothing}(1)
+        finish = Channel{Nothing}(1)
+        server_task = errormonitor(Threads.@spawn begin
+            accepted_conn = NC.accept(listener)
+            reader = HT._ConnReader(accepted_conn)
+            server_encoder = HT.Encoder()
+            server_decoder = HT.Decoder()
+            try
+                _ = _read_exact_h2_tcp!(accepted_conn, length(HT._H2_PREFACE))
+                _ = HT.read_frame!(reader)
+                _write_frame_to_conn!(accepted_conn, HT.SettingsFrame(false, Pair{UInt16, UInt32}[]))
+                _ = HT.read_frame!(reader)
+                headers_frame = _read_next_headers_frame!(reader)
+                hf = headers_frame::HT.HeadersFrame
+                _ = HT.decode_header_block(server_decoder, hf.header_block_fragment)
+                encoded = HT.encode_header_block(server_encoder, HT.HeaderField[HT.HeaderField(":status", "200", false)])
+                _write_frame_to_conn!(accepted_conn, HT.HeadersFrame(hf.stream_id, false, true, encoded))
+                _write_frame_to_conn!(accepted_conn, HT.DataFrame(hf.stream_id, false, collect(codeunits(payload))))
+                take!(allow_reset)
+                _write_frame_to_conn!(accepted_conn, HT.RSTStreamFrame(hf.stream_id, UInt32(0x8)))
+                take!(finish)
+            finally
+                HTTP.@try_ignore NC.close(accepted_conn)
+            end
+            return nothing
+        end)
+        h2_conn = HT.connect_h2!(address; secure = false)
+        try
+            request = HT.Request("GET", "/partial-reset"; host = address, body = HT.EmptyBody(), content_length = 0)
+            response = HT.h2_roundtrip!(h2_conn, request)
+            put!(allow_reset, nothing)
+            state = (response.body::HT.H2Body).state
+            reset_received = timedwait(() -> begin
+                lock(state.lock)
+                try
+                    return state.stream_error !== nothing
+                finally
+                    unlock(state.lock)
+                end
+            end, 5.0; pollint = 0.001)
+            @test reset_received != :timed_out
+
+            buf = Vector{UInt8}(undef, length(payload))
+            @test HT.body_read!(response.body, buf) == length(payload)
+            @test String(copy(buf)) == payload
+            err = try
+                HT.body_read!(response.body, buf)
+                nothing
+            catch e
+                e
+            end
+            @test err isa HT.H2StreamResetError
+            @test (err::HT.H2StreamResetError).error_code == UInt32(0x8)
+            @test HT.body_closed(response.body)
+            @test HT._stream_state(h2_conn, (response.body::HT.H2Body).stream_id) === nothing
+            @test isempty(HT.get_request_context(request).cancel_callbacks)
+            @test HT._h2_conn_reusable(h2_conn)
+        finally
+            isready(allow_reset) || put!(allow_reset, nothing)
+            put!(finish, nothing)
+            _wait_task_h2!(server_task)
+            close(h2_conn)
+            HTTP.@try_ignore NC.close(listener)
+        end
+    end
+end
+
+@testset "HTTP/2 client keeps a complete response after a late reset" begin
+    listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8)
+    laddr = NC.addr(listener)::NC.SocketAddrV4
+    address = ND.join_host_port("127.0.0.1", Int(laddr.port))
+    reset_processed = Channel{Nothing}(1)
+    finish = Channel{Nothing}(1)
+    server_task = errormonitor(Threads.@spawn begin
+        accepted_conn = NC.accept(listener)
+        reader = HT._ConnReader(accepted_conn)
+        server_encoder = HT.Encoder()
+        server_decoder = HT.Decoder()
+        try
+            _ = _read_exact_h2_tcp!(accepted_conn, length(HT._H2_PREFACE))
+            _ = HT.read_frame!(reader)
+            _write_frame_to_conn!(accepted_conn, HT.SettingsFrame(false, Pair{UInt16, UInt32}[]))
+            _ = HT.read_frame!(reader)
+            headers_frame = _read_next_headers_frame!(reader)
+            hf = headers_frame::HT.HeadersFrame
+            _ = HT.decode_header_block(server_decoder, hf.header_block_fragment)
+            encoded = HT.encode_header_block(server_encoder, HT.HeaderField[HT.HeaderField(":status", "200", false)])
+            _write_frame_to_conn!(accepted_conn, HT.HeadersFrame(hf.stream_id, false, true, encoded))
+            _write_frame_to_conn!(accepted_conn, HT.DataFrame(hf.stream_id, true, collect(codeunits("complete"))))
+            _write_frame_to_conn!(accepted_conn, HT.RSTStreamFrame(hf.stream_id, UInt32(0x8)))
+            ping_data = ntuple(UInt8, 8)
+            _write_frame_to_conn!(accepted_conn, HT.PingFrame(false, ping_data))
+            while true
+                frame = HT.read_frame!(reader)
+                frame isa HT.PingFrame && (frame::HT.PingFrame).ack && break
+            end
+            put!(reset_processed, nothing)
+            take!(finish)
+        finally
+            HTTP.@try_ignore NC.close(accepted_conn)
+        end
+        return nothing
+    end)
+    h2_conn = HT.connect_h2!(address; secure = false)
+    try
+        request = HT.Request("GET", "/complete-late-reset"; host = address, body = HT.EmptyBody(), content_length = 0)
+        response = HT.h2_roundtrip!(h2_conn, request)
+        reset_seen = timedwait(() -> isready(reset_processed), 5.0; pollint = 0.001)
+        @test reset_seen != :timed_out
+        reset_seen == :timed_out && error("client did not process the late reset")
+        take!(reset_processed)
+        state = (response.body::HT.H2Body).state
+        response_complete = timedwait(() -> begin
+            lock(state.lock)
+            try
+                return state.stream_done
+            finally
+                unlock(state.lock)
+            end
+        end, 5.0; pollint = 0.001)
+        @test response_complete != :timed_out
+        @test String(_read_all_h2_body(response.body)) == "complete"
+        @test HT.body_closed(response.body)
+        @test HT._stream_state(h2_conn, (response.body::HT.H2Body).stream_id) === nothing
+        @test isempty(HT.get_request_context(request).cancel_callbacks)
+        @test HT._h2_conn_reusable(h2_conn)
+    finally
+        put!(finish, nothing)
+        _wait_task_h2!(server_task)
+        close(h2_conn)
+        HTTP.@try_ignore NC.close(listener)
+    end
+end
+
+@testset "HTTP/2 client keeps a complete response when another stream fails with the connection" begin
+    listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8)
+    laddr = NC.addr(listener)::NC.SocketAddrV4
+    address = ND.join_host_port("127.0.0.1", Int(laddr.port))
+    allow_close = Channel{Nothing}(1)
+    server_task = errormonitor(Threads.@spawn begin
+        accepted_conn = NC.accept(listener)
+        reader = HT._ConnReader(accepted_conn)
+        server_encoder = HT.Encoder()
+        server_decoder = HT.Decoder()
+        try
+            _ = _read_exact_h2_tcp!(accepted_conn, length(HT._H2_PREFACE))
+            _ = HT.read_frame!(reader)
+            _write_frame_to_conn!(accepted_conn, HT.SettingsFrame(false, Pair{UInt16, UInt32}[]))
+            _ = HT.read_frame!(reader)
+            first_headers_frame = _read_next_headers_frame!(reader)
+            first_hf = first_headers_frame::HT.HeadersFrame
+            _ = HT.decode_header_block(server_decoder, first_hf.header_block_fragment)
+            encoded = HT.encode_header_block(server_encoder, HT.HeaderField[HT.HeaderField(":status", "200", false)])
+            _write_frame_to_conn!(accepted_conn, HT.HeadersFrame(first_hf.stream_id, false, true, encoded))
+            _write_frame_to_conn!(accepted_conn, HT.DataFrame(first_hf.stream_id, true, collect(codeunits("complete"))))
+            second_headers_frame = _read_next_headers_frame!(reader)
+            second_hf = second_headers_frame::HT.HeadersFrame
+            _ = HT.decode_header_block(server_decoder, second_hf.header_block_fragment)
+            _write_frame_to_conn!(accepted_conn, HT.HeadersFrame(second_hf.stream_id, false, true, encoded))
+            take!(allow_close)
+        finally
+            HTTP.@try_ignore NC.close(accepted_conn)
+        end
+        return nothing
+    end)
+    h2_conn = HT.connect_h2!(address; secure = false)
+    try
+        complete_request = HT.Request("GET", "/complete"; host = address, body = HT.EmptyBody(), content_length = 0)
+        complete_response = HT.h2_roundtrip!(h2_conn, complete_request)
+        incomplete_request = HT.Request("GET", "/incomplete"; host = address, body = HT.EmptyBody(), content_length = 0)
+        incomplete_response = HT.h2_roundtrip!(h2_conn, incomplete_request)
+        put!(allow_close, nothing)
+        _wait_task_h2!(server_task)
+        complete_state = (complete_response.body::HT.H2Body).state
+        incomplete_state = (incomplete_response.body::HT.H2Body).state
+        failure_seen = timedwait(() -> begin
+            complete_failed = lock(complete_state.lock) do
+                complete_state.conn_errored
+            end
+            return lock(incomplete_state.lock) do
+                complete_failed && incomplete_state.conn_errored
+            end
+        end, 5.0; pollint = 0.001)
+        @test failure_seen != :timed_out
+        @test String(_read_all_h2_body(complete_response.body)) == "complete"
+        @test_throws HT.ProtocolError HT.body_read!(incomplete_response.body, Vector{UInt8}(undef, 1))
+        HT.body_close!(incomplete_response.body)
+        @test HT.body_closed(complete_response.body)
+        @test HT._stream_state(h2_conn, (complete_response.body::HT.H2Body).stream_id) === nothing
+        @test isempty(HT.get_request_context(complete_request).cancel_callbacks)
+        @test !HT._h2_conn_reusable(h2_conn)
+    finally
+        isready(allow_close) || put!(allow_close, nothing)
+        _wait_task_h2!(server_task)
+        close(h2_conn)
+        HTTP.@try_ignore NC.close(listener)
+    end
+end
+
 @testset "HTTP/2 client requires initial SETTINGS before other frames" begin
     listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8)
     laddr = NC.addr(listener)::NC.SocketAddrV4
@@ -1740,6 +1944,9 @@ end
         @test String(_read_all_h2_body(response.body)) == "ok"
         failure = only(failures)::TaskFailedException
         @test failure.task.exception isa HT.H2GoAwayError
+        goaway_error = failure.task.exception::HT.H2GoAwayError
+        @test goaway_error.last_stream_id == UInt32(1)
+        @test sprint(showerror, goaway_error) == "HTTP/2 stream rejected by GOAWAY"
         _wait_task_h2!(server_task)
     finally
         close(h2_conn)