Honor server Keep-Alive timeout hints in the HTTP/1 connection pool

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=<seconds>` 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

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=<s>`, 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=<seconds>`) 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=<seconds>[, max=<n>]` (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=<n>` 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)

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