fix(client): prune a channel when its adapter connection process dies

grpc/lib/grpc/client/connection.ex

@@ -97,6 +97,12 @@ defmodule GRPC.Client.Connection do
       shut down, after its resources were released.
       * Measurements: none
       * Metadata: `:name`, `:target`, `:pid`, `:reason`
+    * `[:grpc, :client, :connection, :channel_pruned]` – a channel's adapter
+      connection process died and the channel was removed from the load
+      balancer, so it is no longer picked.
+      * Measurements: `:remaining` – channels still connected afterwards
+      * Metadata: `:name`, `:target`, `:pid`, `:reason`, `:address` – the
+        address key of the pruned channel
     * `[:grpc, :client, :connection, :await_ready, :start]` – a caller started
       waiting in `await_ready/2`.
       * Measurements: `:system_time`
@@ -163,6 +169,7 @@ defmodule GRPC.Client.Connection do
   @connected_event [:grpc, :client, :connection, :connected]
   @connect_error_event [:grpc, :client, :connection, :connect_error]
   @disconnected_event [:grpc, :client, :connection, :disconnected]
+  @channel_pruned_event [:grpc, :client, :connection, :channel_pruned]
   @await_ready_start_event [:grpc, :client, :connection, :await_ready, :start]
   @await_ready_stop_event [:grpc, :client, :connection, :await_ready, :stop]
 
@@ -573,20 +580,12 @@ defmodule GRPC.Client.Connection do
 
       {:noreply, state}
     else
-      Logger.warning(
-        "#{inspect(__MODULE__)} received :EXIT from #{inspect(pid)} reason: #{inspect(reason)}"
-      )
-
-      {:noreply, state}
+      {:noreply, handle_adapter_exit(pid, reason, state)}
     end
   end
 
   def handle_info({:EXIT, pid, reason}, state) do
-    Logger.warning(
-      "#{inspect(__MODULE__)} received :EXIT from #{inspect(pid)} reason: #{inspect(reason)}"
-    )
-
-    {:noreply, state}
+    {:noreply, handle_adapter_exit(pid, reason, state)}
   end
 
   def handle_info({:DOWN, mon, :process, pid, reason}, state) do
@@ -937,7 +936,7 @@ defmodule GRPC.Client.Connection do
       end
 
     if connected == [] do
-      Logger.warning("No healthy channels available after re-resolution")
+      Logger.warning("No healthy channels available")
     end
 
     %{state | real_channels: real_channels, lb_state: new_lb_state}
@@ -958,6 +957,65 @@ defmodule GRPC.Client.Connection do
   defp channel_alive?({:connected, _}), do: true
   defp channel_alive?(_), do: false
 
+  # An adapter connection process died. It is linked to us, so we are the only
+  # one told: the channel it backs is still `{:connected, ch}` in
+  # `real_channels` and still in the load balancer's list, so `pick_channel/2`
+  # would keep handing out a channel whose `conn_pid` is dead until the next
+  # re-resolution reconciles it. Literal-address targets (`ipv4:`, `ipv6:`,
+  # `unix:`) have no background re-resolution at all, so for those no tick is
+  # ever coming and the dead channel would be served forever.
+  defp handle_adapter_exit(pid, reason, state) do
+    case channel_key_for_conn_pid(state.real_channels, pid) do
+      nil ->
+        Logger.warning(
+          "#{inspect(__MODULE__)} received :EXIT from #{inspect(pid)} reason: #{inspect(reason)}"
+        )
+
+        state
+
+      key ->
+        Logger.warning(
+          "#{inspect(__MODULE__)} adapter connection #{inspect(pid)} exited: " <>
+            "#{inspect(reason)}, pruning its channel from the load balancer"
+        )
+
+        new_state =
+          state.real_channels
+          |> Map.put(key, {:failed, reason})
+          |> rebalance_after_reconcile(state)
+
+        :telemetry.execute(
+          @channel_pruned_event,
+          %{remaining: length(connected_channels(new_state.real_channels))},
+          state
+          |> lifecycle_metadata()
+          |> Map.merge(%{reason: reason, address: key})
+        )
+
+        maybe_reestablish(new_state)
+    end
+  end
+
+  defp channel_key_for_conn_pid(real_channels, pid) do
+    Enum.find_value(real_channels, fn
+      {key, {:connected, %{adapter_payload: %{conn_pid: ^pid}}}} -> key
+      _ -> nil
+    end)
+  end
+
+  # With channels left we simply serve them and let the next reconcile retry
+  # the failed address. With none left there is nothing to pick, so restart the
+  # establish loop from scratch -- `:retry_establish` is a no-op while
+  # `established?` is true, so that has to be cleared for the retry to run.
+  defp maybe_reestablish(state) do
+    if connected_channels(state.real_channels) == [] do
+      Process.send_after(self(), :retry_establish, backoff_delay(0))
+      %{state | established?: false, retry_attempt: 0}
+    else
+      state
+    end
+  end
+
   defp via(ref) do
     {:via, Registry, {GRPC.Client.Registry, {__MODULE__, ref}}}
   end

grpc/test/grpc/client/connection_supervised_test.exs

@@ -145,6 +145,96 @@ defmodule GRPC.Client.ConnectionSupervisedTest do
     end
   end
 
+  describe "adapter connection process exits" do
+    @tag capture_log: true
+    test "stops picking a channel whose adapter process died" do
+      name = unique_name("dead_channel")
+      attach_telemetry([:grpc, :client, :connection, :channel_pruned])
+
+      start_supervised!(
+        {Connection,
+         name: name,
+         target: "ipv4:127.0.0.1:50051,127.0.0.1:50052",
+         adapter: GRPC.Test.ProcessClientAdapter}
+      )
+
+      assert :ok = Connection.await_ready(name, 2_000)
+
+      assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: dead_pid}}} =
+               Connection.pick_channel(%GRPC.Channel{ref: name})
+
+      Process.exit(dead_pid, :kill)
+
+      # The exit signal comes from the dying process, so it is not ordered
+      # against anything this process sends. Sync on the prune itself.
+      assert_receive {:telemetry, [:grpc, :client, :connection, :channel_pruned], %{remaining: 1},
+                      %{name: ^name, reason: :killed}},
+                     1_000
+
+      refute Process.alive?(dead_pid)
+
+      assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: live_pid}}} =
+               Connection.pick_channel(%GRPC.Channel{ref: name})
+
+      assert live_pid != dead_pid
+      assert Process.alive?(live_pid)
+    end
+
+    @tag capture_log: true
+    test "re-establishes when the last remaining channel's adapter process dies" do
+      name = unique_name("last_channel")
+      attach_telemetry([:grpc, :client, :connection, :connected])
+
+      start_supervised!(
+        {Connection,
+         name: name, target: "ipv4:127.0.0.1:50051", adapter: GRPC.Test.ProcessClientAdapter}
+      )
+
+      assert :ok = Connection.await_ready(name, 2_000)
+      assert_receive {:telemetry, [:grpc, :client, :connection, :connected], _, %{name: ^name}}
+
+      assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: dead_pid}}} =
+               Connection.pick_channel(%GRPC.Channel{ref: name})
+
+      Process.exit(dead_pid, :kill)
+
+      # The only channel is gone, so the establish loop has to run again. A
+      # second :connected event can only come from that.
+      assert_receive {:telemetry, [:grpc, :client, :connection, :connected], _, %{name: ^name}},
+                     2_000
+
+      assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: live_pid}}} =
+               Connection.pick_channel(%GRPC.Channel{ref: name})
+
+      assert live_pid != dead_pid
+      assert Process.alive?(live_pid)
+    end
+
+    @tag capture_log: true
+    test "an unrelated linked exit leaves the channels alone" do
+      name = unique_name("unrelated_exit")
+
+      start_supervised!(
+        {Connection,
+         name: name, target: "ipv4:127.0.0.1:50051", adapter: GRPC.Test.ProcessClientAdapter}
+      )
+
+      assert :ok = Connection.await_ready(name, 2_000)
+
+      assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: pid}}} =
+               Connection.pick_channel(%GRPC.Channel{ref: name})
+
+      conn = whereis_connection(name)
+      send(conn, {:EXIT, spawn(fn -> :ok end), :some_crash})
+      :sys.get_state(conn)
+
+      assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: ^pid}}} =
+               Connection.pick_channel(%GRPC.Channel{ref: name})
+
+      assert Process.alive?(pid)
+    end
+  end
+
   describe "await_ready/2 waiter lifecycle" do
     setup do
       %{

grpc/test/support/test_adapter.exs

@@ -11,6 +11,43 @@ defmodule GRPC.Test.ClientAdapter do
   def cancel(stream), do: stream
 end
 
+defmodule GRPC.Test.ProcessClientAdapter do
+  @moduledoc """
+  A test adapter that backs each channel with a real linked process recorded
+  as `conn_pid`, the way the Gun and Mint adapters do.
+
+  `GRPC.Test.ClientAdapter` returns the channel untouched, so its channels have
+  no `conn_pid` and there is no process to kill. This adapter exists so tests
+  can kill the process behind a channel and assert on what the connection does
+  about it.
+  """
+  @behaviour GRPC.Client.Adapter
+
+  def connect(channel, _opts) do
+    {:ok, %{channel | adapter_payload: %{conn_pid: spawn_link(&idle/0)}}}
+  end
+
+  def disconnect(%{adapter_payload: %{conn_pid: pid}} = channel) when is_pid(pid) do
+    send(pid, :stop)
+    {:ok, %{channel | adapter_payload: %{conn_pid: nil}}}
+  end
+
+  def disconnect(channel), do: {:ok, channel}
+
+  defp idle do
+    receive do
+      :stop -> :ok
+    end
+  end
+
+  def send_request(stream, _message, _opts), do: stream
+  def receive_data(_stream, _opts), do: {:ok, nil}
+  def send_data(stream, _message, _opts), do: stream
+  def send_headers(stream, _opts), do: stream
+  def end_stream(stream), do: stream
+  def cancel(stream), do: stream
+end
+
 defmodule GRPC.Test.FailingClientAdapter do
   @moduledoc """
   A test adapter that refuses to connect to selected hosts. All other hosts