fix(mint): add keepalive feature using HTTP/2 pings

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**

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`.

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

grpc/lib/grpc/client/adapters/mint.ex

@@ -40,18 +40,24 @@ 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 =
         channel
         |> connect_opts(opts)
         |> merge_opts(module_opts)
         |> Keyword.put(:retry, retry)
+        |> Keyword.merge(keepalive_opts)
 
       Process.flag(:trap_exit, true)
 

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,14 +191,23 @@ 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 ->
           Logger.debug(fn -> "Received unknown message: " <> inspect(message) end)
           {: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} ->

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
 

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

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")