diff --git a/src/http2_client.jl b/src/http2_client.jl index d0a50037a..1d6e646d0 100644 --- a/src/http2_client.jl +++ b/src/http2_client.jl @@ -38,6 +38,37 @@ function Base.showerror(io::IO, err::H2GoAwayError) return nothing end +# RFC 9113 §7 error codes (0x0–0xd). +const _H2_ERROR_CODE_NAMES = ( + "NO_ERROR", "PROTOCOL_ERROR", "INTERNAL_ERROR", "FLOW_CONTROL_ERROR", + "SETTINGS_TIMEOUT", "STREAM_CLOSED", "FRAME_SIZE_ERROR", "REFUSED_STREAM", + "CANCEL", "COMPRESSION_ERROR", "CONNECT_ERROR", "ENHANCE_YOUR_CALM", + "INADEQUATE_SECURITY", "HTTP_1_1_REQUIRED", +) +const _H2_REFUSED_STREAM = UInt32(0x7) + +function _h2_error_code_name(code::UInt32)::String + code < UInt32(length(_H2_ERROR_CODE_NAMES)) && return _H2_ERROR_CODE_NAMES[Int(code) + 1] + return string("0x", string(code; base = 16)) +end + +""" + H2StreamResetError + +Raised when the peer resets an HTTP/2 stream with `RST_STREAM`. Carries the +RFC 9113 §7 `error_code`. `REFUSED_STREAM` guarantees the stream was closed +prior to any processing (RFC 9113 §8.7), so such requests are retried when +retries are enabled. +""" +struct H2StreamResetError <: HTTPError + error_code::UInt32 +end + +function Base.showerror(io::IO, err::H2StreamResetError) + print(io, "HTTP/2 stream reset by peer: ", _h2_error_code_name(err.error_code)) + return nothing +end + """ H2StreamState @@ -762,7 +793,7 @@ function _process_incoming_frame!(conn::H2Connection, frame::AbstractFrame) rst = frame::RSTStreamFrame state = _stream_state(conn, rst.stream_id) state === nothing && return nothing - _set_stream_error!(state::H2StreamState, ProtocolError("HTTP/2 stream reset by peer")) + _set_stream_error!(state::H2StreamState, H2StreamResetError(rst.error_code)) return nothing elseif frame isa WindowUpdateFrame update = frame::WindowUpdateFrame diff --git a/src/http_client_retry.jl b/src/http_client_retry.jl index c8d5a66d1..bdcbb5e11 100644 --- a/src/http_client_retry.jl +++ b/src/http_client_retry.jl @@ -48,6 +48,10 @@ end function _retryable_request_error(err::Exception)::Bool current = err while true + # RFC 9113 §8.7: a stream reset with REFUSED_STREAM, or rejected by + # GOAWAY (stream_id > last_stream_id), was not processed by the server. + current isa H2StreamResetError && return (current::H2StreamResetError).error_code == _H2_REFUSED_STREAM + current isa H2GoAwayError && return true current isa EOFError && return true current isa SystemError && return true current isa ParseError && return true @@ -72,6 +76,22 @@ end _retryable_request_error(err::RequestRetryError)::Bool = _retryable_request_error(err.err) +# RFC 9113 §8.7: requests on streams reset with REFUSED_STREAM or rejected by +# GOAWAY are guaranteed unprocessed, hence safe to retry regardless of method +# idempotency. The replayable-body gate in `_should_retry_request_attempt` +# still applies. +function _h2_guaranteed_unprocessed(err)::Bool + current = err + while true + if current isa RequestRetryError + current = (current::RequestRetryError).err + continue + end + current isa H2StreamResetError && return (current::H2StreamResetError).error_code == _H2_REFUSED_STREAM + return current isa H2GoAwayError + end +end + function _retry_hook_decision(controller::_RetryController, attempt::Int, err, req::Request, resp) hook = controller.retry_if hook === nothing && return nothing @@ -86,7 +106,8 @@ function _should_retry_request_attempt(controller::_RetryController, attempt::In _retryable_request_body(req) || return false built_in = false if err !== nothing - built_in = _retryable_policy_request(req, controller.retry_non_idempotent) && _retryable_request_error(err) + policy_ok = _retryable_policy_request(req, controller.retry_non_idempotent) || _h2_guaranteed_unprocessed(err) + built_in = policy_ok && _retryable_request_error(err) elseif resp !== nothing built_in = _retryable_policy_request(req, controller.retry_non_idempotent) && _retryable_status((resp::Response).status) end diff --git a/test/http2_refused_stream_retry_tests.jl b/test/http2_refused_stream_retry_tests.jl new file mode 100644 index 000000000..0e61f0d29 --- /dev/null +++ b/test/http2_refused_stream_retry_tests.jl @@ -0,0 +1,223 @@ +using Test +using HTTP +using Reseau + +const HT = HTTP +const ND = Reseau.HostResolvers +const NC = Reseau.TCP + +# RFC 9113 §7 error codes +const _RSR_REFUSED_STREAM = UInt32(0x7) +const _RSR_NO_ERROR = UInt32(0x0) + +function _rsr_write_frame!(conn::NC.Conn, frame::HT.AbstractFrame) + io = IOBuffer() + HT.write_frame!(io, frame) + bytes = take!(io) + total = 0 + while total < length(bytes) + n = write(conn, bytes[(total + 1):end]) + n > 0 || error("expected write progress") + total += n + end + return nothing +end + +function _rsr_read_preface!(conn::NC.Conn) + n = length(HT._H2_PREFACE) + offset = 0 + while offset < n + chunk = Vector{UInt8}(undef, n - offset) + nr = readbytes!(conn, chunk) + nr > 0 || error("unexpected EOF reading client preface") + offset += nr + end + return nothing +end + +function _rsr_next_headers!(reader)::HT.HeadersFrame + while true + frame = HT.read_frame!(reader) + frame isa HT.HeadersFrame && return frame::HT.HeadersFrame + frame isa HT.DataFrame && continue # POST request bodies + frame isa HT.WindowUpdateFrame && continue + frame isa HT.SettingsFrame && continue + frame isa HT.PingFrame && continue + error("expected headers frame, got $(typeof(frame))") + end +end + +# Scripted h2 server: refuses the first `refuse_first` requests (RST_STREAM +# REFUSED_STREAM, or GOAWAY rejecting the in-flight stream), then answers 200. +# An accept loop plus a per-connection headers loop keeps it agnostic to +# whether the client retries on the same connection or on a new one. +function _rsr_serve!(listener; scenario::Symbol, refuse_first::Int = 1, rst_code::UInt32 = _RSR_REFUSED_STREAM) + attempts = Threads.Atomic{Int}(0) + conns = Threads.Atomic{Int}(0) + accept_task = errormonitor(Threads.@spawn begin + while true + conn = try + NC.accept(listener) + catch + break # listener closed: end of test + end + Threads.atomic_add!(conns, 1) + errormonitor(Threads.@spawn begin + try + _rsr_read_preface!(conn) + _rsr_write_frame!(conn, HT.SettingsFrame(false, Pair{UInt16, UInt32}[])) + reader = HT._ConnReader(conn) + encoder = HT.Encoder() + while true + hf = _rsr_next_headers!(reader) + attempt = Threads.atomic_add!(attempts, 1) + 1 + if attempt <= refuse_first + if scenario === :rst + _rsr_write_frame!(conn, HT.RSTStreamFrame(hf.stream_id, rst_code)) + elseif scenario === :goaway + _rsr_write_frame!(conn, HT.GoAwayFrame(UInt32(0), _RSR_NO_ERROR, UInt8[])) + break + else + error("unknown scenario $scenario") + end + else + encoded = HT.encode_header_block(encoder, HT.HeaderField[HT.HeaderField(":status", "200", false)]) + _rsr_write_frame!(conn, HT.HeadersFrame(hf.stream_id, true, true, encoded)) + end + end + catch + # client tore the connection down (expected on error paths) + finally + HTTP.@try_ignore NC.close(conn) + end + end) + end + end) + return attempts, conns, accept_task +end + +function _rsr_request(url; method::String = "GET", body::String = "") + return try + HT.request(method, url, Pair{String, String}[], body; + protocol = :h2, retry = true, connect_timeout = 5, request_timeout = 15) + catch err + err + end +end + +@testset "HTTP/2 retry of guaranteed-unprocessed streams (RFC 9113 §8.7)" begin + +# A stream reset with REFUSED_STREAM is guaranteed unprocessed (RFC 9113 §8.7) +# and must be retried transparently. +@testset "HTTP/2 client retries a stream reset with REFUSED_STREAM" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + laddr = NC.addr(listener)::NC.SocketAddrV4 + url = "http://" * ND.join_host_port("127.0.0.1", Int(laddr.port)) * "/" + attempts, conns, accept_task = _rsr_serve!(listener; scenario = :rst) + try + result = _rsr_request(url) + @info "REFUSED_STREAM scenario" result attempts[] conns[] + @test attempts[] == 2 # the refused attempt was retried + @test result isa HT.Response + result isa HT.Response && @test (result::HT.Response).status == 200 + finally + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(accept_task) + end +end + +# A stream rejected by GOAWAY (stream_id > last_stream_id) is likewise +# unprocessed and must be retried, necessarily on a new connection. +@testset "HTTP/2 client retries a stream rejected by GOAWAY" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + laddr = NC.addr(listener)::NC.SocketAddrV4 + url = "http://" * ND.join_host_port("127.0.0.1", Int(laddr.port)) * "/" + attempts, conns, accept_task = _rsr_serve!(listener; scenario = :goaway) + try + result = _rsr_request(url) + @info "GOAWAY scenario" result attempts[] conns[] + @test attempts[] == 2 + @test conns[] == 2 # GOAWAY drains the first connection + @test result isa HT.Response + result isa HT.Response && @test (result::HT.Response).status == 200 + finally + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(accept_task) + end +end + +# Other reset codes give no unprocessed guarantee: they must surface, not retry. +@testset "HTTP/2 client does not retry a stream reset with CANCEL" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + laddr = NC.addr(listener)::NC.SocketAddrV4 + url = "http://" * ND.join_host_port("127.0.0.1", Int(laddr.port)) * "/" + attempts, conns, accept_task = _rsr_serve!(listener; scenario = :rst, rst_code = UInt32(0x8)) + try + result = _rsr_request(url) + @test attempts[] == 1 # surfaced on first occurrence + @test result isa HT.H2StreamResetError + result isa HT.H2StreamResetError && @test occursin("CANCEL", sprint(showerror, result)) + finally + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(accept_task) + end +end + +# §8.7 lifts the idempotency gate: requests on guaranteed-unprocessed streams +# are retried even when non-idempotent (the body, here in-memory, is replayable). +@testset "HTTP/2 client retries a refused POST (non-idempotent)" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + laddr = NC.addr(listener)::NC.SocketAddrV4 + url = "http://" * ND.join_host_port("127.0.0.1", Int(laddr.port)) * "/" + attempts, conns, accept_task = _rsr_serve!(listener; scenario = :rst) + try + result = _rsr_request(url; method = "POST", body = "ping") + @test attempts[] == 2 + @test result isa HT.Response + result isa HT.Response && @test (result::HT.Response).status == 200 + finally + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(accept_task) + end +end + +@testset "HTTP/2 client retries a POST rejected by GOAWAY" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + laddr = NC.addr(listener)::NC.SocketAddrV4 + url = "http://" * ND.join_host_port("127.0.0.1", Int(laddr.port)) * "/" + attempts, conns, accept_task = _rsr_serve!(listener; scenario = :goaway) + try + result = _rsr_request(url; method = "POST", body = "ping") + @test attempts[] == 2 + @test conns[] == 2 + @test result isa HT.Response + result isa HT.Response && @test (result::HT.Response).status == 200 + finally + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(accept_task) + end +end + +# A cancelled POST has no unprocessed guarantee: no retry, the reset surfaces. +@testset "HTTP/2 client does not retry a cancelled POST" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + laddr = NC.addr(listener)::NC.SocketAddrV4 + url = "http://" * ND.join_host_port("127.0.0.1", Int(laddr.port)) * "/" + attempts, conns, accept_task = _rsr_serve!(listener; scenario = :rst, rst_code = UInt32(0x8)) + try + result = _rsr_request(url; method = "POST", body = "ping") + @test attempts[] == 1 + @test result isa HT.H2StreamResetError + finally + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(accept_task) + end +end + +@testset "H2StreamResetError names RFC 9113 §7 error codes" begin + @test sprint(showerror, HT.H2StreamResetError(UInt32(0x7))) == "HTTP/2 stream reset by peer: REFUSED_STREAM" + @test sprint(showerror, HT.H2StreamResetError(UInt32(0xb))) == "HTTP/2 stream reset by peer: ENHANCE_YOUR_CALM" + @test endswith(sprint(showerror, HT.H2StreamResetError(UInt32(0xff))), ": 0xff") +end + +end # parent testset diff --git a/test/runtests.jl b/test/runtests.jl index 7da50caac..beb273826 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -112,6 +112,7 @@ test_files = [ "hpack_tests.jl", "http2_frame_tests.jl", "http2_client_tests.jl", + "http2_refused_stream_retry_tests.jl", "http2_server_tests.jl", "http_integration_tests.jl", "http_parity_tests.jl",