Fix/mint client side timeout

grpc/CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+### 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`.
+
 ## v1.0.3 (2026-07-27)
 
 ### Enhancements

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

@@ -27,6 +27,7 @@ if Code.ensure_loaded?(Mint.HTTP) do
       max_frame_size: 8_000_000
     ]
     @default_transport_opts [timeout: :infinity]
+    @cleanup_timeout 1_000
 
     @doc """
     Connects using Mint based on the provided configs. Options
@@ -192,24 +193,80 @@ if Code.ensure_loaded?(Mint.HTTP) do
     end
 
     defp do_receive_data(
-           %{payload: %{stream_response_pid: pid}},
+           %{payload: %{stream_response_pid: pid}} = stream,
            request_type,
            opts
          )
          when request_type in [:client_stream, :unary] do
-      responses = pid |> StreamResponseProcess.build_stream() |> Enum.to_list()
+      responses =
+        pid
+        |> StreamResponseProcess.build_stream(true, recv_timeout(request_type, opts))
+        |> Enum.to_list()
 
-      with :ok <- check_for_error(responses) do
-        data = Keyword.fetch!(responses, :ok)
+      case check_for_error(responses) do
+        :ok ->
+          data = Keyword.fetch!(responses, :ok)
 
-        if opts[:return_headers] do
-          {:ok, data, get_headers_and_trailers(responses)}
-        else
-          {:ok, data}
-        end
+          if opts[:return_headers] do
+            {:ok, data, get_headers_and_trailers(responses)}
+          else
+            {:ok, data}
+          end
+
+        {:error, :deadline_exceeded} ->
+          give_up_on_request(stream, pid)
+
+          {:error, GRPC.RPCError.exception(GRPC.Status.deadline_exceeded(), "deadline exceeded")}
+
+        {:error, error} ->
+          {:error, error}
       end
     end
 
+    # Only unary receives are bounded. Server and bidirectional streams are
+    # consumed lazily by the caller, where a gap between messages is expected
+    # rather than a failure, and a client stream awaits its response through a
+    # separate `GRPC.Stub.recv/2` call, which `GRPC.Stub` documents as unbounded
+    # even though it fills in the same 10s default.
+    defp recv_timeout(:unary, opts) do
+      # An explicit `:deadline` wins: `GRPC.Stub` always fills in a `:timeout`,
+      # so a deadline could never take effect otherwise. Both arrive here as a
+      # float number of milliseconds, already negative for a deadline in the
+      # past, while a receive timeout has to be a non-negative integer.
+      case opts[:deadline] || opts[:timeout] do
+        nil -> :infinity
+        :infinity -> :infinity
+        milliseconds when is_number(milliseconds) -> max(0, round(milliseconds))
+      end
+    end
+
+    defp recv_timeout(_request_type, _opts), do: :infinity
+
+    # The caller stopped waiting on a request that is still open. Reset it so the
+    # server stops working on it, and stop the process that buffers the response:
+    # it is linked to the caller rather than to the connection, so nothing else
+    # shuts it down — least of all when the connection process is the very thing
+    # that went away.
+    defp give_up_on_request(stream, stream_response_pid) do
+      %{
+        channel: %{adapter_payload: %{conn_pid: conn_pid}},
+        payload: %{response: {:ok, %{request_ref: request_ref}}}
+      } = stream
+
+      quietly(fn -> ConnectionProcess.cancel(conn_pid, request_ref, @cleanup_timeout) end)
+      quietly(fn -> GenServer.stop(stream_response_pid, :normal, @cleanup_timeout) end)
+    end
+
+    # Cleanup runs after the deadline already elapsed, against processes that may
+    # be gone or wedged — which is what the deadline exists for. Failing to tidy
+    # up must neither replace the error the caller is about to get nor keep it
+    # waiting much longer.
+    defp quietly(cleanup) do
+      cleanup.()
+    catch
+      :exit, _reason -> :ok
+    end
+
     def handle_errors_receive_data(%GRPC.Client.Stream{payload: %{response: response}}, _opts) do
       {:error,
        GRPC.RPCError.exception(

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

@@ -56,9 +56,12 @@ if Code.ensure_loaded?(Mint.HTTP) do
 
     @doc """
     cancels an open request request
+
+    `timeout` bounds the wait for the connection process to acknowledge the
+    cancellation, for callers that are already past a deadline of their own.
     """
-    def cancel(pid, request_ref) do
-      GenServer.call(pid, {:cancel_request, request_ref})
+    def cancel(pid, request_ref, timeout \\ 5_000) do
+      GenServer.call(pid, {:cancel_request, request_ref}, timeout)
     end
 
     ## Callbacks

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

@@ -29,25 +29,47 @@ defmodule GRPC.Client.Adapters.Mint.StreamResponseProcess do
 
   @doc """
   Given a pid from this process, build an Elixir.Stream that will consume the accumulated
-  data inside this process
+  data inside this process.
+
+  `timeout` bounds how long the stream waits for each response. When it elapses
+  the stream emits `{:error, :deadline_exceeded}` and halts, rather than waiting
+  for a response that may never arrive — the connection process notifies this
+  process, so anything that stops it from doing so (a connection that goes down
+  without cleaning up, for instance) would otherwise block the caller forever.
+  Defaults to `:infinity` to keep the previous behaviour for callers that do not
+  ask for a deadline.
   """
-  def build_stream(pid, produce_trailers? \\ true) do
-    Stream.unfold(pid, fn pid ->
-      pid
-      |> GenServer.call(:get_response, :infinity)
-      |> process_response(produce_trailers?, pid)
+  def build_stream(pid, produce_trailers? \\ true, timeout \\ :infinity) do
+    Stream.unfold(pid, fn
+      :halt ->
+        nil
+
+      pid ->
+        pid
+        |> get_response(timeout)
+        |> process_response(produce_trailers?, pid, timeout)
     end)
   end
 
-  defp process_response(nil = _response, _produce_trailers, _pid), do: nil
+  defp get_response(pid, timeout) do
+    GenServer.call(pid, :get_response, timeout)
+  catch
+    :exit, {:timeout, {GenServer, :call, _args}} -> :deadline_exceeded
+  end
+
+  defp process_response(nil = _response, _produce_trailers, _pid, _timeout), do: nil
+
+  defp process_response(:deadline_exceeded, _produce_trailers, _pid, _timeout) do
+    {{:error, :deadline_exceeded}, :halt}
+  end
 
-  defp process_response({:trailers, _trailers}, false = produce_trailers?, pid) do
+  defp process_response({:trailers, _trailers}, false = produce_trailers?, pid, timeout) do
     pid
-    |> GenServer.call(:get_response, :infinity)
-    |> process_response(produce_trailers?, pid)
+    |> get_response(timeout)
+    |> process_response(produce_trailers?, pid, timeout)
   end
 
-  defp process_response(response, _produce_trailers, pid) do
+  defp process_response(response, _produce_trailers, pid, _timeout) do
     {response, pid}
   end
 

grpc/test/grpc/adapters/mint/stream_response_process_test.exs

@@ -451,4 +451,28 @@ defmodule GRPC.Client.Adapters.Mint.StreamResponseProcessTest do
       do: [{:headers}, {:trailers}]
     )
   end
+
+  describe "build_stream/3 - deadline" do
+    setup do
+      {:ok, pid} = StreamResponseProcess.start_link(build(:client_stream), true)
+
+      %{pid: pid}
+    end
+
+    test "emits an error and halts when no response arrives in time", %{pid: pid} do
+      stream = StreamResponseProcess.build_stream(pid, true, 10)
+
+      assert Enum.to_list(stream) == [error: :deadline_exceeded]
+    end
+
+    test "does not emit an error when the response arrives within the deadline", %{pid: pid} do
+      data = <<0, 0, 0, 0, 12, 10, 10, 72, 101, 108, 108, 111, 32, 76, 117, 105, 115>>
+      stream = StreamResponseProcess.build_stream(pid, true, :timer.seconds(5))
+
+      StreamResponseProcess.consume(pid, :data, data)
+      StreamResponseProcess.done(pid)
+
+      assert Enum.to_list(stream) == [ok: build(:hello_reply_rpc)]
+    end
+  end
 end

grpc/test/grpc/adapters/mint_test.exs

@@ -137,6 +137,96 @@ defmodule GRPC.Client.Adapters.MintTest do
     end
   end
 
+  describe "receive_data/2 - deadline" do
+    setup do
+      {:ok, stream_response_pid} =
+        GRPC.Client.Adapters.Mint.StreamResponseProcess.start_link(build(:client_stream), true)
+
+      # A connection process that is already gone: nothing will ever notify the
+      # stream response process, which is what used to block the caller forever.
+      dead_conn_pid = spawn(fn -> :ok end)
+      ref = Process.monitor(dead_conn_pid)
+      assert_receive {:DOWN, ^ref, :process, ^dead_conn_pid, _reason}
+
+      stream =
+        build(:client_stream,
+          channel: build(:channel, adapter: Mint, adapter_payload: %{conn_pid: dead_conn_pid}),
+          payload: %{
+            stream_response_pid: stream_response_pid,
+            response: {:ok, %{request_ref: make_ref()}}
+          }
+        )
+
+      %{stream: stream, stream_response_pid: stream_response_pid}
+    end
+
+    # Without a deadline these would block until the ExUnit timeout, so keep that
+    # wait short enough to read as a failure rather than as a stuck suite.
+    @describetag timeout: 5_000
+
+    test "returns DEADLINE_EXCEEDED when no response arrives in time", %{stream: stream} do
+      assert {:error, %GRPC.RPCError{status: status, message: message}} =
+               Mint.receive_data(stream, timeout: 10)
+
+      assert status == GRPC.Status.deadline_exceeded()
+      assert message == "deadline exceeded"
+    end
+
+    test "stops the stream response process it gave up on", %{
+      stream: stream,
+      stream_response_pid: stream_response_pid
+    } do
+      assert {:error, %GRPC.RPCError{}} = Mint.receive_data(stream, timeout: 10)
+
+      refute Process.alive?(stream_response_pid)
+    end
+
+    test "accepts the float milliseconds a :deadline is resolved into", %{stream: stream} do
+      timeout = GRPC.TimeUtils.to_relative(DateTime.add(DateTime.utc_now(), 20, :millisecond))
+
+      assert is_float(timeout)
+
+      assert {:error, %GRPC.RPCError{status: status}} =
+               Mint.receive_data(stream, timeout: timeout)
+
+      assert status == GRPC.Status.deadline_exceeded()
+    end
+
+    test "treats a deadline that has already passed as an immediate one", %{stream: stream} do
+      timeout = GRPC.TimeUtils.to_relative(DateTime.add(DateTime.utc_now(), -5, :second))
+
+      assert timeout < 0
+
+      assert {:error, %GRPC.RPCError{status: status}} =
+               Mint.receive_data(stream, timeout: timeout)
+
+      assert status == GRPC.Status.deadline_exceeded()
+    end
+
+    test "lets an explicit :deadline override the timeout GRPC.Stub fills in", %{stream: stream} do
+      assert {:error, %GRPC.RPCError{status: status}} =
+               Mint.receive_data(stream, timeout: :timer.minutes(1), deadline: 10)
+
+      assert status == GRPC.Status.deadline_exceeded()
+    end
+  end
+
+  describe "receive_data/2 - deadline through GRPC.Stub" do
+    test "a :deadline on a unary call reaches the server instead of raising", %{port: port} do
+      {:ok, channel} = GRPC.Stub.connect("localhost:#{port}", adapter: Mint)
+      on_exit(fn -> GRPC.Stub.disconnect(channel) end)
+
+      point = %Routeguide.Point{latitude: 409_146_138, longitude: -746_188_906}
+
+      assert {:ok, feature} =
+               Routeguide.RouteGuide.Stub.get_feature(channel, point,
+                 deadline: DateTime.add(DateTime.utc_now(), 30, :second)
+               )
+
+      assert feature == %Routeguide.Feature{location: point, name: "409146138,-746188906"}
+    end
+  end
+
   describe "connect/2 with retry option" do
     test "passes retry option to ConnectionProcess state", %{port: port} do
       channel = build(:channel, adapter: Mint, port: port, host: "localhost")