fix(reverse_proxy): release the HTTP client on every outcome

Three paths obtained a client ref and abandoned it without closing:
the content-length rejection in call/3, the invalid-status clause in
request/4, and the success branch of response/6.

Under Hackney 4 an abandoned ref is a permanent process leak. A conn
that finishes or errors without an explicit stop parks in `closed` or
`connected` and never terminates: closed/3 only arms its stop timer for
pooled conns, `connected` gets no state_timeout when idle_timeout is
infinity, and a non-pooled conn monitors hackney_conn_sup rather than
its caller, so the owner-DOWN cleanup never fires. Each parked process
pins every refc binary it read, which for proxied video is tens of MB.

Releasing on every outcome means the gun client must stop releasing in
stream_body/1, or the ref would be removed twice.

Hackney does reclaim a pooled connection itself for a response that
cannot carry a body (204, 304, HEAD), so those refs are marked and left
alone; stopping one would evict a connection another request may
already hold. The mark is gated on the request actually being pooled,
since a direct or proxy-tunnelled conn is never checked in and leaks
unless we stop it.

changelog.d/reverse-proxy-client-leak.fix

@@ -0,0 +1 @@
+Release the reverse proxy HTTP client on every outcome, fixing a memory leak where abandoned Hackney connection processes retained the whole response body

lib/pleroma/reverse_proxy.ex

@@ -118,7 +118,8 @@ defmodule Pleroma.ReverseProxy do
     with {:ok, nil} <- @cachex.get(:failed_proxy_url_cache, url),
          {:ok, code, headers, client} <- request(method, url, req_headers, client_opts),
          :ok <-
-           header_length_constraint(
+           enforce_header_length(
+             client,
              headers,
              Keyword.get(opts, :max_body_length, @max_body_length)
            ) do
@@ -176,7 +177,8 @@ defmodule Pleroma.ReverseProxy do
       {:ok, code, headers} when code in @valid_resp_codes ->
         {:ok, code, downcase_headers(headers)}
 
-      {:ok, code, _, _} ->
+      {:ok, code, _, client} ->
+        client().close(client)
         {:error, {:invalid_http_response, code}}
 
       {:ok, code, _} ->
@@ -197,12 +199,16 @@ defmodule Pleroma.ReverseProxy do
       |> send_chunked(status)
       |> chunk_reply(client, opts)
 
+    # Release the ref on every outcome, including a fully streamed body. The
+    # connection process behind it is not tied to this request and keeps the
+    # response body's binaries alive until it is explicitly closed.
+    client().close(client)
+
     case result do
       {:ok, conn} ->
         halt(conn)
 
       {:error, :closed, conn} ->
-        client().close(client)
         halt(conn)
 
       {:error, error, conn} ->
@@ -210,7 +216,6 @@ defmodule Pleroma.ReverseProxy do
           "#{__MODULE__} request to #{url} failed while reading/chunking: #{inspect(error)}"
         )
 
-        client().close(client)
         halt(conn)
     end
   end
@@ -396,6 +401,19 @@ defmodule Pleroma.ReverseProxy do
     end
   end
 
+  # Rejecting on content-length happens after the upstream response was already
+  # accepted, so the ref is live and has to be released before we bail out.
+  defp enforce_header_length(client, headers, limit) do
+    case header_length_constraint(headers, limit) do
+      :ok ->
+        :ok
+
+      {:error, _} = error ->
+        client().close(client)
+        error
+    end
+  end
+
   defp header_length_constraint(headers, limit) when is_integer(limit) and limit > 0 do
     with {_, size} <- List.keyfind(headers, "content-length", 0),
          {size, _} <- Integer.parse(size),

lib/pleroma/reverse_proxy/client/hackney.ex

@@ -54,6 +54,8 @@ defmodule Pleroma.ReverseProxy.Client.Hackney do
   end
 
   @impl true
+  def stream_body(%{released: true}), do: :done
+
   def stream_body(%{ref: ref, recv_timeout: recv_timeout} = client) do
     _ = :hackney.stream_next(ref)
 
@@ -73,6 +75,8 @@ defmodule Pleroma.ReverseProxy.Client.Hackney do
   end
 
   @impl true
+  def close(%{released: true}), do: :ok
+
   def close(%{ref: ref}), do: :hackney.close(ref)
 
   def close(ref), do: :hackney.close(ref)
@@ -89,7 +93,7 @@ defmodule Pleroma.ReverseProxy.Client.Hackney do
       result =
         with {:ok, {:status, code, _reason}} <- receive_response(ref, recv_timeout),
              {:ok, {:headers, resp_headers}} <- receive_response(ref, recv_timeout) do
-          {:ok, code, resp_headers, %{ref: ref, recv_timeout: recv_timeout}}
+          {:ok, code, resp_headers, build_client(ref, recv_timeout, method, code, opts)}
         end
 
       case result do
@@ -104,6 +108,43 @@ defmodule Pleroma.ReverseProxy.Client.Hackney do
     end
   end
 
+  # A response that cannot carry a body (204, 304, or a reply to HEAD) is
+  # complete as soon as the headers are parsed. Hackney queues `done` for us and,
+  # if the connection is pooled, checks it straight back in, so the ref stops
+  # being ours at that point: pulling the stream or stopping the connection
+  # afterwards would disturb whichever request checks it out next. Mark it so
+  # `stream_body/1` and `close/1` leave it alone.
+  #
+  # This relies on hackney never running its own redirect handling for us --
+  # `request/5` forces `follow_redirect: false` -- otherwise a redirect status
+  # would be checked back in too and would need the same treatment.
+  defp build_client(ref, recv_timeout, method, code, opts) do
+    client = %{ref: ref, recv_timeout: recv_timeout}
+
+    if released?(method, code) and pooled?(opts) do
+      flush_responses(ref)
+      Map.put(client, :released, true)
+    else
+      client
+    end
+  end
+
+  defp released?(method, _code) when method in [:head, :HEAD, "head", "HEAD"], do: true
+  defp released?(_method, code) when code in [204, 304], do: true
+  defp released?(_method, _code), do: false
+
+  # Mirrors hackney's own pooling decision: :hackney.use_pool/1, plus the fact
+  # that a proxy tunnel always bypasses the pool. An unpooled connection is never
+  # checked in, and parks in `connected` with no idle timeout, so it leaks
+  # forever unless we stop it ourselves.
+  defp pooled?(opts) do
+    is_nil(Keyword.get(opts, :proxy)) and use_pool?(Keyword.get(opts, :pool))
+  end
+
+  defp use_pool?(false), do: false
+  defp use_pool?(nil), do: Application.get_env(:hackney, :use_default_pool, true)
+  defp use_pool?(_pool_name), do: true
+
   defp receive_response(ref, recv_timeout) do
     receive do
       {:hackney_response, ^ref, {:error, error}} -> {:error, error}

lib/pleroma/reverse_proxy/client/tesla.ex

@@ -42,13 +42,14 @@ defmodule Pleroma.ReverseProxy.Client.Tesla do
     end
   end
 
+  # Reaching the end of the body does not release the ref: callers must always
+  # call close/1, exactly once, whatever the outcome. Releasing here as well
+  # would remove the client twice and crash the pool worker on the missing
+  # monitor.
   @impl true
   @spec stream_body(map()) ::
           {:ok, binary(), map()} | {:error, atom() | String.t()} | :done | no_return()
-  def stream_body(%{pid: pid, fin: true}) do
-    ConnectionPool.release_conn(pid)
-    :done
-  end
+  def stream_body(%{fin: true}), do: :done
 
   def stream_body(client) do
     case read_chunk!(client) do

test/pleroma/http/hackney_follow_redirect_regression_test.exs

@@ -147,6 +147,87 @@ defmodule Pleroma.HTTP.HackneyFollowRedirectRegressionTest do
              Pleroma.ReverseProxy.Client.Hackney.request(:get, url, [], "", opts)
   end
 
+  # A pooled connection serving a bodyless response is checked back in by hackney
+  # at header time, so stopping that ref afterwards would evict a connection
+  # another request may already hold. These go direct rather than through the
+  # CONNECT proxy, because a proxy tunnel is never pooled.
+  test "hackney reverse proxy marks a pooled bodyless response as already released",
+       %{tls_server: tls_server} do
+    url = "#{tls_server.base_url}/no_content"
+
+    assert {:ok, 204, _headers, client} =
+             Pleroma.ReverseProxy.Client.Hackney.request(
+               :get,
+               url,
+               [],
+               "",
+               direct_opts(tls_server)
+             )
+
+    assert %{released: true} = client
+    assert :done = Pleroma.ReverseProxy.Client.Hackney.stream_body(client)
+    assert :ok = Pleroma.ReverseProxy.Client.Hackney.close(client)
+  end
+
+  test "hackney reverse proxy still owns a response that carries a body",
+       %{tls_server: tls_server} do
+    url = "#{tls_server.base_url}/final"
+
+    assert {:ok, 200, _headers, client} =
+             Pleroma.ReverseProxy.Client.Hackney.request(
+               :get,
+               url,
+               [],
+               "",
+               direct_opts(tls_server)
+             )
+
+    refute Map.has_key?(client, :released)
+    assert "ok" == collect_body(client)
+    Pleroma.ReverseProxy.Client.Hackney.close(client)
+  end
+
+  # An unpooled connection is never checked in and parks forever with no idle
+  # timeout, so it stays ours to stop even when the response has no body.
+  test "hackney reverse proxy still owns a bodyless response over a proxy tunnel", %{
+    tls_server: tls_server,
+    proxy: proxy
+  } do
+    url = "#{tls_server.base_url}/no_content"
+
+    opts = Keyword.put(direct_opts(tls_server), :proxy, proxy.proxy_url)
+
+    assert {:ok, 204, _headers, client} =
+             Pleroma.ReverseProxy.Client.Hackney.request(:get, url, [], "", opts)
+
+    refute Map.has_key?(client, :released)
+    Pleroma.ReverseProxy.Client.Hackney.close(client)
+  end
+
+  test "hackney reverse proxy still owns a bodyless response when pooling is off", %{
+    tls_server: tls_server
+  } do
+    url = "#{tls_server.base_url}/no_content"
+
+    opts = Keyword.put(direct_opts(tls_server), :pool, false)
+
+    assert {:ok, 204, _headers, client} =
+             Pleroma.ReverseProxy.Client.Hackney.request(:get, url, [], "", opts)
+
+    refute Map.has_key?(client, :released)
+    Pleroma.ReverseProxy.Client.Hackney.close(client)
+  end
+
+  defp direct_opts(tls_server) do
+    [
+      pool: :media,
+      ssl_options: tls_client_options(tls_server),
+      connect_timeout: 1_000,
+      recv_timeout: 1_000,
+      follow_redirect: false
+    ]
+  end
+
   defp collect_body(ref, acc \\ "") do
     case Pleroma.ReverseProxy.Client.Hackney.stream_body(ref) do
       :done -> acc
@@ -216,6 +297,9 @@ defmodule Pleroma.HTTP.HackneyFollowRedirectRegressionTest do
         "/closed" ->
           :ok
 
+        "/no_content" ->
+          send_ssl_response(ssl_socket, 204, "No Content", [], "")
+
         "/delayed" ->
           Process.sleep(100)
           send_ssl_response(ssl_socket, 200, "OK", [], "ok")

test/pleroma/reverse_proxy_test.exs

@@ -18,6 +18,21 @@ defmodule Pleroma.ReverseProxyTest do
 
   setup :verify_on_exit!
 
+  # Every path that obtains a client ref releases it, so tests that do not assert
+  # on the release still need close/1 to be callable. Recording each call lets
+  # the tests below pin *exactly* once: a Mox `expect` would silently fall
+  # through to this stub on a second call.
+  setup do
+    test_pid = self()
+
+    stub(ClientMock, :close, fn client ->
+      send(test_pid, {:closed, client})
+      :ok
+    end)
+
+    :ok
+  end
+
   defp request_mock(invokes) do
     ClientMock
     |> expect(:request, fn :get, url, headers, _body, _opts ->
@@ -91,6 +106,81 @@ defmodule Pleroma.ReverseProxyTest do
     end
   end
 
+  # The connection behind a client ref is not tied to the lifetime of this
+  # request. Hackney 4 in particular parks a non-pooled `hackney_conn` process
+  # in the `closed` state forever when nobody stops it, pinning every refc
+  # binary it read, so abandoning a ref leaks the whole response body.
+  describe "client ref release" do
+    test "closes the client after the body is fully streamed", %{conn: conn} do
+      ClientMock
+      |> expect(:request, fn :get, "/release", _, _, _ ->
+        {:ok, 200, [{"content-type", "image/png"}], %{url: "/release"}}
+      end)
+      |> expect(:stream_body, fn _ -> :done end)
+
+      conn = ReverseProxy.call(conn, "/release")
+
+      assert conn.halted
+      assert_received {:closed, %{url: "/release"}}
+      refute_received {:closed, _}
+    end
+
+    test "closes the client when content-length exceeds max_body_length", %{conn: conn} do
+      ClientMock
+      |> expect(:request, fn :get, "/release-huge", _, _, _ ->
+        {:ok, 200, [{"content-length", "102227307"}], %{url: "/release-huge"}}
+      end)
+
+      assert capture_log(fn ->
+               ReverseProxy.call(conn, "/release-huge", max_body_length: 25 * 1_048_576)
+             end) =~ ":body_too_large"
+
+      assert_received {:closed, %{url: "/release-huge"}}
+      refute_received {:closed, _}
+    end
+
+    test "closes the client when the upstream status is not proxiable", %{conn: conn} do
+      ClientMock
+      |> expect(:request, fn :get, "/release-418", _, _, _ ->
+        {:ok, 418, [], %{url: "/release-418"}}
+      end)
+
+      assert capture_log(fn -> ReverseProxy.call(conn, "/release-418") end) =~
+               "failed with HTTP status 418"
+
+      assert_received {:closed, %{url: "/release-418"}}
+      refute_received {:closed, _}
+    end
+
+    # The hackney client answers HEAD with a ref too, so this path must release
+    # it as well; only the ref-less 3-tuple reply has nothing to close.
+    test "closes the client when a HEAD response comes back with a ref", %{conn: conn} do
+      ClientMock
+      |> expect(:request, fn :head, "/release-head", _, _, _ ->
+        {:ok, 200, [{"content-type", "image/png"}], %{url: "/release-head"}}
+      end)
+      |> expect(:stream_body, fn _ -> :done end)
+
+      conn = ReverseProxy.call(Map.put(conn, :method, "HEAD"), "/release-head")
+
+      assert conn.halted
+      assert_received {:closed, %{url: "/release-head"}}
+      refute_received {:closed, _}
+    end
+
+    test "has nothing to close when the reply carries no ref", %{conn: conn} do
+      ClientMock
+      |> expect(:request, fn :head, "/release-headless", _, _, _ ->
+        {:ok, 200, [{"content-type", "image/png"}]}
+      end)
+
+      conn = ReverseProxy.call(Map.put(conn, :method, "HEAD"), "/release-headless")
+
+      assert conn.status == 200
+      refute_received {:closed, _}
+    end
+  end
+
   test "use Pleroma's user agent in the request; don't pass the client's", %{conn: conn} do
     request_mock(2)
 
@@ -116,7 +206,7 @@ defmodule Pleroma.ReverseProxyTest do
     assert conn.halted
   end
 
-  defp stream_mock(invokes, with_close? \\ false) do
+  defp stream_mock(invokes) do
     ClientMock
     |> expect(:request, fn :get, "/stream-bytes/" <> length, _, _, _ ->
       Registry.register(ClientMock, "/stream-bytes/" <> length, 0)
@@ -142,10 +232,6 @@ defmodule Pleroma.ReverseProxyTest do
           :done
       end
     end)
-
-    if with_close? do
-      expect(ClientMock, :close, fn _ -> :ok end)
-    end
   end
 
   describe "max_body" do
@@ -165,12 +251,15 @@ defmodule Pleroma.ReverseProxyTest do
     end
 
     test "max_body_length returns error if streaming body more than that option", %{conn: conn} do
-      stream_mock(3, true)
+      stream_mock(3)
 
       assert capture_log(fn ->
                ReverseProxy.call(conn, "/stream-bytes/50", max_body_length: 30)
              end) =~
                "Elixir.Pleroma.ReverseProxy request to /stream-bytes/50 failed while reading/chunking: :body_too_large"
+
+      assert_received {:closed, %{url: "/stream-bytes/50"}}
+      refute_received {:closed, _}
     end
   end