Stop dead-end connections instead of parking them in closed (#918)

NEWS.md

@@ -1,5 +1,24 @@
 # NEWS
 
+unreleased
+----------
+
+### Fixed
+
+- A response body cut short by the peer closing mid-transfer no longer leaks
+  the connection process. `read_full_body/2` hands back `socket = undefined`,
+  so the connection went straight to `closed` and never reached the reuse
+  check added for #902. An unpooled connection arms no grace timer there and,
+  when started under `hackney_conn_sup`, has the supervisor as its `owner`, so
+  the owner-DOWN clause never fired either: the process parked forever holding
+  every refc binary it had read. Callers could not clean up, since a
+  synchronous request returns the body directly and the truncated read still
+  reports `{ok, Body}` (#918). The same applies to a failed body read and to
+  bodyless (204/304) responses.
+- `hackney_conn:get_location/1` and `set_location/2` no longer exit with
+  `noproc` when the connection has already stopped, which would otherwise
+  propagate out of `hackney:request/5` on the redirect path.
+
 4.7.2 - 2026-07-17
 ------------------
 

src/hackney_conn.erl

@@ -510,12 +510,15 @@ response_headers(Pid) ->
 %% @doc Get the stored location (final URL after redirects).
 -spec get_location(pid()) -> binary() | undefined.
 get_location(Pid) ->
-    gen_statem:call(Pid, get_location).
+    case safe_call(Pid, get_location) of
+        {error, closed} -> undefined;
+        Location -> Location
+    end.
 
 %% @doc Set the location (used after following redirects).
--spec set_location(pid(), binary()) -> ok.
+-spec set_location(pid(), binary()) -> ok | {error, closed}.
 set_location(Pid, Location) ->
-    gen_statem:call(Pid, {set_location, Location}).
+    safe_call(Pid, {set_location, Location}).
 
 %% @doc Send data through the connection process.
 %% This is a low-level function used by hackney_request.
@@ -1553,13 +1556,12 @@ receiving({call, From}, body, Data) ->
     case read_full_body(Data, <<>>) of
         {ok, Body, #conn_data{socket = undefined} = NewData} ->
             %% Socket was closed during body read (e.g., no Content-Length)
-            %% Transition to closed state instead of connected
-            {next_state, closed, NewData, [{reply, From, {ok, Body}}]};
+            finish_dead_request(From, {ok, Body}, NewData);
         {ok, Body, NewData} ->
             %% 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}}]}
+            finish_dead_request(From, {error, Reason}, Data)
     end;
 
 receiving({call, From}, stream_body, Data) ->
@@ -1568,12 +1570,12 @@ receiving({call, From}, stream_body, Data) ->
         {ok, Chunk, NewData} ->
             {keep_state, NewData, [{reply, From, {ok, Chunk}}]};
         {done, #conn_data{socket = undefined} = NewData} ->
-            %% Socket was closed during body read - transition to closed state
-            {next_state, closed, NewData, [{reply, From, done}]};
+            %% Socket was closed during body read
+            finish_dead_request(From, done, NewData);
         {done, NewData} ->
             finish_sync_request(From, done, NewData);
         {error, Reason} ->
-            {next_state, closed, Data, [{reply, From, {error, Reason}}]}
+            finish_dead_request(From, {error, Reason}, Data)
     end;
 
 receiving({call, From}, get_state, _Data) ->
@@ -2024,6 +2026,27 @@ finish_sync_request(From, Reply, #conn_data{transport = Transport, socket = Sock
              Data#conn_data{socket = undefined}}
     end.
 
+%% @private The request is over and the connection cannot carry another one:
+%% the body was cut short (the peer closed mid-transfer, so read_full_body/2
+%% hands back `socket = undefined'), the response had no body to read, or the
+%% read failed outright. None of these reach finish_sync_request/3 (#902).
+%%
+%% A pooled conn parks in `closed', keeping the grace window added for #836 so
+%% late calls still get a proper reply before it stops. An unpooled conn has no
+%% such timer, and when it was started under hackney_conn_sup its `owner' is the
+%% supervisor, so the owner-DOWN clause never fires either: parking there leaks
+%% the process along with every refc binary it read (#918). Stop instead.
+finish_dead_request(From, Reply, #conn_data{transport = Transport, socket = Socket,
+                                            pool_pid = PoolPid} = Data) ->
+    case PoolPid of
+        undefined ->
+            ok = close_socket(Transport, Socket),
+            {stop_and_reply, normal, [{reply, From, Reply}],
+             Data#conn_data{socket = undefined}};
+        _ ->
+            {next_state, closed, Data, [{reply, From, Reply}]}
+    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,

test/hackney_conn_tests.erl

@@ -68,6 +68,11 @@ hackney_conn_integration_test_() ->
       {"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}},
+      %% Truncated body: bypasses finish_sync_request/3 and lands in `closed' (#918)
+      {"truncated body on an unpooled conn stops it", {timeout, 30, fun test_truncated_body_unpooled_stops/0}},
+      {"truncated stream_body on an unpooled conn stops it", {timeout, 30, fun test_truncated_stream_body_unpooled_stops/0}},
+      {"truncated body on a pooled conn keeps the grace window", {timeout, 30, fun test_truncated_body_pooled_keeps_grace/0}},
+      {"location accessors are noproc-safe", {timeout, 30, fun test_location_accessors_noproc_safe/0}},
       %% 1XX response handling
       {"skip 1XX informational responses", {timeout, 30, fun test_skip_1xx_responses/0}}
      ]}.
@@ -955,3 +960,109 @@ stream_all(Pid, Acc) ->
         done ->
             Acc
     end.
+
+%% A response whose body is cut short (peer closes mid-body) leaves
+%% read_full_body/2 with `socket = undefined', so `receiving' goes straight to
+%% `closed' and never reaches finish_sync_request/3 -- the #902 fix. For an
+%% unpooled connection `closed(enter)' arms no timer and the owner is
+%% hackney_conn_sup, so the process parked forever holding the partial body.
+test_truncated_body_unpooled_stops() ->
+    {LSock, Port} = start_truncating_server(1024 * 64),
+    try
+        Pid = truncating_conn(Port),
+        MRef = erlang:monitor(process, Pid),
+        %% The short read still reports success, so the caller has no reason
+        %% (and under hackney 4 no handle) to close anything.
+        {ok, Body} = hackney_conn:body(Pid),
+        ?assertEqual(1024 * 64, byte_size(Body)),
+        ?assertEqual(normal, wait_down_reason(Pid, MRef))
+    after
+        catch gen_tcp:close(LSock)
+    end.
+
+%% Same, draining through stream_body/1 rather than body/1.
+test_truncated_stream_body_unpooled_stops() ->
+    {LSock, Port} = start_truncating_server(1024 * 64),
+    try
+        Pid = truncating_conn(Port),
+        MRef = erlang:monitor(process, Pid),
+        _ = stream_all(Pid, <<>>),
+        ?assertEqual(normal, wait_down_reason(Pid, MRef))
+    after
+        catch gen_tcp:close(LSock)
+    end.
+
+%% A pooled connection must still park in `closed' so the #836 grace window
+%% answers late calls; the pool's own timer stops it shortly after.
+test_truncated_body_pooled_keeps_grace() ->
+    {LSock, Port} = start_truncating_server(1024 * 64),
+    Pool = spawn(fun() -> receive stop -> ok end end),
+    try
+        Pid = truncating_conn(Port, #{pool_pid => Pool}),
+        {ok, _Body} = hackney_conn:body(Pid),
+        ?assertEqual({ok, closed}, hackney_conn:get_state(Pid)),
+        ?assert(is_process_alive(Pid))
+    after
+        Pool ! stop,
+        catch gen_tcp:close(LSock)
+    end.
+
+truncating_conn(Port) ->
+    truncating_conn(Port, #{}).
+
+truncating_conn(Port, Extra) ->
+    Opts = maps:merge(#{
+        host => "127.0.0.1",
+        port => Port,
+        transport => hackney_tcp,
+        connect_timeout => 5000,
+        recv_timeout => 5000
+    }, Extra),
+    {ok, Pid} = hackney_conn:start_link(Opts),
+    ok = hackney_conn:connect(Pid),
+    {ok, _Status, _Headers} = hackney_conn:request(Pid, <<"GET">>, <<"/truncated">>, [], <<>>),
+    Pid.
+
+%% Announces a Content-Length far larger than what it sends, then closes.
+start_truncating_server(SendBytes) ->
+    {ok, LSock} = gen_tcp:listen(0, [binary, {active, false}, {reuseaddr, true}]),
+    {ok, Port} = inet:port(LSock),
+    spawn(fun() ->
+        case gen_tcp:accept(LSock, 5000) of
+            {ok, Sock} ->
+                _ = gen_tcp:recv(Sock, 0, 5000),
+                Headers = ["HTTP/1.1 200 OK\r\n",
+                           "Content-Type: application/octet-stream\r\n",
+                           "Content-Length: 100000000\r\n\r\n"],
+                _ = gen_tcp:send(Sock, Headers),
+                _ = gen_tcp:send(Sock, binary:copy(<<"x">>, SendBytes)),
+                gen_tcp:close(Sock);
+            _ ->
+                ok
+        end
+    end),
+    {LSock, Port}.
+
+wait_down_reason(Pid, MRef) ->
+    receive
+        {'DOWN', MRef, process, Pid, Reason} -> Reason
+    after 5000 ->
+        timeout
+    end.
+
+%% hackney:request/5 calls set_location/2 on the original connection after a
+%% redirect chain returns. Now that a dead-end connection stops rather than
+%% parks, that pid can already be gone, so the accessors must not exit with
+%% noproc and propagate out of the caller.
+test_location_accessors_noproc_safe() ->
+    Opts = #{
+        host => "127.0.0.1",
+        port => ?PORT,
+        transport => hackney_tcp,
+        connect_timeout => 5000
+    },
+    {ok, Pid} = hackney_conn:start_link(Opts),
+    ok = hackney_conn:stop(Pid),
+    ?assertNot(is_process_alive(Pid)),
+    ?assertEqual(undefined, hackney_conn:get_location(Pid)),
+    ?assertEqual({error, closed}, hackney_conn:set_location(Pid, <<"http://127.0.0.1/final">>)).