StreamResponseProcess leaks when the connection process refuses a request

[arrangabriel]

A unary RPC call may take this path:

1. do_request/3 spawns a StreamResponseProcess linked to the caller.
2. ConnectionProcess.request/6 replies with an error:
   • the conn is closed, so the call short-circuits with {:error, "the connection is closed"} (connection_process.ex:109), or
   • Mint.HTTP.request/5 itself errors, and that clause only updates the conn before replying {:error, reason}.
3. receive_data/2 routes to handle_errors_receive_data/2, which just returns the error.

The response process is left running as long as the caller stays up.

Reproduction Steps

Mix.install([
  {:grpc, github: "elixir-grpc/grpc", sparse: "grpc"},
  {:mint, "~> 1.9"},
  {:protobuf, "~> 0.17"},
  {:bandit, "~> 1.0"}
])

Logger.configure(level: :error)

defmodule Msg do
  use Protobuf, syntax: :proto3
  field(:text, 1, type: :string)
end

defmodule Echo do
  use GRPC.Service, name: "repro.Echo"
  rpc(:Echo, Msg, Msg)
end

defmodule Echo.Stub do
  use GRPC.Stub, service: Echo
end

defmodule Ok do
  def init(opts), do: opts
  def call(conn, _opts), do: Plug.Conn.send_resp(conn, 200, "")
end

{:ok, server} = Bandit.start_link(plug: Ok, scheme: :http, port: 50_557)

{:ok, channel} =
  GRPC.Stub.connect("ipv4:127.0.0.1:50557",
    adapter: GRPC.Client.Adapters.Mint,
    adapter_opts: [retry: 1_000_000]
  )

Supervisor.stop(server)
Process.sleep(1_000)

links = fn -> self() |> Process.info(:links) |> elem(1) |> length() end

before = links.()
for _ <- 1..20, do: Echo.Stub.echo(channel, struct(Msg))

IO.puts("links before=#{before} after=#{links.()}")

links before=2 after=22

The 20 new links are all {GRPC.Client.Adapters.Mint.StreamResponseProcess, :init, 1}.

Expected Behavior

The response process for a request should be stopped when the request is not accepted by the connection.

Environment Details

• Elixir 1.20.0
• Erlang OTP 28
• grpc (master, 26bd3ec)
• Mint 1.9.3