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

grpc/lib/grpc/client/application.ex

@@ -3,9 +3,16 @@ defmodule GRPC.Client.Application do
   use Application
 
   def start(_type, _args) do
+    supervisor_config = Application.get_env(:grpc, __MODULE__, [])
+
     children = [
       {Registry, [keys: :unique, name: GRPC.Client.Registry]},
-      {DynamicSupervisor, [name: GRPC.Client.Supervisor]}
+      {DynamicSupervisor,
+       [
+         name: GRPC.Client.Supervisor,
+         hibernate_after: Keyword.get(supervisor_config, :hibernate_after, 15_000),
+         spawn_opt: [fullsweep_after: Keyword.get(supervisor_config, :fullsweep_after, 20)]
+       ]}
     ]
 
     opts = [strategy: :one_for_one, name: GRPC.Supervisor]

grpc/lib/grpc/client/connection.ex

@@ -52,6 +52,20 @@ defmodule GRPC.Client.Connection do
   first establishment attempt finishes and returns `{:error, reason}` (tearing
   the process down) if that attempt fails.
 
+  ## Supervisor memory
+
+  Connection start arguments (for example large default headers) can linger in
+  the `GRPC.Client.Supervisor` process heap after children exit. That supervisor
+  therefore hibernates when idle and runs fullsweep GCs more often. Tune via
+  config:
+
+      config :grpc, GRPC.Client.Application,
+        hibernate_after: 15_000,
+        fullsweep_after: 20
+
+  * `:hibernate_after` – idle ms before the supervisor hibernates (default: `15_000`)
+  * `:fullsweep_after` – minor GCs between fullsweeps on the supervisor (default: `20`)
+
   ## Target syntax
 
   The `target` argument to `connect/2` accepts URI-like strings that are resolved
@@ -283,7 +297,16 @@ defmodule GRPC.Client.Connection do
       {:ok, pid} ->
         case ready_status(name, validated.resolver_target, timeout) do
           :ok ->
-            finalize_connection(name)
+            case finalize_connection(name) do
+              {:ok, %Channel{}} = result ->
+                result
+
+              {:error, _reason} = error ->
+                # A connection that reported ready but cannot hand out a
+                # channel is unusable. Keeping it in the supervisor could leak memory.
+                :ok = DynamicSupervisor.terminate_child(GRPC.Client.Supervisor, pid)
+                error
+            end
 
           {:error, reason} ->
             _ = DynamicSupervisor.terminate_child(GRPC.Client.Supervisor, pid)

grpc/test/grpc/client/connection_test.exs

@@ -16,12 +16,31 @@ defmodule GRPC.Client.ConnectionTest do
     end
   end
 
+  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
 
@@ -40,6 +59,33 @@ 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
+      monitor_ref = Process.monitor(pid)
+      assert_receive {:DOWN, ^monitor_ref, :process, ^pid, _reason}, 500
+
+      assert DynamicSupervisor.count_children(GRPC.Client.Supervisor).active ==
+               supervisor_children
     end
   end
 
@@ -92,16 +138,20 @@ 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)
 
       pid = whereis_name(ref)
+      ref_mon = Process.monitor(pid)
 
       {:ok, _} = Connection.disconnect(channel)
 
-      ref_mon = Process.monitor(pid)
       assert_receive {:DOWN, ^ref_mon, :process, ^pid, _reason}, 500
+
+      assert DynamicSupervisor.count_children(GRPC.Client.Supervisor).active ==
+               supervisor_children
     end
 
     test "pick_channel returns {:error, :no_connection} after disconnect (persistent_term entry is erased)",
@@ -114,23 +164,6 @@ defmodule GRPC.Client.ConnectionTest do
     end
   end
 
-  describe "terminate/2 - persistent_term cleanup on process kill" do
-    test "persistent_term entry is erased when process is killed without disconnect", %{
-      ref: ref,
-      target: target,
-      adapter: adapter
-    } 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
-
-      assert {:error, :no_connection} = Connection.pick_channel(channel)
-    end
-  end
-
   describe "LB ETS table lifecycle" do
     test "disconnect/1 exits the GenServer and the ETS table is freed", %{
       ref: ref,
@@ -215,6 +248,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