diff --git a/grpc/CHANGELOG.md b/grpc/CHANGELOG.md index b09f98cf..ca5eac2b 100644 --- a/grpc/CHANGELOG.md +++ b/grpc/CHANGELOG.md @@ -1,5 +1,10 @@ # 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.4 (2026-0-15) ### Bug Fixes diff --git a/grpc/lib/grpc/client/adapters/mint.ex b/grpc/lib/grpc/client/adapters/mint.ex index 8be79bd3..bfa8414e 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 610630e6..1b60b740 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 fe6d37a9..f6510efc 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 e6770110..1dc0b6eb 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 e7af10a0..e5e23e63 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")