From b5d67c2b9b75dbca1712809bef36edc32720fc56 Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Thu, 23 Jul 2026 08:47:44 -0600 Subject: [PATCH 1/4] fix(http2): surface partial response resets Preserve buffered response DATA after RST_STREAM, then raise the terminal stream error instead of reporting clean EOF. Close and unregister the body when the error surfaces. Cover partial bodies across small and compaction-boundary payload sizes. --- src/http2_client.jl | 21 +++++++++-- test/http2_client_tests.jl | 72 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 91 insertions(+), 2 deletions(-) diff --git a/src/http2_client.jl b/src/http2_client.jl index f7e86a137..789ee5459 100644 --- a/src/http2_client.jl +++ b/src/http2_client.jl @@ -1598,6 +1598,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 +1619,18 @@ 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 + elseif body.state.conn_errored + terminal_error = _stream_conn_error(body.conn) + 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 +1644,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))) diff --git a/test/http2_client_tests.jl b/test/http2_client_tests.jl index 603780612..bed9f5151 100644 --- a/test/http2_client_tests.jl +++ b/test/http2_client_tests.jl @@ -269,6 +269,78 @@ 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 requires initial SETTINGS before other frames" begin listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) laddr = NC.addr(listener)::NC.SocketAddrV4 From 267cd4a5d19aa0bdd9a3c4a973d98748e3b686ff Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Wed, 5 Aug 2026 17:24:28 -0600 Subject: [PATCH 2/4] fix(http2): preserve completed response after reset Ignore a late stream reset after END_STREAM completed the response normally. Add a synchronized wire-level regression that proves a non-NO_ERROR reset cannot discard the buffered complete body. --- src/http2_client.jl | 7 ++-- test/http2_client_tests.jl | 67 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 3 deletions(-) diff --git a/src/http2_client.jl b/src/http2_client.jl index 789ee5459..f37ba0a12 100644 --- a/src/http2_client.jl +++ b/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 diff --git a/test/http2_client_tests.jl b/test/http2_client_tests.jl index bed9f5151..17cea0753 100644 --- a/test/http2_client_tests.jl +++ b/test/http2_client_tests.jl @@ -341,6 +341,73 @@ 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 requires initial SETTINGS before other frames" begin listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) laddr = NC.addr(listener)::NC.SocketAddrV4 From 07b0ace13cdb52e21abf4cb1cd58bdb058fc67ff Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Wed, 5 Aug 2026 19:56:20 -0600 Subject: [PATCH 3/4] fix(http2): keep completed streams after conn error A connection failure caused by another multiplexed stream must not invalidate a response that already received END_STREAM. Remove the impossible incomplete-response branch and cover the completed-stream ownership case with a raw two-stream peer. --- src/http2_client.jl | 2 -- test/http2_client_tests.jl | 65 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 2 deletions(-) diff --git a/src/http2_client.jl b/src/http2_client.jl index f37ba0a12..9ee4f2a6c 100644 --- a/src/http2_client.jl +++ b/src/http2_client.jl @@ -1626,8 +1626,6 @@ function body_read!(body::H2Body, dst::Vector{UInt8})::Int # like a successful clean EOF. if body.state.stream_error !== nothing terminal_error = body.state.stream_error - elseif body.state.conn_errored - terminal_error = _stream_conn_error(body.conn) else _publish_h2_response_trailers!(body.state) done = true diff --git a/test/http2_client_tests.jl b/test/http2_client_tests.jl index 17cea0753..2f0dd86a8 100644 --- a/test/http2_client_tests.jl +++ b/test/http2_client_tests.jl @@ -408,6 +408,71 @@ end 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 From 2c824e0dd3c839879444e4e131a7af59afb92230 Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Wed, 5 Aug 2026 20:30:28 -0600 Subject: [PATCH 4/4] test(http2): verify GOAWAY error details Assert the terminal GOAWAY error retains the peer last-stream identifier and renders its user-facing message. This pins the typed error contract used by response lifecycle failures. --- test/http2_client_tests.jl | 3 +++ 1 file changed, 3 insertions(+) diff --git a/test/http2_client_tests.jl b/test/http2_client_tests.jl index 2f0dd86a8..5665ffc1e 100644 --- a/test/http2_client_tests.jl +++ b/test/http2_client_tests.jl @@ -1944,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)