fix(client): harden supervised connection recovery edge cases

grpc/lib/grpc/client/connection.ex

@@ -191,7 +191,7 @@ defmodule GRPC.Client.Connection do
             desired_addresses: [],
             repair_attempt: 0,
             repair_scheduled?: false,
-            retry_scheduled?: false,
+            retry_timer: nil,
             waiters: []
 
   @doc """
@@ -533,11 +533,11 @@ defmodule GRPC.Client.Connection do
 
   @impl GenServer
   def handle_info(:retry_establish, %__MODULE__{established?: true} = state) do
-    {:noreply, %{state | retry_scheduled?: false}}
+    {:noreply, %{state | retry_timer: nil}}
   end
 
   def handle_info(:retry_establish, state) do
-    state = %{state | retry_scheduled?: false}
+    state = %{state | retry_timer: nil}
 
     # A background resolver update may have reconnected channels while this
     # retry was pending; adopt them instead of dialing a duplicate set that
@@ -569,7 +569,6 @@ defmodule GRPC.Client.Connection do
         {: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
@@ -688,7 +687,11 @@ defmodule GRPC.Client.Connection do
       delay = if flaps == 0, do: 0, else: backoff_delay(flaps)
       state = schedule_retry(state, delay)
 
-      {:noreply, %{state | established?: false, last_error: reason, flaps: flaps}}
+      # Seed the dial-failure ladder from the flap count so an establish
+      # failure after a flap continues the backoff instead of restarting it
+      # from the shortest delay (adopt_established reset retry_attempt to 0).
+      {:noreply,
+       %{state | established?: false, last_error: reason, flaps: flaps, retry_attempt: flaps}}
     else
       # Other channels keep serving; ask the resolver for an early
       # re-resolution and start the repair loop so the dead endpoint is
@@ -724,11 +727,18 @@ defmodule GRPC.Client.Connection do
 
   defp request_reresolve(%__MODULE__{resolver: resolver, resolver_state: rs} = state)
        when not is_nil(rs) do
-    # 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
+        {:ok, new_rs} ->
+          %{state | resolver_state: new_rs}
+
+        other ->
+          Logger.warning(
+            "Resolver update failed for #{state.resolver_target}, " <>
+              "keeping previous resolver state: #{inspect(other)}"
+          )
+
+          state
       end
     else
       state
@@ -748,11 +758,27 @@ defmodule GRPC.Client.Connection do
     %{state | repair_scheduled?: true}
   end
 
-  defp schedule_retry(%__MODULE__{retry_scheduled?: true} = state, _delay), do: state
-
+  # Cancel-and-reschedule instead of first-timer-wins: a pending retry may
+  # carry a stale delay (e.g. a long boot backoff scheduled before a resolver
+  # update recovered the connection), and a later death that computes a short
+  # flap delay must not wait it out.
   defp schedule_retry(state, delay) do
-    Process.send_after(self(), :retry_establish, delay)
-    %{state | retry_scheduled?: true}
+    state = cancel_retry_timer(state)
+    %{state | retry_timer: Process.send_after(self(), :retry_establish, delay)}
+  end
+
+  defp cancel_retry_timer(%__MODULE__{retry_timer: nil} = state), do: state
+
+  defp cancel_retry_timer(%__MODULE__{retry_timer: ref} = state) do
+    unless Process.cancel_timer(ref) do
+      receive do
+        :retry_establish -> :ok
+      after
+        0 -> :ok
+      end
+    end
+
+    %{state | retry_timer: nil}
   end
 
   defp any_failed?(real_channels) do
@@ -768,6 +794,10 @@ defmodule GRPC.Client.Connection do
 
     reply_waiters(state, state.waiters, :ok, :ok)
 
+    # A retry scheduled before this recovery is now stale; cancel it so a
+    # later death schedules its own delay instead of deduping against it.
+    state = cancel_retry_timer(state)
+
     state = %{
       state
       | established?: true,
@@ -1154,12 +1184,22 @@ defmodule GRPC.Client.Connection do
       Logger.warning("No healthy channels available for #{state.resolver_target}")
     end
 
+    # Built-in balancers mutate their ETS table in place, but the behaviour
+    # contract allows update/2 to return a fresh state; republish so pickers
+    # never read a stale one.
+    if state.lb_mod && new_lb_state != state.lb_state do
+      :persistent_term.put(lb_key(state.virtual_channel.ref), {state.lb_mod, new_lb_state})
+    end
+
     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
+    # recovery instead of the retry backoff. Liveness is checked (as in the
+    # retry handler) because a just-connected transport may already be dead
+    # with its death signal still queued.
+    if not state.established? and
+         Enum.any?(connected, &channel_alive?({:connected, &1})) do
       adopt_established(state)
     else
       state

grpc/lib/grpc/stub.ex

@@ -278,13 +278,6 @@ 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
@@ -313,12 +306,9 @@ defmodule GRPC.Stub do
 
     case resolve_channel(channel, opts) do
       {:error, %GRPC.RPCError{} = error} ->
-        unavailable_result(error, stream, request, req_mod, res_mod, req_stream)
+        unavailable_result(error, stream, request, req_mod, res_mod, req_stream, opts)
 
       {: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 =
@@ -351,27 +341,37 @@ defmodule GRPC.Stub do
   # 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), do: resolve_channel(channel, opts, @resolve_attempts, nil)
 
-  defp resolve_channel(channel, _opts, 0) do
+  defp resolve_channel(channel, _opts, 0, _last_pid) 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
+  defp resolve_channel(channel, opts, attempts, last_pid) 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 local_process_alive?(conn_pid) do
-          {:ok, ch}
-        else
-          resolve_channel(channel, opts, attempts - 1)
+      {:ok, %Channel{adapter_payload: %{conn_pid: pid}} = ch} ->
+        cond do
+          local_process_alive?(pid) ->
+            {:ok, ch}
+
+          # A repeated pick means the policy is not rotating (e.g. PickFirst);
+          # further picks would return the same dead entry.
+          pid == last_pid ->
+            fallback_channel(channel)
+
+          true ->
+            resolve_channel(channel, opts, attempts - 1, pid)
         end
 
+      {:ok, %Channel{adapter_payload: payload} = ch} when is_map(payload) ->
+        # An adapter that exposes no transport pid cannot be liveness-checked;
+        # treat it as usable, matching channel_alive?/1 on the connection side.
+        {:ok, ch}
+
       _ ->
         fallback_channel(channel)
     end
@@ -388,20 +388,17 @@ defmodule GRPC.Stub do
   # 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
-    case payload do
-      %{conn_pid: pid} when is_pid(pid) ->
-        if local_process_alive?(pid) do
-          {:ok, channel}
-        else
-          unavailable_error(channel.ref)
-        end
-
-      _ ->
-        {:ok, channel}
+  defp fallback_channel(%Channel{adapter_payload: %{conn_pid: pid}} = channel) do
+    if local_process_alive?(pid) do
+      {:ok, channel}
+    else
+      unavailable_error(channel.ref)
     end
   end
 
+  defp fallback_channel(%Channel{adapter_payload: payload} = channel) when is_map(payload),
+    do: {:ok, channel}
+
   defp fallback_channel(%Channel{ref: ref}), do: unavailable_error(ref)
 
   defp unavailable_error(ref) do
@@ -412,23 +409,36 @@ defmodule GRPC.Stub do
      )}
   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
+         req_stream,
+         opts
        ) do
-    stream = %{stream | request_mod: req_mod, response_mod: res_mod}
+    config_ch =
+      case Connection.get_channel(channel.ref) do
+        {:ok, %Channel{} = virtual_channel} -> virtual_channel
+        _ -> channel
+      end
+
+    compressor = Keyword.get(opts, :compressor, config_ch.compressor)
+
+    stream = %{
+      stream
+      | request_mod: req_mod,
+        response_mod: res_mod,
+        codec: Keyword.get(opts, :codec, config_ch.codec),
+        compressor: compressor,
+        accepted_compressors:
+          Keyword.get(opts, :accepted_compressors, config_ch.accepted_compressors)
+    }
 
     GRPC.Telemetry.client_span(stream, request, fn ->
       last = fn _stream, _request -> {:error, error} end
-      result = run_interceptors(channel, last).(stream, request)
+      result = run_interceptors(config_ch, 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