HTTP/2: retry streams reset with REFUSED_STREAM or rejected by GOAWAY (RFC 9113 §8.7)

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

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

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

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",