From 1b0564c7744d4d1d46f7d93da46bed0bd8a8ec32 Mon Sep 17 00:00:00 2001
From: Erik Nilsen <enilsen16@live.com>
Date: Thu, 30 Jul 2026 08:05:38 -0700
Subject: [PATCH 1/4] fix: recover supervised connections when the transport
 process dies

When an underlying transport process died, GRPC.Client.Connection kept
the dead channel in the load-balancing rotation. Every subsequent RPC
picked the stale channel, fell back to the payload-less virtual handle,
and crashed with a FunctionClauseError in the Gun adapter - permanently,
until the connection process was restarted. Worse, the orchestrator had
no death signal at all for Gun transports: they live under the adapter's
DynamicSupervisor (restart: :temporary), not linked to the orchestrator,
so the log-only :EXIT clauses never fired either.

Connection changes:

- Monitor every successfully connected transport (connect_real_channel).
  A :DOWN (or :EXIT, for adapters that link like Mint) from a tracked
  conn_pid marks its channel {:failed, reason} and rebalances so pickers
  stop seeing it. Deliberate disconnects remove the channel from state
  before the signal arrives, so they never match; their trailing
  :DOWN :normal is ignored quietly.
- When the last healthy channel is gone, flip established? and schedule
  :retry_establish instead of dialing inside the signal handler. Deaths
  within 10s of establishing count as flaps and back off exponentially;
  stable connections redial immediately.
- :retry_establish adopts channels a background resolver update already
  reconnected instead of dialing a duplicate set that would orphan them.
- When other channels remain, request an early re-resolution and run a
  self-scheduling repair loop that redials {:failed, _} entries from the
  last resolved address set, so static multi-address targets recover the
  dead endpoint too.
- Re-establishment reuses the ETS-backed LB state via lb_mod.update/2;
  a policy flip disposes the old balancer via a new optional terminate/1
  callback on GRPC.Client.LoadBalancing.

Stub changes:

- Re-pick (bounded) when the picked channel's conn_pid is dead, so
  rotating policies advance past the dead entry during the rebalance
  window.
- With no healthy channel resolvable for a named connection's virtual
  handle, fail with UNAVAILABLE instead of handing the adapter an
  unusable channel: error tuple for unary/server-streaming, raised
  GRPC.RPCError for request-streaming calls (their return value is a
  stream). The failure still flows through the interceptor chain and
  client_span telemetry. GRPC.Stub.connect/2 channels keep the existing
  fallback behavior.
---
 grpc/lib/grpc/client/connection.ex            | 277 +++++++++++++++---
 grpc/lib/grpc/client/load_balacing.ex         |   9 +
 .../grpc/client/load_balacing/pick_first.ex   |   8 +
 .../grpc/client/load_balacing/round_robin.ex  |   8 +
 grpc/lib/grpc/stub.ex                         | 142 ++++++---
 .../client/connection_supervised_test.exs     | 260 ++++++++++++++++
 6 files changed, 621 insertions(+), 83 deletions(-)

diff --git a/grpc/lib/grpc/client/connection.ex b/grpc/lib/grpc/client/connection.ex
index 0f4dd7b1..88985d16 100644
--- a/grpc/lib/grpc/client/connection.ex
+++ b/grpc/lib/grpc/client/connection.ex
@@ -46,7 +46,14 @@ defmodule GRPC.Client.Connection do
 
   The channel handle returned by `get_channel/1` is valid as soon as the
   process is running, even while the connection is still being established;
-  RPCs return `{:error, :no_connection}` style errors until then.
+  RPCs fail with an `UNAVAILABLE` `GRPC.RPCError` until then.
+
+  If an established underlying connection later goes down, its channel is
+  removed from the load-balancing rotation as soon as the connection process
+  exits. When it was the last healthy channel, the process re-enters the
+  establishment loop and retries with the same exponential backoff as at
+  boot; RPCs fail with `UNAVAILABLE` in the meantime instead of being routed
+  to a dead connection.
 
   `connect/2` keeps its historical fail-fast contract: it blocks until the
   first establishment attempt finishes and returns `{:error, reason}` (tearing
@@ -177,8 +184,13 @@ defmodule GRPC.Client.Connection do
           connect_opts: keyword(),
           resolver_state: term() | nil,
           established?: boolean(),
+          established_at: integer() | nil,
           last_error: term() | nil,
           retry_attempt: non_neg_integer(),
+          flaps: non_neg_integer(),
+          desired_addresses: [%{address: term(), port: :inet.port_number()}],
+          repair_attempt: non_neg_integer(),
+          repair_scheduled?: boolean(),
           waiters: [{pid(), GenServer.from(), reference(), integer()}]
         }
 
@@ -192,8 +204,13 @@ defmodule GRPC.Client.Connection do
             connect_opts: [],
             resolver_state: nil,
             established?: false,
+            established_at: nil,
             last_error: nil,
             retry_attempt: 0,
+            flaps: 0,
+            desired_addresses: [],
+            repair_attempt: 0,
+            repair_scheduled?: false,
             waiters: []
 
   @doc """
@@ -541,20 +558,93 @@ defmodule GRPC.Client.Connection do
     {:noreply, state}
   end
 
-  def handle_info(:retry_establish, state), do: attempt_establish(state)
+  def handle_info(:retry_establish, state) do
+    # A background resolver update may have reconnected channels while this
+    # retry was pending; adopt them instead of dialing a duplicate set that
+    # would orphan the live ones.
+    case connected_channels(state.real_channels) do
+      [] -> attempt_establish(state)
+      _connected -> {:noreply, adopt_established(state)}
+    end
+  end
 
   def handle_info({:resolver_update, result}, state) do
     state = handle_resolve_result(result, state)
     {:noreply, state}
   end
 
-  def handle_info({:EXIT, _pid, :normal}, state), do: {:noreply, state}
+  def handle_info(:repair_channels, state) do
+    state = %{state | repair_scheduled?: false}
 
-  def handle_info({:EXIT, pid, reason}, %{resolver: resolver, resolver_state: rs} = state)
-      when not is_nil(rs) do
-    # Adapter connection processes are linked too, so re-init must be gated
-    # on the resolver worker's own pid: re-initializing on any linked exit
-    # would spawn a duplicate worker and orphan the live one.
+    if state.established? and any_failed?(state.real_channels) do
+      state =
+        reconcile_channels(state.desired_addresses, state.adapter, state.connect_opts, state)
+
+      if any_failed?(state.real_channels) do
+        {:noreply, schedule_repair(%{state | repair_attempt: state.repair_attempt + 1})}
+      else
+        {:noreply, %{state | repair_attempt: 0}}
+      end
+    else
+      # Either everything healed or a full re-establish owns recovery now.
+      {:noreply, %{state | repair_attempt: 0}}
+    end
+  end
+
+  def handle_info({:EXIT, pid, reason}, state) do
+    # Transport death is primarily detected via the monitor set in
+    # connect_real_channel/5, but adapters that link their transport to this
+    # process (e.g. Mint's start_link) deliver an :EXIT too — route it through
+    # the same death handling. A tracked channel whose process exited — for
+    # any reason, including :normal — is unusable and must leave the rotation;
+    # deliberate disconnects remove the channel from state before the signal
+    # is processed, so they never match.
+    case down_channel_key(state.real_channels, pid) do
+      nil -> handle_unrelated_exit(pid, reason, state)
+      key -> handle_channel_down(key, reason, state)
+    end
+  end
+
+  def handle_info({:DOWN, mon, :process, pid, reason}, state) do
+    case Enum.split_with(state.waiters, fn {_pid, _from, m, _started_at} -> m == mon end) do
+      {[], _} ->
+        case down_channel_key(state.real_channels, pid) do
+          nil ->
+            # Deliberately disconnected transports still deliver a monitor
+            # :DOWN after their channel left the state; stay quiet for those.
+            unless reason == :normal do
+              Logger.warning(
+                "#{inspect(__MODULE__)} received :DOWN from #{inspect(pid)} with reason: #{inspect(reason)}"
+              )
+            end
+
+            {:noreply, state}
+
+          key ->
+            handle_channel_down(key, reason, state)
+        end
+
+      {dropped, remaining} ->
+        Enum.each(dropped, fn {caller_pid, _from, _mon, started_at} ->
+          emit_await_ready_stop(state, caller_pid, started_at, :abandoned)
+        end)
+
+        {:noreply, %{state | waiters: remaining}}
+    end
+  end
+
+  def handle_info(msg, state) do
+    Logger.warning("#{inspect(__MODULE__)} received unexpected message: #{inspect(msg)}")
+
+    {:noreply, state}
+  end
+
+  defp handle_unrelated_exit(_pid, :normal, state), do: {:noreply, state}
+
+  defp handle_unrelated_exit(pid, reason, %{resolver: resolver, resolver_state: rs} = state)
+       when not is_nil(rs) do
+    # Re-init must be gated on the resolver worker's own pid: re-initializing
+    # on any linked exit would spawn a duplicate worker and orphan the live one.
     if pid == resolver_worker_pid(rs) do
       Logger.warning("Resolver worker exited: #{inspect(reason)}, re-initializing")
 
@@ -581,7 +671,7 @@ defmodule GRPC.Client.Connection do
     end
   end
 
-  def handle_info({:EXIT, pid, reason}, state) do
+  defp handle_unrelated_exit(pid, reason, state) do
     Logger.warning(
       "#{inspect(__MODULE__)} received :EXIT from #{inspect(pid)} reason: #{inspect(reason)}"
     )
@@ -589,28 +679,86 @@ defmodule GRPC.Client.Connection do
     {:noreply, state}
   end
 
-  def handle_info({:DOWN, mon, :process, pid, reason}, state) do
-    case Enum.split_with(state.waiters, fn {_pid, _from, m, _started_at} -> m == mon end) do
-      {[], _} ->
-        Logger.warning(
-          "#{inspect(__MODULE__)} received :DOWN from #{inspect(pid)} with reason: #{inspect(reason)}"
-        )
+  # A connection that dies within this window of establishing counts as a
+  # flap, and each consecutive flap backs the redial off further.
+  @flap_window 10_000
 
-        {:noreply, state}
+  defp handle_channel_down(key, reason, state) do
+    Logger.warning(
+      "gRPC connection #{key} for #{state.resolver_target} went down: #{inspect(reason)}"
+    )
 
-      {dropped, remaining} ->
-        Enum.each(dropped, fn {caller_pid, _from, _mon, started_at} ->
-          emit_await_ready_stop(state, caller_pid, started_at, :abandoned)
-        end)
+    real_channels = Map.put(state.real_channels, key, {:failed, reason})
+    state = rebalance_after_reconcile(real_channels, state)
 
-        {:noreply, %{state | waiters: remaining}}
+    if connected_channels(real_channels) == [] do
+      # Last healthy channel gone: flip back to establishing and schedule the
+      # boot retry loop instead of dialing here, so this handler stays
+      # non-blocking. A stable connection redials immediately; a flapping one
+      # backs off.
+      flaps = if uptime_ms(state) < @flap_window, do: state.flaps + 1, else: 0
+      delay = if flaps == 0, do: 0, else: backoff_delay(flaps)
+      Process.send_after(self(), :retry_establish, delay)
+
+      {:noreply, %{state | established?: false, last_error: reason, flaps: flaps}}
+    else
+      # Other channels keep serving; ask the resolver for an early
+      # re-resolution and start the repair loop so the dead endpoint is
+      # redialed even without a background resolver.
+      {:noreply, state |> request_reresolve() |> schedule_repair()}
     end
   end
 
-  def handle_info(msg, state) do
-    Logger.warning("#{inspect(__MODULE__)} received unexpected message: #{inspect(msg)}")
+  defp down_channel_key(real_channels, pid) do
+    Enum.find_value(real_channels, fn
+      {key, {:connected, %{adapter_payload: %{conn_pid: ^pid}}}} -> key
+      _ -> nil
+    end)
+  end
 
-    {:noreply, state}
+  defp uptime_ms(%__MODULE__{established_at: nil}), do: 0
+
+  defp uptime_ms(%__MODULE__{established_at: at}),
+    do: System.monotonic_time(:millisecond) - at
+
+  defp request_reresolve(%__MODULE__{resolver: resolver, resolver_state: rs} = state)
+       when not is_nil(rs) do
+    case resolver.update(rs, :resolve_now) do
+      {:ok, new_rs} -> %{state | resolver_state: new_rs}
+      _ -> state
+    end
+  end
+
+  defp request_reresolve(state), do: state
+
+  defp schedule_repair(%__MODULE__{repair_scheduled?: true} = state), do: state
+
+  defp schedule_repair(state) do
+    Process.send_after(self(), :repair_channels, backoff_delay(state.repair_attempt))
+    %{state | repair_scheduled?: true}
+  end
+
+  defp any_failed?(real_channels) do
+    Enum.any?(real_channels, &match?({_key, {:failed, _}}, &1))
+  end
+
+  defp adopt_established(state) do
+    :telemetry.execute(
+      @connected_event,
+      %{retry_attempt: state.retry_attempt},
+      lifecycle_metadata(state)
+    )
+
+    reply_waiters(state, state.waiters, :ok, :ok)
+
+    %{
+      state
+      | established?: true,
+        established_at: System.monotonic_time(:millisecond),
+        waiters: [],
+        retry_attempt: 0,
+        last_error: nil
+    }
   end
 
   @impl GenServer
@@ -673,22 +821,7 @@ defmodule GRPC.Client.Connection do
   defp attempt_establish(state) do
     case establish(state) do
       {:ok, established_state} ->
-        :telemetry.execute(
-          @connected_event,
-          %{retry_attempt: state.retry_attempt},
-          lifecycle_metadata(state)
-        )
-
-        reply_waiters(established_state, established_state.waiters, :ok, :ok)
-
-        {:noreply,
-         %{
-           established_state
-           | established?: true,
-             waiters: [],
-             retry_attempt: 0,
-             last_error: nil
-         }}
+        {:noreply, adopt_established(established_state)}
 
       {:error, reason} ->
         delay = backoff_delay(state.retry_attempt)
@@ -730,7 +863,7 @@ defmodule GRPC.Client.Connection do
 
     real_channels = build_real_channels(addresses, state.virtual_channel, norm_opts, adapter)
 
-    case init_lb(lb_mod, real_channels, adapter) do
+    case init_or_update_lb(lb_mod, real_channels, adapter, state) do
       {:ok, lb_state} ->
         resolver_state = maybe_init_resolver(state)
         :persistent_term.put(lb_key(state.virtual_channel.ref), {lb_mod, lb_state})
@@ -739,6 +872,7 @@ defmodule GRPC.Client.Connection do
          %__MODULE__{
            state
            | real_channels: real_channels,
+             desired_addresses: addresses,
              lb_mod: lb_mod,
              lb_state: lb_state,
              resolver_state: resolver_state
@@ -749,6 +883,51 @@ defmodule GRPC.Client.Connection do
     end
   end
 
+  # Re-establishment after a connection loss reuses the existing LB state
+  # (its ETS table is what pickers read through :persistent_term), both so
+  # in-flight pickers see the swap immediately and so repeated reconnects
+  # don't leak one table per cycle.
+  defp init_or_update_lb(
+         lb_mod,
+         real_channels,
+         adapter,
+         %__MODULE__{lb_mod: lb_mod, lb_state: lb_state}
+       )
+       when not is_nil(lb_state) do
+    case connected_channels(real_channels) do
+      [] ->
+        disconnect_real_channels(real_channels, adapter)
+        {:error, first_failure(real_channels) || :no_addresses}
+
+      connected ->
+        case lb_mod.update(lb_state, connected) do
+          {:ok, new_lb_state} ->
+            {:ok, new_lb_state}
+
+          {:error, reason} ->
+            disconnect_real_channels(real_channels, adapter)
+            {:error, reason}
+        end
+    end
+  end
+
+  defp init_or_update_lb(lb_mod, real_channels, adapter, state) do
+    # A policy flip discards the previous LB; drop its ETS-backed state so
+    # repeated re-establishments can't leak one table per flip.
+    maybe_terminate_lb(state.lb_mod, state.lb_state)
+    init_lb(lb_mod, real_channels, adapter)
+  end
+
+  defp maybe_terminate_lb(lb_mod, lb_state)
+       when is_nil(lb_mod)
+       when is_nil(lb_state),
+       do: :ok
+
+  defp maybe_terminate_lb(lb_mod, lb_state) do
+    if function_exported?(lb_mod, :terminate, 1), do: lb_mod.terminate(lb_state)
+    :ok
+  end
+
   defp init_lb(lb_mod, real_channels, adapter) do
     case connected_channels(real_channels) do
       [] ->
@@ -866,6 +1045,7 @@ defmodule GRPC.Client.Connection do
   defp handle_resolve_result({:ok, %{addresses: []}}, state), do: state
 
   defp handle_resolve_result({:ok, %{addresses: new_addresses}}, state) do
+    state = %{state | desired_addresses: new_addresses}
     reconcile_channels(new_addresses, state.adapter, state.connect_opts, state)
   end
 
@@ -937,7 +1117,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 for #{state.resolver_target}")
     end
 
     %{state | real_channels: real_channels, lb_state: new_lb_state}
@@ -1129,8 +1309,17 @@ defmodule GRPC.Client.Connection do
   defp choose_lb(_), do: GRPC.Client.LoadBalancing.PickFirst
 
   defp connect_real_channel(%Channel{} = vc, host, port, opts, adapter) do
-    %Channel{vc | host: host, port: port}
-    |> adapter.connect(opts[:adapter_opts])
+    result =
+      %Channel{vc | host: host, port: port}
+      |> adapter.connect(opts[:adapter_opts])
+
+    with {:ok, %Channel{adapter_payload: %{conn_pid: pid}}} when is_pid(pid) <- result do
+      # The transport may be linked to this process (Mint) or owned by the
+      # adapter's supervisor (Gun); a monitor is the only death signal that
+      # covers both.
+      Process.monitor(pid)
+      result
+    end
   end
 
   defp init_interceptors(interceptors) do
diff --git a/grpc/lib/grpc/client/load_balacing.ex b/grpc/lib/grpc/client/load_balacing.ex
index 7dcf363d..8a3ad8d6 100644
--- a/grpc/lib/grpc/client/load_balacing.ex
+++ b/grpc/lib/grpc/client/load_balacing.ex
@@ -8,4 +8,13 @@ defmodule GRPC.Client.LoadBalancing do
 
   @callback update(state :: any(), new_channels :: [struct()]) ::
               {:ok, new_state :: any()} | {:error, reason :: any()}
+
+  @doc """
+  Releases any resources held by the balancer state (e.g. ETS tables).
+
+  Called when a connection replaces its balancer with a different policy.
+  """
+  @callback terminate(state :: any()) :: :ok
+
+  @optional_callbacks terminate: 1
 end
diff --git a/grpc/lib/grpc/client/load_balacing/pick_first.ex b/grpc/lib/grpc/client/load_balacing/pick_first.ex
index bfc04aae..d16cd972 100644
--- a/grpc/lib/grpc/client/load_balacing/pick_first.ex
+++ b/grpc/lib/grpc/client/load_balacing/pick_first.ex
@@ -39,4 +39,12 @@ defmodule GRPC.Client.LoadBalancing.PickFirst do
     :ets.insert(tid, {@current_key, nil})
     {:ok, state}
   end
+
+  @impl true
+  def terminate(%{tid: tid}) do
+    :ets.delete(tid)
+    :ok
+  rescue
+    ArgumentError -> :ok
+  end
 end
diff --git a/grpc/lib/grpc/client/load_balacing/round_robin.ex b/grpc/lib/grpc/client/load_balacing/round_robin.ex
index a11a1e6a..748230b4 100644
--- a/grpc/lib/grpc/client/load_balacing/round_robin.ex
+++ b/grpc/lib/grpc/client/load_balacing/round_robin.ex
@@ -42,4 +42,12 @@ defmodule GRPC.Client.LoadBalancing.RoundRobin do
     :atomics.put(aref, 1, 0)
     {:ok, state}
   end
+
+  @impl true
+  def terminate(%{tid: tid}) do
+    :ets.delete(tid)
+    :ok
+  rescue
+    ArgumentError -> :ok
+  end
 end
diff --git a/grpc/lib/grpc/stub.ex b/grpc/lib/grpc/stub.ex
index f1aeddb3..f6c7cdc1 100644
--- a/grpc/lib/grpc/stub.ex
+++ b/grpc/lib/grpc/stub.ex
@@ -289,6 +289,13 @@ defmodule GRPC.Stub do
   #    * Client streaming. A `GRPC.Client.Stream`
   #    * Server streaming. `{:ok, Enumerable.t} | {:ok, Enumerable.t, trailers_map} | {:error, error}`
   #
+  #  Any call made through a named connection's virtual channel fails with
+  #  UNAVAILABLE while the connection has no healthy underlying channel to
+  #  resolve to: `{:error, %GRPC.RPCError{status: 14}}` for unary and
+  #  server-streaming calls, raised as `GRPC.RPCError` for request-streaming
+  #  calls (their return value is a stream). Both flow through the channel's
+  #  interceptors and client telemetry.
+  #
   #  Options
   #
   #    * `:timeout` - request timeout. Default is 10s for unary calls and `:infinity` for
@@ -301,59 +308,116 @@ defmodule GRPC.Stub do
   def call(_service_mod, rpc, %{channel: channel} = stream, request, opts) do
     {_, {req_mod, req_stream}, {res_mod, response_stream}, _rpc_options} = rpc
 
-    ch =
-      case Connection.pick_channel(channel, opts) do
-        {:ok, %Channel{adapter_payload: adapter_payload} = ch} when is_map(adapter_payload) ->
-          conn_pid = Map.get(adapter_payload, :conn_pid)
+    case resolve_channel(channel, opts) do
+      {:error, %GRPC.RPCError{} = error} ->
+        unavailable_result(error, stream, request, req_mod, res_mod, req_stream)
+
+      {:ok, ch} ->
+        stream = %{stream | channel: ch, request_mod: req_mod, response_mod: res_mod}
 
-          if is_pid(conn_pid) and Process.alive?(conn_pid) do
-            ch
+        opts =
+          if req_stream || response_stream do
+            parse_req_opts([{:timeout, :infinity} | opts])
           else
-            Logger.warning(
-              "The connection process #{inspect(conn_pid)} is not alive, " <>
-                "please create a new channel via GRPC.Stub.connect/2"
-            )
+            parse_req_opts([{:timeout, @default_timeout} | opts])
+          end
+
+        compressor = Keyword.get(opts, :compressor, ch.compressor)
+        accepted_compressors = Keyword.get(opts, :accepted_compressors, ch.accepted_compressors)
 
-            channel
+        if not is_list(accepted_compressors) do
+          raise ArgumentError, "accepted_compressors is not a list"
+        end
+
+        accepted_compressors =
+          if compressor do
+            Enum.uniq([compressor | accepted_compressors])
+          else
+            accepted_compressors
           end
 
-        _ ->
-          # fallback to the channel in the stream
-          channel
-      end
+        stream = %{
+          stream
+          | codec: Keyword.get(opts, :codec, ch.codec),
+            compressor: compressor,
+            accepted_compressors: accepted_compressors
+        }
 
-    stream = %{stream | channel: ch, request_mod: req_mod, response_mod: res_mod}
+        GRPC.Telemetry.client_span(stream, request, fn ->
+          do_call(req_stream, stream, request, opts)
+        end)
+    end
+  end
 
-    opts =
-      if req_stream || response_stream do
-        parse_req_opts([{:timeout, :infinity} | opts])
-      else
-        parse_req_opts([{:timeout, @default_timeout} | opts])
-      end
+  # Bounded re-picks let rotating policies advance past a dead entry during
+  # the window before the connection process rebalances it away.
+  @resolve_attempts 3
+
+  defp resolve_channel(channel, opts), do: resolve_channel(channel, opts, @resolve_attempts)
+
+  defp resolve_channel(channel, _opts, 0), do: fallback_channel(channel)
 
-    compressor = Keyword.get(opts, :compressor, ch.compressor)
-    accepted_compressors = Keyword.get(opts, :accepted_compressors, ch.accepted_compressors)
+  defp resolve_channel(channel, opts, attempts) do
+    case Connection.pick_channel(channel, opts) do
+      {:ok, %Channel{adapter_payload: adapter_payload} = ch} when is_map(adapter_payload) ->
+        conn_pid = Map.get(adapter_payload, :conn_pid)
 
-    if not is_list(accepted_compressors) do
-      raise ArgumentError, "accepted_compressors is not a list"
+        if is_pid(conn_pid) and Process.alive?(conn_pid) do
+          {:ok, ch}
+        else
+          Logger.warning(
+            "The connection process #{inspect(conn_pid)} is not alive, picking another channel"
+          )
+
+          resolve_channel(channel, opts, attempts - 1)
+        end
+
+      _ ->
+        fallback_channel(channel)
     end
+  end
 
-    accepted_compressors =
-      if compressor do
-        Enum.uniq([compressor | accepted_compressors])
-      else
-        accepted_compressors
-      end
+  # A channel built by connect/2 carries its own adapter_payload and can serve
+  # the RPC directly. The virtual handle of a named connection has none: with
+  # no healthy underlying connection to resolve to, fail with UNAVAILABLE
+  # instead of handing the adapter a channel it cannot use.
+  defp fallback_channel(%Channel{adapter_payload: payload} = channel) when is_map(payload) do
+    {:ok, channel}
+  end
 
-    stream = %{
-      stream
-      | codec: Keyword.get(opts, :codec, ch.codec),
-        compressor: compressor,
-        accepted_compressors: accepted_compressors
-    }
+  defp fallback_channel(%Channel{ref: ref}) do
+    {:error,
+     GRPC.RPCError.exception(
+       GRPC.Status.unavailable(),
+       "no healthy connection available for #{inspect(ref)}"
+     )}
+  end
+
+  # Fail without a usable channel while preserving the calling contract: the
+  # failure still flows through the interceptor chain and client_span
+  # telemetry, and request-streaming calls raise — their return value is a
+  # `GRPC.Client.Stream`, so an error tuple cannot express failure to them.
+  defp unavailable_result(
+         error,
+         %{channel: channel} = stream,
+         request,
+         req_mod,
+         res_mod,
+         req_stream
+       ) do
+    stream = %{stream | request_mod: req_mod, response_mod: res_mod}
 
     GRPC.Telemetry.client_span(stream, request, fn ->
-      do_call(req_stream, stream, request, opts)
+      last = fn _stream, _request -> {:error, error} end
+
+      next =
+        Enum.reduce(channel.interceptors, last, fn {interceptor, opts}, acc ->
+          fn s, r -> interceptor.call(s, r, acc, opts) end
+        end)
+
+      result = next.(stream, request)
+
+      if req_stream, do: raise(error), else: result
     end)
   end
 
diff --git a/grpc/test/grpc/client/connection_supervised_test.exs b/grpc/test/grpc/client/connection_supervised_test.exs
index 3eceb0b5..15bb34b6 100644
--- a/grpc/test/grpc/client/connection_supervised_test.exs
+++ b/grpc/test/grpc/client/connection_supervised_test.exs
@@ -30,6 +30,70 @@ defmodule GRPC.Client.ConnectionSupervisedTest do
     defp test_pid, do: Application.get_env(:grpc, :tracking_resolver_test_pid)
   end
 
+  defmodule TransportProcessAdapter do
+    @moduledoc false
+    # Mirrors the Gun adapter's process shape: the "transport" process is NOT
+    # linked to the connection process (Gun's ConnectionProcess lives under
+    # the adapter's DynamicSupervisor), so its death is only observable via
+    # the monitor the connection sets in connect_real_channel/5. Its pid is
+    # exposed as adapter_payload.conn_pid; killing it simulates the transport
+    # dying mid-flight (e.g. gun giving up after internal retries). Accepts
+    # FailingClientAdapter-style :failing_hosts adapter options so tests can
+    # flip reachability while a connection is down.
+    @behaviour GRPC.Client.Adapter
+
+    def connect(%{host: host} = channel, opts) do
+      if host in failing_hosts(opts) do
+        {:error, :connection_refused}
+      else
+        pid =
+          spawn(fn ->
+            receive do
+              :stop -> :ok
+            end
+          end)
+
+        {:ok, %{channel | adapter_payload: %{conn_pid: pid}}}
+      end
+    end
+
+    def disconnect(%{adapter_payload: %{conn_pid: pid}} = channel) when is_pid(pid) do
+      if Process.alive?(pid), do: send(pid, :stop)
+
+      {:ok, %{channel | adapter_payload: %{conn_pid: nil}}}
+    end
+
+    def disconnect(channel), do: {:ok, channel}
+
+    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
+
+    defp failing_hosts(opts) do
+      case Keyword.get(opts || [], :failing_hosts, []) do
+        fun when is_function(fun, 0) -> fun.()
+        hosts when is_list(hosts) -> hosts
+      end
+    end
+  end
+
+  defmodule TwoAddressResolver do
+    @moduledoc false
+    def resolve(_target) do
+      {:ok,
+       %{
+         addresses: [
+           %{address: "127.0.0.1", port: 50051},
+           %{address: "127.0.0.2", port: 50052}
+         ],
+         service_config: nil
+       }}
+    end
+  end
+
   describe "child_spec/1 and start_link/1" do
     test "starts a named connection from an inline child spec" do
       name = unique_name("inline")
@@ -145,6 +209,173 @@ defmodule GRPC.Client.ConnectionSupervisedTest do
     end
   end
 
+  describe "underlying connection process death" do
+    @tag capture_log: true
+    test "re-establishes in place when the last connection process dies" do
+      name = unique_name("conn_death")
+      attach_telemetry([:grpc, :client, :connection, :connected])
+
+      start_supervised!(
+        {Connection, name: name, target: "ipv4:127.0.0.1:50051", adapter: TransportProcessAdapter}
+      )
+
+      assert :ok = Connection.await_ready(name, 2_000)
+      assert_receive {:telemetry, [:grpc, :client, :connection, :connected], _, %{name: ^name}}
+
+      conn = whereis_connection(name)
+
+      assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: pid1}}} =
+               Connection.pick_channel(%GRPC.Channel{ref: name})
+
+      Process.exit(pid1, :kill)
+
+      # Recovery must come from the same orchestrator process reconnecting,
+      # not from a supervisor restart.
+      assert_receive {:telemetry, [:grpc, :client, :connection, :connected], _, %{name: ^name}},
+                     2_000
+
+      assert whereis_connection(name) == conn
+      assert :ok = Connection.await_ready(name, 2_000)
+
+      assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: pid2}}} =
+               Connection.pick_channel(%GRPC.Channel{ref: name})
+
+      assert pid2 != pid1
+      assert Process.alive?(pid2)
+    end
+
+    @tag capture_log: true
+    test "RPCs fail with UNAVAILABLE while down and succeed after recovery" do
+      name = unique_name("conn_death_rpc")
+      hosts = start_supervised!({Agent, fn -> [] end})
+
+      start_supervised!(
+        {Connection,
+         name: name,
+         target: "ipv4:127.0.0.1:50051",
+         adapter: TransportProcessAdapter,
+         adapter_opts: [failing_hosts: fn -> Agent.get(hosts, & &1) end]}
+      )
+
+      assert :ok = Connection.await_ready(name, 2_000)
+      conn = whereis_connection(name)
+      {:ok, handle} = Connection.get_channel(name)
+      request = %Helloworld.HelloRequest{name: "ping"}
+
+      assert {:ok, _} = Helloworld.Greeter.Stub.say_hello(handle, request)
+
+      assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: pid1}}} =
+               Connection.pick_channel(handle)
+
+      # Make redials fail, then kill the transport: the connection enters the
+      # retry loop and RPCs must fail clean instead of crashing in the adapter
+      # (previously a FunctionClauseError on the payload-less virtual handle).
+      Agent.update(hosts, fn _ -> ["127.0.0.1"] end)
+      kill_and_await(pid1)
+      wait_until(fn -> not :sys.get_state(conn).established? end)
+
+      unavailable = GRPC.Status.unavailable()
+
+      assert {:error, %GRPC.RPCError{status: ^unavailable}} =
+               Helloworld.Greeter.Stub.say_hello(handle, request)
+
+      Agent.update(hosts, fn _ -> [] end)
+      send(conn, :retry_establish)
+
+      assert :ok = Connection.await_ready(name, 2_000)
+      assert {:ok, _} = Helloworld.Greeter.Stub.say_hello(handle, request)
+    end
+
+    @tag capture_log: true
+    test "request-streaming calls raise UNAVAILABLE instead of returning an error tuple" do
+      name = unique_name("conn_death_stream")
+      hosts = start_supervised!({Agent, fn -> ["127.0.0.1"] end})
+
+      start_supervised!(
+        {Connection,
+         name: name,
+         target: "ipv4:127.0.0.1:50051",
+         adapter: TransportProcessAdapter,
+         adapter_opts: [failing_hosts: fn -> Agent.get(hosts, & &1) end]}
+      )
+
+      {:ok, handle} = Connection.get_channel(name)
+
+      # A stream return value cannot express failure, so the stub must raise
+      # rather than hand back an error tuple that send_request/3 would crash on.
+      assert_raise GRPC.RPCError, ~r/no healthy connection/, fn ->
+        Routeguide.RouteGuide.Stub.record_route(handle)
+      end
+    end
+
+    @tag capture_log: true
+    test "keeps serving from the remaining channels and redials the dead one" do
+      name = unique_name("partial_death")
+
+      start_supervised!(
+        {Connection,
+         name: name,
+         target: "dns://multi.test:50051",
+         resolver: TwoAddressResolver,
+         lb_policy: :round_robin,
+         adapter: TransportProcessAdapter}
+      )
+
+      assert :ok = Connection.await_ready(name, 2_000)
+      conn = whereis_connection(name)
+      handle = %GRPC.Channel{ref: name}
+
+      pids =
+        for _ <- 1..4, uniq: true do
+          {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: pid}}} =
+            Connection.pick_channel(handle)
+
+          pid
+        end
+
+      assert length(pids) == 2
+      [dead, _survivor] = pids
+
+      kill_and_await(dead)
+      wait_until(fn -> is_nil(connected_key_for(conn, dead)) end)
+
+      # Still established and no full re-establish: picks never see the dead
+      # channel again.
+      assert :ok = Connection.await_ready(name, 100)
+      assert whereis_connection(name) == conn
+
+      for _ <- 1..4 do
+        assert {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: pid}}} =
+                 Connection.pick_channel(handle)
+
+        assert pid != dead
+        assert Process.alive?(pid)
+      end
+
+      # The repair loop redials the dead endpoint even though this resolver
+      # has no background worker to trigger a re-resolution.
+      wait_until(fn ->
+        pids =
+          for _ <- 1..4, uniq: true do
+            {:ok, %GRPC.Channel{adapter_payload: %{conn_pid: pid}}} =
+              Connection.pick_channel(handle)
+
+            pid
+          end
+
+        length(pids) == 2 and Enum.all?(pids, &Process.alive?/1)
+      end)
+    end
+
+    test "stub calls through an unresolvable virtual handle return UNAVAILABLE" do
+      handle = %GRPC.Channel{ref: :no_such_connection_name}
+      unavailable = GRPC.Status.unavailable()
+
+      assert {:error, %GRPC.RPCError{status: ^unavailable}} =
+               Helloworld.Greeter.Stub.say_hello(handle, %Helloworld.HelloRequest{name: "x"})
+    end
+  end
+
   describe "await_ready/2 waiter lifecycle" do
     setup do
       %{
@@ -315,6 +546,35 @@ defmodule GRPC.Client.ConnectionSupervisedTest do
 
   defp unique_name(prefix), do: :"#{prefix}_#{System.unique_integer([:positive])}"
 
+  # Kills a transport and blocks until it is actually gone, so its monitor
+  # signal to the connection process has been dispatched.
+  defp kill_and_await(pid) do
+    ref = Process.monitor(pid)
+    Process.exit(pid, :kill)
+    assert_receive {:DOWN, ^ref, :process, ^pid, _}, 1_000
+  end
+
+  defp wait_until(fun, tries \\ 200) do
+    cond do
+      fun.() ->
+        :ok
+
+      tries == 0 ->
+        flunk("condition not met within the wait budget")
+
+      true ->
+        Process.sleep(10)
+        wait_until(fun, tries - 1)
+    end
+  end
+
+  defp connected_key_for(conn, pid) do
+    Enum.find_value(:sys.get_state(conn).real_channels, fn
+      {key, {:connected, %{adapter_payload: %{conn_pid: ^pid}}}} -> key
+      _ -> nil
+    end)
+  end
+
   defp whereis_connection(name) do
     case Registry.lookup(GRPC.Client.Registry, {Connection, name}) do
       [{pid, _value}] -> pid

From 6c3666fbcb493744273c5b67e95e62a37e310093 Mon Sep 17 00:00:00 2001
From: Erik Nilsen <enilsen16@live.com>
Date: Thu, 30 Jul 2026 09:19:30 -0700
Subject: [PATCH 2/4] fix: harden re-establish and UNAVAILABLE paths from
 review findings

- terminate the old LB only after the replacement initializes, so a failed
  policy-flip attempt can't leave lb_state pointing at a deleted ETS table
  (later update calls would crash the connection process); dedup the
  init/update failure shells into run_lb/3
- on a resolve failure during re-establishment, redial the last known
  address set under the existing LB instead of silently downgrading a
  round_robin connection to a single PickFirst endpoint
- schedule the repair loop from adopt_established when some addresses
  failed to dial, so a partially successful establish doesn't strand the
  failed endpoints when there is no background resolver
- flush the sibling :EXIT/:DOWN signal when handling a channel death so
  link-based adapters (Mint) don't log a spurious unrelated-signal warning
  for every transport death
- validate call options before resolving the channel so configuration
  errors raise deterministically instead of being masked as UNAVAILABLE
  while the connection is down
- fail fallback_channel with UNAVAILABLE when a payload-carrying channel's
  conn_pid is dead instead of handing the adapter a dead transport
- honor an interceptor-transformed result on the request-streaming
  UNAVAILABLE path, raising only when the chain still returns an error;
  dedup the interceptor fold into run_interceptors/2
---
 grpc/lib/grpc/client/connection.ex |  94 +++++++++++++++--------
 grpc/lib/grpc/stub.ex              | 115 +++++++++++++++++------------
 2 files changed, 132 insertions(+), 77 deletions(-)

diff --git a/grpc/lib/grpc/client/connection.ex b/grpc/lib/grpc/client/connection.ex
index 88985d16..5ba1ec80 100644
--- a/grpc/lib/grpc/client/connection.ex
+++ b/grpc/lib/grpc/client/connection.ex
@@ -601,7 +601,7 @@ defmodule GRPC.Client.Connection do
     # is processed, so they never match.
     case down_channel_key(state.real_channels, pid) do
       nil -> handle_unrelated_exit(pid, reason, state)
-      key -> handle_channel_down(key, reason, state)
+      key -> handle_channel_down(key, pid, reason, state)
     end
   end
 
@@ -621,7 +621,7 @@ defmodule GRPC.Client.Connection do
             {:noreply, state}
 
           key ->
-            handle_channel_down(key, reason, state)
+            handle_channel_down(key, pid, reason, state)
         end
 
       {dropped, remaining} ->
@@ -683,7 +683,13 @@ defmodule GRPC.Client.Connection do
   # flap, and each consecutive flap backs the redial off further.
   @flap_window 10_000
 
-  defp handle_channel_down(key, reason, state) do
+  defp handle_channel_down(key, pid, reason, state) do
+    # Adapters that link their transport (Mint) deliver both an :EXIT and a
+    # monitor :DOWN for the same death; both are enqueued at exit time, so
+    # drop the sibling now — once the channel is marked {:failed, _} the
+    # second signal would be logged as an unrelated message.
+    flush_sibling_death_signal(pid)
+
     Logger.warning(
       "gRPC connection #{key} for #{state.resolver_target} went down: #{inspect(reason)}"
     )
@@ -709,6 +715,19 @@ defmodule GRPC.Client.Connection do
     end
   end
 
+  defp flush_sibling_death_signal(pid) do
+    receive do
+      {:DOWN, _mon, :process, ^pid, _} -> :ok
+    after
+      0 ->
+        receive do
+          {:EXIT, ^pid, _} -> :ok
+        after
+          0 -> :ok
+        end
+    end
+  end
+
   defp down_channel_key(real_channels, pid) do
     Enum.find_value(real_channels, fn
       {key, {:connected, %{adapter_payload: %{conn_pid: ^pid}}}} -> key
@@ -751,7 +770,7 @@ defmodule GRPC.Client.Connection do
 
     reply_waiters(state, state.waiters, :ok, :ok)
 
-    %{
+    state = %{
       state
       | established?: true,
         established_at: System.monotonic_time(:millisecond),
@@ -759,6 +778,15 @@ defmodule GRPC.Client.Connection do
         retry_attempt: 0,
         last_error: nil
     }
+
+    # A partially successful establish (some addresses failed to dial) still
+    # needs the repair loop; without a background resolver nothing else would
+    # ever redial the failed endpoints.
+    if any_failed?(state.real_channels) do
+      schedule_repair(state)
+    else
+      state
+    end
   end
 
   @impl GenServer
@@ -854,11 +882,21 @@ defmodule GRPC.Client.Connection do
           {addresses, choose_lb_mod(config, norm_opts[:lb_policy])}
 
         {:error, _reason} ->
-          # Fall back to treating the target as a single direct endpoint. Any
-          # LB policy would only have one address to choose from, so PickFirst
-          # is the only meaningful choice.
-          {host, port} = EndpointResolver.split_host_port(state.resolver_target)
-          {[%{address: host, port: port}], GRPC.Client.LoadBalancing.PickFirst}
+          case state do
+            %__MODULE__{desired_addresses: [_ | _] = addresses, lb_mod: lb_mod}
+            when not is_nil(lb_mod) ->
+              # A transient resolve failure during re-establishment must not
+              # downgrade the policy: redial the last known address set under
+              # the existing LB instead of collapsing to a single endpoint.
+              {addresses, lb_mod}
+
+            _ ->
+              # Fall back to treating the target as a single direct endpoint.
+              # Any LB policy would only have one address to choose from, so
+              # PickFirst is the only meaningful choice.
+              {host, port} = EndpointResolver.split_host_port(state.resolver_target)
+              {[%{address: host, port: port}], GRPC.Client.LoadBalancing.PickFirst}
+          end
       end
 
     real_channels = build_real_channels(addresses, state.virtual_channel, norm_opts, adapter)
@@ -894,28 +932,22 @@ defmodule GRPC.Client.Connection do
          %__MODULE__{lb_mod: lb_mod, lb_state: lb_state}
        )
        when not is_nil(lb_state) do
-    case connected_channels(real_channels) do
-      [] ->
-        disconnect_real_channels(real_channels, adapter)
-        {:error, first_failure(real_channels) || :no_addresses}
-
-      connected ->
-        case lb_mod.update(lb_state, connected) do
-          {:ok, new_lb_state} ->
-            {:ok, new_lb_state}
-
-          {:error, reason} ->
-            disconnect_real_channels(real_channels, adapter)
-            {:error, reason}
-        end
-    end
+    run_lb(real_channels, adapter, &lb_mod.update(lb_state, &1))
   end
 
   defp init_or_update_lb(lb_mod, real_channels, adapter, state) do
-    # A policy flip discards the previous LB; drop its ETS-backed state so
-    # repeated re-establishments can't leak one table per flip.
-    maybe_terminate_lb(state.lb_mod, state.lb_state)
-    init_lb(lb_mod, real_channels, adapter)
+    case init_lb(lb_mod, real_channels, adapter) do
+      {:ok, lb_state} ->
+        # A policy flip discards the previous LB; drop its ETS-backed state so
+        # repeated re-establishments can't leak one table per flip. Terminate
+        # only after the new LB is up: on failure state.lb_state must stay
+        # usable for later update calls.
+        maybe_terminate_lb(state.lb_mod, state.lb_state)
+        {:ok, lb_state}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
   end
 
   defp maybe_terminate_lb(lb_mod, lb_state)
@@ -929,13 +961,17 @@ defmodule GRPC.Client.Connection do
   end
 
   defp init_lb(lb_mod, real_channels, adapter) do
+    run_lb(real_channels, adapter, &lb_mod.init(channels: &1))
+  end
+
+  defp run_lb(real_channels, adapter, lb_call) do
     case connected_channels(real_channels) do
       [] ->
         disconnect_real_channels(real_channels, adapter)
         {:error, first_failure(real_channels) || :no_addresses}
 
       connected ->
-        case lb_mod.init(channels: connected) do
+        case lb_call.(connected) do
           {:ok, lb_state} ->
             {:ok, lb_state}
 
diff --git a/grpc/lib/grpc/stub.ex b/grpc/lib/grpc/stub.ex
index f6c7cdc1..ed5d1ff7 100644
--- a/grpc/lib/grpc/stub.ex
+++ b/grpc/lib/grpc/stub.ex
@@ -308,37 +308,45 @@ defmodule GRPC.Stub do
   def call(_service_mod, rpc, %{channel: channel} = stream, request, opts) do
     {_, {req_mod, req_stream}, {res_mod, response_stream}, _rpc_options} = rpc
 
-    case resolve_channel(channel, opts) do
-      {:error, %GRPC.RPCError{} = error} ->
-        unavailable_result(error, stream, request, req_mod, res_mod, req_stream)
+    # Options are validated before the channel is resolved so a configuration
+    # error raises the same ArgumentError whether or not the connection is
+    # healthy, instead of being masked as a retriable UNAVAILABLE. Real
+    # channels inherit codec/compressor fields from the virtual channel, so
+    # defaults can be read from `channel` here.
+    opts =
+      if req_stream || response_stream do
+        parse_req_opts([{:timeout, :infinity} | opts])
+      else
+        parse_req_opts([{:timeout, @default_timeout} | opts])
+      end
 
-      {:ok, ch} ->
-        stream = %{stream | channel: ch, request_mod: req_mod, response_mod: res_mod}
+    compressor = Keyword.get(opts, :compressor, channel.compressor)
 
-        opts =
-          if req_stream || response_stream do
-            parse_req_opts([{:timeout, :infinity} | opts])
-          else
-            parse_req_opts([{:timeout, @default_timeout} | opts])
-          end
+    accepted_compressors =
+      Keyword.get(opts, :accepted_compressors, channel.accepted_compressors)
 
-        compressor = Keyword.get(opts, :compressor, ch.compressor)
-        accepted_compressors = Keyword.get(opts, :accepted_compressors, ch.accepted_compressors)
+    if not is_list(accepted_compressors) do
+      raise ArgumentError, "accepted_compressors is not a list"
+    end
 
-        if not is_list(accepted_compressors) do
-          raise ArgumentError, "accepted_compressors is not a list"
-        end
+    accepted_compressors =
+      if compressor do
+        Enum.uniq([compressor | accepted_compressors])
+      else
+        accepted_compressors
+      end
 
-        accepted_compressors =
-          if compressor do
-            Enum.uniq([compressor | accepted_compressors])
-          else
-            accepted_compressors
-          end
+    case resolve_channel(channel, opts) do
+      {:error, %GRPC.RPCError{} = error} ->
+        unavailable_result(error, stream, request, req_mod, res_mod, req_stream)
 
+      {:ok, ch} ->
         stream = %{
           stream
-          | codec: Keyword.get(opts, :codec, ch.codec),
+          | channel: ch,
+            request_mod: req_mod,
+            response_mod: res_mod,
+            codec: Keyword.get(opts, :codec, ch.codec),
             compressor: compressor,
             accepted_compressors: accepted_compressors
         }
@@ -378,14 +386,27 @@ defmodule GRPC.Stub do
   end
 
   # A channel built by connect/2 carries its own adapter_payload and can serve
-  # the RPC directly. The virtual handle of a named connection has none: with
-  # no healthy underlying connection to resolve to, fail with UNAVAILABLE
-  # instead of handing the adapter a channel it cannot use.
+  # the RPC directly — but only while its transport process is alive; a stale
+  # snapshot of a re-establishing connection must fail with UNAVAILABLE
+  # instead of handing the adapter a dead conn_pid. The virtual handle of a
+  # named connection has no payload at all and always fails here.
   defp fallback_channel(%Channel{adapter_payload: payload} = channel) when is_map(payload) do
-    {:ok, channel}
+    case payload do
+      %{conn_pid: pid} when is_pid(pid) ->
+        if Process.alive?(pid) do
+          {:ok, channel}
+        else
+          unavailable_error(channel.ref)
+        end
+
+      _ ->
+        {:ok, channel}
+    end
   end
 
-  defp fallback_channel(%Channel{ref: ref}) do
+  defp fallback_channel(%Channel{ref: ref}), do: unavailable_error(ref)
+
+  defp unavailable_error(ref) do
     {:error,
      GRPC.RPCError.exception(
        GRPC.Status.unavailable(),
@@ -409,15 +430,23 @@ defmodule GRPC.Stub do
 
     GRPC.Telemetry.client_span(stream, request, fn ->
       last = fn _stream, _request -> {:error, error} end
+      result = run_interceptors(channel, last).(stream, request)
+
+      # Request-streaming calls return a stream, so an error tuple cannot
+      # express failure to them and errors raise instead. An interceptor may
+      # have rescued the failure into a usable result; honor it the same way
+      # do_call honors the chain's return value.
+      case {req_stream, result} do
+        {true, {:error, %GRPC.RPCError{} = transformed}} -> raise transformed
+        {true, {:error, _other}} -> raise error
+        _ -> result
+      end
+    end)
+  end
 
-      next =
-        Enum.reduce(channel.interceptors, last, fn {interceptor, opts}, acc ->
-          fn s, r -> interceptor.call(s, r, acc, opts) end
-        end)
-
-      result = next.(stream, request)
-
-      if req_stream, do: raise(error), else: result
+  defp run_interceptors(channel, last) do
+    Enum.reduce(channel.interceptors, last, fn {interceptor, opts}, acc ->
+      fn s, r -> interceptor.call(s, r, acc, opts) end
     end)
   end
 
@@ -436,12 +465,7 @@ defmodule GRPC.Stub do
       |> recv(opts)
     end
 
-    next =
-      Enum.reduce(channel.interceptors, last, fn {interceptor, opts}, acc ->
-        fn s, r -> interceptor.call(s, r, acc, opts) end
-      end)
-
-    next.(stream, request)
+    run_interceptors(channel, last).(stream, request)
   end
 
   defp do_call(true, %{channel: channel} = stream, req, opts) do
@@ -449,12 +473,7 @@ defmodule GRPC.Stub do
       channel.adapter.send_headers(s, opts)
     end
 
-    next =
-      Enum.reduce(channel.interceptors, last, fn {interceptor, opts}, acc ->
-        fn s, r -> interceptor.call(s, r, acc, opts) end
-      end)
-
-    next.(stream, req)
+    run_interceptors(channel, last).(stream, req)
   end
 
   @doc """

From 2cc1f1666507fc3b6c7436efccc1e12ad46f49d5 Mon Sep 17 00:00:00 2001
From: Erik Nilsen <enilsen16@live.com>
Date: Thu, 30 Jul 2026 11:40:15 -0700
Subject: [PATCH 3/4] feat: make the flap-detection window configurable

Expose the previously hard-coded 10s flap window as a :flap_window
connection option, validated and defaulted like the other timing knobs.
---
 grpc/lib/grpc/client/connection.ex | 14 +++++++++-----
 1 file changed, 9 insertions(+), 5 deletions(-)

diff --git a/grpc/lib/grpc/client/connection.ex b/grpc/lib/grpc/client/connection.ex
index 5ba1ec80..98016bdf 100644
--- a/grpc/lib/grpc/client/connection.ex
+++ b/grpc/lib/grpc/client/connection.ex
@@ -162,6 +162,7 @@ defmodule GRPC.Client.Connection do
   @default_max_resolve_interval 300_000
   @default_min_resolve_interval 5_000
   @default_connect_timeout 15_000
+  @default_flap_window 10_000
   @backoff_initial 100
   @backoff_multiplier 1.6
   @backoff_max 120_000
@@ -283,6 +284,9 @@ defmodule GRPC.Client.Connection do
     * `:headers` – default metadata headers
     * `:connect_timeout` – how long `connect/2` waits for the first
       establishment attempt in ms (default: 15000)
+    * `:flap_window` – a connection that dies within this many ms of
+      establishing counts as a flap, and consecutive flaps back the redial
+      off exponentially instead of redialing immediately (default: 10000)
     * `:resolve_interval` – DNS re-resolution interval in ms (default: 30000)
     * `:max_resolve_interval` – backoff cap in ms (default: 300000)
     * `:min_resolve_interval` – rate-limit floor in ms (default: 5000)
@@ -679,10 +683,6 @@ defmodule GRPC.Client.Connection do
     {:noreply, state}
   end
 
-  # A connection that dies within this window of establishing counts as a
-  # flap, and each consecutive flap backs the redial off further.
-  @flap_window 10_000
-
   defp handle_channel_down(key, pid, reason, state) do
     # Adapters that link their transport (Mint) deliver both an :EXIT and a
     # monitor :DOWN for the same death; both are enqueued at exit time, so
@@ -702,7 +702,10 @@ defmodule GRPC.Client.Connection do
       # boot retry loop instead of dialing here, so this handler stays
       # non-blocking. A stable connection redials immediately; a flapping one
       # backs off.
-      flaps = if uptime_ms(state) < @flap_window, do: state.flaps + 1, else: 0
+      # A connection that dies within this window of establishing counts as a
+      # flap, and each consecutive flap backs the redial off further.
+      flap_window = Keyword.get(state.connect_opts, :flap_window, @default_flap_window)
+      flaps = if uptime_ms(state) < flap_window, do: state.flaps + 1, else: 0
       delay = if flaps == 0, do: 0, else: backoff_delay(flaps)
       Process.send_after(self(), :retry_establish, delay)
 
@@ -1214,6 +1217,7 @@ defmodule GRPC.Client.Connection do
         headers: [],
         lb_policy: nil,
         connect_timeout: @default_connect_timeout,
+        flap_window: @default_flap_window,
         resolver: GRPC.Client.Resolver,
         resolve_interval: @default_resolve_interval,
         max_resolve_interval: @default_max_resolve_interval,

From 6ebc9b6b2e1cc1b64890b6978e97ad937b040ae3 Mon Sep 17 00:00:00 2001
From: Erik Nilsen <enilsen16@live.com>
Date: Thu, 30 Jul 2026 12:05:33 -0700
Subject: [PATCH 4/4] fix: harden recovery lifecycle from second review round

- guard the optional Resolver.update/2 callback in request_reresolve so a
  partial transport death can't crash the connection process when a custom
  resolver omits it; handle_cast(:resolve_now) now delegates to the same
  helper instead of keeping a divergent copy
- dedup :retry_establish timers behind a retry_scheduled? flag so flap
  cycles during an outage can't accumulate concurrent retry loops
- adopt immediately when a resolver update reconnects channels while a
  delayed retry is pending, so await_ready/connect track actual recovery
  instead of the retry backoff
- liveness-check channels on the :retry_establish adopt path so a
  reconnected-then-dead channel whose death signal is still queued can't
  trigger a false adopt
- only schedule the repair loop for connections without a background
  resolver; resolver ticks already redial failed endpoints
- read codec/compression defaults from the picked channel again: bare
  %Channel{ref: name} handles carry none of the connection's config
  (keeps option validation ahead of resolution)
- log one warning per RPC when re-picks are exhausted instead of one per
  attempt
---
 grpc/lib/grpc/client/connection.ex | 62 +++++++++++++++++++++---------
 grpc/lib/grpc/stub.ex              | 45 ++++++++++++----------
 2 files changed, 68 insertions(+), 39 deletions(-)

diff --git a/grpc/lib/grpc/client/connection.ex b/grpc/lib/grpc/client/connection.ex
index 98016bdf..ba85b54c 100644
--- a/grpc/lib/grpc/client/connection.ex
+++ b/grpc/lib/grpc/client/connection.ex
@@ -192,6 +192,7 @@ defmodule GRPC.Client.Connection do
           desired_addresses: [%{address: term(), port: :inet.port_number()}],
           repair_attempt: non_neg_integer(),
           repair_scheduled?: boolean(),
+          retry_scheduled?: boolean(),
           waiters: [{pid(), GenServer.from(), reference(), integer()}]
         }
 
@@ -212,6 +213,7 @@ defmodule GRPC.Client.Connection do
             desired_addresses: [],
             repair_attempt: 0,
             repair_scheduled?: false,
+            retry_scheduled?: false,
             waiters: []
 
   @doc """
@@ -499,13 +501,7 @@ defmodule GRPC.Client.Connection do
   end
 
   @impl GenServer
-  def handle_cast(:resolve_now, %{resolver: resolver, resolver_state: rs} = state)
-      when not is_nil(rs) do
-    {:ok, new_rs} = resolver.update(rs, :resolve_now)
-    {:noreply, %{state | resolver_state: new_rs}}
-  end
-
-  def handle_cast(:resolve_now, state), do: {:noreply, state}
+  def handle_cast(:resolve_now, state), do: {:noreply, request_reresolve(state)}
 
   @impl GenServer
   def handle_call({:ready_status, expected_target}, _from, state) do
@@ -559,16 +555,21 @@ defmodule GRPC.Client.Connection do
 
   @impl GenServer
   def handle_info(:retry_establish, %__MODULE__{established?: true} = state) do
-    {:noreply, state}
+    {:noreply, %{state | retry_scheduled?: false}}
   end
 
   def handle_info(:retry_establish, state) do
+    state = %{state | retry_scheduled?: false}
+
     # A background resolver update may have reconnected channels while this
     # retry was pending; adopt them instead of dialing a duplicate set that
-    # would orphan the live ones.
-    case connected_channels(state.real_channels) do
-      [] -> attempt_establish(state)
-      _connected -> {:noreply, adopt_established(state)}
+    # would orphan the live ones. Liveness is checked because a reconnected
+    # channel may already be dead with its death signal still queued behind
+    # this message.
+    if Enum.any?(state.real_channels, fn {_key, entry} -> channel_alive?(entry) end) do
+      {:noreply, adopt_established(state)}
+    else
+      attempt_establish(state)
     end
   end
 
@@ -707,7 +708,7 @@ defmodule GRPC.Client.Connection do
       flap_window = Keyword.get(state.connect_opts, :flap_window, @default_flap_window)
       flaps = if uptime_ms(state) < flap_window, do: state.flaps + 1, else: 0
       delay = if flaps == 0, do: 0, else: backoff_delay(flaps)
-      Process.send_after(self(), :retry_establish, delay)
+      state = schedule_retry(state, delay)
 
       {:noreply, %{state | established?: false, last_error: reason, flaps: flaps}}
     else
@@ -745,9 +746,14 @@ defmodule GRPC.Client.Connection do
 
   defp request_reresolve(%__MODULE__{resolver: resolver, resolver_state: rs} = state)
        when not is_nil(rs) do
-    case resolver.update(rs, :resolve_now) do
-      {:ok, new_rs} -> %{state | resolver_state: new_rs}
-      _ -> state
+    # update/2 is an optional callback of GRPC.Client.Resolver.
+    if function_exported?(resolver, :update, 2) do
+      case resolver.update(rs, :resolve_now) do
+        {:ok, new_rs} -> %{state | resolver_state: new_rs}
+        _ -> state
+      end
+    else
+      state
     end
   end
 
@@ -755,11 +761,22 @@ defmodule GRPC.Client.Connection do
 
   defp schedule_repair(%__MODULE__{repair_scheduled?: true} = state), do: state
 
+  # A background resolver already redials failed endpoints on its resolve
+  # ticks; the repair loop exists for connections without one.
+  defp schedule_repair(%__MODULE__{resolver_state: rs} = state) when not is_nil(rs), do: state
+
   defp schedule_repair(state) do
     Process.send_after(self(), :repair_channels, backoff_delay(state.repair_attempt))
     %{state | repair_scheduled?: true}
   end
 
+  defp schedule_retry(%__MODULE__{retry_scheduled?: true} = state, _delay), do: state
+
+  defp schedule_retry(state, delay) do
+    Process.send_after(self(), :retry_establish, delay)
+    %{state | retry_scheduled?: true}
+  end
+
   defp any_failed?(real_channels) do
     Enum.any?(real_channels, &match?({_key, {:failed, _}}, &1))
   end
@@ -870,7 +887,7 @@ defmodule GRPC.Client.Connection do
             "#{inspect(reason)}, retrying in #{delay}ms"
         )
 
-        Process.send_after(self(), :retry_establish, delay)
+        state = schedule_retry(state, delay)
         {:noreply, %{state | last_error: reason, retry_attempt: state.retry_attempt + 1}}
     end
   end
@@ -1159,7 +1176,16 @@ defmodule GRPC.Client.Connection do
       Logger.warning("No healthy channels available for #{state.resolver_target}")
     end
 
-    %{state | real_channels: real_channels, lb_state: new_lb_state}
+    state = %{state | real_channels: real_channels, lb_state: new_lb_state}
+
+    # A resolver update can reconnect channels while a delayed retry is still
+    # pending; adopt immediately so await_ready and connect/2 track actual
+    # recovery instead of the retry backoff.
+    if connected != [] and not state.established? do
+      adopt_established(state)
+    else
+      state
+    end
   end
 
   defp reconcile_lb(lb_mod, lb_state, new_channels) do
diff --git a/grpc/lib/grpc/stub.ex b/grpc/lib/grpc/stub.ex
index ed5d1ff7..69f435b2 100644
--- a/grpc/lib/grpc/stub.ex
+++ b/grpc/lib/grpc/stub.ex
@@ -310,9 +310,7 @@ defmodule GRPC.Stub do
 
     # Options are validated before the channel is resolved so a configuration
     # error raises the same ArgumentError whether or not the connection is
-    # healthy, instead of being masked as a retriable UNAVAILABLE. Real
-    # channels inherit codec/compressor fields from the virtual channel, so
-    # defaults can be read from `channel` here.
+    # healthy, instead of being masked as a retriable UNAVAILABLE.
     opts =
       if req_stream || response_stream do
         parse_req_opts([{:timeout, :infinity} | opts])
@@ -320,27 +318,30 @@ defmodule GRPC.Stub do
         parse_req_opts([{:timeout, @default_timeout} | opts])
       end
 
-    compressor = Keyword.get(opts, :compressor, channel.compressor)
-
-    accepted_compressors =
-      Keyword.get(opts, :accepted_compressors, channel.accepted_compressors)
-
-    if not is_list(accepted_compressors) do
+    if not is_list(Keyword.get(opts, :accepted_compressors, [])) do
       raise ArgumentError, "accepted_compressors is not a list"
     end
 
-    accepted_compressors =
-      if compressor do
-        Enum.uniq([compressor | accepted_compressors])
-      else
-        accepted_compressors
-      end
-
     case resolve_channel(channel, opts) do
       {:error, %GRPC.RPCError{} = error} ->
         unavailable_result(error, stream, request, req_mod, res_mod, req_stream)
 
       {:ok, ch} ->
+        # Codec/compression defaults come from the picked channel: a caller
+        # may hold a bare %Channel{ref: name} handle that carries none of the
+        # connection's configuration.
+        compressor = Keyword.get(opts, :compressor, ch.compressor)
+
+        accepted_compressors =
+          Keyword.get(opts, :accepted_compressors, ch.accepted_compressors)
+
+        accepted_compressors =
+          if compressor do
+            Enum.uniq([compressor | accepted_compressors])
+          else
+            accepted_compressors
+          end
+
         stream = %{
           stream
           | channel: ch,
@@ -363,7 +364,13 @@ defmodule GRPC.Stub do
 
   defp resolve_channel(channel, opts), do: resolve_channel(channel, opts, @resolve_attempts)
 
-  defp resolve_channel(channel, _opts, 0), do: fallback_channel(channel)
+  defp resolve_channel(channel, _opts, 0) do
+    Logger.warning(
+      "no live connection process after #{@resolve_attempts} picks for #{inspect(channel.ref)}"
+    )
+
+    fallback_channel(channel)
+  end
 
   defp resolve_channel(channel, opts, attempts) do
     case Connection.pick_channel(channel, opts) do
@@ -373,10 +380,6 @@ defmodule GRPC.Stub do
         if is_pid(conn_pid) and Process.alive?(conn_pid) do
           {:ok, ch}
         else
-          Logger.warning(
-            "The connection process #{inspect(conn_pid)} is not alive, picking another channel"
-          )
-
           resolve_channel(channel, opts, attempts - 1)
         end