From 750869c8557d1c1c71a067cf714648ae91b79672 Mon Sep 17 00:00:00 2001 From: Benoit Chesneau Date: Wed, 15 Jul 2026 09:17:39 +0200 Subject: [PATCH] Stop non-reusable connections when a request completes --- src/hackney_conn.erl | 64 +++++++++++++++--------- test/hackney_conn_tests.erl | 97 +++++++++++++++++++++++++++++++++++++ 2 files changed, 139 insertions(+), 22 deletions(-) diff --git a/src/hackney_conn.erl b/src/hackney_conn.erl index 8bf711fc..4995e282 100644 --- a/src/hackney_conn.erl +++ b/src/hackney_conn.erl @@ -1518,8 +1518,8 @@ receiving({call, From}, body, Data) -> %% Transition to closed state instead of connected {next_state, closed, NewData, [{reply, From, {ok, Body}}]}; {ok, Body, NewData} -> - %% Socket still valid - return to connected state - {next_state, connected, NewData, [{reply, From, {ok, Body}}]}; + %% Socket still valid - reuse it, or stop if not reusable. + finish_sync_request(From, {ok, Body}, NewData); {error, Reason} -> {next_state, closed, Data, [{reply, From, {error, Reason}}]} end; @@ -1533,7 +1533,7 @@ receiving({call, From}, stream_body, Data) -> %% Socket was closed during body read - transition to closed state {next_state, closed, NewData, [{reply, From, done}]}; {done, NewData} -> - {next_state, connected, NewData, [{reply, From, done}]}; + finish_sync_request(From, done, NewData); {error, Reason} -> {next_state, closed, Data, [{reply, From, {error, Reason}}]} end; @@ -1955,27 +1955,47 @@ should_close_connection(#conn_data{version = Version, response_headers = Headers request_close = RequestClose}) -> hackney_keepalive:should_close(Version, Headers, RequestClose). -%% @private Finish async streaming - close or return to connected based on Connection header -finish_async_streaming(Data) -> - #conn_data{transport = Transport, socket = Socket, pool_pid = PoolPid} = Data, - case should_close_connection(Data) of +%% @private Whether a connection whose request/response cycle just finished may +%% be reused. Shared by the sync and async completion paths so the reuse test +%% lives in one place: a connection flagged no_reuse (proxy tunnel, SSL upgrade, +%% pool disabled) or whose response asked to close (RFC 7230) is not reusable. +connection_reusable(#conn_data{no_reuse = true}) -> + false; +connection_reusable(Data) -> + not should_close_connection(Data). + +%% @private Close the socket if we still hold one. Returns ok. +close_socket(_Transport, undefined) -> ok; +close_socket(Transport, Socket) -> Transport:close(Socket), ok. + +%% @private Finish a completed synchronous request (blocking body / stream_body). +%% A reusable connection returns to `connected': a pooled one is handed back to +%% the pool from connected(enter, receiving), a direct one stays alive for the +%% owner to reuse. A non-reusable one is closed and the process stopped, so a +%% no_reuse / Connection: close conn no longer parks in `connected' forever +%% (#902). Mirrors finish_async_streaming/1 for the sync path. +finish_sync_request(From, Reply, #conn_data{transport = Transport, socket = Socket} = Data) -> + case connection_reusable(Data) of true -> - %% Connection: close - close socket and stop process - case Socket of - undefined -> ok; - _ -> Transport:close(Socket) - end, - {stop, normal, reset_async(Data#conn_data{socket = undefined})}; - false when PoolPid =:= undefined -> - %% No pool and no Connection: close - still stop since no reuse - case Socket of - undefined -> ok; - _ -> Transport:close(Socket) - end, - {stop, normal, reset_async(Data#conn_data{socket = undefined})}; + {next_state, connected, Data, [{reply, From, Reply}]}; + false -> + ok = close_socket(Transport, Socket), + {stop_and_reply, normal, [{reply, From, Reply}], + Data#conn_data{socket = undefined}} + end. + +%% @private Finish async streaming. Reuse only a reusable, pooled connection +%% (direct async conns are not re-driven, so they stop rather than park); +%% otherwise close and stop. Now also honours no_reuse via connection_reusable/1, +%% closing the gap where a no_reuse pooled async conn returned to `connected'. +finish_async_streaming(#conn_data{transport = Transport, socket = Socket, + pool_pid = PoolPid} = Data) -> + case connection_reusable(Data) andalso PoolPid =/= undefined of + true -> + {next_state, connected, reset_async(Data)}; false -> - %% Has pool or keep-alive - return to connected for reuse - {next_state, connected, reset_async(Data)} + ok = close_socket(Transport, Socket), + {stop, normal, reset_async(Data#conn_data{socket = undefined})} end. %% @private Handle receiving response status and headers diff --git a/test/hackney_conn_tests.erl b/test/hackney_conn_tests.erl index 8ae1071e..a7d3828d 100644 --- a/test/hackney_conn_tests.erl +++ b/test/hackney_conn_tests.erl @@ -63,6 +63,11 @@ hackney_conn_integration_test_() -> {"set_owner while receiving body", {timeout, 30, fun test_set_owner_while_receiving/0}}, {"set_owner while streaming async", {timeout, 30, fun test_set_owner_while_streaming/0}}, {"set_owner API unchanged in other states", {timeout, 30, fun test_set_owner_api_unchanged/0}}, + %% Non-reusable connections stop at request completion (#902) + {"sync body on a non-reusable conn stops it", {timeout, 30, fun test_sync_body_no_reuse_stops/0}}, + {"sync stream_body on a non-reusable conn stops it", {timeout, 30, fun test_sync_stream_body_no_reuse_stops/0}}, + {"async on a non-reusable pooled conn stops it", {timeout, 30, fun test_async_no_reuse_pooled_stops/0}}, + {"sync body on a reusable conn keeps it connected", {timeout, 30, fun test_sync_body_reusable_stays_connected/0}}, %% 1XX response handling {"skip 1XX informational responses", {timeout, 30, fun test_skip_1xx_responses/0}} ]}. @@ -776,6 +781,98 @@ test_set_owner_api_unchanged() -> hackney_conn:stop(Pid). +%% A connection flagged no_reuse must be closed and its process stopped once the +%% synchronous body read completes, instead of parking in `connected' forever +%% (#902). Uses a keep-alive endpoint so the socket stays valid at completion; +%% the no_reuse flag alone must drive termination. +test_sync_body_no_reuse_stops() -> + Opts = #{ + host => "127.0.0.1", + port => ?PORT, + transport => hackney_tcp, + connect_timeout => 5000, + recv_timeout => 5000, + no_reuse => true + }, + {ok, Pid} = hackney_conn:start_link(Opts), + ok = hackney_conn:connect(Pid), + {ok, _Status, _Headers} = hackney_conn:request(Pid, <<"GET">>, <<"/get">>, [], <<>>), + MRef = erlang:monitor(process, Pid), + {ok, Body} = hackney_conn:body(Pid), + ?assert(is_binary(Body)), + ?assert(wait_down(Pid, MRef)), + ?assertNot(is_process_alive(Pid)). + +%% Same, but draining the body through stream_body/1 to `done'. +test_sync_stream_body_no_reuse_stops() -> + Opts = #{ + host => "127.0.0.1", + port => ?PORT, + transport => hackney_tcp, + connect_timeout => 5000, + recv_timeout => 5000, + no_reuse => true + }, + {ok, Pid} = hackney_conn:start_link(Opts), + ok = hackney_conn:connect(Pid), + {ok, _Status, _Headers} = hackney_conn:request(Pid, <<"GET">>, <<"/get">>, [], <<>>), + MRef = erlang:monitor(process, Pid), + _Body = stream_all(Pid, <<>>), + ?assert(wait_down(Pid, MRef)), + ?assertNot(is_process_alive(Pid)). + +%% Async path parity: a no_reuse pooled connection used to return to `connected' +%% after streaming (the sync path's sibling gap). It must now stop too. +test_async_no_reuse_pooled_stops() -> + Pool = spawn(fun() -> receive stop -> ok end end), + Opts = #{ + host => "127.0.0.1", + port => ?PORT, + transport => hackney_tcp, + connect_timeout => 5000, + recv_timeout => 5000, + no_reuse => true, + pool_pid => Pool + }, + {ok, Pid} = hackney_conn:start_link(Opts), + ok = hackney_conn:connect(Pid), + MRef = erlang:monitor(process, Pid), + {ok, Ref} = hackney_conn:request_async(Pid, <<"GET">>, <<"/get">>, [], <<>>, true), + Messages = receive_all_async(Ref, []), + ?assert(lists:member(done, Messages)), + ?assert(wait_down(Pid, MRef)), + ?assertNot(is_process_alive(Pid)), + Pool ! stop. + +%% Non-regression: a reusable connection (no no_reuse, no Connection: close) must +%% still return to `connected' and be usable for a second request. +test_sync_body_reusable_stays_connected() -> + Opts = #{ + host => "127.0.0.1", + port => ?PORT, + transport => hackney_tcp, + connect_timeout => 5000, + recv_timeout => 5000 + }, + {ok, Pid} = hackney_conn:start_link(Opts), + ok = hackney_conn:connect(Pid), + {ok, _S1, _H1} = hackney_conn:request(Pid, <<"GET">>, <<"/get">>, [], <<>>), + {ok, _B1} = hackney_conn:body(Pid), + ?assertEqual({ok, connected}, hackney_conn:get_state(Pid)), + ?assert(is_process_alive(Pid)), + %% Reuse the same connection for a second request. + {ok, _S2, _H2} = hackney_conn:request(Pid, <<"GET">>, <<"/get">>, [], <<>>), + {ok, _B2} = hackney_conn:body(Pid), + ?assert(is_process_alive(Pid)), + hackney_conn:stop(Pid). + +%% Wait for a connection process to terminate normally after completion. +wait_down(Pid, MRef) -> + receive + {'DOWN', MRef, process, Pid, _Reason} -> true + after 2000 -> false + end. + %%==================================================================== %% 1XX Response Tests %%====================================================================