Stop non-reusable connections when a request completes

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

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
 %%====================================================================