diff --git a/grpc/lib/grpc/client/connection.ex b/grpc/lib/grpc/client/connection.ex index c9cdbb38..82949caa 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 @@ -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 diff --git a/grpc/lib/grpc/stub.ex b/grpc/lib/grpc/stub.ex index abba1c5c..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 @@ -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,9 +341,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 +351,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 +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