From 96956ea1c1219d4df63b0af40339de8d12f53b4f Mon Sep 17 00:00:00 2001 From: Benoit Chesneau Date: Mon, 21 Sep 2026 20:29:19 +0200 Subject: [PATCH 1/2] Keep pooled HTTP/2 connections alive when their dialer exits A pooled HTTP/2 connection stayed owned by the caller that dialed it, so when that caller exited the connection stopped and every other caller's request on it failed with {error, closed}. A shared connection now has no owner. hackney shares it before registering it with the pool, each stream monitors its own caller and is reset if that caller dies, and the connection closes itself once it has had no open stream for the pool timeout. The pool stays a registry and releases the per-host slot when the connection stops. Connections that are not pooled keep their owner. Unregistering a checked-out HTTP/2 connection also dropped the pool's monitor on it, so its per-host slot was never released; it now keeps the monitor. Reported and diagnosed by @smartinio in #937, whose per-stream tracking this builds on. --- NEWS.md | 9 + src/hackney.erl | 4 +- src/hackney_conn.erl | 261 +++++++++++++++----- src/hackney_pool.erl | 21 +- test/hackney_http2_shared_conn_tests.erl | 296 +++++++++++++++++++++++ 5 files changed, 520 insertions(+), 71 deletions(-) create mode 100644 test/hackney_http2_shared_conn_tests.erl diff --git a/NEWS.md b/NEWS.md index 575033f7..f226df12 100644 --- a/NEWS.md +++ b/NEWS.md @@ -5,6 +5,15 @@ unreleased ### Fixed +- A pooled HTTP/2 connection no longer closes when the caller that opened it + exits. It stayed owned by that caller, so its exit failed every other + caller's request on the connection with `{error, closed}`. A shared + connection now has no owner: each stream is tied to its own caller and is + reset if that caller dies, and the connection closes itself once it has had + no open stream for the pool `timeout` (#937, thanks @smartinio). +- Unregistering a pooled HTTP/2 connection no longer leaks its per-host slot. + The pool dropped its monitor on the connection, so the slot was never + released when the connection stopped. - A request that races a peer-initiated close now returns `{error, closed}` instead of `{error, invalid_state}`. A connection that sees the peer close stays alive briefly so late calls get an answer, and during that window every diff --git a/src/hackney.erl b/src/hackney.erl index 889e5ba1..b7cf6449 100644 --- a/src/hackney.erl +++ b/src/hackney.erl @@ -437,7 +437,9 @@ connect_pool_ssl(Transport, Host, Port, Options, FinalSslOpts, PoolHandler) -> maybe_register_h2(ConnPid, Host, Port, Transport, Options, PoolHandler) -> try hackney_conn:get_protocol(ConnPid) of http2 -> - %% HTTP/2 negotiated - register for connection sharing + %% HTTP/2 negotiated - register for connection sharing. Share it first + %% so it no longer dies with this caller while other callers use it. + _ = hackney_conn:share_h2(ConnPid), PoolHandler:register_h2(Host, Port, Transport, ConnPid, Options); http1 -> ok; diff --git a/src/hackney_conn.erl b/src/hackney_conn.erl index ca54fb99..ab1c1430 100644 --- a/src/hackney_conn.erl +++ b/src/hackney_conn.erl @@ -87,6 +87,7 @@ set_owner/2, set_owner/3, set_owner_async/2, + share_h2/1, %% Protocol info get_protocol/1 ]). @@ -139,7 +140,7 @@ %% State data record -record(conn_data, { %% Connection owner - owner :: pid(), + owner :: pid() | undefined, owner_mon :: reference() | undefined, %% Connection identity @@ -228,6 +229,11 @@ %% {stream, body_full, Status, Headers, Acc, From} %% {stream, done, Status, Headers, Buffer} h2_streams = #{} :: #{pos_integer() => {term(), tuple()}}, + %% Shared through the pool (share_h2/1): no owner, each stream monitors + %% its caller, and the connection closes itself once idle. + h2_shared = false :: boolean(), + %% Per-stream caller monitors of a shared connection: StreamId => ref + h2_stream_monitors = #{} :: #{pos_integer() => reference()}, %% Current HTTP/2 stream ID for streaming body mode (body = stream) h2_stream_id :: pos_integer() | undefined, %% Per-stream recv_timeout watchdog timers (sync one-shot reads): @@ -614,6 +620,15 @@ set_owner(Pid, NewOwner, Timeout) -> set_owner_async(Pid, NewOwner) -> gen_statem:cast(Pid, {set_owner, NewOwner}). +%% @doc Share a pooled HTTP/2 connection between callers. +%% The connection stops monitoring its owner, so the caller that dialed it +%% can exit without tearing down other callers' streams. Each stream is tied +%% to its own caller instead, and the connection closes itself after +%% `idle_timeout' with no stream open. Only pooled connections can be shared. +-spec share_h2(pid()) -> ok | {error, term()}. +share_h2(Pid) -> + safe_call(Pid, share_h2, 5000). + %% @doc Check if the connection's socket is still healthy. %% Returns ok if socket is open, {error, closed} otherwise. -spec verify_socket(pid()) -> ok | {error, closed | term()}. @@ -863,11 +878,29 @@ connected(enter, OldState, #conn_data{transport = Transport, socket = Socket, %% Transfer ownership back to pool and notify it auto_release_to_pool(Data) end, - case Timeout of - infinity -> {keep_state, Data2}; + case {Timeout, Data2#conn_data.protocol} of + {infinity, _} -> {keep_state, Data2}; + %% HTTP/2 multiplexes: the connection idles when its last stream ends, + %% not when it enters `connected' (#836). + {_, http2} -> {keep_state, Data2, h2_idle_actions(Data2)}; _ -> {keep_state, Data2, [{state_timeout, Timeout, idle_timeout}]} end; +connected({call, From}, share_h2, #conn_data{h2_shared = true}) -> + {keep_state_and_data, [{reply, From, ok}]}; + +connected({call, From}, share_h2, #conn_data{protocol = http2, pool_pid = PoolPid, + owner_mon = OwnerMon} = Data) + when is_pid(PoolPid) -> + %% The dialing caller no longer owns the connection: its exit must not + %% close streams other callers opened. Streams opened from now on monitor + %% their own caller, and the pool releases the per-host slot when the + %% connection stops. + demonitor(OwnerMon, [flush]), + Data2 = Data#conn_data{owner = undefined, owner_mon = undefined, + h2_shared = true}, + {keep_state, Data2, [{reply, From, ok} | h2_idle_actions(Data2)]}; + connected({call, From}, release_to_pool, #conn_data{pool_pid = PoolPid, owner_mon = OldMon, transport = Transport, socket = Socket} = Data) -> %% Reset owner to pool before notifying, to avoid deadlock @@ -890,6 +923,13 @@ connected({call, From}, release_to_pool, #conn_data{pool_pid = PoolPid, owner_mo notify_pool_available_sync(Data2), {keep_state, Data2, [{reply, From, ok}]}; +connected({call, From}, {set_owner, _NewOwner}, #conn_data{h2_shared = true}) -> + %% A shared connection has no owner to hand over. + {keep_state_and_data, [{reply, From, ok}]}; + +connected(cast, {set_owner, _NewOwner}, #conn_data{h2_shared = true}) -> + keep_state_and_data; + connected({call, From}, {set_owner, NewOwner}, #conn_data{owner_mon = OldMon} = Data) -> %% Update owner - demonitor old, monitor new demonitor(OldMon, [flush]), @@ -1895,6 +1935,28 @@ handle_common(cast, stop, _State, Data) -> %% Async stop - used by pool to avoid deadlock during sync checkin {stop, normal, Data}; +handle_common(info, {'DOWN', Ref, process, _Pid, _Reason}, State, Data) -> + handle_h2_stream_owner_down(Ref, State, Data); + +handle_common({timeout, h2_idle}, h2_idle, connected, + #conn_data{h2_shared = true, h2_streams = Streams, + h2_conn = H2Conn, h2_mon = H2Mon} = Data) + when map_size(Streams) =:= 0 -> + %% Idle with no stream open: close like a GOAWAY. The conn stops after + %% the closed grace window and the pool drops it on DOWN. + _ = case H2Mon of + undefined -> ok; + _ -> erlang:demonitor(H2Mon, [flush]) + end, + close_h2(H2Conn), + {next_state, closed, Data#conn_data{h2_conn = undefined, h2_mon = undefined, + socket = undefined}}; + +handle_common({timeout, h2_idle}, h2_idle, _State, _Data) -> + %% A stream opened since the timer was armed; the last one to finish + %% arms it again. + keep_state_and_data; + handle_common(cast, _Msg, _State, _Data) -> keep_state_and_data; @@ -2998,7 +3060,8 @@ start_h2_connection(Socket, Data, From, Origin) -> NewData = Data#conn_data{ h2_conn = H2Conn, h2_mon = Mon, - h2_streams = #{} + h2_streams = #{}, + h2_stream_monitors = #{} }, %% Cancel any pending idle_timeout armed by the %% TCP-first connected(enter): HTTP/2 connections @@ -3034,6 +3097,68 @@ h2_start_failure(after_upgrade, From, Reason) -> close_h2(H2Conn) -> try h2_connection:close(H2Conn) catch _:_ -> ok end. +%% An unshared connection lives and dies with its owner, so its streams need +%% no monitor of their own. +track_h2_stream(StreamId, Owner, StreamState, + #conn_data{h2_shared = false, h2_streams = Streams} = Data) -> + Data#conn_data{h2_streams = maps:put(StreamId, {Owner, StreamState}, Streams)}; +track_h2_stream(StreamId, Owner, StreamState, + #conn_data{h2_streams = Streams, + h2_stream_monitors = Monitors} = Data) -> + OwnerPid = h2_stream_owner_pid(Owner), + Monitor = erlang:monitor(process, OwnerPid), + Data#conn_data{ + h2_streams = maps:put(StreamId, {Owner, StreamState}, Streams), + h2_stream_monitors = maps:put(StreamId, Monitor, Monitors) + }. + +h2_stream_owner_pid({Pid, _Tag}) when is_pid(Pid) -> Pid; +h2_stream_owner_pid(Pid) when is_pid(Pid) -> Pid. + +drop_h2_stream(StreamId, + #conn_data{h2_streams = Streams, + h2_stream_monitors = Monitors} = Data) -> + Data1 = cancel_h2_timer(StreamId, Data), + Monitors2 = case maps:take(StreamId, Monitors) of + {Monitor, Rest} -> + _ = erlang:demonitor(Monitor, [flush]), + Rest; + error -> + Monitors + end, + Data1#conn_data{ + h2_streams = maps:remove(StreamId, Streams), + h2_stream_monitors = Monitors2 + }. + +clear_h2_stream_monitors(#conn_data{h2_stream_monitors = Monitors} = Data) -> + _ = maps:fold(fun(_StreamId, Monitor, ok) -> + _ = erlang:demonitor(Monitor, [flush]), + ok + end, ok, Monitors), + Data#conn_data{h2_stream_monitors = #{}}. + +handle_h2_stream_owner_down(Ref, State, + #conn_data{h2_stream_monitors = Monitors} = Data) -> + case [StreamId || {StreamId, Monitor} <- maps:to_list(Monitors), + Monitor =:= Ref] of + [StreamId] -> + _ = cancel_h2_stream(Data#conn_data.h2_conn, StreamId), + Data1 = drop_h2_stream(StreamId, Data), + h2_stream_owner_down_result(State, StreamId, Data1); + [] -> + keep_state_and_data + end. + +h2_stream_owner_down_result(streaming_body, StreamId, + #conn_data{h2_stream_id = StreamId} = Data) -> + %% The caller streaming a request body died: its stream is gone, so the + %% connection can take requests again. + {next_state, connected, + Data#conn_data{h2_stream_id = undefined, request_from = undefined}}; +h2_stream_owner_down_result(_State, _StreamId, Data) -> + h2_stream_result(Data, []). + %% @private Arm a per-stream recv_timeout watchdog for a sync HTTP/2 read so a %% lost frame fails fast with {error, timeout} instead of blocking until the %% connection dies. No-op when recv_timeout is infinity. @@ -3079,7 +3204,6 @@ handle_h2_recv_timeout(StreamId, TRef, h2_conn = H2Conn} = Data) -> case maps:get(StreamId, Timers, undefined) of TRef -> - Timers2 = maps:remove(StreamId, Timers), case maps:get(StreamId, Streams, undefined) of {From, Inner} when is_tuple(Inner), element(1, Inner) =:= sync -> %% RST_STREAM(CANCEL) the stalled stream so the peer stops @@ -3087,13 +3211,15 @@ handle_h2_recv_timeout(StreamId, TRef, %% pooled connection would be reused with an orphaned stream %% still open (h2_conn_usable only checks the conn state). _ = cancel_h2_stream(H2Conn, StreamId), - Streams2 = maps:remove(StreamId, Streams), - {keep_state, - Data#conn_data{h2_streams = Streams2, h2_timers = Timers2, - request_from = undefined}, - [{reply, From, {error, timeout}}]}; + Data2 = drop_h2_stream( + StreamId, + Data#conn_data{request_from = undefined}), + h2_stream_result( + Data2, + [{reply, From, {error, timeout}}]); _ -> - {keep_state, Data#conn_data{h2_timers = Timers2}} + {keep_state, + Data#conn_data{h2_timers = maps:remove(StreamId, Timers)}} end; _ -> {keep_state, Data} @@ -3167,10 +3293,8 @@ do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, SendTimeout, Da sync -> From; {async, _Ref0, StreamTo0, _AsyncMode0} -> StreamTo0 end, - Streams = maps:put(StreamId, {Owner, StreamState}, - Data#conn_data.h2_streams), - NewData0 = Data#conn_data{ - h2_streams = Streams, + NewData0 = track_h2_stream(StreamId, Owner, StreamState, Data), + NewData1 = NewData0#conn_data{ method = MethodBin, path = PathBin }, @@ -3178,9 +3302,9 @@ do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, SendTimeout, Da sync -> %% Watchdog the response so a lost frame fails fast rather %% than blocking on the infinity gen_statem:call. - arm_h2_timer(StreamId, NewData0#conn_data{request_from = From}); + arm_h2_timer(StreamId, NewData1#conn_data{request_from = From}); {async, Ref, StreamTo, AsyncMode} -> - NewData0#conn_data{ + NewData1#conn_data{ async = AsyncMode, async_ref = Ref, stream_to = StreamTo @@ -3198,7 +3322,7 @@ do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, SendTimeout, Da %% END_STREAM and transition to streaming_body so the caller can push body %% chunks via send_body_chunk/finish_send_body. Mirrors do_h3_send_headers/5. do_h2_send_headers(From, Method, Path, Headers, ReqOpts, Data) -> - #conn_data{h2_conn = H2Conn, h2_streams = Streams} = Data, + #conn_data{h2_conn = H2Conn} = Data, {MethodBin, PathBin, H2Headers} = build_h2_request_headers(Method, Path, Headers, Data), %% Effective send_timeout for this stream's body chunks. Stored in the @@ -3215,8 +3339,9 @@ do_h2_send_headers(From, Method, Path, Headers, ReqOpts, Data) -> end, case SendRes of {ok, StreamId} -> - NewData = Data#conn_data{ - h2_streams = maps:put(StreamId, {undefined, {stream, sending}}, Streams), + Owner = element(1, From), + NewData0 = track_h2_stream(StreamId, Owner, {stream, sending}, Data), + NewData = NewData0#conn_data{ h2_stream_id = StreamId, req_send_timeout = SendTimeout, method = MethodBin, @@ -3309,8 +3434,8 @@ handle_h2_stream_body(From, #conn_data{h2_stream_id = StreamId, h2_streams = Str [{reply, From, {ok, Buffer}}]} end; {_, {stream, done, _Status, _Hdrs, <<>>}} -> - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}, [{reply, From, done}]}; + h2_stream_result(drop_h2_stream(StreamId, Data), + [{reply, From, done}]); {_, {stream, done, Status, Hdrs, Buffer}} -> %% Hand back the last buffered chunk; next call returns done. Streams2 = maps:put(StreamId, {undefined, {stream, done, Status, Hdrs, <<>>}}, Streams), @@ -3329,8 +3454,8 @@ handle_h2_read_body(From, #conn_data{h2_stream_id = StreamId, h2_streams = Strea Streams), {keep_state, Data#conn_data{h2_streams = Streams2}}; {_, {stream, done, _Status, _Hdrs, Buffer}} -> - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}, [{reply, From, {ok, Buffer}}]}; + h2_stream_result(drop_h2_stream(StreamId, Data), + [{reply, From, {ok, Buffer}}]); _ -> {keep_state_and_data, [{reply, From, {error, no_stream}}]} end. @@ -3482,11 +3607,10 @@ deliver_once_item(StreamId, StreamTo, Ref, [{data, Body} | Rest], Data) -> {keep_state, Data#conn_data{h2_streams = Streams2}}; deliver_once_item(StreamId, StreamTo, Ref, [done], Data) -> StreamTo ! {hackney_response, Ref, done}, - Streams2 = maps:remove(StreamId, Data#conn_data.h2_streams), - {keep_state, Data#conn_data{h2_streams = Streams2, - async = false, - async_ref = undefined, - stream_to = undefined}}; + Data1 = drop_h2_stream(StreamId, Data), + h2_stream_result(Data1#conn_data{async = false, + async_ref = undefined, + stream_to = undefined}, []); deliver_once_item(StreamId, StreamTo, Ref, [], Data) -> Streams2 = maps:put(StreamId, {StreamTo, {async_once, StreamTo, Ref, [], 1}}, @@ -3516,12 +3640,11 @@ h2_on_data(StreamId, Body, EndStream, Data) -> NewAcc = <>, case EndStream of true -> - Streams2 = maps:remove(StreamId, Streams), - Data2 = cancel_h2_timer(StreamId, - Data#conn_data{h2_streams = Streams2, - request_from = undefined}), - {keep_state, Data2, - [{reply, From, {ok, Status, Headers, NewAcc}}]}; + Data2 = drop_h2_stream( + StreamId, + Data#conn_data{request_from = undefined}), + h2_stream_result(Data2, + [{reply, From, {ok, Status, Headers, NewAcc}}]); false -> Streams2 = maps:put(StreamId, {From, {sync, body, Status, Headers, NewAcc}}, @@ -3553,12 +3676,11 @@ h2_on_data(StreamId, Body, EndStream, Data) -> case EndStream of true -> StreamTo ! {hackney_response, Ref, done}, - Streams2 = maps:remove(StreamId, Streams), - {keep_state, - Data#conn_data{h2_streams = Streams2, - async = false, - async_ref = undefined, - stream_to = undefined}}; + Data2 = drop_h2_stream(StreamId, Data), + h2_stream_result( + Data2#conn_data{async = false, + async_ref = undefined, + stream_to = undefined}, []); false -> NewState = {async, AsyncMode, StreamTo, Ref, streaming, Status, Headers}, @@ -3586,9 +3708,8 @@ h2_on_data(StreamId, Body, EndStream, Data) -> [{reply, From, {ok, NewBuffer}}]}; From when EndStream -> %% Parked caller, no buffered bytes, stream ended -> done. - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}, - [{reply, From, done}]}; + h2_stream_result(drop_h2_stream(StreamId, Data), + [{reply, From, done}]); _From -> %% Empty DATA frame without END_STREAM: keep the caller parked. {keep_state, Data} @@ -3598,9 +3719,8 @@ h2_on_data(StreamId, Body, EndStream, Data) -> NewAcc = <>, case EndStream of true -> - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}, - [{reply, From, {ok, NewAcc}}]}; + h2_stream_result(drop_h2_stream(StreamId, Data), + [{reply, From, {ok, NewAcc}}]); false -> Streams2 = maps:put(StreamId, {From, {stream, body_full, Status, Headers, NewAcc, From}}, @@ -3615,34 +3735,32 @@ h2_on_stream_reset(StreamId, ErrorCode, Data) -> #conn_data{h2_streams = Streams} = Data, case maps:get(StreamId, Streams, undefined) of {From, Inner} when is_tuple(Inner), element(1, Inner) =:= sync -> - Streams2 = maps:remove(StreamId, Streams), - Data2 = cancel_h2_timer(StreamId, - Data#conn_data{h2_streams = Streams2, - request_from = undefined}), - {keep_state, Data2, - [{reply, From, {error, {stream_error, ErrorCode}}}]}; + Data2 = drop_h2_stream( + StreamId, + Data#conn_data{request_from = undefined}), + h2_stream_result(Data2, + [{reply, From, {error, {stream_error, ErrorCode}}}]); {StreamTo, {async, _, StreamTo, Ref, _, _, _}} -> StreamTo ! {hackney_response, Ref, {error, {stream_error, ErrorCode}}}, - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}}; + h2_stream_result(drop_h2_stream(StreamId, Data), []); {StreamTo, {async, _, StreamTo, Ref, _}} -> StreamTo ! {hackney_response, Ref, {error, {stream_error, ErrorCode}}}, - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}}; + h2_stream_result(drop_h2_stream(StreamId, Data), []); {StreamTo, {async_once, StreamTo, Ref, _, _}} -> StreamTo ! {hackney_response, Ref, {error, {stream_error, ErrorCode}}}, - Streams2 = maps:remove(StreamId, Streams), - {keep_state, Data#conn_data{h2_streams = Streams2}}; + h2_stream_result(drop_h2_stream(StreamId, Data), []); {_, Inner} when is_tuple(Inner), element(1, Inner) =:= stream -> %% Streaming-body stream: reply to any parked caller and drop it so a %% later stream_body/start_response sees {error, no_stream}. - Streams2 = maps:remove(StreamId, Streams), Replies = case h2_stream_parked_from(Inner) of undefined -> []; From -> [{reply, From, {error, {stream_error, ErrorCode}}}] end, - {keep_state, Data#conn_data{h2_streams = Streams2, request_from = undefined}, - Replies}; + h2_stream_result( + drop_h2_stream( + StreamId, + Data#conn_data{request_from = undefined}), + Replies); _ -> {keep_state, Data} end. @@ -3654,6 +3772,21 @@ h2_stream_parked_from({stream, headers, _, _, _, From}) -> From; h2_stream_parked_from({stream, body_full, _, _, _, From}) -> From; h2_stream_parked_from(_) -> undefined. +%% @private Result for a handler that just removed an HTTP/2 stream: a shared +%% connection left with no stream arms its idle timer. +h2_stream_result(Data, Replies) -> + {keep_state, Data, Replies ++ h2_idle_actions(Data)}. + +%% @private Arm the idle timer of a shared HTTP/2 connection with no stream +%% open. Re-arming replaces the previous timer, so it always counts from the +%% moment the last stream finished. +h2_idle_actions(#conn_data{h2_shared = true, h2_streams = Streams, + idle_timeout = Timeout}) + when map_size(Streams) =:= 0, Timeout =/= infinity -> + [{{timeout, h2_idle}, Timeout, h2_idle}]; +h2_idle_actions(_Data) -> + []. + h2_on_goaway(ErrorCode, #conn_data{h2_conn = H2Conn, h2_mon = H2Mon} = Data) -> %% A GOAWAY means the peer will not service new streams on this connection. %% AWS ALBs recycle connections this way, sending GOAWAY but keeping the @@ -3707,7 +3840,9 @@ collect_h2_aborts(Err, #conn_data{h2_streams = Streams} = Data) -> end; (_, _, Acc) -> Acc end, [], Streams), - {Replies, Data#conn_data{h2_streams = #{}, request_from = undefined}}. + Data1 = clear_h2_stream_monitors( + Data#conn_data{h2_streams = #{}, request_from = undefined}), + {Replies, Data1}. %%==================================================================== diff --git a/src/hackney_pool.erl b/src/hackney_pool.erl index b48ce0bc..d4522ecb 100644 --- a/src/hackney_pool.erl +++ b/src/hackney_pool.erl @@ -1253,7 +1253,8 @@ h2_conn_usable(Pid) -> %% @private Remove an HTTP/2 connection from the pool do_unregister_h2(Pid, State) -> - #state{h2_connections = H2Conns, pid_monitors = PidMonitors} = State, + #state{h2_connections = H2Conns, in_use = InUse, + pid_monitors = PidMonitors} = State, %% Find and remove the connection H2Conns2 = maps:fold( fun(Key, ConnPid, Acc) -> @@ -1265,12 +1266,18 @@ do_unregister_h2(Pid, State) -> H2Conns, H2Conns ), - %% Demonitor if no longer tracked - PidMonitors2 = case maps:take(Pid, PidMonitors) of - {MonRef, PM} -> - erlang:demonitor(MonRef, [flush]), - PM; - error -> PidMonitors + %% Demonitor if no longer tracked. A checked-out conn keeps its monitor: + %% it holds a per-host slot until it stops, and only the DOWN releases it. + PidMonitors2 = case maps:is_key(Pid, InUse) of + true -> + PidMonitors; + false -> + case maps:take(Pid, PidMonitors) of + {MonRef, PM} -> + erlang:demonitor(MonRef, [flush]), + PM; + error -> PidMonitors + end end, State#state{h2_connections = H2Conns2, pid_monitors = PidMonitors2}. diff --git a/test/hackney_http2_shared_conn_tests.erl b/test/hackney_http2_shared_conn_tests.erl new file mode 100644 index 00000000..3e26a325 --- /dev/null +++ b/test/hackney_http2_shared_conn_tests.erl @@ -0,0 +1,296 @@ +%%% Lifetime of pooled HTTP/2 connections shared between callers (#937). +%%% +%%% A pooled HTTP/2 connection used to stay owned by the caller that dialed +%%% it: when that caller exited, the connection stopped and every other +%%% caller's stream on it failed with {error, closed}. A shared connection +%%% now has no owner. Each stream is tied to its own caller, and the +%%% connection closes itself once it has been idle for the pool timeout. +-module(hackney_http2_shared_conn_tests). + +-include_lib("eunit/include/eunit.hrl"). + +shared_conn_test_() -> + [{timeout, 30, {Title, Fun}} || {Title, Fun} <- [ + {"dialer exiting does not close other callers' streams", + fun dialer_exit_keeps_other_streams/0}, + {"a killed caller resets only its own stream", + fun killed_caller_resets_its_stream/0}, + {"a dead async consumer resets only its own stream", + fun dead_async_consumer/0}, + {"a dead uploader frees the connection for new requests", + fun dead_uploader/0}, + {"stream monitors are removed when streams end", + fun stream_monitors_removed/0}, + {"an idle shared connection closes and releases its slot", + fun idle_conn_closes/0}, + {"the idle timer waits for open streams", + fun idle_waits_for_streams/0}, + {"an unregistered connection still releases its slot", + fun unregistered_conn_releases_slot/0}, + {"an unpooled connection is not shared", + fun unpooled_conn_not_shared/0} + ]]. + +%%==================================================================== +%% Tests +%%==================================================================== + +dialer_exit_keeps_other_streams() -> + with_server([], fun(URL, _Port, Opts) -> + {First, FirstRef} = spawn_monitor(fun() -> + Result = hackney:request(get, <>, [], <<>>, Opts), + exit({result, Result}) + end), + {FirstHandler, SConn} = started(<<"/first">>), + _ = sync_request(second, <>, Opts), + {SecondHandler, SConn2} = started(<<"/second">>), + ?assertEqual(SConn, SConn2), + FirstHandler ! respond, + receive + {'DOWN', FirstRef, process, First, {result, FirstResult}} -> + ?assertMatch({ok, 200, _, <<"ok">>}, FirstResult) + after 5000 -> error(no_first_result) + end, + SecondHandler ! respond, + ?assertMatch({ok, 200, _, <<"ok">>}, result(second)) + end). + +killed_caller_resets_its_stream() -> + with_server([], fun(URL, Port, Opts) -> + First = sync_request(first, <>, Opts), + {_FirstHandler, SConn} = started(<<"/first">>), + Conn = shared_conn(Opts, Port), + ?assert(lists:member(First, monitored(Conn))), + exit(First, kill), + ok = wait_until(fun() -> not lists:member(First, monitored(Conn)) end), + _ = sync_request(second, <>, Opts), + {SecondHandler, SConn2} = started(<<"/second">>), + ?assertEqual(SConn, SConn2), + SecondHandler ! respond, + ?assertMatch({ok, 200, _, <<"ok">>}, result(second)), + ?assertEqual(Conn, shared_conn(Opts, Port)) + end). + +dead_async_consumer() -> + with_server([], fun(URL, Port, Opts) -> + Parent = self(), + Consumer = spawn(fun() -> + {ok, _Ref} = hackney:request(get, <>, [], <<>>, + [async | Opts]), + Parent ! async_sent, + receive stop -> ok end + end), + receive async_sent -> ok after 5000 -> error(no_async_request) end, + {AsyncHandler, _} = started(<<"/async">>), + Conn = shared_conn(Opts, Port), + ?assert(lists:member(Consumer, monitored(Conn))), + exit(Consumer, kill), + ok = wait_until(fun() -> not lists:member(Consumer, monitored(Conn)) end), + %% The stream was reset, so the handler can no longer answer it. + AsyncHandler ! respond, + ?assertMatch({ok, 200, _, <<"ok">>}, + hackney:request(get, <>, [], <<>>, Opts)), + ?assertEqual(Conn, shared_conn(Opts, Port)) + end). + +dead_uploader() -> + with_server([], fun(URL, Port, Opts) -> + Parent = self(), + Uploader = spawn(fun() -> + {ok, UploadConn} = hackney:request(post, <>, [], + stream, Opts), + ok = hackney:send_body(UploadConn, <<"chunk">>), + Parent ! {uploading, UploadConn}, + receive stop -> ok end + end), + Conn = receive {uploading, C} -> C after 5000 -> error(no_upload) end, + {_UploadHandler, _} = started(<<"/upload">>), + ?assertEqual({ok, streaming_body}, hackney_conn:get_state(Conn)), + exit(Uploader, kill), + ok = wait_until(fun() -> + hackney_conn:get_state(Conn) =:= {ok, connected} + end), + ?assertMatch({ok, 200, _, <<"ok">>}, + hackney:request(get, <>, [], <<>>, Opts)), + ?assertEqual(Conn, shared_conn(Opts, Port)) + end). + +stream_monitors_removed() -> + with_server([], fun(URL, Port, Opts) -> + {ok, 200, _, <<"ok">>} = + hackney:request(get, <>, [], <<>>, Opts), + Conn = shared_conn(Opts, Port), + Baseline = monitored(Conn), + Callers = [begin + Pid = sync_request({fast, N}, <>, Opts), + ?assertMatch({ok, 200, _, _}, result({fast, N})), + Pid + end || N <- lists:seq(1, 20)], + %% The recv_timeout watchdog path drops the stream monitor too. + TimeoutOpts = lists:keystore(recv_timeout, 1, Opts, {recv_timeout, 100}), + ?assertEqual({error, timeout}, + hackney:request(get, <>, [], <<>>, + TimeoutOpts)), + ?assertEqual(Baseline, monitored(Conn)), + ?assertEqual([], [P || P <- Callers, lists:member(P, monitored(Conn))]), + ?assertEqual(Conn, shared_conn(Opts, Port)) + end). + +idle_conn_closes() -> + with_server([{timeout, 100}], fun(URL, Port, Opts) -> + {ok, 200, _, <<"ok">>} = + hackney:request(get, <>, [], <<>>, Opts), + Conn = shared_conn(Opts, Port), + Ref = erlang:monitor(process, Conn), + receive {'DOWN', Ref, process, Conn, _} -> ok + after 5000 -> error(conn_did_not_idle_out) + end, + ok = wait_until(fun() -> + hackney_load_regulation:current("localhost", Port) =:= 0 + end), + ?assertMatch({ok, 200, _, <<"ok">>}, + hackney:request(get, <>, [], <<>>, Opts)) + end). + +idle_waits_for_streams() -> + with_server([{timeout, 100}], fun(URL, Port, Opts) -> + _ = sync_request(slow, <>, Opts), + {Handler, _} = started(<<"/slow">>), + Conn = shared_conn(Opts, Port), + %% Hold the stream open well past the idle timeout. + receive after 400 -> ok end, + ?assertEqual({ok, connected}, hackney_conn:get_state(Conn)), + Handler ! respond, + ?assertMatch({ok, 200, _, <<"ok">>}, result(slow)) + end). + +unregistered_conn_releases_slot() -> + with_server([], fun(URL, Port, Opts) -> + {ok, 200, _, <<"ok">>} = + hackney:request(get, <>, [], <<>>, Opts), + Conn = shared_conn(Opts, Port), + ?assertEqual(1, hackney_load_regulation:current("localhost", Port)), + ok = hackney_pool:unregister_h2(Conn, Opts), + ok = wait_until(fun() -> no_shared_conn(Opts, Port) end), + hackney_conn:stop(Conn), + ok = wait_until(fun() -> + hackney_load_regulation:current("localhost", Port) =:= 0 + end) + end). + +unpooled_conn_not_shared() -> + with_server([], fun(URL, _Port, Opts) -> + Parent = self(), + UnpooledOpts = lists:keystore(pool, 1, Opts, {pool, false}), + Uploader = spawn(fun() -> + {ok, UploadConn} = hackney:request(post, <>, [], + stream, UnpooledOpts), + Parent ! {uploading, UploadConn}, + receive stop -> ok end + end), + Conn = receive {uploading, C} -> C after 5000 -> error(no_conn) end, + {_Handler, _} = started(<<"/upload">>), + ?assertEqual({error, invalid_state}, hackney_conn:share_h2(Conn)), + %% Its lifetime stays with its owner, so streams are not monitored. + ?assertNot(lists:member(Uploader, monitored(Conn))), + exit(Uploader, kill), + hackney_conn:stop(Conn) + end). + +%%==================================================================== +%% Helpers +%%==================================================================== + +%% Start an h2 server and a pool of one connection per host. `/fast' +%% answers at once and `/never' never answers; any other path reports +%% {request_started, Path, Handler, ServerConn} and answers on `respond'. +with_server(PoolOpts, Fun) -> + _ = application:ensure_all_started(hackney), + _ = application:ensure_all_started(h2), + Parent = self(), + Handler = fun(SConn, Sid, _Method, Path, _Headers) -> + case Path of + <<"/fast">> -> ok; + <<"/never">> -> receive after infinity -> ok end; + _ -> + Parent ! {request_started, Path, self(), SConn}, + receive respond -> ok end + end, + ok = h2:send_response(SConn, Sid, 200, + [{<<"content-type">>, <<"text/plain">>}]), + ok = h2:send_data(SConn, Sid, <<"ok">>, true) + end, + Certs = cert_dir(), + {ok, Server} = h2:start_server(0, #{ + cert => filename:join(Certs, "server.pem"), + key => filename:join(Certs, "server.key"), + handler => Handler}), + Port = h2:server_port(Server), + Pool = list_to_atom("hackney_h2_shared_" ++ integer_to_list(Port)), + ok = hackney_pool:start_pool(Pool, [{max_connections, 1} | PoolOpts]), + URL = iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port)]), + Opts = [{pool, Pool}, {protocols, [http2]}, {recv_timeout, 5000}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}], + try + Fun(URL, Port, Opts) + after + _ = hackney_pool:stop_pool(Pool), + _ = h2:stop_server(Server), + hackney_load_regulation:reset("localhost", Port) + end. + +cert_dir() -> + BeamDir = filename:dirname(code:which(?MODULE)), + Root = filename:join([BeamDir, "..", "..", "..", "..", ".."]), + filename:join([filename:absname(Root), "test", "certs"]). + +sync_request(Tag, URL, Opts) -> + Parent = self(), + spawn(fun() -> + Parent ! {result, Tag, hackney:request(get, URL, [], <<>>, Opts)} + end). + +result(Tag) -> + receive {result, Tag, Result} -> Result + after 5000 -> error({no_result, Tag}) + end. + +started(Path) -> + receive {request_started, Path, Handler, SConn} -> {Handler, SConn} + after 5000 -> error({not_started, Path}) + end. + +%% The pool keys shared connections by the TLS options hash, so read the one +%% registered for this port from the pool state instead of rebuilding the key. +shared_conn(Opts, Port) -> + [Conn] = shared_conns(Opts, Port), + Conn. + +no_shared_conn(Opts, Port) -> + shared_conns(Opts, Port) =:= []. + +shared_conns(Opts, Port) -> + PoolPid = hackney_pool:find_pool(proplists:get_value(pool, Opts)), + State = sys:get_state(PoolPid), + [Pid || Field <- tuple_to_list(State), is_map(Field), + {{_Host, P, hackney_ssl, _TlsKey}, Pid} <- maps:to_list(Field), + P =:= Port, is_pid(Pid)]. + +monitored(Pid) -> + {monitors, Monitors} = process_info(Pid, monitors), + lists:sort([P || {process, P} <- Monitors]). + +wait_until(Fun) -> + wait_until(Fun, erlang:monotonic_time(millisecond) + 5000). + +wait_until(Fun, Deadline) -> + case Fun() of + true -> ok; + false -> + case erlang:monotonic_time(millisecond) > Deadline of + true -> error(wait_until_timeout); + false -> + receive after 5 -> ok end, + wait_until(Fun, Deadline) + end + end. From 22321499332bc440e6f9a3fa4e972b17e008efe1 Mon Sep 17 00:00:00 2001 From: Benoit Chesneau Date: Mon, 21 Sep 2026 20:36:55 +0200 Subject: [PATCH 2/2] Do not answer on the reset stream in the dead consumer test The handler answered the stream the client had just reset. On slow runners that response crossed the RST_STREAM, and h2 drops a header block for a reset stream without decoding it, so the HPACK tables fell out of sync and the next response on the connection failed to decode. That is an h2 issue of its own; this test only needs the stream reset. --- test/hackney_http2_shared_conn_tests.erl | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/test/hackney_http2_shared_conn_tests.erl b/test/hackney_http2_shared_conn_tests.erl index 3e26a325..9ac5671b 100644 --- a/test/hackney_http2_shared_conn_tests.erl +++ b/test/hackney_http2_shared_conn_tests.erl @@ -81,13 +81,13 @@ dead_async_consumer() -> receive stop -> ok end end), receive async_sent -> ok after 5000 -> error(no_async_request) end, - {AsyncHandler, _} = started(<<"/async">>), + %% The handler never answers: a response crossing the client's + %% RST_STREAM is a separate h2 HPACK issue, not what this covers. + {_AsyncHandler, _} = started(<<"/async">>), Conn = shared_conn(Opts, Port), ?assert(lists:member(Consumer, monitored(Conn))), exit(Consumer, kill), ok = wait_until(fun() -> not lists:member(Consumer, monitored(Conn)) end), - %% The stream was reset, so the handler can no longer answer it. - AsyncHandler ! respond, ?assertMatch({ok, 200, _, <<"ok">>}, hackney:request(get, <>, [], <<>>, Opts)), ?assertEqual(Conn, shared_conn(Opts, Port))