From 82e1c9443750049f618d9fc5b4c9b3b0d5180a2b Mon Sep 17 00:00:00 2001 From: Erik Nilsen Date: Mon, 31 Aug 2026 09:33:22 -0700 Subject: [PATCH 1/3] fix(client): harden supervised connection recovery edge cases Follow-ups to #568: - cancel pending retry timers on adopt/reschedule so a stale long-backoff timer can't delay redial after a fresh transport death - guard rebalance_after_reconcile adoption with channel_alive?, matching the :retry_establish handler, to avoid adopting an already-dead channel - republish LB state to :persistent_term when lb_mod.update/2 returns a new state, so pickers don't read stale state from behaviour-compliant LBs - seed retry_attempt from the flap count so a dial failure after a flap-death continues the backoff ladder instead of restarting it - log non-ok resolver update/2 results instead of silently ignoring them - liveness-check fallback channels carrying conn_pid (e.g. the documented disconnect/1 return) and return UNAVAILABLE instead of crashing in the adapter - align resolve_channel with channel_alive? on payloads without conn_pid so pid-less adapters aren't treated as permanently dead - resolve connection config in unavailable_result so failure results run the connection's interceptors with the caller's codec/compressor - stop re-picking when the LB policy returns the same channel (PickFirst) --- grpc/lib/grpc/client/connection.ex | 68 ++++++++++++++++++++----- grpc/lib/grpc/stub.ex | 80 ++++++++++++++++++++---------- 2 files changed, 111 insertions(+), 37 deletions(-) diff --git a/grpc/lib/grpc/client/connection.ex b/grpc/lib/grpc/client/connection.ex index c9cdbb38..c6f2dce1 100644 --- a/grpc/lib/grpc/client/connection.ex +++ b/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 @@ -688,7 +688,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 @@ -727,8 +731,16 @@ defmodule GRPC.Client.Connection 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 +760,29 @@ 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 + # The timer already fired; drop its queued message so the retry loop + # cannot run twice. + receive do + :retry_establish -> :ok + after + 0 -> :ok + end + end + + %{state | retry_timer: nil} end defp any_failed?(real_channels) do @@ -768,6 +798,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 +1188,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 diff --git a/grpc/lib/grpc/stub.ex b/grpc/lib/grpc/stub.ex index abba1c5c..249b5d44 100644 --- a/grpc/lib/grpc/stub.ex +++ b/grpc/lib/grpc/stub.ex @@ -313,7 +313,7 @@ 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 @@ -351,9 +351,9 @@ 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)}" ) @@ -361,17 +361,27 @@ defmodule GRPC.Stub do 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 +398,19 @@ 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 + # Covers dead and remote pids as well as the `conn_pid: nil` payload + # that Connection.disconnect/1 returns. + 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 @@ -422,13 +431,34 @@ defmodule GRPC.Stub do request, req_mod, res_mod, - req_stream + req_stream, + opts ) do - stream = %{stream | request_mod: req_mod, response_mod: res_mod} + # A bare %Channel{ref: name} handle carries none of the connection's + # configuration, so interceptors/codec/compressor are resolved through the + # stored virtual channel — the same config a picked real channel inherits — + # keeping the failure path consistent with the healthy one. + 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 From b0de97e1f925cae52dbcef2c33c740ef3162469d Mon Sep 17 00:00:00 2001 From: Erik Nilsen Date: Tue, 1 Sep 2026 12:41:03 -0700 Subject: [PATCH 2/3] chore: remove unnecessary comment in unavailable_result --- grpc/lib/grpc/stub.ex | 4 ---- 1 file changed, 4 deletions(-) diff --git a/grpc/lib/grpc/stub.ex b/grpc/lib/grpc/stub.ex index 249b5d44..0556cdd1 100644 --- a/grpc/lib/grpc/stub.ex +++ b/grpc/lib/grpc/stub.ex @@ -434,10 +434,6 @@ defmodule GRPC.Stub do req_stream, opts ) do - # A bare %Channel{ref: name} handle carries none of the connection's - # configuration, so interceptors/codec/compressor are resolved through the - # stored virtual channel — the same config a picked real channel inherits — - # keeping the failure path consistent with the healthy one. config_ch = case Connection.get_channel(channel.ref) do {:ok, %Channel{} = virtual_channel} -> virtual_channel From a5bfa20ef4d58fcf86f18c65339aa9b83554e42a Mon Sep 17 00:00:00 2001 From: Erik Nilsen Date: Tue, 1 Sep 2026 12:55:53 -0700 Subject: [PATCH 3/3] chore: drop redundant comments added in this PR --- grpc/lib/grpc/client/connection.ex | 4 ---- grpc/lib/grpc/stub.ex | 16 ---------------- 2 files changed, 20 deletions(-) diff --git a/grpc/lib/grpc/client/connection.ex b/grpc/lib/grpc/client/connection.ex index c6f2dce1..82949caa 100644 --- a/grpc/lib/grpc/client/connection.ex +++ b/grpc/lib/grpc/client/connection.ex @@ -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 @@ -728,7 +727,6 @@ 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} -> @@ -773,8 +771,6 @@ defmodule GRPC.Client.Connection do defp cancel_retry_timer(%__MODULE__{retry_timer: ref} = state) do unless Process.cancel_timer(ref) do - # The timer already fired; drop its queued message so the retry loop - # cannot run twice. receive do :retry_establish -> :ok after diff --git a/grpc/lib/grpc/stub.ex b/grpc/lib/grpc/stub.ex index 0556cdd1..a5ccfafd 100644 --- a/grpc/lib/grpc/stub.ex +++ b/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 @@ -316,9 +309,6 @@ defmodule GRPC.Stub do 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 = @@ -402,8 +392,6 @@ defmodule GRPC.Stub do if local_process_alive?(pid) do {:ok, channel} else - # Covers dead and remote pids as well as the `conn_pid: nil` payload - # that Connection.disconnect/1 returns. unavailable_error(channel.ref) end end @@ -421,10 +409,6 @@ 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,