diff --git a/CHANGELOG.md b/CHANGELOG.md index afe725271..e5bc65469 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- The HTTP/1 connection pool now honors a server-advertised + `Keep-Alive: timeout=` response header: the connection is not reused + once that window (less a one-second safety margin) has elapsed since it went + idle, whichever of it and `idle_timeout_ns` is tighter, and a hint with no + headroom (`timeout=1` or less) makes the connection non-reusable. Previously + a request could be written onto a connection the server had already closed + at its own keep-alive deadline. ([#1155]) - Added `HTTP2Settings` to configure HTTP/2 receive flow-control windows (per-stream `initial_window_size` and connection-level `connection_window_size`). Pass it via the `http2_settings` keyword on `Client`, `Server`, `listen!`, `serve!`, `serve`, and `connect_h2!`. Defaults preserve the protocol-default 65535-byte windows, and the per-stream receive buffer cap is derived from the window. Raising the windows improves single-stream throughput on links with non-trivial latency. - Added `HTTP.peeraddr(::HTTP.Stream)`, returning the remote (client) `SocketAddr` of a server stream for both plain-TCP and TLS connections and both HTTP/1 and HTTP/2. This is the supported way to obtain the client IP (for rate limiting, audit logging, and per-client policy) without reaching into transport internals, and restores the capability `Sockets.getpeername(::HTTP.Stream)` provided in HTTP.jl 1.x. - Added `HTTP.RetrySkippedEvent`, a request trace event emitted when the retry @@ -870,4 +877,5 @@ See changes for 0.9.15: this release is equivalent to 0.9.15 with [#752] reverte [#1127]: https://github.com/JuliaWeb/HTTP.jl/issues/1127 [#1277]: https://github.com/JuliaWeb/HTTP.jl/issues/1277 [#1342]: https://github.com/JuliaWeb/HTTP.jl/issues/1342 +[#1155]: https://github.com/JuliaWeb/HTTP.jl/issues/1155 [#1353]: https://github.com/JuliaWeb/HTTP.jl/issues/1353 diff --git a/src/http_transport.jl b/src/http_transport.jl index 28842e0f6..cfefe8bba 100644 --- a/src/http_transport.jl +++ b/src/http_transport.jl @@ -156,6 +156,9 @@ mutable struct Conn @atomic closed::Bool @atomic slot_released::Bool last_used_ns::Int64 + # Server-advertised keep-alive idle bound (`Keep-Alive: timeout=`, less + # `_KEEPALIVE_TIMEOUT_MARGIN_NS`) in ns; `0` when the peer gave no hint. + keepalive_timeout_ns::Int64 end const _CONN_WAITER_WAITING = UInt8(0) @@ -207,6 +210,12 @@ connection may be handed out again; it also governs the owning `Client`'s pooled HTTP/2 connections. Keep it under the network's silent idle-drop window (NAT/load-balancer idle timeouts are commonly a few minutes) so a request is never written onto a connection whose peer has silently vanished. + +A server that advertises its own idle window on a response +(`Keep-Alive: timeout=`) additionally bounds that HTTP/1 connection's +reuse to the advertised time less a one-second safety margin; the tighter of +the two limits applies. A hint that leaves no headroom (`timeout=1` or less) +makes the connection non-reusable. """ mutable struct Transport host_resolver::_TransportHostResolver @@ -992,7 +1001,7 @@ function _new_conn_tcp!( connect_deadline_ns::Int64=Int64(0), )::Conn tcp = _new_tcp_conn!(plan, address, host_resolver, connect_deadline_ns) - return Conn(plan.pool_key, plan.first_hop_address, false, tcp, nothing, _ConnReader(tcp), IOBuffer(), false, false, false, time_ns()) + return Conn(plan.pool_key, plan.first_hop_address, false, tcp, nothing, _ConnReader(tcp), IOBuffer(), false, false, false, time_ns(), Int64(0)) end function _new_conn_tls!( @@ -1010,7 +1019,7 @@ function _new_conn_tls!( tls = TLS.client(tcp, cfg) connect_deadline_ns == 0 || TLS.set_deadline!(tls, connect_deadline_ns) TLS.handshake!(tls) - return Conn(plan.pool_key, plan.first_hop_address, true, tcp, tls, _ConnReader(tls), IOBuffer(), false, false, false, time_ns()) + return Conn(plan.pool_key, plan.first_hop_address, true, tcp, tls, _ConnReader(tls), IOBuffer(), false, false, false, time_ns(), Int64(0)) catch err @try_ignore TCP.close(tcp) # Type TLS failures at the site where the phase is known: a TLSError @@ -1028,7 +1037,9 @@ function _evict_expired_idle_locked!(transport::Transport, key::String, now_ns:: kept = Conn[] stale = Conn[] for conn in idle_list::Vector{Conn} - expired = transport.idle_timeout_ns > 0 && (now_ns - conn.last_used_ns) > transport.idle_timeout_ns + age_ns = now_ns - conn.last_used_ns + expired = (transport.idle_timeout_ns > 0 && age_ns > transport.idle_timeout_ns) || + (conn.keepalive_timeout_ns > 0 && age_ns > conn.keepalive_timeout_ns) if _conn_closed(conn) || expired @atomic :acquire_release transport.idle_total -= 1 push!(stale, conn) @@ -1505,6 +1516,75 @@ end ) end +# Servers that bound keep-alive idleness advertise it as +# `Keep-Alive: timeout=[, max=]` (RFC 2068 §19.7.1; still emitted by +# Apache, nginx, HAProxy and others). Reusing a pooled connection after that +# window has elapsed races the server's close: an idempotent request burns a +# retry, anything else fails outright. The hint is therefore honored as a +# per-connection idle bound alongside `idle_timeout_ns`. The server's clock +# started when it finished *sending* the response, before the client finished +# reading it, so a safety margin is subtracted (as Node's undici does with its +# `keepAliveTimeoutThreshold`); a hint that leaves no headroom marks the +# connection as not reusable at all. The `max=` parameter is ignored: a +# server that exhausts it answers with `Connection: close`, which is already +# honored. +const _KEEPALIVE_TIMEOUT_MARGIN_NS = Int64(1_000_000_000) +const _KEEPALIVE_TIMEOUT_MAX_S = typemax(Int64) ÷ Int64(1_000_000_000) + +""" + _keepalive_timeout_hint_ns(hdrs::Headers) -> Int64 + +Return the `timeout` parameter of the response's `Keep-Alive` header in +nanoseconds, or `-1` when the header is absent or carries no well-formed +(unsigned decimal) `timeout`. Parameter names are case-insensitive, a +quoted value is accepted, and an absurdly large value saturates. +""" +function _keepalive_timeout_hint_ns(hdrs::Headers)::Int64 + for value in headers(hdrs, "Keep-Alive") + for param in eachsplit(value, ',') + eq = findfirst(==('='), param) + eq === nothing && continue + name = strip(SubString(param, firstindex(param), prevind(param, eq))) + _ascii_lowercase_equal(name, "timeout") || continue + raw = strip(SubString(param, nextind(param, eq))) + if ncodeunits(raw) >= 2 && startswith(raw, '"') && endswith(raw, '"') + raw = strip(SubString(raw, nextind(raw, firstindex(raw)), prevind(raw, lastindex(raw)))) + end + (!isempty(raw) && all(isdigit, raw)) || continue + secs = tryparse(Int64, raw) + secs === nothing && (secs = _KEEPALIVE_TIMEOUT_MAX_S) + return min(secs, _KEEPALIVE_TIMEOUT_MAX_S) * Int64(1_000_000_000) + end + end + return Int64(-1) +end + +@inline function _ascii_lowercase_equal(a::AbstractString, lower::String)::Bool + ncodeunits(a) == ncodeunits(lower) || return false + for (ca, cb) in zip(codeunits(a), codeunits(lower)) + (ca == cb || (UInt8('A') <= ca <= UInt8('Z') && ca + 0x20 == cb)) || return false + end + return true +end + +""" + _note_keepalive_hint!(conn::Conn, hdrs::Headers) -> Bool + +Record the server-advertised keep-alive idle bound from a response on `conn` +and return whether the connection may be pooled at all. Without a hint the +pool's `idle_timeout_ns` alone applies. +""" +function _note_keepalive_hint!(conn::Conn, hdrs::Headers)::Bool + hint_ns = _keepalive_timeout_hint_ns(hdrs) + bound_ns = hint_ns < 0 ? Int64(0) : hint_ns - _KEEPALIVE_TIMEOUT_MARGIN_NS + if hint_ns >= 0 && bound_ns <= 0 + conn.keepalive_timeout_ns = Int64(0) + return false + end + conn.keepalive_timeout_ns = bound_ns + return true +end + @inline function _response_reusable(response::_IncomingResponse, request::Request)::Bool response.head.close && return false _transport_request_wants_close(request) && return false @@ -1944,6 +2024,7 @@ function _roundtrip_incoming!( end reusable = _response_reusable(raw_response, attempt_request) early_final && (reusable = false) + reusable && (reusable = _note_keepalive_hint!((raw_response.rawbody::H1Body).conn, raw_response.head.headers)) body = _arm_h1_body!(raw_response.rawbody::H1Body, reusable, request_ctx, cancel_cb) if _body_immediately_empty(body) body_close!(body) diff --git a/test/http_client_transport_tests.jl b/test/http_client_transport_tests.jl index a0f1243c9..d4bdb4f51 100644 --- a/test/http_client_transport_tests.jl +++ b/test/http_client_transport_tests.jl @@ -1665,3 +1665,145 @@ end wait(server) end end + +@testset "Keep-Alive timeout hint parsing (#1155)" begin + hint = value -> HT._keepalive_timeout_hint_ns(HT.Headers("Keep-Alive" => value)) + @test hint("timeout=5") == 5_000_000_000 + @test hint("timeout=5, max=100") == 5_000_000_000 + @test hint("max=100, timeout=7") == 7_000_000_000 + @test hint("Timeout = 3") == 3_000_000_000 + @test hint("timeout=\"4\"") == 4_000_000_000 + @test hint("timeout=0") == 0 + @test hint("max=100") == -1 + @test hint("timeout=abc") == -1 + @test hint("timeout=-5") == -1 + @test hint("timeout=+5") == -1 + @test hint("timeout=1.5") == -1 + @test hint("timeout=") == -1 + @test hint("timeout") == -1 + @test hint("") == -1 + @test HT._keepalive_timeout_hint_ns(HT.Headers()) == -1 + @test hint("timeout=99999999999999999999") == HT._KEEPALIVE_TIMEOUT_MAX_S * 1_000_000_000 + # Multiple header lines: the first well-formed timeout wins. + multi = HT.Headers("Keep-Alive" => "max=5", "Keep-Alive" => "timeout=2") + @test HT._keepalive_timeout_hint_ns(multi) == 2_000_000_000 +end + +@testset "HTTP client transport evicts an idle conn past its server keep-alive window (#1155)" begin + listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8) + address = ND.join_host_port("127.0.0.1", Int((NC.addr(listener)::NC.SocketAddrV4).port)) + accepted = Channel{NC.Conn}(2) + server_task = Threads.@spawn begin + for _ in 1:2 + put!(accepted, NC.accept(listener)) + end + return nothing + end + # The pool-wide idle timeout is disabled, so only the per-connection + # server hint can expire a pooled connection here. + transport = HT.Transport( + max_idle_per_host = 2, + max_idle_total = 2, + max_conns_per_host = 2, + idle_timeout_ns = 0, + ) + plan = HT._proxy_plan(transport.proxy, false, address) + stale_conn = nothing + fresh_conn = nothing + acquired_conn = nothing + server_conns = NC.Conn[] + try + stale_conn = HT._acquire_conn!(transport, plan, address, false, nothing) + push!(server_conns, take!(accepted)) + fresh_conn = HT._acquire_conn!(transport, plan, address, false, nothing) + push!(server_conns, take!(accepted)) + _wait_task!(server_task) + + HT._put_idle_conn!(transport, stale_conn::HT.Conn) + HT._put_idle_conn!(transport, fresh_conn::HT.Conn) + (stale_conn::HT.Conn).keepalive_timeout_ns = 1 + (stale_conn::HT.Conn).last_used_ns = 0 + (fresh_conn::HT.Conn).keepalive_timeout_ns = 1 + (fresh_conn::HT.Conn).last_used_ns = typemax(Int64) + + acquired_conn = HT._acquire_conn!(transport, plan, address, false, nothing) + @test acquired_conn === fresh_conn + @test HT._conn_closed(stale_conn::HT.Conn) + @test (@atomic transport.idle_total) == 0 + + HT._close_owned_conn!(transport, acquired_conn::HT.Conn) + acquired_conn = nothing + fresh_conn = nothing + finally + acquired_conn === nothing || HT._close_owned_conn!(transport, acquired_conn::HT.Conn) + fresh_conn === nothing || HT._close_owned_conn!(transport, fresh_conn::HT.Conn) + stale_conn === nothing || HT._close_owned_conn!(transport, stale_conn::HT.Conn) + close(transport) + for conn in server_conns + HTTP.@try_ignore NC.close(conn) + end + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(server_task) + end +end + +@testset "HTTP client transport honors the server Keep-Alive timeout hint end to end (#1155)" begin + for (hint, expect_reuse, expect_bound_ns) in ( + ("timeout=5, max=100", true, 4_000_000_000), + ("max=100", true, 0), + ("timeout=1", false, 0), + ) + 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)) + response_bytes = collect(codeunits("HTTP/1.1 200 OK\r\nContent-Length: 2\r\nKeep-Alive: $(hint)\r\n\r\nok")) + server_task = errormonitor(Threads.@spawn begin + nconns = 0 + served = 0 + while served < 2 + conn = NC.accept(listener) + nconns += 1 + served_here = 0 + reader = HT._ConnReader(conn) + try + while served < 2 + request = HT.read_request(reader) + _read_all_transport_body_bytes(request.body) + _write_all_tcp!(conn, response_bytes) + served += 1 + served_here += 1 + end + catch err + # A non-reusable client closes between requests. + served_here > 0 || rethrow() + finally + HTTP.@try_ignore NC.close(conn) + end + end + return nconns + end) + transport = HT.Transport(max_idle_per_host = 2, max_idle_total = 2) + try + for i in 1:2 + request = HT.Request("GET", "/$(i)"; host = address, body = HT.EmptyBody(), content_length = 0) + response = HT.roundtrip!(transport, address, request) + @test String(_read_all_transport_body_bytes(response.body)) == "ok" + HT.body_close!(response.body) + if expect_reuse + @test HT.idle_connection_count(transport) == 1 + pooled = lock(transport.lock) do + first(first(values(transport.idle))) + end + @test pooled.keepalive_timeout_ns == expect_bound_ns + else + @test HT.idle_connection_count(transport) == 0 + end + end + @test fetch(server_task) == (expect_reuse ? 1 : 2) + finally + close(transport) + HTTP.@try_ignore NC.close(listener) + HTTP.@try_ignore wait(server_task) + end + end +end