From 7c71931d0a793bc6322b1c05d12349ec2524dfa5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arran=20=C3=98ystein=20Kostveit=20Gabriel?= Date: Thu, 8 Oct 2026 09:42:57 +0200 Subject: [PATCH] fix(mint): add keepalive feature using HTTP/2 pings --- README.md | 10 +++ grpc/CHANGELOG.md | 5 ++ grpc/README.md | 10 +++ grpc/lib/grpc/client/adapters/mint.ex | 6 ++ .../connection_process/connection_process.ex | 63 +++++++++++++++- .../adapters/mint/connection_process/state.ex | 9 ++- .../adapters/mint/connection_process_test.exs | 75 ++++++++++++++++++- grpc/test/grpc/adapters/mint_test.exs | 27 +++++++ 8 files changed, 198 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 2b77d232..b9670acd 100644 --- a/README.md +++ b/README.md @@ -432,6 +432,16 @@ By default, `:retry` is `0` (no reconnection attempts). > **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. +#### Keepalive + +To detect connections that died without being closed, the Mint adapter can send HTTP/2 pings: + +```elixir +adapter_opts: [keepalive: 30_000, keepalive_tolerance: 2] +``` + +This pings every 30 seconds and closes the connection, triggering `:retry`, when 2 pings are left unacknowledged. + --- ### **HTTP Transcoding** diff --git a/grpc/CHANGELOG.md b/grpc/CHANGELOG.md index 495cd22e..71265003 100644 --- a/grpc/CHANGELOG.md +++ b/grpc/CHANGELOG.md @@ -2,6 +2,11 @@ ## Unreleased +### Enhancements + +- The Mint adapter can now be configured to send HTTP/2 keepalive pings. `:keepalive` sets the interval in milliseconds, and `:keepalive_tolerance` the number of unacknowledged pings after which the connection is treated as dead, and forcibly closed. + Both default to `nil`, keeping the previous behavior. + ### 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`. diff --git a/grpc/README.md b/grpc/README.md index bb519c90..99e925dd 100644 --- a/grpc/README.md +++ b/grpc/README.md @@ -222,6 +222,16 @@ By default, `:retry` is `0` (no reconnection attempts). > **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. +#### Keepalive + +To detect connections that died without being closed, the Mint adapter can send HTTP/2 pings: + +```elixir +adapter_opts: [keepalive: 30_000, keepalive_tolerance: 2] +``` + +This pings every 30 seconds and closes the connection, triggering `:retry`, when 2 pings are left unacknowledged. + --- ## Contributing diff --git a/grpc/lib/grpc/client/adapters/mint.ex b/grpc/lib/grpc/client/adapters/mint.ex index b37518c2..30c5a794 100644 --- a/grpc/lib/grpc/client/adapters/mint.ex +++ b/grpc/lib/grpc/client/adapters/mint.ex @@ -40,11 +40,16 @@ if Code.ensure_loaded?(Mint.HTTP) do 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). Uses exponential backoff with jitter between attempts. + * `:keepalive`: Interval in milliseconds between HTTP/2 PING frames sent to the server. + Defaults to `nil` (no pings). + * `:keepalive_tolerance`: Number of unacknowledged pings after which the connection is closed, + triggering `:retry` if configured. Defaults to `nil` (never close). Has no effect without `:keepalive`. """ @impl true def connect(%{host: host, port: port} = channel, opts \\ []) do {config_opts, opts} = Keyword.pop(opts, :config_options, []) {retry, opts} = Keyword.pop(opts, :retry, 0) + {keepalive_opts, opts} = Keyword.split(opts, [:keepalive, :keepalive_tolerance]) module_opts = Application.get_env(:grpc, __MODULE__, config_opts) opts = @@ -52,6 +57,7 @@ if Code.ensure_loaded?(Mint.HTTP) do |> connect_opts(opts) |> merge_opts(module_opts) |> Keyword.put(:retry, retry) + |> Keyword.merge(keepalive_opts) Process.flag(:trap_exit, 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 0e865d65..52d2b2ea 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 @@ -69,6 +69,7 @@ if Code.ensure_loaded?(Mint.HTTP) do @impl true def init({scheme, host, port, opts}) do {retry, opts} = Keyword.pop(opts, :retry, 0) + {keepalive_opts, opts} = Keyword.split(opts, [:keepalive, :keepalive_tolerance]) case Mint.HTTP.connect(scheme, host, port, opts) do {:ok, conn} -> @@ -81,7 +82,7 @@ if Code.ensure_loaded?(Mint.HTTP) do retry: retry ] - {:ok, State.new(conn, state_opts)} + {:ok, conn |> State.new(state_opts ++ keepalive_opts) |> schedule_keepalive()} {:error, reason} -> Logger.error( @@ -190,6 +191,12 @@ if Code.ensure_loaded?(Mint.HTTP) do attempt_reconnect(state) end + def handle_info(:keepalive, state) do + state + |> schedule_keepalive() + |> handle_keepalive() + end + def handle_info(message, state) do case Mint.HTTP.stream(state.conn, message) do :unknown -> @@ -197,7 +204,10 @@ if Code.ensure_loaded?(Mint.HTTP) do {:noreply, state} {:ok, conn, responses} -> - state = State.update_conn(state, conn) + state = + state + |> State.update_conn(conn) + |> acknowledge_pongs(responses) state = case state.requests do @@ -241,6 +251,53 @@ if Code.ensure_loaded?(Mint.HTTP) do :normal end + defp schedule_keepalive(%{keepalive: interval} = state) when is_integer(interval) do + Process.send_after(self(), :keepalive, interval) + state + end + + defp schedule_keepalive(state), do: state + + defp handle_keepalive(%{conn: %Mint.HTTP2{state: :closed}} = state), do: {:noreply, state} + + defp handle_keepalive(%{pings_unacked: unacked, keepalive_tolerance: tolerance} = state) + when is_integer(tolerance) and unacked >= tolerance do + Logger.warning( + "Closing connection to #{state.scheme}://#{state.host}:#{state.port} after #{unacked} unacknowledged keepalive pings" + ) + + close_connection(state) + end + + defp handle_keepalive(state) do + case Mint.HTTP2.ping(state.conn) do + {:ok, conn, _ref} -> + {:noreply, %{State.update_conn(state, conn) | pings_unacked: state.pings_unacked + 1}} + + {:error, conn, reason} -> + Logger.warning( + "Closing connection to #{state.scheme}://#{state.host}:#{state.port} after failing to send keepalive ping: #{inspect(reason)}" + ) + + state + |> State.update_conn(conn) + |> close_connection() + end + end + + defp close_connection(state) do + {:ok, conn} = Mint.HTTP.close(state.conn) + + state + |> State.update_conn(conn) + |> finish_all_pending_requests() + end + + defp acknowledge_pongs(state, responses) do + pongs = Enum.count(responses, &match?({:pong, _ref}, &1)) + %{state | pings_unacked: state.pings_unacked - pongs} + end + # Frames may still arrive for a stream whose state was already dropped # (e.g. cancelled after its response process died mid-batch). Ignore them # instead of crashing the whole connection. @@ -492,7 +549,7 @@ if Code.ensure_loaded?(Mint.HTTP) do reconnect_metadata(state, next_attempt) ) - new_state = %{state | conn: conn, retry_attempt: 0} + new_state = %{state | conn: conn, retry_attempt: 0, pings_unacked: 0} {:noreply, new_state} {:error, reason} -> 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 6afab3ee..09cc962a 100644 --- a/grpc/lib/grpc/client/adapters/mint/connection_process/state.ex +++ b/grpc/lib/grpc/client/adapters/mint/connection_process/state.ex @@ -10,10 +10,13 @@ if Code.ensure_loaded?(Mint.HTTP) do :port, :connect_opts, :retry_timeout_ms, + :keepalive, + :keepalive_tolerance, requests: %{}, request_stream_queue: :queue.new(), retry: 0, - retry_attempt: 0 + retry_attempt: 0, + pings_unacked: 0 ] def new(conn, opts) do @@ -25,7 +28,9 @@ if Code.ensure_loaded?(Mint.HTTP) do host: opts[:host], port: opts[:port], connect_opts: opts[:connect_opts] || [], - retry: opts[:retry] || 0 + retry: opts[:retry] || 0, + keepalive: opts[:keepalive], + keepalive_tolerance: opts[:keepalive_tolerance] } end diff --git a/grpc/test/grpc/adapters/mint/connection_process_test.exs b/grpc/test/grpc/adapters/mint/connection_process_test.exs index d097d561..15dab8cf 100644 --- a/grpc/test/grpc/adapters/mint/connection_process_test.exs +++ b/grpc/test/grpc/adapters/mint/connection_process_test.exs @@ -625,6 +625,71 @@ defmodule GRPC.Client.Adapters.Mint.ConnectionProcessTest do end end + describe "keepalive" do + test "keeps the connection open while the server acknowledges pings", %{port: port} do + %{process_pid: pid} = valid_connection(%{port: port}, keepalive: 20, keepalive_tolerance: 1) + Process.sleep(250) + assert Mint.HTTP.open?(:sys.get_state(pid).conn) + end + + @tag :capture_log + test "closes the connection when the server stops acknowledging pings", ctx do + %{process_pid: pid} = quiet_connection(ctx, keepalive: 20, keepalive_tolerance: 2) + assert_receive {:elixir_grpc, :connection_down, ^pid}, 1_000 + end + end + + describe "handle_info - :keepalive" do + setup :quiet_connection + + test "each pong acknowledges one ping", %{state: state} do + ping_ack = {:tcp, state.conn.socket, <<8::24, 0x06, 0x01, 0::32, 0::64>>} + + {:noreply, state} = ConnectionProcess.handle_info(:keepalive, state) + + assert {:noreply, %{pings_unacked: 2} = state} = + ConnectionProcess.handle_info(:keepalive, state) + + assert {:noreply, %{pings_unacked: 1} = state} = + ConnectionProcess.handle_info(ping_ack, state) + + assert {:noreply, %{pings_unacked: 0}} = ConnectionProcess.handle_info(ping_ack, state) + end + + @tag :capture_log + test "closes the connection when the ping cannot be sent", %{state: state} do + :ok = :gen_tcp.close(state.conn.socket) + + assert {:stop, :normal, %{pings_unacked: 0, conn: conn}} = + ConnectionProcess.handle_info(:keepalive, state) + + refute Mint.HTTP.open?(conn) + assert_receive {:elixir_grpc, :connection_down, _pid} + end + end + + describe "handle_info - :keepalive - with retry" do + setup ctx, do: valid_connection(ctx, retry: 3, keepalive: 60_000, keepalive_tolerance: 1) + + @tag :capture_log + test "reconnects with a fresh ping count when the tolerance is exceeded", %{state: state} do + state = %{state | pings_unacked: 1} + + assert {:noreply, %{pings_unacked: 0, conn: conn}} = + ConnectionProcess.handle_info(:keepalive, state) + + assert Mint.HTTP.open?(conn) + refute_received {:elixir_grpc, :connection_down, _pid} + end + + test "does nothing while reconnecting", %{state: state} do + {:ok, conn} = Mint.HTTP.close(state.conn) + state = %{state | conn: conn} + + assert {:noreply, ^state} = ConnectionProcess.handle_info(:keepalive, state) + end + end + describe "retry_timeout/1" do test "returns exponentially increasing timeouts" do t1 = ConnectionProcess.retry_timeout(1) @@ -779,7 +844,7 @@ defmodule GRPC.Client.Adapters.Mint.ConnectionProcessTest do # Like valid_connection/2, but against a bare TCP server that only completes # the HTTP/2 preface (server SETTINGS + ack) and then stays silent — so no # real server frames race the ones injected by the test. - defp quiet_connection(_ctx) do + defp quiet_connection(_ctx, opts \\ []) do {:ok, listen_socket} = :gen_tcp.listen(0, [:binary, active: false, reuseaddr: true]) {:ok, port} = :inet.port(listen_socket) @@ -795,7 +860,13 @@ defmodule GRPC.Client.Adapters.Mint.ConnectionProcessTest do on_exit(fn -> :gen_tcp.close(listen_socket) end) - {:ok, pid} = ConnectionProcess.start_link(:http, "localhost", port, protocols: [:http2]) + {:ok, pid} = + ConnectionProcess.start_link( + :http, + "localhost", + port, + Keyword.merge([protocols: [:http2]], opts) + ) # The state snapshot must only be taken once the connection has processed # the server preface: HTTP/2 requires SETTINGS to be the first server diff --git a/grpc/test/grpc/adapters/mint_test.exs b/grpc/test/grpc/adapters/mint_test.exs index 52c52654..ff4921ba 100644 --- a/grpc/test/grpc/adapters/mint_test.exs +++ b/grpc/test/grpc/adapters/mint_test.exs @@ -83,6 +83,33 @@ defmodule GRPC.Client.Adapters.MintTest do end end + describe "connect/2 - keepalive" do + setup(%{port: port}) do + %{channel: build(:channel, adapter: Mint, port: port, host: "localhost")} + end + + test "is disabled by default", %{channel: channel} do + assert {:ok, result} = Mint.connect(channel, []) + + assert %{keepalive: nil, keepalive_tolerance: nil} = + :sys.get_state(result.adapter_payload.conn_pid) + end + + test "passes keepalive options to the connection process", %{channel: channel} do + keepalive = 15_000 + keepalive_tolerance = 3 + + assert {:ok, result} = + Mint.connect(channel, + keepalive: keepalive, + keepalive_tolerance: keepalive_tolerance + ) + + assert %{keepalive: ^keepalive, keepalive_tolerance: ^keepalive_tolerance} = + :sys.get_state(result.adapter_payload.conn_pid) + end + end + describe "disconnect/1" do test "keeps adapter_payload as a map with conn_pid set to nil", %{port: port} do channel = build(:channel, adapter: Mint, port: port, host: "localhost")