From 3334de3a4908f75143454f4b0d59f11259a2ee91 Mon Sep 17 00:00:00 2001 From: Sam Martin Date: Wed, 2 Sep 2026 03:09:23 +0200 Subject: [PATCH 1/2] test: cover pooled HTTP/2 connection ownership --- test/hackney_http2_async_once_tests.erl | 37 +++ test/hackney_http2_concurrency_tests.erl | 381 +++++++++++++++++++++++ test/hackney_pool_h2h3_fault_tests.erl | 96 +++++- test/hackney_pool_tests.erl | 125 +++++++- test/hackney_race_pool.erl | 59 +++- 5 files changed, 685 insertions(+), 13 deletions(-) diff --git a/test/hackney_http2_async_once_tests.erl b/test/hackney_http2_async_once_tests.erl index 5533db71..533a6ae7 100644 --- a/test/hackney_http2_async_once_tests.erl +++ b/test/hackney_http2_async_once_tests.erl @@ -30,6 +30,8 @@ async_once_test_() -> {timeout, 60, fun t_once_stream_reset/0}}, {"once: connection teardown surfaces as an error message", {timeout, 60, fun t_once_conn_teardown/0}}, + {"once: dead consumer does not block connection retirement", + {timeout, 60, fun t_once_dead_consumer/0}}, {"legacy bare stream_next atom still routes", {timeout, 60, fun t_legacy_bare_stream_next/0}}]. @@ -145,6 +147,41 @@ t_once_conn_teardown() -> repro_h2_raw_server:stop(element(1, Srv)) end. +t_once_dead_consumer() -> + _ = application:ensure_all_started(hackney), + _ = application:ensure_all_started(h2), + Srv = repro_h2_raw_server:start(#{ + body_size => ?BODY_SIZE, + frame_count => ?FRAME_COUNT + }), + Port = repro_h2_raw_server:port(Srv), + Pool = hackney_h2_once_dead_consumer_pool, + _ = hackney_pool:start_pool(Pool, [{max_connections, 1}]), + try + Url = iolist_to_binary([<<"https://localhost:">>, + integer_to_list(Port), <<"/">>]), + Opts = [{async, once}, {pool, Pool}, {protocols, [http2]}, + {recv_timeout, 15000}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}], + Parent = self(), + {Worker, WorkerRef} = spawn_monitor(fun() -> + {ok, Ref} = hackney:request(get, Url, [], <<>>, Opts), + Parent ! {once_connection, Ref} + end), + Conn = receive {once_connection, Ref} -> Ref end, + receive {'DOWN', WorkerRef, process, Worker, normal} -> ok end, + ConnRef = monitor(process, Conn), + ok = hackney_conn:retire_h2(Conn), + receive + {'DOWN', ConnRef, process, Conn, normal} -> ok + after 1000 -> + ?assert(false) + end + after + catch hackney_pool:stop_pool(Pool), + repro_h2_raw_server:stop(element(1, Srv)) + end. + %% The pre-upgrade bare stream_next atom (no caller pid) still pulls, and is %% a no-op once no once-mode stream remains. t_legacy_bare_stream_next() -> diff --git a/test/hackney_http2_concurrency_tests.erl b/test/hackney_http2_concurrency_tests.erl index 9597a257..be642a7c 100644 --- a/test/hackney_http2_concurrency_tests.erl +++ b/test/hackney_http2_concurrency_tests.erl @@ -21,6 +21,345 @@ cert_dir() -> concurrent_tight_loop_test_() -> {timeout, 30, fun run_concurrent_tight_loop/0}. +pooled_sync_requests_share_connection_test_() -> + {timeout, 30, fun run_pooled_sync_requests_share_connection/0}. + +busy_connection_retires_after_drain_test_() -> + {timeout, 30, fun run_busy_connection_retires_after_drain/0}. + +registration_failure_does_not_hang_test_() -> + {timeout, 30, fun run_registration_failure_does_not_hang/0}. + +busy_connection_becomes_reusable_test_() -> + {timeout, 30, fun run_busy_connection_becomes_reusable/0}. + +abandoned_upload_does_not_block_retirement_test_() -> + {timeout, 30, fun run_abandoned_upload_does_not_block_retirement/0}. + +checkout_timeout_honored_while_h2_busy_test_() -> + {timeout, 30, fun run_checkout_timeout_honored_while_h2_busy/0}. + +run_checkout_timeout_honored_while_h2_busy() -> + _ = application:ensure_all_started(hackney), + PreviousPoolHandler = application:get_env(hackney, pool_handler), + Pool = hackney_h2_checkout_deadline_pool, + Port = 49199, + Parent = self(), + Busy = spawn(fun() -> busy_then_slow_h2(Parent, 0) end), + _ = hackney_pool:start_pool(Pool, [{max_connections, 1}]), + application:set_env(hackney, race_dead_pid, Busy), + application:set_env(hackney, pool_handler, hackney_race_pool), + ok = hackney_load_regulation:acquire("localhost", Port, 1, 0), + try + URL = iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port)]), + Opts = [{pool, Pool}, + {protocols, [http2]}, + {max_per_host, 1}, + {checkout_timeout, 25}, + {connect_timeout, 1000}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}], + ?assertEqual({error, checkout_timeout}, + hackney:request(get, URL, [], <<>>, Opts)), + receive + {unexpected_h2_probe, Busy} -> ?assert(false) + after 0 -> + ok + end + after + hackney_load_regulation:release("localhost", Port), + restore_pool_handler(PreviousPoolHandler), + application:unset_env(hackney, race_dead_pid), + exit(Busy, kill), + catch hackney_pool:stop_pool(Pool) + end. + +run_busy_connection_becomes_reusable() -> + _ = application:ensure_all_started(hackney), + _ = application:ensure_all_started(h2), + PreviousPoolHandler = application:get_env(hackney, pool_handler), + Parent = self(), + Handler = fun(Conn, Sid, _Method, Path, _Headers) -> + Parent ! {request_started, Path, Conn}, + case Path of + <<"/upload">> -> + ok = h2:set_stream_handler(Conn, Sid, self()), + _ = recv_request_body(Conn, Sid); + _ -> + ok + end, + send_ok(Conn, Sid) + 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 = hackney_h2_busy_reuse_pool, + _ = hackney_pool:start_pool(Pool, [{max_connections, 1}]), + try + URL = iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port)]), + Opts = [{pool, Pool}, + {protocols, [http2]}, + {max_per_host, 1}, + {checkout_timeout, 500}, + {recv_timeout, 5000}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}], + {ok, First} = hackney:request(post, <>, + [], stream, Opts), + {request_started, <<"/upload">>, Conn} = receive + UploadStarted = {request_started, <<"/upload">>, _} -> UploadStarted + end, + Second = spawn(fun() -> + receive start -> ok end, + Parent ! {second_result, + hackney:request(get, <>, + [], <<>>, Opts)} + end), + application:set_env(hackney, race_h2_checkout_observer, + {Parent, Second}), + application:set_env(hackney, pool_handler, hackney_race_pool), + Second ! start, + receive + {h2_checkout, Second, CheckoutResult} -> + ?assertEqual(none, CheckoutResult) + after 5000 -> + ?assert(false) + end, + ok = hackney:finish_send_body(First), + {ok, 200, _, First} = hackney:start_response(First), + {ok, <<"ok">>} = hackney:body(First), + receive + {second_result, SecondResult} -> + ?assertMatch({ok, 200, _, <<"ok">>}, SecondResult) + after 1000 -> + ?assert(false) + end, + receive + {request_started, <<"/second">>, SecondConn} -> + ?assertEqual(Conn, SecondConn) + after 1000 -> + ?assert(false) + end + after + restore_pool_handler(PreviousPoolHandler), + application:unset_env(hackney, race_h2_checkout_observer), + catch hackney_pool:stop_pool(Pool), + catch h2:stop_server(Server) + end. + +run_abandoned_upload_does_not_block_retirement() -> + _ = application:ensure_all_started(hackney), + _ = application:ensure_all_started(h2), + Parent = self(), + Handler = fun(Conn, Sid, _Method, _Path, _Headers) -> + ok = h2:set_stream_handler(Conn, Sid, self()), + Parent ! upload_handler_ready, + _ = recv_request_body(Conn, Sid), + send_ok(Conn, Sid) + 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 = hackney_h2_abandoned_upload_pool, + _ = hackney_pool:start_pool(Pool, [{max_connections, 1}]), + try + URL = iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port)]), + Opts = [{pool, Pool}, + {protocols, [http2]}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}], + {Worker, WorkerRef} = spawn_monitor(fun() -> + {ok, ConnPid} = hackney:request(post, URL, [], stream, Opts), + Parent ! {upload_connection, ConnPid}, + receive abandon_upload -> ok end + end), + Conn = receive {upload_connection, ConnPid} -> ConnPid end, + receive upload_handler_ready -> ok end, + Worker ! abandon_upload, + receive {'DOWN', WorkerRef, process, Worker, normal} -> ok end, + ConnRef = monitor(process, Conn), + ok = hackney_conn:retire_h2(Conn), + receive + {'DOWN', ConnRef, process, Conn, normal} -> ok + after 1000 -> + ?assert(false) + end + after + catch hackney_pool:stop_pool(Pool), + catch h2:stop_server(Server) + end. + +run_registration_failure_does_not_hang() -> + _ = application:ensure_all_started(hackney), + PreviousPoolHandler = application:get_env(hackney, pool_handler), + Pool = hackney_h2_registration_failure_pool, + Host = "h2-registration-failure.invalid", + Port = 443, + ok = hackney_load_regulation:reset(Host, Port), + _ = hackney_pool:start_pool(Pool, [{max_connections, 1}]), + try + Opts = [{pool, Pool}, + {protocols, [http2]}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}], + application:set_env(hackney, race_register_h2_error, self()), + application:set_env(hackney, pool_handler, hackney_race_pool), + Parent = self(), + spawn(fun() -> + Parent ! {request_result, + hackney:request(get, + <<"https://h2-registration-failure.invalid/">>, + [], <<>>, Opts)} + end), + Conn = receive + {h2_registration_candidate, Pid} -> Pid + after 1000 -> + ?assert(false) + end, + ConnRef = monitor(process, Conn), + receive + {request_result, Result} -> + ?assertEqual({error, set_owner_failed}, Result) + after 1000 -> + ?assert(false) + end, + receive + {'DOWN', ConnRef, process, Conn, _Reason} -> ok + after 1000 -> + ?assert(false) + end + after + restore_pool_handler(PreviousPoolHandler), + application:unset_env(hackney, race_register_h2_error), + catch hackney_pool:stop_pool(Pool), + hackney_load_regulation:reset(Host, Port) + end. + +run_busy_connection_retires_after_drain() -> + _ = application:ensure_all_started(hackney), + _ = application:ensure_all_started(h2), + PreviousPoolHandler = application:get_env(hackney, pool_handler), + Handler = fun(Conn, Sid, _Method, Path, _Headers) -> + case Path of + <<"/upload">> -> + ok = h2:set_stream_handler(Conn, Sid, self()), + _ = recv_request_body(Conn, Sid), + send_ok(Conn, Sid); + _ -> + send_ok(Conn, Sid) + end + 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 = hackney_h2_retire_pool, + _ = hackney_pool:start_pool(Pool, [{max_connections, 2}]), + try + 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}]}], + {ok, First} = hackney:request(post, <>, + [], stream, Opts), + FirstRef = monitor(process, First), + application:set_env(hackney, race_dead_pid, First), + application:set_env(hackney, pool_handler, hackney_race_pool), + ?assertMatch({ok, 200, _, <<"ok">>}, + hackney:request(get, <>, + [], <<>>, Opts)), + ok = hackney:finish_send_body(First), + {ok, 200, _, First} = hackney:start_response(First), + {ok, <<"ok">>} = hackney:body(First), + receive + {'DOWN', FirstRef, process, First, normal} -> ok + after 1000 -> + ?assert(false) + end, + ok = wait_until(fun() -> + HostStats = hackney_pool:host_stats(Pool, "localhost", Port), + case {proplists:get_value(in_use, HostStats), + proplists:get_value(free, HostStats)} of + {1, 0} -> ok; + _ -> false + end + end, 1000) + after + restore_pool_handler(PreviousPoolHandler), + application:unset_env(hackney, race_dead_pid), + catch hackney_pool:stop_pool(Pool), + catch h2:stop_server(Server) + end. + +run_pooled_sync_requests_share_connection() -> + _ = application:ensure_all_started(hackney), + _ = application:ensure_all_started(h2), + Parent = self(), + Handler = fun(Conn, Sid, _Method, Path, _Headers) -> + Parent ! {request_started, Path, self(), Conn, Sid}, + receive respond -> ok end, + ok = h2:send_response(Conn, Sid, 200, + [{<<"content-type">>, <<"text/plain">>}]), + ok = h2:send_data(Conn, 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 = hackney_h2_sync_pool, + _ = hackney_pool:start_pool(Pool, [{max_connections, 1}]), + try + 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}]}], + {First, FirstRef} = spawn_monitor(fun() -> + Parent ! {request_result, first, + hackney:request(get, <>, [], <<>>, Opts)} + end), + {request_started, <<"/first">>, FirstHandler, Conn, FirstSid} = + receive FirstStarted = {request_started, <<"/first">>, _, _, _} -> + FirstStarted + end, + spawn(fun() -> + Parent ! {request_result, second, + hackney:request(get, <>, [], <<>>, Opts)} + end), + {request_started, <<"/second">>, SecondHandler, Conn, SecondSid} = + receive SecondStarted = {request_started, <<"/second">>, _, _, _} -> + SecondStarted + end, + ?assertNotEqual(FirstSid, SecondSid), + + FirstHandler ! respond, + receive + {request_result, first, FirstResult} -> + ?assertMatch({ok, 200, _, <<"ok">>}, FirstResult) + end, + receive {'DOWN', FirstRef, process, First, normal} -> ok end, + + SecondHandler ! respond, + receive + {request_result, second, SecondResult} -> + ?assertMatch({ok, 200, _, <<"ok">>}, SecondResult) + end + after + catch hackney_pool:stop_pool(Pool), + catch h2:stop_server(Server) + end. + run_concurrent_tight_loop() -> _ = application:ensure_all_started(hackney), _ = application:ensure_all_started(h2), @@ -79,3 +418,45 @@ run_concurrent_tight_loop() -> catch hackney_pool:stop_pool(Pool), catch h2:stop_server(Server) end. + +recv_request_body(Conn, Sid) -> + receive + {h2, Conn, {data, Sid, _Data, true}} -> ok; + {h2, Conn, {data, Sid, _Data, false}} -> recv_request_body(Conn, Sid) + end. + +send_ok(Conn, Sid) -> + ok = h2:send_response(Conn, Sid, 200, + [{<<"content-type">>, <<"text/plain">>}]), + h2:send_data(Conn, Sid, <<"ok">>, true). + +restore_pool_handler({ok, Handler}) -> + application:set_env(hackney, pool_handler, Handler); +restore_pool_handler(undefined) -> + application:unset_env(hackney, pool_handler). + +wait_until(Fun, Timeout) -> + wait_until(Fun, Timeout, erlang:monotonic_time(millisecond)). + +wait_until(Fun, Timeout, Start) -> + case Fun() of + false -> + case erlang:monotonic_time(millisecond) - Start > Timeout of + true -> erlang:error({timeout_waiting_for, Fun}); + false -> + timer:sleep(20), + wait_until(Fun, Timeout, Start) + end; + Value -> + Value + end. + +busy_then_slow_h2(Parent, StateChecks) -> + receive + {'$gen_call', From, get_state} when StateChecks =:= 0 -> + gen_statem:reply(From, {ok, streaming_body}), + busy_then_slow_h2(Parent, 1); + {'$gen_call', _From, get_state} -> + Parent ! {unexpected_h2_probe, self()}, + busy_then_slow_h2(Parent, StateChecks) + end. diff --git a/test/hackney_pool_h2h3_fault_tests.erl b/test/hackney_pool_h2h3_fault_tests.erl index db7b895b..2d49af05 100644 --- a/test/hackney_pool_h2h3_fault_tests.erl +++ b/test/hackney_pool_h2h3_fault_tests.erl @@ -37,6 +37,12 @@ h2h3_fault_test_() -> fun teardown/1, [ {"a healthy h2 connection is handed out", fun t_h2_healthy/0}, + {"duplicate h2 registration keeps the first connection", + fun t_h2_duplicate_registration/0}, + {"a replaced busy h2 connection is retired", + fun t_h2_busy_registration/0}, + {"a slow h2 connection drains before retirement", + fun t_h2_slow_registration/0}, {"a wedged h2 connection is dropped without stalling the pool", fun t_h2_wedged/0}, {"an h2 connection that dies when probed does not crash the pool", @@ -84,6 +90,65 @@ t_h2_healthy() -> ?assertEqual({ok, Pid}, checkout_h2()), assert_pool_healthy(). +t_h2_duplicate_registration() -> + First = live_conn(), + Second = live_conn(), + SecondRef = monitor(process, Second), + _ = hackney_pool:register_h2(?HOST, ?PORT, hackney_tcp, First, opts()), + ?assertEqual({ok, First}, checkout_h2()), + ok = sys:suspend(Second), + try + Registration = hackney_pool:register_h2(?HOST, ?PORT, hackney_tcp, + Second, opts()), + StopResult = receive + {'DOWN', SecondRef, process, Second, _Reason} -> stopped + after 1000 -> alive + end, + ?assertEqual({ok, First}, Registration), + ?assertEqual(stopped, StopResult), + ?assertEqual({ok, First}, checkout_h2()), + assert_pool_healthy() + after + case is_process_alive(Second) of + true -> + catch sys:resume(Second), + catch hackney_conn:stop(Second, 100); + false -> + ok + end + end. + +t_h2_busy_registration() -> + First = fake_h2_conn(streaming_body, self()), + Second = fake_h2_conn(connected, self()), + ok = hackney_pool:register_h2(?HOST, ?PORT, hackney_tcp, First, opts()), + ok = hackney_pool:register_h2(?HOST, ?PORT, hackney_tcp, Second, opts()), + receive + {retired, First} -> ok + after 1000 -> + ?assert(false) + end, + First ! stop, + Second ! stop, + assert_pool_healthy(). + +t_h2_slow_registration() -> + First = live_conn(), + FirstRef = monitor(process, First), + Second = live_conn(), + ok = hackney_pool:register_h2(?HOST, ?PORT, hackney_tcp, First, opts()), + ok = sys:suspend(First), + ok = hackney_pool:register_h2(?HOST, ?PORT, hackney_tcp, Second, opts()), + ?assert(is_process_alive(First)), + ok = sys:resume(First), + receive + {'DOWN', FirstRef, process, First, normal} -> ok + after 1000 -> + ?assert(false) + end, + ?assertEqual({ok, Second}, checkout_h2()), + assert_pool_healthy(). + %% The connection is alive but answers nothing. Before `get_state' took a %% timeout the pool sat on the default 5s call for every caller of that host. t_h2_wedged() -> @@ -98,7 +163,7 @@ t_h2_wedged() -> %% Alive when registered, gone by the time the pool asks it anything. t_h2_dies_when_probed() -> - Pid = spawn(fun() -> receive _ -> exit(probed) end end), + Pid = spawn(fun dies_when_probed/0), ok = hackney_pool:register_h2(?HOST, ?PORT, hackney_tcp, Pid, opts()), ?assertEqual(none, checkout_h2()), assert_pool_healthy(). @@ -145,6 +210,35 @@ checkout_h2() -> checkout_h3() -> hackney_pool:checkout_h3(?HOST, ?PORT, hackney_tcp, opts()). +dies_when_probed() -> + receive + {'$gen_call', From, {set_owner, _Owner}} -> + gen_statem:reply(From, ok), + receive + {'$gen_call', _StateFrom, get_state} -> exit(probed) + end + end. + +fake_h2_conn(State, Parent) -> + spawn(fun() -> fake_h2_loop(State, Parent) end). + +fake_h2_loop(State, Parent) -> + receive + {'$gen_call', From, {set_owner, _Owner}} -> + gen_statem:reply(From, ok), + fake_h2_loop(State, Parent); + {'$gen_call', From, get_state} -> + gen_statem:reply(From, {ok, State}), + fake_h2_loop(State, Parent); + {'$gen_cast', retire_h2} -> + Parent ! {retired, self()}, + fake_h2_loop(State, Parent); + {'$gen_cast', stop} -> + ok; + stop -> + ok + end. + %% A real connection process against the test server, outside the pool's %% checkout bookkeeping: these tests are about the shared-connection map. live_conn() -> diff --git a/test/hackney_pool_tests.erl b/test/hackney_pool_tests.erl index c999f370..96d8814c 100644 --- a/test/hackney_pool_tests.erl +++ b/test/hackney_pool_tests.erl @@ -31,7 +31,9 @@ hackney_pool_unit_test_() -> {"start custom pool", fun test_custom_pool/0}, {"pool stats", fun test_pool_stats/0}, {"max connections setting", fun test_max_connections/0}, - {"timeout setting", fun test_timeout_setting/0} + {"timeout setting", fun test_timeout_setting/0}, + {"failed h2 ownership transfer is reported and stops connection", + fun test_h2_owner_transfer_timeout/0} ]}. %% HTTP/2 tls_key bucket tests - no server required @@ -62,6 +64,8 @@ hackney_pool_integration_test_() -> fun test_connect_timeout_does_not_crash_pool/0}, {"connect crash does not crash the pool", fun test_connect_crash_does_not_crash_pool/0}, + {"h2 registration timeout stops the candidate and releases its slot", + fun test_h2_registration_timeout_releases_slot/0}, {"queue timeout", {timeout, 120, fun test_queue_timeout/0}}, {"checkout timeout", {timeout, 120, fun test_checkout_timeout/0}}, {"stop_pool releases in_use load_regulation slots", @@ -217,12 +221,50 @@ test_timeout_setting() -> ?assertEqual(2000, hackney_pool:timeout(test_pool_5)), % Capped at 2s ok = hackney_pool:stop_pool(test_pool_5). +test_h2_owner_transfer_timeout() -> + Pool = test_pool_h2_owner_timeout, + ok = hackney_pool:start_pool(Pool, []), + Parent = self(), + {Conn, ConnMon} = spawn_monitor(fun() -> + receive + {'$gen_call', From, {set_owner, _Owner}} -> + Parent ! {set_owner_started, self()}, + receive + {'$gen_cast', stop} -> gen_statem:reply(From, ok) + end + end + end), + Opts = [{pool, Pool}], + try + Registration = hackney_pool:register_h2("h2-owner.example.com", 443, + hackney_ssl, Conn, Opts), + StartResult = receive + {set_owner_started, Conn} -> started + after 1000 -> not_started + end, + ?assertEqual(started, StartResult), + StopResult = receive + {'DOWN', ConnMon, process, Conn, _Reason} -> stopped + after 1000 -> alive + end, + ?assertEqual(stopped, StopResult), + ?assertMatch({error, _}, Registration), + ?assertEqual(none, hackney_pool:checkout_h2("h2-owner.example.com", 443, + hackney_ssl, Opts)) + after + case is_process_alive(Conn) of + true -> exit(Conn, kill); + false -> ok + end, + catch hackney_pool:stop_pool(Pool) + end. + %%==================================================================== %% HTTP/2 tls_key Bucket Tests %%==================================================================== -%% Dummy connection that answers hackney_conn:get_state/1 (used by the -%% pool's h2_conn_usable liveness check) with {ok, connected}. +%% Dummy connection that implements the state and ownership calls used by +%% HTTP/2 registration and checkout. dummy_h2_conn() -> spawn(fun dummy_h2_loop/0). @@ -231,6 +273,9 @@ dummy_h2_loop() -> {'$gen_call', From, get_state} -> gen_statem:reply(From, {ok, connected}), dummy_h2_loop(); + {'$gen_call', From, {set_owner, _Owner}} -> + gen_statem:reply(From, ok), + dummy_h2_loop(); stop -> ok end. @@ -774,6 +819,80 @@ test_connect_crash_does_not_crash_pool() -> ?assert(is_process_alive(hackney_pool:find_pool(PoolName))), ok = hackney_pool:stop_pool(PoolName). +test_h2_registration_timeout_releases_slot() -> + PoolName = test_pool_h2_registration_timeout, + Host = "localhost", + Port = ?PORT, + ok = hackney_load_regulation:reset(Host, Port), + ok = hackney_pool:start_pool(PoolName, [{pool_size, 1}, + {prewarm_count, 0}]), + Pool = hackney_pool:find_pool(PoolName), + Opts = [{pool, PoolName}, {connect_timeout, 1000}, + {checkout_timeout, 25}], + try + ok = hackney_load_regulation:acquire(Host, Port, 1, 0), + {ok, _PoolInfo, Conn} = hackney_pool:checkout(Host, Port, + hackney_tcp, Opts), + ConnRef = monitor(process, Conn), + ok = sys:suspend(Pool), + Parent = self(), + spawn(fun() -> + Parent ! {registration_result, + hackney_pool:register_h2(Host, Port, hackney_tcp, + Conn, Opts)} + end), + ok = wait_for_registration_call(Pool, 100), + timer:sleep(25), + ok = sys:resume(Pool), + receive + {registration_result, Result} -> + ?assertEqual({error, checkout_timeout}, Result) + after 1000 -> + ?assert(false) + end, + receive + {'DOWN', ConnRef, process, Conn, _Reason} -> ok + after 1000 -> + ?assert(false) + end, + ok = wait_for_load_count(Host, Port, 0, 100), + ?assertEqual(none, + hackney_pool:checkout_h2(Host, Port, hackney_tcp, Opts)) + after + case is_process_alive(Pool) of + true -> catch sys:resume(Pool); + false -> ok + end, + catch hackney_pool:stop_pool(PoolName), + hackney_load_regulation:reset(Host, Port) + end. + +wait_for_registration_call(_Pool, 0) -> + error(registration_call_not_queued); +wait_for_registration_call(Pool, Attempts) -> + {messages, Messages} = process_info(Pool, messages), + Queued = lists:any( + fun({'$gen_call', _, {register_h2, _, _}}) -> true; + ({'$gen_call', _, {register_h2, _, _, _}}) -> true; + (_) -> false + end, Messages), + case Queued of + true -> ok; + false -> + timer:sleep(5), + wait_for_registration_call(Pool, Attempts - 1) + end. + +wait_for_load_count(Host, Port, Expected, 0) -> + ?assertEqual(Expected, hackney_load_regulation:current(Host, Port)); +wait_for_load_count(Host, Port, Expected, Attempts) -> + case hackney_load_regulation:current(Host, Port) of + Expected -> ok; + _ -> + timer:sleep(5), + wait_for_load_count(Host, Port, Expected, Attempts - 1) + end. + %%==================================================================== %% Timeout Tests %%==================================================================== diff --git a/test/hackney_race_pool.erl b/test/hackney_race_pool.erl index 517f1117..c0e54aba 100644 --- a/test/hackney_race_pool.erl +++ b/test/hackney_race_pool.erl @@ -1,9 +1,8 @@ %%% Test pool handler for issue #914. %%% -%%% checkout_h2/4 and checkout_h3/4 return an already-terminated pid so the -%%% checkout get_state liveness probe races connection teardown. Every other -%%% callback delegates to hackney_pool, so the new-connection fallback behaves -%%% exactly as in production. +%%% By default, checkout_h2/4 and checkout_h3/4 return an already-terminated +%%% pid so the checkout probe races connection teardown. Optional modes expose +%%% registration failures and checkout ordering -module(hackney_race_pool). -export([checkout/4, @@ -25,15 +24,42 @@ dead_pid() -> {ok, Pid} = application:get_env(hackney, race_dead_pid), Pid. -checkout_h2(_Host, _Port, _Transport, _Options) -> - {ok, dead_pid()}. +checkout_h2(Host, Port, Transport, Options) -> + case application:get_env(hackney, race_h2_checkout_observer) of + {ok, {Parent, Observed}} -> + Result = hackney_pool:checkout_h2(Host, Port, Transport, Options), + notify_h2_checkout(Parent, Observed, Result), + Result; + undefined -> + case application:get_env(hackney, race_register_h2_error) of + {ok, _} -> hackney_pool:checkout_h2(Host, Port, Transport, Options); + undefined -> {ok, dead_pid()} + end + end. + +notify_h2_checkout(Parent, Observed, Result) -> + Key = {?MODULE, h2_checkout_observed}, + case self() =:= Observed andalso get(Key) =:= undefined of + true -> + put(Key, true), + Parent ! {h2_checkout, self(), Result}; + false -> + ok + end. checkout_h3(_Host, _Port, _Transport, _Options) -> {ok, dead_pid()}. -%% Everything else delegates unchanged. checkout(Host, Port, Transport, Options) -> - hackney_pool:checkout(Host, Port, Transport, Options). + case application:get_env(hackney, race_register_h2_error) of + {ok, Parent} -> + Pid = spawn(fun fake_h2_conn/0), + _ = timer:kill_after(2000, Pid), + Parent ! {h2_registration_candidate, Pid}, + {ok, undefined, Pid}; + undefined -> + hackney_pool:checkout(Host, Port, Transport, Options) + end. checkin(Ref, Options) -> hackney_pool:checkin(Ref, Options). @@ -42,7 +68,22 @@ checkout_ssl(Host, Port, Transport, Options) -> hackney_pool:checkout_ssl(Host, Port, Transport, Options). register_h2(Host, Port, Transport, Pid, Options) -> - hackney_pool:register_h2(Host, Port, Transport, Pid, Options). + case application:get_env(hackney, race_register_h2_error) of + {ok, _Parent} -> + {error, set_owner_failed}; + undefined -> + hackney_pool:register_h2(Host, Port, Transport, Pid, Options) + end. + +fake_h2_conn() -> + receive + {'$gen_call', From, is_upgraded_ssl} -> + gen_statem:reply(From, true), + fake_h2_conn(); + {'$gen_call', From, get_protocol} -> + gen_statem:reply(From, http2), + fake_h2_conn() + end. unregister_h2(Pid, Options) -> hackney_pool:unregister_h2(Pid, Options). From 3e26ad7736d71de43147313b94200cf420fd9cb3 Mon Sep 17 00:00:00 2001 From: Sam Martin Date: Wed, 2 Sep 2026 03:10:17 +0200 Subject: [PATCH 2/2] fix: keep shared HTTP/2 connections pool-owned --- src/hackney.erl | 157 ++++++++++++++++++---- src/hackney_conn.erl | 207 ++++++++++++++++++++--------- src/hackney_pool.erl | 153 ++++++++++++++++----- test/hackney_pool_safety_tests.erl | 2 +- 4 files changed, 394 insertions(+), 125 deletions(-) diff --git a/src/hackney.erl b/src/hackney.erl index 889e5ba1..37d2089f 100644 --- a/src/hackney.erl +++ b/src/hackney.erl @@ -73,6 +73,7 @@ -define(METHOD_TPL(Method), -export([Method/1, Method/2, Method/3, Method/4])). +-define(H2_PROBE_TIMEOUT, 250). -include("hackney_methods.hrl"). -include("hackney.hrl"). @@ -195,21 +196,13 @@ connect_pool(Transport, Host, Port, Options) -> %% Try HTTP/2 multiplexing case PoolHandler:checkout_h2(Host, Port, Transport, Options2) of {ok, H2Pid} -> - %% Verify connection is actually in connected state - %% (OTP 28 on FreeBSD may have timing issues with SSL connections). - %% The probe is a gen_statem:call, which exits if the pooled - %% connection is terminating (idle teardown, GOAWAY, keepalive - %% close) at checkout time; treat that as unusable and fall - %% through to a fresh connection instead of crashing the caller. - %% Mirrors maybe_register_h2/maybe_upgrade_ssl. - GetState = try hackney_conn:get_state(H2Pid) - catch exit:_ -> {error, terminated} end, - case GetState of - {ok, connected} -> - {ok, H2Pid}; - _ -> - %% Connection not ready, unregister and create new - PoolHandler:unregister_h2(H2Pid, Options2), + case h2_checkout_status(H2Pid) of + ready -> + {ok, H2Pid}; + busy -> + connect_pool_new(Transport, Host, Port, Options2, FinalSslOpts, PoolHandler); + unusable -> + stop_conn(H2Pid), connect_pool_new(Transport, Host, Port, Options2, FinalSslOpts, PoolHandler) end; none -> @@ -369,9 +362,10 @@ connect_pool_new(Transport, Host, Port, Options, FinalSslOpts, PoolHandler) -> CheckoutTimeout = proplists:get_value(checkout_timeout, Options, proplists:get_value(connect_timeout, Options, 8000)), - %% 1. Acquire per-host slot (blocks with backoff until available) - case hackney_load_regulation:acquire(Host, Port, MaxPerHost, CheckoutTimeout) of - ok -> + %% 1. Acquire per-host slot, or reuse an H2 connection that becomes ready + case acquire_pool_slot(Transport, Host, Port, Options, PoolHandler, + MaxPerHost, CheckoutTimeout) of + slot -> %% Slot acquired - now get connection from pool SslPooling = proplists:get_value(ssl_pooling, Options, hackney_app:get_app_env(ssl_pooling, false)), @@ -389,8 +383,7 @@ connect_pool_new(Transport, Host, Port, Options, FinalSslOpts, PoolHandler) -> case maybe_upgrade_ssl(Transport, ConnPid, FinalSslOpts) of ok -> %% Check if HTTP/2 was negotiated, register for multiplexing - maybe_register_h2(ConnPid, Host, Port, Transport, Options, PoolHandler), - {ok, ConnPid}; + maybe_register_h2(ConnPid, Host, Port, Transport, Options, PoolHandler); {error, Reason} -> %% Upgrade failed - release slot and close connection hackney_load_regulation:release(Host, Port), @@ -403,10 +396,93 @@ connect_pool_new(Transport, Host, Port, Options, FinalSslOpts, PoolHandler) -> {error, Reason} end end; - {error, timeout} -> + {h2, H2Pid} -> + {ok, H2Pid}; + timeout -> {error, checkout_timeout} end. +acquire_pool_slot(hackney_ssl, Host, Port, Options, PoolHandler, + MaxPerHost, Timeout) -> + case lists:member(http2, proplists:get_value(protocols, Options, + hackney_util:default_protocols())) of + true -> + Deadline = checkout_deadline(Timeout), + acquire_pool_slot_h2(Host, Port, Options, PoolHandler, + MaxPerHost, Deadline); + false -> + acquire_pool_slot_only(Host, Port, MaxPerHost, Timeout) + end; +acquire_pool_slot(_Transport, Host, Port, _Options, _PoolHandler, + MaxPerHost, Timeout) -> + acquire_pool_slot_only(Host, Port, MaxPerHost, Timeout). + +acquire_pool_slot_only(Host, Port, MaxPerHost, Timeout) -> + case hackney_load_regulation:acquire(Host, Port, MaxPerHost, Timeout) of + ok -> slot; + {error, timeout} -> timeout + end. + +acquire_pool_slot_h2(Host, Port, Options, PoolHandler, MaxPerHost, Deadline) -> + Wait = min(?H2_PROBE_TIMEOUT, checkout_time_left(Deadline)), + case hackney_load_regulation:acquire(Host, Port, MaxPerHost, Wait) of + ok -> + slot; + {error, timeout} -> + case checkout_h2_ready(Host, Port, Options, PoolHandler, Deadline) of + {ok, Pid} -> + {h2, Pid}; + timeout -> + timeout; + none when Wait =:= 0 -> + timeout; + none -> + acquire_pool_slot_h2(Host, Port, Options, PoolHandler, + MaxPerHost, Deadline) + end + end. + +checkout_h2_ready(Host, Port, Options, PoolHandler, Deadline) -> + case checkout_time_left(Deadline) of + 0 -> + timeout; + Remaining -> + ProbeTimeout = min(?H2_PROBE_TIMEOUT, Remaining), + ProbeOptions = lists:keystore(connect_timeout, 1, Options, + {connect_timeout, ProbeTimeout}), + case PoolHandler:checkout_h2(Host, Port, hackney_ssl, ProbeOptions) of + {ok, Pid} -> + checkout_h2_candidate(Pid, Deadline); + none -> + none + end + end. + +checkout_h2_candidate(Pid, Deadline) -> + case checkout_time_left(Deadline) of + 0 -> + timeout; + Remaining -> + ProbeTimeout = min(?H2_PROBE_TIMEOUT, Remaining), + case h2_checkout_status(Pid, ProbeTimeout) of + ready -> {ok, Pid}; + busy -> none; + unusable -> + stop_conn(Pid, min(ProbeTimeout, checkout_time_left(Deadline))), + none + end + end. + +checkout_deadline(infinity) -> + infinity; +checkout_deadline(Timeout) -> + erlang:monotonic_time(millisecond) + Timeout. + +checkout_time_left(infinity) -> + ?H2_PROBE_TIMEOUT; +checkout_time_left(Deadline) -> + max(0, Deadline - erlang:monotonic_time(millisecond)). + %% @private SSL-pooling checkout. A `ready' conn is an already-upgraded %% HTTPS/1.1 connection reused on an exact tls_key match; it was registered %% for h2 at creation if applicable, so it is not re-registered here. A @@ -420,8 +496,7 @@ connect_pool_ssl(Transport, Host, Port, Options, FinalSslOpts, PoolHandler) -> case hackney_conn:upgrade_to_ssl(ConnPid, FinalSslOpts, #{final => true, pool_ssl => true}) of ok -> - maybe_register_h2(ConnPid, Host, Port, Transport, Options, PoolHandler), - {ok, ConnPid}; + maybe_register_h2(ConnPid, Host, Port, Transport, Options, PoolHandler); {error, Reason} -> hackney_load_regulation:release(Host, Port), stop_conn(ConnPid), @@ -438,15 +513,36 @@ maybe_register_h2(ConnPid, Host, Port, Transport, Options, PoolHandler) -> try hackney_conn:get_protocol(ConnPid) of http2 -> %% HTTP/2 negotiated - register for connection sharing - PoolHandler:register_h2(Host, Port, Transport, ConnPid, Options); + case PoolHandler:register_h2(Host, Port, Transport, ConnPid, Options) of + ok -> + {ok, ConnPid}; + {ok, RegisteredPid} -> + {ok, RegisteredPid}; + {error, _} = Error -> + stop_conn(ConnPid, ?H2_PROBE_TIMEOUT), + Error + end; http1 -> - ok; + {ok, ConnPid}; http3 -> - ok + {ok, ConnPid} catch - _:_ -> - %% Connection terminated before we could check - ignore - ok + _:Reason -> + stop_conn(ConnPid, ?H2_PROBE_TIMEOUT), + {error, Reason} + end. + +h2_checkout_status(Pid) -> + h2_checkout_status(Pid, ?H2_PROBE_TIMEOUT). + +h2_checkout_status(Pid, Timeout) -> + try hackney_conn:get_state(Pid, Timeout) of + {ok, connected} -> ready; + {ok, streaming_body} -> busy; + _ -> unusable + catch + exit:{timeout, _} -> busy; + _:_ -> unusable end. %% @private Upgrade TCP connection to SSL if needed. @@ -474,6 +570,9 @@ maybe_upgrade_ssl(_, _ConnPid, _FinalSslOpts) -> stop_conn(ConnPid) -> try hackney_conn:stop(ConnPid) catch _:_ -> ok end. +stop_conn(ConnPid, Timeout) -> + try hackney_conn:stop(ConnPid, Timeout) catch _:_ -> ok end. + %% @private Signal the websocket process to shut down, ignoring errors. shutdown_ws(WsPid) -> try exit(WsPid, shutdown) catch _:_ -> ok end. diff --git a/src/hackney_conn.erl b/src/hackney_conn.erl index ca54fb99..560e867a 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, + retire_h2/1, %% Protocol info get_protocol/1 ]). @@ -228,6 +229,10 @@ %% {stream, body_full, Status, Headers, Acc, From} %% {stream, done, Status, Headers, Buffer} h2_streams = #{} :: #{pos_integer() => {term(), tuple()}}, + %% Per-stream caller monitors: StreamId => monitor reference + h2_stream_monitors = #{} :: #{pos_integer() => reference()}, + %% Stop after the last tracked stream is consumed + h2_retiring = false :: boolean(), %% 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 +619,10 @@ set_owner(Pid, NewOwner, Timeout) -> set_owner_async(Pid, NewOwner) -> gen_statem:cast(Pid, {set_owner, NewOwner}). +-spec retire_h2(pid()) -> ok. +retire_h2(Pid) -> + gen_statem:cast(Pid, retire_h2). + %% @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()}. @@ -1895,6 +1904,12 @@ handle_common(cast, stop, _State, Data) -> %% Async stop - used by pool to avoid deadlock during sync checkin {stop, normal, Data}; +handle_common(cast, retire_h2, _State, Data) -> + h2_stream_result(Data#conn_data{h2_retiring = true}, []); + +handle_common(info, {'DOWN', Ref, process, _Pid, _Reason}, State, Data) -> + handle_h2_stream_owner_down(Ref, State, Data); + handle_common(cast, _Msg, _State, _Data) -> keep_state_and_data; @@ -2998,7 +3013,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 +3050,68 @@ h2_start_failure(after_upgrade, From, Reason) -> close_h2(H2Conn) -> try h2_connection:close(H2Conn) catch _:_ -> ok end. +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) -> + Data1 = Data#conn_data{h2_stream_id = undefined, + request_from = undefined}, + case h2_stream_result(Data1, []) of + {keep_state, Data2, []} -> + {next_state, connected, Data2, + [{state_timeout, infinity, idle_timeout}]}; + Stop -> + Stop + end; +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,21 +3157,22 @@ 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 %% sending for it and the h2 layer drops it; otherwise the %% pooled connection would be reused with an orphaned stream - %% still open (h2_conn_usable only checks the conn state). + %% still open (pool readiness 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 +3246,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 +3255,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 +3275,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 +3292,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 +3387,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 +3407,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 +3560,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 +3593,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 +3629,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 +3661,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 +3672,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 +3688,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,15 +3725,21 @@ h2_stream_parked_from({stream, headers, _, _, _, From}) -> From; h2_stream_parked_from({stream, body_full, _, _, _, From}) -> From; h2_stream_parked_from(_) -> undefined. +h2_stream_result(#conn_data{h2_retiring = true, h2_streams = Streams} = Data, + Replies) when map_size(Streams) =:= 0 -> + {stop_and_reply, normal, Replies, Data}; +h2_stream_result(Data, Replies) -> + {keep_state, Data, Replies}. + 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 %% socket open for a drain window. Leaving the conn `connected` and pooled - %% made checkout_h2/h2_conn_usable keep handing it out, so every reused + %% made checkout_h2 keep handing it out, so every reused %% request opened a stream past last_stream_id that the peer ignored and hung %% to recv_timeout. Tear the connection down and transition to `closed` (like - %% h2_on_closed/2): the pool then stops reusing it (h2_conn_usable requires - %% `connected`) and new requests dial a fresh connection. in-flight streams + %% h2_on_closed/2): the pool then stops reusing it, and new requests dial a + %% fresh connection. In-flight streams %% are aborted with the goaway error as before. {Replies, Data1} = collect_h2_aborts({goaway, ErrorCode}, Data), Data2 = cancel_all_h2_timers(Data1), @@ -3707,7 +3784,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..54e5112a 100644 --- a/src/hackney_pool.erl +++ b/src/hackney_pool.erl @@ -94,10 +94,9 @@ -define(DEFAULT_PREWARM_COUNT, 4). % Connections to maintain per host -define(STOP_CONN_TIMEOUT, 100). % Max wait for a conn to stop -define(PREWARM_CONNECT_TIMEOUT, 5000). % Dial budget for a prewarm conn -%% Every question the pool asks a conn about its own health is answered from -%% the conn's state, so a healthy conn answers at once. A conn that does not -%% is wedged, and waiting on it from inside the pool gen_server blocks every -%% caller of the pool, not just the one that asked: treat slow as unusable. +%% Bound calls from the pool to a connection so one slow process does not block +%% every pool caller. HTTP/2 state probe timeouts are treated as busy so active +%% streams can drain -define(PROBE_TIMEOUT, 250). start() -> @@ -210,13 +209,27 @@ checkout_h2(Host, Port, Transport, Options) -> %% @doc Register an HTTP/2 connection in the pool for sharing. %% Called after ALPN negotiation confirms HTTP/2. -spec register_h2(Host :: string(), Port :: non_neg_integer(), - Transport :: module(), Pid :: pid(), Options :: list()) -> ok. + Transport :: module(), Pid :: pid(), Options :: list()) -> + ok | {ok, pid()} | {error, term()}. register_h2(Host, Port, Transport, Pid, Options) -> PoolName = proplists:get_value(pool, Options, default), + ConnectTimeout = proplists:get_value(connect_timeout, Options, 8000), + RegisterTimeout = proplists:get_value(checkout_timeout, Options, + ConnectTimeout), Pool = find_pool(PoolName, Options), Key = h2_connection_key(Host, Port, Transport, Options), - gen_server:cast(Pool, {register_h2, Key, Pid}), - ok. + Deadline = registration_deadline(RegisterTimeout), + CallTimeout = registration_call_timeout(RegisterTimeout), + try + gen_server:call(Pool, {register_h2, Key, Pid, Deadline}, CallTimeout) + catch + exit:{timeout, _} -> + stop_conn(Pid), + {error, checkout_timeout}; + _:_ -> + stop_conn(Pid), + {error, checkout_failure} + end. %% @doc Remove an HTTP/2 connection from the pool (e.g., on GOAWAY). -spec unregister_h2(Pid :: pid(), Options :: list()) -> ok. @@ -633,15 +646,28 @@ handle_call({checkout_h2, Key}, _From, #state{h2_connections = H2Conns} = State) undefined -> {reply, none, State}; Pid -> - case h2_conn_usable(Pid) of - true -> + case h2_conn_status(Pid) of + ready -> {reply, {ok, Pid}, State}; - false -> + busy -> + {reply, none, State}; + unusable -> + stop_conn(Pid), H2Conns2 = maps:remove(Key, H2Conns), {reply, none, State#state{h2_connections = H2Conns2}} end end; +handle_call({register_h2, Key, Pid, Deadline}, _From, + #state{h2_connections = H2Conns} = State) -> + case registration_expired(Deadline) of + true -> + stop_conn(Pid), + {reply, {error, checkout_timeout}, State}; + false -> + do_register_h2(Key, Pid, Deadline, H2Conns, State) + end; + handle_call({checkout_h3, Key}, _From, #state{h3_connections = H3Conns} = State) -> %% HTTP/3 checkout - return existing connection if available case maps:get(Key, H3Conns, undefined) of @@ -701,20 +727,6 @@ handle_cast({prewarm_checkin, Pid, Key}, State) -> Available2 = maps:update_with(Key, fun(Pids) -> [Pid | Pids] end, [Pid], Available), {noreply, State#state{available=Available2, pid_monitors=PidMonitors2}}; -handle_cast({register_h2, Key, Pid}, State) -> - %% Register an HTTP/2 connection for sharing - #state{h2_connections = H2Conns, pid_monitors = PidMonitors} = State, - %% Monitor the connection if not already monitored - PidMonitors2 = case maps:is_key(Pid, PidMonitors) of - true -> PidMonitors; - false -> - MonRef = erlang:monitor(process, Pid), - maps:put(Pid, MonRef, PidMonitors) - end, - %% Store HTTP/2 connection - H2Conns2 = maps:put(Key, Pid, H2Conns), - {noreply, State#state{h2_connections = H2Conns2, pid_monitors = PidMonitors2}}; - handle_cast({unregister_h2, Pid}, State) -> %% Remove an HTTP/2 connection from the pool State2 = do_unregister_h2(Pid, State), @@ -1237,20 +1249,99 @@ pool_has_idle_room(#state{available=Available, max_connections=MaxConn}) -> idle_count(Available) -> maps:fold(fun(_, Pids, Acc) -> Acc + length(Pids) end, 0, Available). -%% @private Check that a pooled HTTP/2 conn is alive and in `connected` state. -%% Short timeout so a stuck conn doesn't wedge the pool; any failure → unusable. -h2_conn_usable(Pid) -> +%% @private Check whether a pooled HTTP/2 conn can accept a new request +h2_conn_status(Pid) -> case erlang:is_process_alive(Pid) of - false -> false; + false -> unusable; true -> try hackney_conn:get_state(Pid, ?PROBE_TIMEOUT) of - {ok, connected} -> true; - _ -> false + {ok, connected} -> ready; + {ok, streaming_body} -> busy; + _ -> unusable catch - _:_ -> false + exit:{timeout, _} -> busy; + _:_ -> unusable end end. +do_register_h2(Key, Pid, Deadline, H2Conns, State) -> + case maps:get(Key, H2Conns, undefined) of + undefined -> + register_h2_connection(Key, Pid, Deadline, State); + Pid -> + {reply, {ok, Pid}, State}; + Existing -> + Status = h2_conn_status(Existing), + case registration_expired(Deadline) of + true -> + stop_conn(Pid), + {reply, {error, checkout_timeout}, State}; + false -> + do_register_h2(Status, Key, Pid, Existing, Deadline, State) + end + end. + +do_register_h2(ready, _Key, Pid, Existing, _Deadline, State) -> + stop_conn(Pid), + {reply, {ok, Existing}, State}; +do_register_h2(busy, Key, Pid, Existing, Deadline, State) -> + case register_h2_connection(Key, Pid, Deadline, State) of + {reply, ok, _} = Result -> + hackney_conn:retire_h2(Existing), + Result; + Error -> + Error + end; +do_register_h2(unusable, Key, Pid, Existing, Deadline, State) -> + stop_conn(Existing), + register_h2_connection(Key, Pid, Deadline, State). + +register_h2_connection(Key, Pid, Deadline, State) -> + #state{h2_connections = H2Conns, pid_monitors = PidMonitors} = State, + case registration_expired(Deadline) of + true -> + stop_conn(Pid), + {reply, {error, checkout_timeout}, State}; + false -> + register_h2_connection(Key, Pid, Deadline, State, + set_owner(Pid, self()), + H2Conns, PidMonitors) + end. + +register_h2_connection(Key, Pid, Deadline, State, ok, + H2Conns, PidMonitors) -> + case registration_expired(Deadline) of + false -> + PidMonitors2 = case maps:is_key(Pid, PidMonitors) of + true -> PidMonitors; + false -> + MonRef = erlang:monitor(process, Pid), + maps:put(Pid, MonRef, PidMonitors) + end, + H2Conns2 = maps:put(Key, Pid, H2Conns), + {reply, ok, State#state{h2_connections = H2Conns2, + pid_monitors = PidMonitors2}}; + true -> + stop_conn(Pid), + {reply, {error, checkout_timeout}, State} + end; +register_h2_connection(_Key, Pid, _Deadline, State, {error, _} = Error, + _H2Conns, _PidMonitors) -> + stop_conn(Pid), + {reply, Error, State}. + +registration_deadline(infinity) -> infinity; +registration_deadline(Timeout) -> + erlang:monotonic_time(millisecond) + Timeout. + +registration_call_timeout(infinity) -> infinity; +registration_call_timeout(Timeout) -> + Timeout + 2 * ?PROBE_TIMEOUT + ?STOP_CONN_TIMEOUT. + +registration_expired(infinity) -> false; +registration_expired(Deadline) -> + erlang:monotonic_time(millisecond) >= Deadline. + %% @private Remove an HTTP/2 connection from the pool do_unregister_h2(Pid, State) -> #state{h2_connections = H2Conns, pid_monitors = PidMonitors} = State, diff --git a/test/hackney_pool_safety_tests.erl b/test/hackney_pool_safety_tests.erl index 4c5ef14b..ed21fc72 100644 --- a/test/hackney_pool_safety_tests.erl +++ b/test/hackney_pool_safety_tests.erl @@ -23,7 +23,7 @@ -include_lib("eunit/include/eunit.hrl"). %% Calls that cannot raise: casts are fire and forget. --define(SAFE_BY_NATURE, [set_owner_async]). +-define(SAFE_BY_NATURE, [set_owner_async, retire_h2]). conn_calls_are_guarded_test() -> Unguarded = [Call || Call <- conn_calls(hackney_pool), unguarded(Call)],