diff --git a/src/http_client.jl b/src/http_client.jl index 946f0f312..794b2b77c 100644 --- a/src/http_client.jl +++ b/src/http_client.jl @@ -1,5 +1,13 @@ # High-level HTTP client orchestration, HTTP/2 integration, cookies, response sinks, and convenience APIs. +# Copy-loop buffer for response bodies that are known (or likely) to be large: +# 64 KiB measured faster than 8 KiB and 256 KiB for high-throughput downloads. +# Loops whose payload is typically small — unknown-size accumulation of +# chunked responses and decompression output — keep 8 KiB buffers so a tiny +# response doesn't pay a 64 KiB allocation per request. +const _RESPONSE_COPY_BUFFER_BYTES = 64 * 1024 +const _RESPONSE_SMALL_COPY_BUFFER_BYTES = 8 * 1024 + """ Client(; ...) @@ -15,7 +23,11 @@ Keyword arguments: followed - `cookiejar`: cookie jar implementation, or `nothing` to disable cookies - `max_redirects`: maximum redirect hops before failing -- `prefer_http2`: whether secure requests should try HTTP/2 when available +- `prefer_http2`: whether secure requests should try HTTP/2 when available. + When automatic negotiation (`protocol = :auto`) learns an origin only + speaks HTTP/1.1, that result is cached and later automatic requests skip + the HTTP/2 attempt; [`close_idle_connections!`](@ref) clears the cache. + An explicit `protocol = :h2` always attempts HTTP/2. - `http2_settings`: an [`HTTP2Settings`](@ref) configuring HTTP/2 receive flow-control windows for connections this client opens - `default_headers`: headers applied to every request issued through this @@ -48,6 +60,12 @@ mutable struct Client{CR} http2_settings::HTTP2Settings h2_lock::ReentrantLock h2_conns::Dict{String,Vector{H2Connection}} + # Origins that negotiated HTTP/1.1 during `protocol = :auto`. Guarded by + # its own lock (never `h2_lock`): `h2_lock` is held across full TCP+TLS + # dials in `_acquire_h2_conn!`, and the hot-path membership check in + # `_use_h2` must not serialize behind those dials. + h1_origins_lock::ReentrantLock + h1_origins::Set{String} default_headers::Headers default_query::Union{Nothing,Vector{Pair{String,String}}} default_basicauth::Any @@ -342,6 +360,8 @@ function Client(; http2_settings, ReentrantLock(), Dict{String,Vector{H2Connection}}(), + ReentrantLock(), + Set{String}(), _normalize_headers_input(default_headers), _normalize_default_query(default_query), default_basicauth, @@ -445,10 +465,16 @@ function _acquire_h2_conn!( address::String, secure::Bool, request::Union{Nothing,Request}=nothing, - server_name::Union{Nothing,String}=nothing, + server_name::Union{Nothing,String}=nothing; allow_h1_alpn::Bool=false, + auto_protocol::Bool=false, )::H2Connection key = _h2_key(plan) + # Fail fast (and without touching `h2_lock`) when another task cached this + # origin as HTTP/1.1 after our caller's `_use_h2` check. + if auto_protocol && _h1_origin_cached(client, key) + throw(H2NegotiationError("http2: origin previously negotiated HTTP/1.1")) + end base_host_resolver = client.transport.host_resolver connect_host_resolver = request === nothing ? base_host_resolver : _request_connect_host_resolver(base_host_resolver, request::Request) connect_deadline_ns = request === nothing ? _phase_deadline_ns(base_host_resolver.timeout_ns, base_host_resolver.deadline_ns) : _request_connect_phase_deadline_ns(base_host_resolver, request::Request) @@ -457,8 +483,15 @@ function _acquire_h2_conn!( # closing an H2Connection waits for its read loop to exit, and that must # not happen while holding the pool lock. to_close = H2Connection[] - lock(client.h2_lock) + waited_for_h2_lock = !trylock(client.h2_lock) + waited_for_h2_lock && lock(client.h2_lock) try + # The origin can be cached while this task waits for `h2_lock`. Only + # contended acquisitions need this second check, so the uncontended h2 + # fast path does not pay for another cache lookup. + if waited_for_h2_lock && auto_protocol && _h1_origin_cached(client, key) + throw(H2NegotiationError("http2: origin previously negotiated HTTP/1.1")) + end conns = get(() -> H2Connection[], client.h2_conns, key) idle_timeout_ns = client.transport.idle_timeout_ns now_ns = Int64(time_ns()) @@ -504,36 +537,43 @@ function _acquire_h2_conn!( nothing end conn = nothing - conn = if plan.mode == _ProxyPlanMode.DIRECT - connect_h2!( - address; - secure=secure, - host_resolver=connect_host_resolver, - tls_config=tls_cfg, - connect_deadline_ns=connect_deadline_ns, - http2_settings=client.http2_settings, - ) - elseif plan.mode == _ProxyPlanMode.HTTP_TUNNEL || _proxy_plan_is_socks(plan) - proxy = plan.proxy - proxy === nothing && throw(ProtocolError("proxy connection is missing proxy config")) - tcp = _new_tcp_conn!(plan, address, connect_host_resolver, connect_deadline_ns) - try + conn = try + if plan.mode == _ProxyPlanMode.DIRECT connect_h2!( - tcp, address; secure=secure, + host_resolver=connect_host_resolver, tls_config=tls_cfg, connect_deadline_ns=connect_deadline_ns, http2_settings=client.http2_settings, ) - catch - @try_ignore begin - TCP.close(tcp) + elseif plan.mode == _ProxyPlanMode.HTTP_TUNNEL || _proxy_plan_is_socks(plan) + proxy = plan.proxy + proxy === nothing && throw(ProtocolError("proxy connection is missing proxy config")) + tcp = _new_tcp_conn!(plan, address, connect_host_resolver, connect_deadline_ns) + try + connect_h2!( + tcp, + address; + secure=secure, + tls_config=tls_cfg, + connect_deadline_ns=connect_deadline_ns, + http2_settings=client.http2_settings, + ) + catch + @try_ignore begin + TCP.close(tcp) + end + rethrow() end - rethrow() + else + throw(ArgumentError("HTTP/2 is not supported for proxy plan mode $(plan.mode)")) end - else - throw(ArgumentError("HTTP/2 is not supported for proxy plan mode $(plan.mode)")) + catch err + if auto_protocol && _should_fallback_h2_to_h1(err) + _cache_h1_origin!(client, key) + end + rethrow() end # Claim a slot on the freshly opened connection up front so subsequent # acquirers that race in see this caller's pending request. @@ -581,7 +621,7 @@ function _drop_h2_conn!(client::Client, plan::_ProxyPlan, target::Union{Nothing, return nothing end -function _use_h2(client::Client, secure::Bool, protocol::Symbol)::Bool +function _use_h2(client::Client, plan::_ProxyPlan, secure::Bool, protocol::Symbol)::Bool protocol == :h1 && return false protocol == :h2 && return true protocol == :auto || throw(ArgumentError("protocol must be :auto, :h1, or :h2")) @@ -594,7 +634,26 @@ function _use_h2(client::Client, secure::Bool, protocol::Symbol)::Bool if cfg !== nothing && !isempty(cfg.alpn_protocols) && !in("h2", cfg.alpn_protocols) return false end - return true + return !_h1_origin_cached(client, _h2_key(plan)) +end + +@inline function _h1_origin_cached(client::Client, key::String)::Bool + lock(client.h1_origins_lock) + try + return key in client.h1_origins + finally + unlock(client.h1_origins_lock) + end +end + +function _cache_h1_origin!(client::Client, key::String)::Nothing + lock(client.h1_origins_lock) + try + push!(client.h1_origins, key) + finally + unlock(client.h1_origins_lock) + end + return nothing end function _host_path_from_request(address::String, request::Request)::Tuple{String,String} @@ -678,11 +737,10 @@ function _store_set_cookies!( end function _clone_bytes_body(body::BytesBody)::BytesBody - remaining = (length(body.data) - body.next_index) + 1 - remaining <= 0 && return BytesBody(UInt8[]) - copied = Vector{UInt8}(undef, remaining) - copyto!(copied, 1, body.data, body.next_index, remaining) - return BytesBody(copied) + # BytesBody retains its backing bytes by contract. A replay only needs an + # independent cursor and closed flag; copying the payload makes every + # buffered upload allocate and copy the complete request body again. + return BytesBody(body.data, body.next_index, false) end function _clone_body(body::AbstractBody)::AbstractBody @@ -792,7 +850,8 @@ function _do_incoming!( send_request = _copy_request_for_send(current_request, retry_attempt == 1) request_url = _request_url(current_secure, current_address, current_request.target) proxy_plan = _proxy_plan(proxy_config, current_secure, current_address) - use_h2 = _use_h2(client, current_secure, protocol) && proxy_plan.mode != _ProxyPlanMode.HTTP_FORWARD + use_h2 = proxy_plan.mode != _ProxyPlanMode.HTTP_FORWARD && + _use_h2(client, proxy_plan, current_secure, protocol) _emit_trace(trace, RequestEvent(send_request, request_url, retry_attempt, redirect_count, use_h2 ? :h2 : :h1)) host, path = _host_path_from_request(current_address, current_request) manual_cookies = cookies === false ? Cookie[] : Cookies.readcookies(send_request.headers, "") @@ -808,8 +867,9 @@ function _do_incoming!( current_address, current_secure, send_request, - current_server_name, - protocol == :auto, + current_server_name; + allow_h1_alpn=(protocol == :auto), + auto_protocol=(protocol == :auto), ) _h2_roundtrip_incoming!(conn::H2Connection, send_request; pending_slot_claimed=true) catch err @@ -1186,6 +1246,14 @@ end function close_idle_connections!(client::Client) close_idle_connections!(client.transport) + # Drop cached HTTP/1.1 negotiation results so long-lived clients re-probe + # origins that may have enabled HTTP/2 since they were first contacted. + lock(client.h1_origins_lock) + try + empty!(client.h1_origins) + finally + unlock(client.h1_origins_lock) + end # Also close pooled HTTP/2 connections with no in-flight streams — these # live on the Client (not the Transport pool) and are equally subject to # silent idle drops by NATs/load balancers (#1331). Connections carrying @@ -1238,7 +1306,7 @@ end function _read_all_response_bytes(io::IO, limit::Int=0)::Vector{UInt8} out = UInt8[] - buf = Vector{UInt8}(undef, 8192) + buf = Vector{UInt8}(undef, _RESPONSE_SMALL_COPY_BUFFER_BYTES) total = 0 while true n = readbytes!(io, buf, length(buf)) @@ -1260,7 +1328,10 @@ function _read_all_response_bytes(body::AbstractBody, content_length_hint::Int64 end out = UInt8[] content_length_hint > 0 && sizehint!(out, Int(min(content_length_hint, _MAX_EAGER_RESPONSE_PREALLOC))) - buf = Vector{UInt8}(undef, 8192) + # A hint above the prealloc cap means a known-large download; without a + # hint (chunked/EOF-framed) the body is usually small, so stay at 8 KiB. + buf_bytes = content_length_hint > _MAX_EAGER_RESPONSE_PREALLOC ? _RESPONSE_COPY_BUFFER_BYTES : _RESPONSE_SMALL_COPY_BUFFER_BYTES + buf = Vector{UInt8}(undef, buf_bytes) while true n = body_read!(body, buf) n == 0 && return out @@ -1269,7 +1340,7 @@ function _read_all_response_bytes(body::AbstractBody, content_length_hint::Int64 end function _copy_response_bytes!(dest::IO, io::IO, limit::Int=0)::Int64 - buf = Vector{UInt8}(undef, 8192) + buf = Vector{UInt8}(undef, _RESPONSE_SMALL_COPY_BUFFER_BYTES) total = Int64(0) while true n = readbytes!(io, buf, length(buf)) @@ -1281,9 +1352,12 @@ function _copy_response_bytes!(dest::IO, io::IO, limit::Int=0)::Int64 end function _copy_response_bytes!(dest::AbstractVector{UInt8}, io::IO, limit::Int=0)::Int64 - buf = Vector{UInt8}(undef, 8192) - total = 0 capacity = length(dest) + # `dest` is preallocated to the expected size; a small response should not + # pay a 64 KiB scratch allocation. The 1-byte floor keeps the capacity + # overflow check below reachable when `dest` is empty. + buf = Vector{UInt8}(undef, min(max(capacity, 1), _RESPONSE_COPY_BUFFER_BYTES)) + total = 0 while true n = readbytes!(io, buf, length(buf)) n == 0 && break @@ -1298,7 +1372,7 @@ function _copy_response_bytes!(dest::AbstractVector{UInt8}, io::IO, limit::Int=0 end function _copy_response_bytes!(dest::IO, body::AbstractBody)::Int64 - buf = Vector{UInt8}(undef, 8192) + buf = Vector{UInt8}(undef, _RESPONSE_COPY_BUFFER_BYTES) total = Int64(0) while true n = body_read!(body, buf) @@ -1309,9 +1383,10 @@ function _copy_response_bytes!(dest::IO, body::AbstractBody)::Int64 end function _copy_response_bytes!(dest::AbstractVector{UInt8}, body::AbstractBody)::Int64 - buf = Vector{UInt8}(undef, 8192) - total = 0 capacity = length(dest) + # See the sizing rationale on the `io::IO` method above. + buf = Vector{UInt8}(undef, min(max(capacity, 1), _RESPONSE_COPY_BUFFER_BYTES)) + total = 0 while true n = body_read!(body, buf) n == 0 && break @@ -1356,7 +1431,7 @@ end end function _pump_response_body!(stream::Base.BufferStream, body::AbstractBody)::Nothing - buf = Vector{UInt8}(undef, 8192) + buf = Vector{UInt8}(undef, _RESPONSE_COPY_BUFFER_BYTES) try while true n = body_read!(body, buf) @@ -1466,7 +1541,7 @@ end function Base.read(io::_BodyIO)::Vector{UInt8} out = UInt8[] - buf = Vector{UInt8}(undef, 8192) + buf = Vector{UInt8}(undef, _RESPONSE_SMALL_COPY_BUFFER_BYTES) while true n = readbytes!(io, buf) n == 0 && break diff --git a/src/http_transport.jl b/src/http_transport.jl index 36284f9ed..660bd9748 100644 --- a/src/http_transport.jl +++ b/src/http_transport.jl @@ -60,7 +60,14 @@ end return reader.stop - reader.next + 1 end +@inline function _request_write_deadline_needs_refresh(request::Request)::Bool + return _request_write_idle_timeout_ns(request) > 0 +end + @inline function _apply_request_write_deadline!(io::_RequestDeadlineWriteIO)::Nothing + # The overall request deadline is applied once when the connection is + # acquired. Only an idle timeout moves after each write and needs refresh. + _request_write_deadline_needs_refresh(io.request) || return nothing _set_conn_write_deadline!(io.conn::Conn, _request_write_deadline_ns(io.request)) return nothing end @@ -1192,7 +1199,9 @@ connection pool. The `Client` and no-argument forms also close the client's pooled HTTP/2 connections that have no in-flight streams; the `Transport` form covers only -the HTTP/1 pool it owns. +the HTTP/1 pool it owns. They additionally clear the client's cache of origins +that negotiated HTTP/1.1 under `protocol = :auto`, so subsequent automatic +requests re-attempt HTTP/2 against origins that may have enabled it since. """ function close_idle_connections!(transport::Transport) to_close = Conn[] diff --git a/test/http2_client_tests.jl b/test/http2_client_tests.jl index 5665ffc1e..93e21d6ba 100644 --- a/test/http2_client_tests.jl +++ b/test/http2_client_tests.jl @@ -2082,7 +2082,8 @@ end client = HT.Client(transport = transport, prefer_http2 = true) try # `_use_h2` should refuse h2 because the ALPN list excludes it. - @test !HT._use_h2(client, true, :auto) + plan = HT._proxy_plan(HT.ProxyConfig(), true, address) + @test !HT._use_h2(client, plan, true, :auto) # Verify end-to-end that protocol=:auto picks h1. response = HT.get!(client, address, "/"; secure = true, protocol = :auto) @test response.status == 200 @@ -2093,6 +2094,122 @@ end end end +@testset "HTTP/2 automatic acquire observes cached HTTP/1.1 origin" begin + client = HT.Client() + address = "cached-h1.example:443" + plan = HT._proxy_plan(HT.ProxyConfig(), true, address) + push!(client.h1_origins, HT._h2_key(plan)) + try + @test_throws HT.H2NegotiationError HT._acquire_h2_conn!( + client, + plan, + address, + true; + allow_h1_alpn = true, + auto_protocol = true, + ) + finally + close(client) + end +end + +@testset "HTTP/2 automatic acquire rechecks HTTP/1.1 origin after waiting" begin + # Reserve a loopback port, then close the listener. A stale HTTP/2 attempt + # will fail with a connection error instead of the expected cached-origin + # negotiation error. + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 1) + laddr = NC.addr(listener)::NC.SocketAddrV4 + address = ND.join_host_port("127.0.0.1", Int(laddr.port)) + NC.close(listener) + client = HT.Client() + plan = HT._proxy_plan(HT.ProxyConfig(), true, address) + key = HT._h2_key(plan) + started = Channel{Nothing}(4) + results = Channel{Any}(4) + tasks = Task[] + lock(client.h2_lock) + try + for _ in 1:4 + task = Threads.@spawn begin + put!(started, nothing) + result = try + HT._acquire_h2_conn!( + client, + plan, + address, + true; + allow_h1_alpn = true, + auto_protocol = true, + ) + catch ex + ex + end + put!(results, result) + end + push!(tasks, errormonitor(task)) + end + for _ in tasks + take!(started) + end + HT._cache_h1_origin!(client, key) + finally + unlock(client.h2_lock) + end + try + foreach(wait, tasks) + @test all(_ -> take!(results) isa HT.H2NegotiationError, tasks) + finally + close(client) + end +end + +@testset "HTTP/2 transient connect failure does not cache HTTP/1.1 origin" begin + # Reserve a loopback port, then close the listener so connects are refused. + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 1) + laddr = NC.addr(listener)::NC.SocketAddrV4 + address = ND.join_host_port("127.0.0.1", Int(laddr.port)) + NC.close(listener) + client = HT.Client() + plan = HT._proxy_plan(HT.ProxyConfig(), true, address) + try + err = try + HT._acquire_h2_conn!( + client, + plan, + address, + true; + allow_h1_alpn = true, + auto_protocol = true, + ) + nothing + catch ex + ex + end + # Only a genuine ALPN h1 negotiation may populate the cache; a refused + # connect must leave the origin eligible for future HTTP/2 attempts. + @test err !== nothing + @test !(err isa HT.H2NegotiationError) + @test isempty(client.h1_origins) + @test HT._use_h2(client, plan, true, :auto) + finally + close(client) + end +end + +@testset "HTTP/2 close_idle_connections! clears cached HTTP/1.1 origins" begin + client = HT.Client() + plan = HT._proxy_plan(HT.ProxyConfig(), true, "cached-h1.example:443") + push!(client.h1_origins, HT._h2_key(plan)) + try + @test !HT._use_h2(client, plan, true, :auto) + HT.close_idle_connections!(client) + @test isempty(client.h1_origins) + @test HT._use_h2(client, plan, true, :auto) + finally + close(client) + end +end + @testset "HTTP/2 client advertises configured flow-control windows" begin listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) laddr = NC.addr(listener)::NC.SocketAddrV4 diff --git a/test/http_client_proxy_tests.jl b/test/http_client_proxy_tests.jl index 1c536e1e2..749b4b9bb 100644 --- a/test/http_client_proxy_tests.jl +++ b/test/http_client_proxy_tests.jl @@ -1313,6 +1313,60 @@ end end end +@testset "HTTP proxy closes tunnel after H2 setup failure" begin + if _http_windows_ci() + @test_skip true + else + proxy_listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + proxy_addr = NC.addr(proxy_listener)::NC.SocketAddrV4 + proxy_address = ND.join_host_port("127.0.0.1", Int(proxy_addr.port)) + server_task = errormonitor(Threads.@spawn begin + conn = NC.accept(proxy_listener) + try + request = HT.read_request(HT._ConnReader(conn)) + @test request.method == "CONNECT" + @test request.target == "origin.invalid:443" + _send_response_proxy!( + conn, + request; + status = 200, + reason = "Connection Established", + headers = HT.Headers(), + ) + finally + HTTP.@try_ignore NC.close(conn) + end + return nothing + end) + client = HT.Client( + transport = HT.Transport( + proxy = HT.ProxyURL("http://$(proxy_address)"), + tls_config = TL.Config(verify_peer = false), + ), + prefer_http2 = true, + ) + try + err = try + HT.get!( + client, + "origin.invalid:443", + "/"; + secure = true, + protocol = :h2, + ) + nothing + catch ex + ex + end + @test err !== nothing + _wait_task_proxy!(server_task) + finally + close(client) + HTTP.@try_ignore NC.close(proxy_listener) + end + end +end + @testset "HTTP SOCKS5H proxy supports tunneled H2 requests" begin if _http_windows_ci() @test_skip true diff --git a/test/http_client_tests.jl b/test/http_client_tests.jl index af4550a2c..eb76c3b4e 100644 --- a/test/http_client_tests.jl +++ b/test/http_client_tests.jl @@ -1806,6 +1806,20 @@ end end end +@testset "HTTP buffered request replay retains payload storage" begin + payload = collect(codeunits("replay-body")) + body = HT.BytesBody(payload) + scratch = Vector{UInt8}(undef, 3) + @test HT.body_read!(body, scratch) == 3 + + replay = HT._clone_bytes_body(body) + @test replay !== body + @test replay.data === payload + @test replay.next_index == body.next_index + @test String(_read_all_body_bytes_client(replay)) == "lay-body" + @test body.next_index == 4 +end + @testset "HTTP high-level request body inputs" begin if _http_windows_ci() @test_skip true @@ -2451,6 +2465,16 @@ end @test legacy_config.response_header_timeout_ns == 0 @test_throws ArgumentError HT._resolve_request_timeout_settings(0, 0, 0, 0.05, 0, nothing, 0.05) + + request = HT.Request("POST", "/upload") + @test !HT._request_write_deadline_needs_refresh(request) + HT._apply_request_timeout_settings!( + HT.get_request_context(request), Int64(1_000_000_000), nothing) + @test !HT._request_write_deadline_needs_refresh(request) + _, write_idle_config = HT._resolve_request_timeout_settings(0, 0, 0, 0, 0.25) + HT._apply_request_timeout_settings!( + HT.get_request_context(request), Int64(0), write_idle_config) + @test HT._request_write_deadline_needs_refresh(request) end @testset "HTTP high-level readtimeout" begin diff --git a/test/http_integration_tests.jl b/test/http_integration_tests.jl index 3be4803ca..763d51c20 100644 --- a/test/http_integration_tests.jl +++ b/test/http_integration_tests.jl @@ -95,8 +95,10 @@ end prefer_http2 = true, ) try + protocols = Symbol[] trace = function(event) if event isa HT.RequestEvent + push!(protocols, event.protocol) HT.setheader(event.request.headers, "Authorization", "Signature trace-time") end return nothing @@ -111,6 +113,17 @@ end @test response.status == 200 @test String(response.body) == "tls-h1:/auto-tls" @test seen_authorization[] == "Signature trace-time" + + response = HT.request( + trace, + "GET", + "https://$(address)/cached-h1"; + client, + protocol = :auto, + ) + @test response.status == 200 + @test String(response.body) == "tls-h1:/cached-h1" + @test protocols == [:h2, :h1] finally close(client) HT.forceclose(server) diff --git a/test/http_server_http1_tests.jl b/test/http_server_http1_tests.jl index 2e6ea3dec..a73aafca3 100644 --- a/test/http_server_http1_tests.jl +++ b/test/http_server_http1_tests.jl @@ -99,9 +99,11 @@ function _raw_http_request( NC.closewrite(sock) end end - # master hardening kept: longer first-byte budget on Windows, - # combined with the re-dial retry below - first_byte_timeout_s = max(Sys.iswindows() ? 5.0 : 2.0, settle_s + 1.0) + # Hosted Linux runners can delay an otherwise immediate local + # response past two seconds under load. Keep a five-second + # first-byte budget on every platform. A persistent no-response + # condition still fails, and re-dial retries remain Windows-only. + first_byte_timeout_s = max(5.0, settle_s + 1.0) return _read_until_quiet( sock; timeout_s = first_byte_timeout_s,