fix: stop leaking client supervisor children on failed and closed connections

grpc/lib/grpc/client/application.ex

@@ -5,7 +5,12 @@ defmodule GRPC.Client.Application do
   def start(_type, _args) do
     children = [
       {Registry, [keys: :unique, name: GRPC.Client.Registry]},
-      {DynamicSupervisor, [name: GRPC.Client.Supervisor]}
+      {DynamicSupervisor,
+       [
+         name: GRPC.Client.Supervisor,
+         hibernate_after: 15_000,
+         spawn_opt: [fullsweep_after: 20]
+       ]}
     ]
 
     opts = [strategy: :one_for_one, name: GRPC.Supervisor]

grpc/lib/grpc/client/connection.ex

@@ -110,7 +110,7 @@ defmodule GRPC.Client.Connection do
       start:
         {GenServer, :start_link,
          [__MODULE__, initial_state, [name: via(initial_state.virtual_channel.ref)]]},
-      restart: :transient,
+      restart: :temporary,
       type: :worker,
       shutdown: 5000
     }
@@ -185,8 +185,15 @@ defmodule GRPC.Client.Connection do
         ch = initial_state.virtual_channel
 
         case DynamicSupervisor.start_child(GRPC.Client.Supervisor, child_spec(initial_state)) do
-          {:ok, _pid} ->
-            finalize_connection(ch, opts)
+          {:ok, pid} ->
+            case finalize_connection(ch, opts) do
+              {:ok, %Channel{}} = result ->
+                result
+
+              {:error, _reason} = error ->
+                DynamicSupervisor.terminate_child(GRPC.Client.Supervisor, pid)
+                error
+            end
 
           {:error, {:already_started, _pid}} ->
             finalize_connection(ch, opts)

grpc/test/grpc/client/connection_test.exs

@@ -6,12 +6,31 @@ defmodule GRPC.Client.ConnectionTest do
 
   @peer if Code.ensure_loaded?(:peer), do: :peer, else: GRPC.Test.PeerShim
 
+  defmodule NoPickResolver do
+    def resolve(_target) do
+      {:ok,
+       %{
+         addresses: [%{address: "127.0.0.1", port: 50051}],
+         service_config: %{}
+       }}
+    end
+
+    def init(_target, _opts) do
+      test_pid = Application.fetch_env!(:grpc, __MODULE__)
+      lb_table = Enum.find(:ets.all(), &(:ets.info(&1, :owner) == self()))
+      true = :ets.delete(lb_table)
+      send(test_pid, {:connection_started, self()})
+      {:ok, nil}
+    end
+  end
+
   setup do
     %{
       ref: make_ref(),
       ip: "127.0.0.1",
       target: "ipv4:127.0.0.1:50051",
-      adapter: GRPC.Test.ClientAdapter
+      adapter: GRPC.Test.ClientAdapter,
+      supervisor_children: DynamicSupervisor.count_children(GRPC.Client.Supervisor).active
     }
   end
 
@@ -30,6 +49,37 @@ defmodule GRPC.Client.ConnectionTest do
       {:ok, channel} = Connection.connect(target, adapter: adapter, name: ref)
 
       assert {:ok, ^channel} = Connection.pick_channel(%Channel{ref: ref})
+
+      Connection.disconnect(channel)
+    end
+  end
+
+  describe "connect/2 - failed finalize" do
+    test "terminates a newly started child when no channel can be picked", %{
+      ref: ref,
+      adapter: adapter,
+      supervisor_children: supervisor_children
+    } do
+      Application.put_env(:grpc, NoPickResolver, self())
+      on_exit(fn -> Application.delete_env(:grpc, NoPickResolver) end)
+
+      assert {:error, :no_connection} =
+               Connection.connect("test://connection",
+                 adapter: adapter,
+                 name: ref,
+                 resolver: NoPickResolver
+               )
+
+      assert_receive {:connection_started, pid}, 500
+      assert_eventually(fn -> not Process.alive?(pid) end)
+
+      assert_eventually(fn ->
+        Registry.lookup(GRPC.Client.Registry, {Connection, ref}) == []
+      end)
+
+      assert_eventually(fn ->
+        DynamicSupervisor.count_children(GRPC.Client.Supervisor).active == supervisor_children
+      end)
     end
   end
 
@@ -69,7 +119,8 @@ defmodule GRPC.Client.ConnectionTest do
     test "GenServer process is no longer alive after disconnect", %{
       ref: ref,
       target: target,
-      adapter: adapter
+      adapter: adapter,
+      supervisor_children: supervisor_children
     } do
       {:ok, channel} = Connection.connect(target, adapter: adapter, name: ref)
 
@@ -79,6 +130,14 @@ defmodule GRPC.Client.ConnectionTest do
 
       ref_mon = Process.monitor(pid)
       assert_receive {:DOWN, ^ref_mon, :process, ^pid, _reason}, 500
+
+      assert_eventually(fn ->
+        Registry.lookup(GRPC.Client.Registry, {Connection, ref}) == []
+      end)
+
+      assert_eventually(fn ->
+        DynamicSupervisor.count_children(GRPC.Client.Supervisor).active == supervisor_children
+      end)
     end
 
     test "pick_channel returns {:error, :no_connection} after disconnect (persistent_term entry is erased)",
@@ -92,19 +151,28 @@ defmodule GRPC.Client.ConnectionTest do
   end
 
   describe "terminate/2 - persistent_term cleanup on process kill" do
-    test "persistent_term entry is erased when process is killed without disconnect", %{
+    test "an abnormally stopped process is cleaned up without being restarted", %{
       ref: ref,
       target: target,
-      adapter: adapter
+      adapter: adapter,
+      supervisor_children: supervisor_children
     } do
       {:ok, channel} = Connection.connect(target, adapter: adapter, name: ref)
 
       pid = whereis_name(ref)
       ref_mon = Process.monitor(pid)
-      GenServer.stop(pid, :shutdown)
-      assert_receive {:DOWN, ^ref_mon, :process, ^pid, :shutdown}, 500
+      GenServer.stop(pid, :connection_crashed)
+      assert_receive {:DOWN, ^ref_mon, :process, ^pid, :connection_crashed}, 500
 
       assert {:error, :no_connection} = Connection.pick_channel(channel)
+
+      assert_eventually(fn ->
+        Registry.lookup(GRPC.Client.Registry, {Connection, ref}) == []
+      end)
+
+      assert_eventually(fn ->
+        DynamicSupervisor.count_children(GRPC.Client.Supervisor).active == supervisor_children
+      end)
     end
   end
 
@@ -192,6 +260,35 @@ defmodule GRPC.Client.ConnectionTest do
   end
 
   describe "resource leaks over repeated connect/disconnect" do
+    test "50 binary-heavy cycles leave supervisor memory bounded after garbage collection", %{
+      target: target,
+      adapter: adapter,
+      supervisor_children: supervisor_children
+    } do
+      supervisor = Process.whereis(GRPC.Client.Supervisor)
+      :erlang.garbage_collect(supervisor)
+      {:memory, before_memory} = Process.info(supervisor, :memory)
+
+      for cycle <- 1..50 do
+        ref = make_ref()
+        header = {"x-test-data", :binary.copy(<<cycle>>, 100_000)}
+
+        {:ok, channel} =
+          Connection.connect(target, adapter: adapter, name: ref, headers: [header])
+
+        {:ok, _} = Connection.disconnect(channel)
+      end
+
+      :erlang.garbage_collect(supervisor)
+      {:memory, after_memory} = Process.info(supervisor, :memory)
+
+      assert %{active: ^supervisor_children} =
+               DynamicSupervisor.count_children(GRPC.Client.Supervisor)
+
+      assert after_memory <= before_memory + 100_000,
+             "supervisor memory grew: before=#{before_memory} after=#{after_memory}"
+    end
+
     test "500 cycles leave persistent_term clean and no per-LB tables leak", %{
       target: target,
       adapter: adapter
@@ -252,6 +349,19 @@ defmodule GRPC.Client.ConnectionTest do
     Enum.count(:persistent_term.get(), &match?({{Connection, _}, _}, &1))
   end
 
+  defp assert_eventually(fun, attempts \\ 50)
+
+  defp assert_eventually(fun, attempts) when attempts > 0 do
+    if fun.() do
+      :ok
+    else
+      Process.sleep(10)
+      assert_eventually(fun, attempts - 1)
+    end
+  end
+
+  defp assert_eventually(fun, 0), do: assert(fun.())
+
   defp lb_tid(ref) do
     pid = whereis_name(ref)
     %{lb_state: %{tid: tid}} = :sys.get_state(pid)