diff --git a/grpc/lib/grpc/client/connection.ex b/grpc/lib/grpc/client/connection.ex index 0f4dd7b1..ba85b54c 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 @@ -155,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 @@ -177,8 +185,14 @@ 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(), + retry_scheduled?: boolean(), waiters: [{pid(), GenServer.from(), reference(), integer()}] } @@ -192,8 +206,14 @@ 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, + retry_scheduled?: false, waiters: [] @doc """ @@ -266,6 +286,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) @@ -478,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 @@ -538,23 +555,101 @@ 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: attempt_establish(state) + 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. 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 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} + + 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, pid, 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, pid, reason, state) + end - 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. + {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 +676,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 +684,129 @@ 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)}" - ) + 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) - {:noreply, state} + 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) + + 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. + # 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) + state = schedule_retry(state, 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 - {:noreply, %{state | waiters: remaining}} + 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 - 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 + # 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 + + defp request_reresolve(state), do: state + + 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 + + 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 = %{ + state + | established?: true, + established_at: System.monotonic_time(:millisecond), + waiters: [], + 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 @@ -673,22 +869,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) @@ -706,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 @@ -721,16 +902,26 @@ 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) - 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 +930,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,14 +941,57 @@ 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 + run_lb(real_channels, adapter, &lb_mod.update(lb_state, &1)) + end + + defp init_or_update_lb(lb_mod, real_channels, adapter, state) do + 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) + 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 + 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} @@ -866,6 +1101,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,10 +1173,19 @@ 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} + 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 @@ -998,6 +1243,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, @@ -1129,8 +1375,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..69f435b2 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,29 +308,9 @@ 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) - - if is_pid(conn_pid) and Process.alive?(conn_pid) do - ch - else - Logger.warning( - "The connection process #{inspect(conn_pid)} is not alive, " <> - "please create a new channel via GRPC.Stub.connect/2" - ) - - channel - end - - _ -> - # fallback to the channel in the stream - channel - end - - stream = %{stream | channel: ch, request_mod: req_mod, response_mod: res_mod} - + # 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. opts = if req_stream || response_stream do parse_req_opts([{:timeout, :infinity} | opts]) @@ -331,29 +318,138 @@ defmodule GRPC.Stub do 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) - - 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, + request_mod: req_mod, + response_mod: res_mod, + codec: Keyword.get(opts, :codec, ch.codec), + compressor: compressor, + accepted_compressors: accepted_compressors + } + + GRPC.Telemetry.client_span(stream, request, fn -> + do_call(req_stream, stream, request, opts) + end) + end + 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 + 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 + {:ok, %Channel{adapter_payload: adapter_payload} = ch} when is_map(adapter_payload) -> + conn_pid = Map.get(adapter_payload, :conn_pid) + + if is_pid(conn_pid) and Process.alive?(conn_pid) do + {:ok, ch} + else + resolve_channel(channel, opts, attempts - 1) + end + + _ -> + fallback_channel(channel) + end + end + + # A channel built by connect/2 carries its own adapter_payload and can serve + # 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 + 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: unavailable_error(ref) - stream = %{ - stream - | codec: Keyword.get(opts, :codec, ch.codec), - compressor: compressor, - accepted_compressors: accepted_compressors - } + defp unavailable_error(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 + 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 + + 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 @@ -372,12 +468,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 @@ -385,12 +476,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 """ 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