From 64da7106c5d2f4af2b5068f52d66af7bdc770ab4 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Sat, 11 Jul 2026 00:47:36 -0700 Subject: [PATCH 1/2] fix: stop leaking supervisor children when client connections close --- grpc/lib/grpc/client/application.ex | 7 +- grpc/lib/grpc/client/connection.ex | 13 ++- grpc/test/grpc/client/connection_test.exs | 110 +++++++++++++++++++++- 3 files changed, 123 insertions(+), 7 deletions(-) diff --git a/grpc/lib/grpc/client/application.ex b/grpc/lib/grpc/client/application.ex index b131a8e9..3c3c29a6 100644 --- a/grpc/lib/grpc/client/application.ex +++ b/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] diff --git a/grpc/lib/grpc/client/connection.ex b/grpc/lib/grpc/client/connection.ex index 9a4c586e..7a6d46b6 100644 --- a/grpc/lib/grpc/client/connection.ex +++ b/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) diff --git a/grpc/test/grpc/client/connection_test.exs b/grpc/test/grpc/client/connection_test.exs index 00cc036c..59d510a8 100644 --- a/grpc/test/grpc/client/connection_test.exs +++ b/grpc/test/grpc/client/connection_test.exs @@ -6,6 +6,24 @@ 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(), @@ -30,6 +48,36 @@ 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 + } 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 == 0 + end) end end @@ -79,6 +127,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 == 0 + end) end test "pick_channel returns {:error, :no_connection} after disconnect (persistent_term entry is erased)", @@ -92,7 +148,7 @@ 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 @@ -101,10 +157,18 @@ defmodule GRPC.Client.ConnectionTest do 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 == 0 + end) end end @@ -192,6 +256,33 @@ 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 + } 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(<>, 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: 0} = 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 +343,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) From 5a6f752bdd3d5db50d19f81667c3c160031bd2e9 Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Sun, 12 Jul 2026 20:57:57 -0700 Subject: [PATCH 2/2] test: baseline supervisor child count instead of asserting zero The connection tests asserted DynamicSupervisor active == 0 on the shared GRPC.Client.Supervisor, which fails under the full suite where other tests leave children behind (CI showed active: 44). Capture the pre-existing count in setup and assert the count returns to that baseline. --- grpc/test/grpc/client/connection_test.exs | 24 ++++++++++++++--------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/grpc/test/grpc/client/connection_test.exs b/grpc/test/grpc/client/connection_test.exs index 59d510a8..efad4b81 100644 --- a/grpc/test/grpc/client/connection_test.exs +++ b/grpc/test/grpc/client/connection_test.exs @@ -29,7 +29,8 @@ defmodule GRPC.Client.ConnectionTest 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 @@ -56,7 +57,8 @@ defmodule GRPC.Client.ConnectionTest do describe "connect/2 - failed finalize" do test "terminates a newly started child when no channel can be picked", %{ ref: ref, - adapter: adapter + adapter: adapter, + supervisor_children: supervisor_children } do Application.put_env(:grpc, NoPickResolver, self()) on_exit(fn -> Application.delete_env(:grpc, NoPickResolver) end) @@ -76,7 +78,7 @@ defmodule GRPC.Client.ConnectionTest do end) assert_eventually(fn -> - DynamicSupervisor.count_children(GRPC.Client.Supervisor).active == 0 + DynamicSupervisor.count_children(GRPC.Client.Supervisor).active == supervisor_children end) end end @@ -117,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) @@ -133,7 +136,7 @@ defmodule GRPC.Client.ConnectionTest do end) assert_eventually(fn -> - DynamicSupervisor.count_children(GRPC.Client.Supervisor).active == 0 + DynamicSupervisor.count_children(GRPC.Client.Supervisor).active == supervisor_children end) end @@ -151,7 +154,8 @@ defmodule GRPC.Client.ConnectionTest do 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) @@ -167,7 +171,7 @@ defmodule GRPC.Client.ConnectionTest do end) assert_eventually(fn -> - DynamicSupervisor.count_children(GRPC.Client.Supervisor).active == 0 + DynamicSupervisor.count_children(GRPC.Client.Supervisor).active == supervisor_children end) end end @@ -258,7 +262,8 @@ defmodule GRPC.Client.ConnectionTest do describe "resource leaks over repeated connect/disconnect" do test "50 binary-heavy cycles leave supervisor memory bounded after garbage collection", %{ target: target, - adapter: adapter + adapter: adapter, + supervisor_children: supervisor_children } do supervisor = Process.whereis(GRPC.Client.Supervisor) :erlang.garbage_collect(supervisor) @@ -277,7 +282,8 @@ defmodule GRPC.Client.ConnectionTest do :erlang.garbage_collect(supervisor) {:memory, after_memory} = Process.info(supervisor, :memory) - assert %{active: 0} = DynamicSupervisor.count_children(GRPC.Client.Supervisor) + 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}"