Evict idle pooled HTTP/2 connections; cover them in close_idle_connections! (#1331)

src/http2_client.jl

@@ -187,6 +187,14 @@ mutable struct H2Connection
     # `_h2_conn_available` so a second caller doesn't decide a peer-capped
     # connection is reusable while the first caller is still in flight.
     pending_stream_count::Int
+    # `time_ns()` at the moment the connection last became fully idle (no
+    # registered streams and no pending reservations); stamped at creation and
+    # on each transition into that state. Guarded by `state_lock`. The pool
+    # uses it to evict connections idle past the transport's `idle_timeout_ns`:
+    # NATs and load balancers silently drop idle flows, and a request written
+    # onto such a connection stalls until the kernel's retransmission limit
+    # (~15 min on Linux) because no RST/FIN ever arrives (#1331).
+    idle_since_ns::Int64
     @atomic closed::Bool
 end
 
@@ -539,13 +547,59 @@ function _release_h2_pending_slot!(conn::H2Connection)
     lock(conn.state_lock)
     try
         conn.pending_stream_count -= 1
+        _h2_stamp_idle_locked!(conn)
         notify(conn.stream_condition; all=true)
     finally
         unlock(conn.state_lock)
     end
     return nothing
 end
 
+# Record the moment the connection becomes fully idle (no registered streams,
+# no pending reservations). Callers must hold `state_lock`. Stamping only on
+# the transition into the idle state keeps `idle_since_ns` monotone with real
+# inactivity: a connection that keeps serving traffic is never considered idle,
+# and one that goes quiet ages from the moment its last stream finished.
+@inline function _h2_stamp_idle_locked!(conn::H2Connection)
+    if isempty(conn.streams) && conn.pending_stream_count <= 0
+        conn.idle_since_ns = Int64(time_ns())
+    end
+    return nothing
+end
+
+"""
+    _h2_conn_idle_expired(conn, idle_timeout_ns, now_ns) -> Bool
+
+`true` when `conn` has no registered streams or pending reservations and has
+been idle longer than `idle_timeout_ns` (`0` disables the check). Pooled HTTP/2
+connections idle past the network's silent-drop window (NAT/load-balancer idle
+timeouts are commonly a few minutes) must not be handed out again: the peer is
+gone but no RST/FIN ever arrives, so the next request stalls until the kernel
+retransmission limit (#1331).
+"""
+function _h2_conn_idle_expired(conn::H2Connection, idle_timeout_ns::Int64, now_ns::Int64)::Bool
+    idle_timeout_ns > 0 || return false
+    lock(conn.state_lock)
+    try
+        isempty(conn.streams) || return false
+        conn.pending_stream_count > 0 && return false
+        return (now_ns - conn.idle_since_ns) > idle_timeout_ns
+    finally
+        unlock(conn.state_lock)
+    end
+end
+
+# `true` when the connection carries no registered streams and no pending
+# reservations right now (regardless of how long it has been that way).
+function _h2_conn_fully_idle(conn::H2Connection)::Bool
+    lock(conn.state_lock)
+    try
+        return isempty(conn.streams) && conn.pending_stream_count <= 0
+    finally
+        unlock(conn.state_lock)
+    end
+end
+
 function _h2_conn_reusable(conn::H2Connection)::Bool
     lock(conn.state_lock)
     try
@@ -695,6 +749,7 @@ function _unregister_stream!(conn::H2Connection, stream_id::UInt32)
         pop!(conn.streams, stream_id, nothing)
         pop!(conn.stream_send_window, stream_id, nothing)
         delete!(conn.send_closed_streams, stream_id)
+        _h2_stamp_idle_locked!(conn)
         notify(conn.stream_condition; all=true)
     finally
         unlock(conn.state_lock)
@@ -1195,6 +1250,7 @@ function _connect_h2_from_tcp!(
             nothing,
             UInt8[],
             0,
+            Int64(time_ns()),
             false,
         )
         _verify_h2_alpn!(conn)

src/http_client.jl

@@ -453,18 +453,31 @@ function _acquire_h2_conn!(
     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)
     tls_handshake_timeout_ns = request === nothing ? Int64(0) : _request_connect_phase_timeout_ns(base_host_resolver, request::Request)
+    # Connections culled from the pool are closed after `h2_lock` is released:
+    # 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)
     try
         conns = get(() -> H2Connection[], client.h2_conns, key)
+        idle_timeout_ns = client.transport.idle_timeout_ns
+        now_ns = Int64(time_ns())
         i = 1
         while i <= length(conns)
             existing = conns[i]
+            # Evict connections idle past the pool's idle timeout before
+            # considering them for reuse: a NAT or load balancer may have
+            # silently dropped the flow, and the first read on such a
+            # connection stalls until the kernel retransmission limit (#1331).
+            if _h2_conn_idle_expired(existing::H2Connection, idle_timeout_ns, now_ns)
+                deleteat!(conns, i)
+                push!(to_close, existing::H2Connection)
+                continue
+            end
             if !_h2_conn_available(existing::H2Connection)
                 if !_h2_conn_reusable(existing::H2Connection)
                     deleteat!(conns, i)
-                    @try_ignore begin
-                        close(existing::H2Connection)
-                    end
+                    push!(to_close, existing::H2Connection)
                     continue
                 end
             else
@@ -530,6 +543,11 @@ function _acquire_h2_conn!(
         return conn::H2Connection
     finally
         unlock(client.h2_lock)
+        for culled in to_close
+            @try_ignore begin
+                close(culled)
+            end
+        end
     end
 end
 
@@ -1167,7 +1185,42 @@ function _default_client!()::Client{Nothing}
     end
 end
 
-close_idle_connections!(client::Client) = close_idle_connections!(client.transport)
+function close_idle_connections!(client::Client)
+    close_idle_connections!(client.transport)
+    # 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
+    # active streams or pending reservations are left untouched. Closing
+    # happens after `h2_lock` is released (close waits for the read loop).
+    to_close = H2Connection[]
+    lock(client.h2_lock)
+    try
+        for key in collect(keys(client.h2_conns))
+            conns = client.h2_conns[key]
+            kept = H2Connection[]
+            for conn in conns
+                if _h2_conn_fully_idle(conn)
+                    push!(to_close, conn)
+                else
+                    push!(kept, conn)
+                end
+            end
+            if isempty(kept)
+                pop!(client.h2_conns, key, nothing)
+            else
+                client.h2_conns[key] = kept
+            end
+        end
+    finally
+        unlock(client.h2_lock)
+    end
+    for conn in to_close
+        @try_ignore begin
+            close(conn)
+        end
+    end
+    return nothing
+end
 
 function close_idle_connections!()
     lock(_DEFAULT_CLIENT_LOCK)
@@ -1177,7 +1230,7 @@ function close_idle_connections!()
         unlock(_DEFAULT_CLIENT_LOCK)
     end
     client === nothing && return nothing
-    return close_idle_connections!(client.transport)
+    return close_idle_connections!(client)
 end
 
 function _status_throws(resp::Response)::Bool

src/http_transport.jl

@@ -193,6 +193,12 @@ versus closed.
 bound the total live HTTP/1 connections (idle, in-flight, and dialing) for one
 pool key and cause additional acquires to wait for direct handoff or a freed
 dial slot.
+
+`idle_timeout_ns` (default 90s, `0` disables) bounds how long an idle pooled
+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.
 """
 mutable struct Transport
     host_resolver::_TransportHostResolver
@@ -1183,6 +1189,10 @@ The no-argument form closes idle connections held by the default client used by
 `HTTP.get`, `HTTP.post`, `HTTP.request`, and friends (a no-op if no request has
 been made yet). Pass an `HTTP.Client` or `HTTP.Transport` to target a specific
 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.
 """
 function close_idle_connections!(transport::Transport)
     to_close = Conn[]

test/http2_client_tests.jl

@@ -2102,6 +2102,7 @@ function _build_bare_h2_connection()
         nothing,
         UInt8[],
         0,
+        Int64(time_ns()),
         false,
     )
     cleanup = () -> begin
@@ -2271,3 +2272,161 @@ end
         cleanup()
     end
 end
+
+# ── #1331: pooled h2 connections must not outlive the idle timeout ──────────
+
+function _h2_serve_one_get!(c::NC.Conn, reader, enc, dec)
+    sid = UInt32(0)
+    while true
+        f = HT.read_frame!(reader)
+        if f isa HT.HeadersFrame
+            hf = f::HT.HeadersFrame
+            _ = HT.decode_header_block(dec, hf.header_block_fragment)
+            sid = hf.stream_id
+            hf.end_stream && break
+        elseif f isa HT.DataFrame && (f::HT.DataFrame).end_stream
+            break
+        end
+    end
+    blk = HT.encode_header_block(enc, HT.HeaderField[HT.HeaderField(":status", "200", false)])
+    _write_frame_to_conn!(c, HT.HeadersFrame(sid, true, true, blk))
+    return nothing
+end
+
+function _h2_handshake_server!(c::NC.Conn)
+    reader = HT._ConnReader(c)
+    _ = _read_exact_h2_tcp!(c, length(HT._H2_PREFACE))
+    _ = HT.read_frame!(reader)
+    _write_frame_to_conn!(c, HT.SettingsFrame(false, Pair{UInt16, UInt32}[]))
+    return reader
+end
+
+@testset "HTTP/2 pool evicts connections idle past the transport idle timeout (#1331)" begin
+    # NATs/load balancers silently drop idle flows; a pooled connection older
+    # than the idle timeout must be discarded at checkout, not handed out (the
+    # first read on a silently-dead connection stalls until the kernel
+    # retransmission limit, ~15 min on Linux).
+    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))
+    accepts = Threads.Atomic{Int}(0)
+    server_task = errormonitor(Threads.@spawn begin
+        # conn 1: answer one request, then go mute (stand-in for a dead path)
+        c1 = NC.accept(listener)
+        Threads.atomic_add!(accepts, 1)
+        reader1 = _h2_handshake_server!(c1)
+        enc1 = HT.Encoder(); dec1 = HT.Decoder()
+        _h2_serve_one_get!(c1, reader1, enc1, dec1)
+        # conn 2: the client must arrive here on a fresh connection
+        c2 = NC.accept(listener)
+        Threads.atomic_add!(accepts, 1)
+        reader2 = _h2_handshake_server!(c2)
+        enc2 = HT.Encoder(); dec2 = HT.Decoder()
+        _h2_serve_one_get!(c2, reader2, enc2, dec2)
+        HTTP.@try_ignore NC.close(c1)
+        HTTP.@try_ignore NC.close(c2)
+        return nothing
+    end)
+    client = HT.Client(transport = HT.Transport(idle_timeout_ns = Int64(150_000_000)))  # 150ms
+    try
+        r1 = HT.get("http://$(address)/one"; client = client, protocol = :h2)
+        @test r1.status == 200
+        pooled = lock(client.h2_lock) do
+            only(only(values(client.h2_conns)))
+        end
+        sleep(0.4)   # exceed the idle timeout
+        started = time()
+        r2 = HT.get("http://$(address)/two"; client = client, protocol = :h2, retry = false)
+        elapsed = time() - started
+        @test r2.status == 200
+        @test elapsed < 5.0            # a reused dead connection would hang
+        @test accepts[] == 2           # request 2 ran on a fresh connection
+        # the evicted connection was closed, and the pool holds only the new one
+        @test timedwait(() -> (@atomic :acquire pooled.closed), 5.0; pollint = 0.01) == :ok
+        n = lock(client.h2_lock) do
+            sum(length(v) for v in values(client.h2_conns); init = 0)
+        end
+        @test n == 1
+        _wait_task_h2!(server_task)
+    finally
+        close(client)
+        HTTP.@try_ignore NC.close(listener)
+    end
+end
+
+@testset "HTTP/2 pool reuses fresh connections and honors idle_timeout_ns=0 (#1331)" 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))
+    accepts = Threads.Atomic{Int}(0)
+    server_task = errormonitor(Threads.@spawn begin
+        c = NC.accept(listener)
+        Threads.atomic_add!(accepts, 1)
+        reader = _h2_handshake_server!(c)
+        enc = HT.Encoder(); dec = HT.Decoder()
+        for _ in 1:3
+            _h2_serve_one_get!(c, reader, enc, dec)
+        end
+        HTTP.@try_ignore NC.close(c)
+        return nothing
+    end)
+    # idle_timeout_ns = 0 disables eviction entirely (matching the h1 pool),
+    # and a fresh connection inside the window must be reused, not evicted.
+    client = HT.Client(transport = HT.Transport(idle_timeout_ns = Int64(0)))
+    try
+        for i in 1:3
+            r = HT.get("http://$(address)/r$(i)"; client = client, protocol = :h2)
+            @test r.status == 200
+            sleep(0.05)
+        end
+        @test accepts[] == 1   # every request reused the single pooled conn
+        _wait_task_h2!(server_task)
+    finally
+        close(client)
+        HTTP.@try_ignore NC.close(listener)
+    end
+end
+
+@testset "close_idle_connections! closes idle pooled HTTP/2 connections (#1331)" 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))
+    server_task = errormonitor(Threads.@spawn begin
+        c = NC.accept(listener)
+        reader = _h2_handshake_server!(c)
+        enc = HT.Encoder(); dec = HT.Decoder()
+        _h2_serve_one_get!(c, reader, enc, dec)
+        # linger so closing is client-initiated
+        HTTP.@try_ignore begin
+            while true
+                _ = HT.read_frame!(reader)
+            end
+        end
+        HTTP.@try_ignore NC.close(c)
+        return nothing
+    end)
+    client = HT.Client()
+    try
+        r = HT.get("http://$(address)/warm"; client = client, protocol = :h2)
+        @test r.status == 200
+        conn = lock(client.h2_lock) do
+            only(only(values(client.h2_conns)))
+        end
+        # a connection with a pending reservation is active: it must be kept
+        @test HT._try_claim_h2_pending_slot!(conn)
+        HT.close_idle_connections!(client)
+        kept = lock(client.h2_lock) do
+            sum(length(v) for v in values(client.h2_conns); init = 0)
+        end
+        @test kept == 1
+        @test !(@atomic :acquire conn.closed)
+        # once fully idle again, it is closed and removed
+        HT._release_h2_pending_slot!(conn)
+        HT.close_idle_connections!(client)
+        remaining = lock(client.h2_lock) do
+            sum(length(v) for v in values(client.h2_conns); init = 0)
+        end
+        @test remaining == 0
+        @test timedwait(() -> (@atomic :acquire conn.closed), 5.0; pollint = 0.01) == :ok
+    finally
+        close(client)
+        HTTP.@try_ignore NC.close(listener)
+    end
+end