From 6a87dcab4f50c5b747212b5b0d689a3f5d8ac084 Mon Sep 17 00:00:00 2001 From: VovaGula Date: Fri, 14 Aug 2026 10:23:16 +0200 Subject: [PATCH 1/2] fix(mint): enforce the requested timeout on unary receives MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Mint adapter accepts and parses `:timeout` but never applies it: unary receives wait on the stream response process with `:infinity`, so the option only ever reaches the server as the `grpc-timeout` header. When no response can arrive — the connection going down without notifying the pending request, for instance — the caller blocks forever even though it asked for a deadline. Pass the requested timeout down to `build_stream/3` and translate an elapsed deadline into `DEADLINE_EXCEEDED`. Along the way: * `:deadline` now takes precedence over `:timeout`, as documented. `recv/2` always fills in the 10s default under `:timeout`, so an explicit deadline could otherwise never take effect. * Both options are resolved by `GRPC.TimeUtils.to_relative/2`, which returns a float — and a negative one for a deadline already in the past — so the value is rounded and clamped before it reaches a receive timeout. * Giving up resets the request, so the server stops working on it, and stops the process buffering the response, which is linked to the caller rather than to the connection and would otherwise outlive the call. Both steps are bounded and best effort: they run after the deadline elapsed, against processes that may themselves be gone or wedged. 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. A client stream awaits its response through a separate `recv/2` call, which `GRPC.Stub` documents as unbounded even though it fills in the same 10s default — worth settling separately from this fix. Note this changes the default behaviour of a unary call that never receives a response: it now fails after the 10s documented in `GRPC.Stub.call/5` instead of blocking indefinitely. --- grpc/lib/grpc/client/adapters/mint.ex | 75 ++++++++++++++-- .../connection_process/connection_process.ex | 7 +- .../adapters/mint/stream_response_process.ex | 44 ++++++--- .../mint/stream_response_process_test.exs | 24 +++++ grpc/test/grpc/adapters/mint_test.exs | 90 +++++++++++++++++++ 5 files changed, 218 insertions(+), 22 deletions(-) diff --git a/grpc/lib/grpc/client/adapters/mint.ex b/grpc/lib/grpc/client/adapters/mint.ex index 8be79bd38..bfa8414e6 100644 --- a/grpc/lib/grpc/client/adapters/mint.ex +++ b/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( 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 610630e65..1b60b740b 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 @@ -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 diff --git a/grpc/lib/grpc/client/adapters/mint/stream_response_process.ex b/grpc/lib/grpc/client/adapters/mint/stream_response_process.ex index fe6d37a97..f6510efc3 100644 --- a/grpc/lib/grpc/client/adapters/mint/stream_response_process.ex +++ b/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 diff --git a/grpc/test/grpc/adapters/mint/stream_response_process_test.exs b/grpc/test/grpc/adapters/mint/stream_response_process_test.exs index e67701102..1dc0b6ebe 100644 --- a/grpc/test/grpc/adapters/mint/stream_response_process_test.exs +++ b/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 diff --git a/grpc/test/grpc/adapters/mint_test.exs b/grpc/test/grpc/adapters/mint_test.exs index e7af10a09..e5e23e633 100644 --- a/grpc/test/grpc/adapters/mint_test.exs +++ b/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") From 684774e15e25f43cdc650ee5eb5aeee25f90e128 Mon Sep 17 00:00:00 2001 From: VovaGula Date: Fri, 14 Aug 2026 13:24:48 +0200 Subject: [PATCH 2/2] docs(changelog): record the unary receive deadline change The enforced deadline alters a documented default, which is the kind of change this changelog records under `### Behavior Changes`. There was no unreleased heading to file it under, so add one rather than assume the next version number. --- grpc/CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/grpc/CHANGELOG.md b/grpc/CHANGELOG.md index 9b61b4a46..735a32c3e 100644 --- a/grpc/CHANGELOG.md +++ b/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