From a22b64cd780042e2344b42f7d031339290b7ee5a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arran=20=C3=98ystein=20Kostveit=20Gabriel?= Date: Thu, 27 Aug 2026 09:56:59 +0200 Subject: [PATCH] fix(mint): explicitly support infinite retries --- README.md | 2 +- grpc/CHANGELOG.md | 7 +++ grpc/README.md | 2 +- grpc/lib/grpc/client/adapter.ex | 4 ++ grpc/lib/grpc/client/adapters/mint.ex | 58 ++++++++++++------- .../connection_process/connection_process.ex | 22 +++---- .../adapters/mint/connection_process/state.ex | 3 + grpc/lib/grpc/client/connection.ex | 15 ++++- .../adapters/mint/connection_process_test.exs | 20 +++++++ grpc/test/grpc/adapters/mint_test.exs | 39 +++++++++++++ grpc/test/grpc/client/connection_test.exs | 9 +++ 11 files changed, 144 insertions(+), 37 deletions(-) diff --git a/README.md b/README.md index c042c4391..078085fe7 100644 --- a/README.md +++ b/README.md @@ -428,7 +428,7 @@ iex> {:ok, channel} = GRPC.Stub.connect("localhost:50051", When the connection drops, the adapter will attempt to reconnect up to `retry` times using **exponential backoff with jitter**. The delay starts at ~1 second and grows up to a maximum of 120 seconds. If all attempts are exhausted, the parent process receives a `{:elixir_grpc, :connection_down, pid}` message. -By default, `:retry` is `0` (no reconnection attempts). +By default, `:retry` is `0` (no reconnection attempts). It accepts any non-negative integer or `:infinity`. > **Note:** Any in-flight requests at the time of the drop will fail immediately. Reconnection only re-establishes the transport connection — it does not replay requests. diff --git a/grpc/CHANGELOG.md b/grpc/CHANGELOG.md index 0c53d9a0d..812d794d8 100644 --- a/grpc/CHANGELOG.md +++ b/grpc/CHANGELOG.md @@ -2,13 +2,20 @@ ## Unreleased +### Enhancements + + * The Mint adapter's `:retry` option explicitly supports `:infinity`, reconnecting for as long as the connection process lives. This previously worked due Erlang term ordering (every atom sorts above every integer). + * `GRPC.Client.Adapter` gained an optional `validate_opts/1` callback. Adapters that implement it have their `:adapter_opts` validated in the caller by `GRPC.Client.Connection.connect/2`, so configuration errors raise there instead of crashing the spawned connection process. + ### Behavior Changes * The Mint adapter now enforces the requested `:timeout`/`:deadline` on unary receives. A unary call that never receives a response fails with `DEADLINE_EXCEEDED` after the documented 10s default instead of blocking indefinitely, and an explicit `:deadline` now takes precedence over `:timeout`. + * Test suites that define a Mox mock for the `GRPC.Client.Adapter` behaviour must stub the new validation callback. Mox generates optional callbacks on mocks, so `GRPC.Client.Connection` calls `validate_opts/1` on the mock and Mox raises `UnexpectedCallError` when it is not stubbed. Adapters that implement the behaviour with `@behaviour` and do not define `validate_opts/1` are unaffected. ### Bug Fixes * The Gun adapter no longer shares a named channel's connection process across Erlang nodes. It was registered in `:global`, so a node connecting with a `:name` already used on another node adopted the remote connection process; `connect/2` returned `{:ok, channel}`, but every RPC on it then raised `ArgumentError` because `GRPC.Stub.call/5` calls `Process.alive?/1` on the connection pid and that raises for remote pids. Connection processes are now registered in the node-local `GRPC.Client.Registry`, so reuse is per node. + * Invalid Mint `:retry` values now raise `ArgumentError`. Because the old comparisons relied on term ordering, bad input failed silently and in two different directions: a negative integer behaved as no-retry, while *any* atom lead to infinite reconnection. ## v1.0.4 (2026-0-15) diff --git a/grpc/README.md b/grpc/README.md index 45cf38b3c..073716a31 100644 --- a/grpc/README.md +++ b/grpc/README.md @@ -218,7 +218,7 @@ iex> {:ok, channel} = GRPC.Stub.connect("localhost:50051", When the connection drops, the adapter will attempt to reconnect up to `retry` times using **exponential backoff with jitter**. The delay starts at ~1 second and grows up to a maximum of 120 seconds. If all attempts are exhausted, the parent process receives a `{:elixir_grpc, :connection_down, pid}` message. -By default, `:retry` is `0` (no reconnection attempts). +By default, `:retry` is `0` (no reconnection attempts). It accepts any non-negative integer or `:infinity`. > **Note:** Any in-flight requests at the time of the drop will fail immediately. Reconnection only re-establishes the transport connection — it does not replay requests. diff --git a/grpc/lib/grpc/client/adapter.ex b/grpc/lib/grpc/client/adapter.ex index 3438b4b2c..ebf116c42 100644 --- a/grpc/lib/grpc/client/adapter.ex +++ b/grpc/lib/grpc/client/adapter.ex @@ -7,6 +7,8 @@ defmodule GRPC.Client.Adapter do @callback connect(channel :: struct(), opts :: keyword()) :: {:ok, struct()} | {:error, any()} + @callback validate_opts(opts :: keyword()) :: :ok | {:error, String.t()} + @callback disconnect(channel :: struct()) :: {:ok, struct()} | {:error, any()} @callback send_request(stream :: Stream.t(), contents :: iodata(), opts :: keyword()) :: @@ -43,4 +45,6 @@ defmodule GRPC.Client.Adapter do Cancel a stream in a streaming client. """ @callback cancel(stream :: Stream.t()) :: :ok | {:error, any()} + + @optional_callbacks validate_opts: 1 end diff --git a/grpc/lib/grpc/client/adapters/mint.ex b/grpc/lib/grpc/client/adapters/mint.ex index bfa8414e6..cb9a4e02d 100644 --- a/grpc/lib/grpc/client/adapters/mint.ex +++ b/grpc/lib/grpc/client/adapters/mint.ex @@ -38,38 +38,26 @@ if Code.ensure_loaded?(Mint.HTTP) do window size ensures that the number of packages exchanges is smaller, thus speeding up the requests by reducing the amount of networks round trip, with the cost of having larger packages reaching the server per connection. Check [Mint.HTTP2.setting() type](https://hexdocs.pm/mint/Mint.HTTP2.html#t:setting/0) for additional configs. - * `:retry`: Number of reconnection attempts when the connection drops. Defaults to `0` (no retries). + * `:retry`: Number of reconnection attempts when the connection drops, or `:infinity` + to keep reconnecting for as long as the process lives. Defaults to `0` (no retries). Uses exponential backoff with jitter between attempts. """ @impl true - def connect(%{host: host, port: port} = channel, opts \\ []) do - {config_opts, opts} = Keyword.pop(opts, :config_options, []) + def connect(%Channel{} = channel, opts \\ []) do {retry, opts} = Keyword.pop(opts, :retry, 0) - module_opts = Application.get_env(:grpc, __MODULE__, config_opts) - - opts = - channel - |> connect_opts(opts) - |> merge_opts(module_opts) - |> Keyword.put(:retry, retry) - - Process.flag(:trap_exit, true) - - channel - |> mint_scheme() - |> ConnectionProcess.start_link(host, port, opts) - |> case do - {:ok, pid} -> - {:ok, %{channel | adapter_payload: %{conn_pid: pid}}} - error -> - {:error, "Error while opening connection: #{inspect(error)}"} + with :ok <- validate_retry(retry), + {:ok, pid} <- start_connection_process(channel, opts, retry) do + {:ok, %{channel | adapter_payload: %{conn_pid: pid}}} end catch :exit, reason -> {:error, "Error while opening connection: #{inspect(reason)}"} end + @impl true + def validate_opts(opts), do: validate_retry(opts[:retry]) + @impl true def disconnect(%{adapter_payload: %{conn_pid: pid}} = channel) when is_pid(pid) do @@ -143,6 +131,13 @@ if Code.ensure_loaded?(Mint.HTTP) do ConnectionProcess.cancel(conn_pid, request_ref) end + defp validate_retry(nil), do: :ok + defp validate_retry(:infinity), do: :ok + defp validate_retry(retry) when is_integer(retry) and retry >= 0, do: :ok + + defp validate_retry(retry), + do: {:error, ":retry must be a non-negative integer or :infinity, got: #{inspect(retry)}"} + defp connect_opts(%Channel{scheme: "https"} = channel, opts) do %Credential{ssl: ssl} = Map.get(channel, :cred) || %Credential{} @@ -177,6 +172,27 @@ if Code.ensure_loaded?(Mint.HTTP) do defp mint_scheme(%Channel{scheme: "https"} = _channel), do: :https defp mint_scheme(_channel), do: :http + defp start_connection_process(channel, opts, retry) do + {config_opts, opts} = Keyword.pop(opts, :config_options, []) + module_opts = Application.get_env(:grpc, __MODULE__, config_opts) + + opts = + channel + |> connect_opts(opts) + |> merge_opts(module_opts) + |> Keyword.put(:retry, retry) + + Process.flag(:trap_exit, true) + + channel + |> mint_scheme() + |> ConnectionProcess.start_link(channel.host, channel.port, opts) + |> case do + {:ok, _} = ok -> ok + error -> {:error, "Error while opening connection: #{inspect(error)}"} + end + end + defp do_receive_data(%{payload: %{stream_response_pid: pid}}, request_type, opts) when request_type in [:bidirectional_stream, :server_stream] do produce_trailers? = opts[:return_headers] == true diff --git a/grpc/lib/grpc/client/adapters/mint/connection_process/connection_process.ex b/grpc/lib/grpc/client/adapters/mint/connection_process/connection_process.ex index 1b60b740b..4f7694f69 100644 --- a/grpc/lib/grpc/client/adapters/mint/connection_process/connection_process.ex +++ b/grpc/lib/grpc/client/adapters/mint/connection_process/connection_process.ex @@ -187,7 +187,7 @@ if Code.ensure_loaded?(Mint.HTTP) do @impl true def handle_info(:reconnect, state) do - attempt_reconnect(state) + maybe_attempt_reconnect(state) end def handle_info(message, state) do @@ -444,12 +444,7 @@ if Code.ensure_loaded?(Mint.HTTP) do clean_state = State.update_request_stream_queue(%{new_state | requests: %{}}, :queue.new()) - if clean_state.retry > 0 do - attempt_reconnect(clean_state) - else - send(clean_state.parent, {:elixir_grpc, :connection_down, self()}) - {:noreply, clean_state} - end + maybe_attempt_reconnect(clean_state) end defp end_stream_response(pid, error) do @@ -457,10 +452,15 @@ if Code.ensure_loaded?(Mint.HTTP) do StreamResponseProcess.done(pid) end - defp attempt_reconnect(%{retry: max, retry_attempt: attempt} = state) - when attempt >= max do + defp maybe_attempt_reconnect(%{retry: 0} = state) do + send(state.parent, {:elixir_grpc, :connection_down, self()}) + {:noreply, state} + end + + defp maybe_attempt_reconnect(%{retry_attempt: attempt} = state) + when State.retries_exhausted?(state) do Logger.warning( - "Connection retry exhausted (#{attempt}/#{max}) for #{state.scheme}://#{state.host}:#{state.port}" + "Connection retry exhausted (#{attempt}/#{state.retry}) for #{state.scheme}://#{state.host}:#{state.port}" ) :telemetry.execute( @@ -473,7 +473,7 @@ if Code.ensure_loaded?(Mint.HTTP) do {:noreply, state} end - defp attempt_reconnect(state) do + defp maybe_attempt_reconnect(state) do next_attempt = state.retry_attempt + 1 Logger.info( diff --git a/grpc/lib/grpc/client/adapters/mint/connection_process/state.ex b/grpc/lib/grpc/client/adapters/mint/connection_process/state.ex index 6afab3ee9..078331ca6 100644 --- a/grpc/lib/grpc/client/adapters/mint/connection_process/state.ex +++ b/grpc/lib/grpc/client/adapters/mint/connection_process/state.ex @@ -29,6 +29,9 @@ if Code.ensure_loaded?(Mint.HTTP) do } end + defguard retries_exhausted?(state) + when state.retry != :infinity and state.retry_attempt >= state.retry + def update_conn(state, conn) do %{state | conn: conn} end diff --git a/grpc/lib/grpc/client/connection.ex b/grpc/lib/grpc/client/connection.ex index c9cdbb38c..79fef6e60 100644 --- a/grpc/lib/grpc/client/connection.ex +++ b/grpc/lib/grpc/client/connection.ex @@ -1231,7 +1231,7 @@ defmodule GRPC.Client.Connection do resolver = Keyword.get(opts, :resolver, GRPC.Client.Resolver) adapter = Keyword.get(opts, :adapter, GRPC.Client.Adapters.Gun) - validate_adapter_opts!(opts[:adapter_opts]) + validate_adapter_opts!(adapter, opts[:adapter_opts]) {norm_target, norm_opts, scheme} = normalize_target_and_opts(target, opts) cred = resolve_credential(norm_opts[:cred], scheme) @@ -1268,9 +1268,18 @@ defmodule GRPC.Client.Connection do defp resolve_credential(nil, _scheme), do: nil defp resolve_credential(other, _scheme), do: other - defp validate_adapter_opts!(opts) when is_list(opts), do: :ok + defp validate_adapter_opts!(adapter, opts) when is_list(opts) do + if Code.ensure_loaded?(adapter) and function_exported?(adapter, :validate_opts, 1) do + case adapter.validate_opts(opts) do + :ok -> :ok + {:error, message} -> raise ArgumentError, message + end + end + + :ok + end - defp validate_adapter_opts!(_), + defp validate_adapter_opts!(_adapter, _), do: raise(ArgumentError, ":adapter_opts must be a keyword list if present") defp build_compressor_list(compressor, accepted) when is_list(accepted) do diff --git a/grpc/test/grpc/adapters/mint/connection_process_test.exs b/grpc/test/grpc/adapters/mint/connection_process_test.exs index eadad2ac0..f4cb95949 100644 --- a/grpc/test/grpc/adapters/mint/connection_process_test.exs +++ b/grpc/test/grpc/adapters/mint/connection_process_test.exs @@ -395,6 +395,7 @@ defmodule GRPC.Client.Adapters.Mint.ConnectionProcessTest do describe "handle_info - connection_closed - no requests" do setup :valid_connection + setup :attach_reconnect_telemetry test "send a message to parent process to inform the connection is down", %{ state: state @@ -419,6 +420,7 @@ defmodule GRPC.Client.Adapters.Mint.ConnectionProcessTest do assert new_state.conn.state == :closed assert new_state.retry == 0 assert_receive {:elixir_grpc, :connection_down, _pid}, 500 + refute_received {:telemetry, [:grpc, :client, :mint, :reconnect, :exhausted], _, _} end end @@ -468,6 +470,22 @@ defmodule GRPC.Client.Adapters.Mint.ConnectionProcessTest do end end + describe "handle_info - connection_closed - retry: :infinity" do + setup :valid_connection_with_infinite_retry + setup :attach_reconnect_telemetry + + test "reconnects when the connection drops", %{state: state} do + tcp_message = {:tcp_closed, state.conn.socket} + + assert {:noreply, new_state} = ConnectionProcess.handle_info(tcp_message, state) + assert Mint.HTTP.open?(new_state.conn) + assert new_state.retry_attempt == 0 + + refute_received {:elixir_grpc, :connection_down, _pid} + refute_received {:telemetry, [:grpc, :client, :mint, :reconnect, :exhausted], _, _} + end + end + describe "handle_info - connection_closed - with request" do setup :valid_connection setup :valid_stream_request @@ -833,6 +851,8 @@ defmodule GRPC.Client.Adapters.Mint.ConnectionProcessTest do defp valid_connection_with_retry(ctx), do: valid_connection(ctx, retry: 3) + defp valid_connection_with_infinite_retry(ctx), do: valid_connection(ctx, retry: :infinity) + defp attach_reconnect_telemetry(_ctx) do test_pid = self() handler_id = "test-reconnect-telemetry-#{inspect(test_pid)}" diff --git a/grpc/test/grpc/adapters/mint_test.exs b/grpc/test/grpc/adapters/mint_test.exs index ede5b958b..7ee74acf5 100644 --- a/grpc/test/grpc/adapters/mint_test.exs +++ b/grpc/test/grpc/adapters/mint_test.exs @@ -249,5 +249,44 @@ defmodule GRPC.Client.Adapters.MintTest do assert state.retry == 0 end + + test "accepts :infinity as the retry budget", %{port: port} do + channel = build(:channel, adapter: Mint, port: port, host: "localhost") + + {:ok, connected} = Mint.connect(channel, retry: :infinity) + state = :sys.get_state(connected.adapter_payload.conn_pid) + + assert state.retry == :infinity + end + + test "rejects a retry budget that is neither a non-negative integer nor :infinity", %{ + port: port + } do + channel = build(:channel, adapter: Mint, port: port, host: "localhost") + + assert {:error, message} = Mint.connect(channel, retry: -1) + assert message =~ ":retry must be a non-negative integer or :infinity" + + assert {:error, message} = Mint.connect(channel, retry: :forever) + assert message =~ ":retry must be a non-negative integer or :infinity" + end + end + + describe "validate_opts/1" do + test "accepts an unset retry, a non-negative integer or :infinity" do + assert :ok == Mint.validate_opts([]) + assert :ok == Mint.validate_opts(retry: nil) + assert :ok == Mint.validate_opts(retry: 0) + assert :ok == Mint.validate_opts(retry: 3) + assert :ok == Mint.validate_opts(retry: :infinity) + end + + test "returns an error for anything else" do + assert {:error, message} = Mint.validate_opts(retry: -1) + assert message =~ ":retry must be a non-negative integer or :infinity" + + assert {:error, message} = Mint.validate_opts(retry: :forever) + assert message =~ ":retry must be a non-negative integer or :infinity" + end end end diff --git a/grpc/test/grpc/client/connection_test.exs b/grpc/test/grpc/client/connection_test.exs index 9615db94c..b8aef6b2d 100644 --- a/grpc/test/grpc/client/connection_test.exs +++ b/grpc/test/grpc/client/connection_test.exs @@ -332,6 +332,15 @@ defmodule GRPC.Client.ConnectionTest do end end + test "raises in the caller on invalid adapter-specific options" do + assert_raise ArgumentError, ~r/:retry must be a non-negative integer or :infinity/, fn -> + Connection.connect("ipv4:127.0.0.1:50051", + adapter: GRPC.Client.Adapters.Mint, + adapter_opts: [retry: -1] + ) + end + end + test "interceptor init/1 runs once per connect", %{ ref: ref, target: target,