diff --git a/grpc/CHANGELOG.md b/grpc/CHANGELOG.md index 0c53d9a0..39ba36d0 100644 --- a/grpc/CHANGELOG.md +++ b/grpc/CHANGELOG.md @@ -5,6 +5,7 @@ ### 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`. + * Connection errors under the Mint adapter now uniformly surface errors with gRPC status `UNAVAILABLE` instead of `UNKNOWN`. ### Bug Fixes diff --git a/grpc/lib/grpc/client/adapters/mint.ex b/grpc/lib/grpc/client/adapters/mint.ex index bfa8414e..585dc60b 100644 --- a/grpc/lib/grpc/client/adapters/mint.ex +++ b/grpc/lib/grpc/client/adapters/mint.ex @@ -218,8 +218,8 @@ if Code.ensure_loaded?(Mint.HTTP) do {:error, GRPC.RPCError.exception(GRPC.Status.deadline_exceeded(), "deadline exceeded")} - {:error, error} -> - {:error, error} + {:error, _reason} = error -> + {:error, rpc_error(error)} end end @@ -268,11 +268,46 @@ if Code.ensure_loaded?(Mint.HTTP) do end def handle_errors_receive_data(%GRPC.Client.Stream{payload: %{response: response}}, _opts) do - {:error, - GRPC.RPCError.exception( - GRPC.Status.unknown(), - "error occurred while receiving data: #{inspect(response)}" - )} + {:error, rpc_error(response)} + end + + # A connection that has dropped, or was never there, is a transient condition + # that should map to the gRPC UNAVAILABLE error status. + defp rpc_error({:error, %GRPC.RPCError{} = error}), do: error + + defp rpc_error({:error, %Mint.TransportError{} = error}) do + unavailable(error) + end + + defp rpc_error({:error, %Mint.HTTPError{reason: :closed} = error}) do + unavailable(error) + end + + defp rpc_error({:error, %Mint.HTTPError{reason: :closed_for_writing} = error}) do + unavailable(error) + end + + defp rpc_error({:error, %Mint.HTTPError{reason: :unprocessed} = error}) do + unavailable(error) + end + + defp rpc_error({:error, %Mint.HTTPError{reason: {:server_closed_connection, _, _}} = error}) do + unavailable(error) + end + + defp rpc_error({:error, :closed}) do + GRPC.RPCError.exception(GRPC.Status.unavailable(), "the connection is closed") + end + + defp rpc_error(response) do + GRPC.RPCError.exception( + GRPC.Status.unknown(), + "error occurred while receiving data: #{inspect(response)}" + ) + end + + defp unavailable(error) do + GRPC.RPCError.exception(GRPC.Status.unavailable(), Exception.message(error)) end defp success_response?(%GRPC.Client.Stream{ 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 1b60b740..32971301 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 @@ -15,7 +15,10 @@ if Code.ensure_loaded?(Mint.HTTP) do require Logger require State - @connection_closed_error "the connection is closed" + @connection_closed_error GRPC.RPCError.exception( + GRPC.Status.unavailable(), + "the connection is closed" + ) @stream_response_dead_event [:grpc, :client, :mint, :stream_response, :dead] @reconnect_stop_event [:grpc, :client, :mint, :reconnect, :stop] @reconnect_error_event [:grpc, :client, :mint, :reconnect, :error] @@ -107,7 +110,7 @@ if Code.ensure_loaded?(Mint.HTTP) do end def handle_call(_request, _from, %{conn: %Mint.HTTP2{state: :closed}} = state) do - {:reply, {:error, "the connection is closed"}, state} + {:reply, {:error, @connection_closed_error}, state} end def handle_call( diff --git a/grpc/test/grpc/adapters/mint/connection_process_test.exs b/grpc/test/grpc/adapters/mint/connection_process_test.exs index eadad2ac..3a5908a3 100644 --- a/grpc/test/grpc/adapters/mint/connection_process_test.exs +++ b/grpc/test/grpc/adapters/mint/connection_process_test.exs @@ -101,7 +101,11 @@ defmodule GRPC.Client.Adapters.Mint.ConnectionProcessTest do assert {:reply, {:error, error}, new_state} = response assert state.conn != new_state.conn - assert "the connection is closed" == error + + assert %GRPC.RPCError{ + status: GRPC.Status.unavailable(), + message: "the connection is closed" + } == error end test "returns error response when mint returns an error when starting stream request", %{ @@ -157,7 +161,11 @@ defmodule GRPC.Client.Adapters.Mint.ConnectionProcessTest do assert {:reply, {:error, error}, new_state} = response assert state.conn != new_state.conn - assert "the connection is closed" == error + + assert %GRPC.RPCError{ + status: GRPC.Status.unavailable(), + message: "the connection is closed" + } == error end test "returns error response when mint returns an error when starting stream request", %{ @@ -486,7 +494,14 @@ defmodule GRPC.Client.Adapters.Mint.ConnectionProcessTest do assert new_state.conn.state == :closed assert_receive {:elixir_grpc, :connection_down, pid}, 500 response_state = :sys.get_state(response_pid) - assert :queue.to_list(response_state.responses) == [error: "the connection is closed"] + + assert :queue.to_list(response_state.responses) == [ + error: %GRPC.RPCError{ + status: GRPC.Status.unavailable(), + message: "the connection is closed" + } + ] + assert true == response_state.done assert pid == self() end @@ -514,7 +529,14 @@ defmodule GRPC.Client.Adapters.Mint.ConnectionProcessTest do assert new_state.conn.state == :closed assert_receive {:elixir_grpc, :connection_down, pid}, 500 response_state = :sys.get_state(response_pid) - assert :queue.to_list(response_state.responses) == [error: "the connection is closed"] + + assert :queue.to_list(response_state.responses) == [ + error: %GRPC.RPCError{ + status: GRPC.Status.unavailable(), + message: "the connection is closed" + } + ] + assert true == response_state.done assert pid == self() end diff --git a/grpc/test/grpc/adapters/mint_test.exs b/grpc/test/grpc/adapters/mint_test.exs index ede5b958..8db9d055 100644 --- a/grpc/test/grpc/adapters/mint_test.exs +++ b/grpc/test/grpc/adapters/mint_test.exs @@ -1,6 +1,8 @@ defmodule GRPC.Client.Adapters.MintTest do use GRPC.Client.DataCase, async: false + alias Elixir.Mint.HTTPError + alias Elixir.Mint.TransportError alias GRPC.Client.Adapters.Mint setup do @@ -125,8 +127,51 @@ defmodule GRPC.Client.Adapters.MintTest do end describe "handle_errors_receive_data/2" do - test "returns a GRPC.RPCError with unknown status" do - response = {:error, :closed} + test "returns UNAVAILABLE when the connection is gone" do + stream = build(:client_stream, payload: %{response: {:error, :closed}}) + + assert {:error, %GRPC.RPCError{status: status, message: message}} = + Mint.handle_errors_receive_data(stream, []) + + assert status == GRPC.Status.unavailable() + assert message == "the connection is closed" + end + + test "returns UNAVAILABLE for a transport error" do + error = %TransportError{reason: :econnrefused} + stream = build(:client_stream, payload: %{response: {:error, error}}) + + assert {:error, %GRPC.RPCError{status: status, message: message}} = + Mint.handle_errors_receive_data(stream, []) + + assert status == GRPC.Status.unavailable() + assert message == Exception.message(error) + end + + test "returns UNAVAILABLE when the server closed the connection" do + error = %HTTPError{ + module: Elixir.Mint.HTTP2, + reason: {:server_closed_connection, :no_error, "shutting down"} + } + + stream = build(:client_stream, payload: %{response: {:error, error}}) + + assert {:error, %GRPC.RPCError{status: status, message: message}} = + Mint.handle_errors_receive_data(stream, []) + + assert status == GRPC.Status.unavailable() + assert message == Exception.message(error) + end + + test "passes an existing GRPC.RPCError through untouched" do + error = GRPC.RPCError.exception(GRPC.Status.unavailable(), "the connection is closed") + stream = build(:client_stream, payload: %{response: {:error, error}}) + + assert {:error, ^error} = Mint.handle_errors_receive_data(stream, []) + end + + test "returns UNKNOWN for a reason it cannot classify" do + response = {:error, :something_else} stream = build(:client_stream, payload: %{response: response}) assert {:error, %GRPC.RPCError{status: status, message: message}} = @@ -137,6 +182,53 @@ defmodule GRPC.Client.Adapters.MintTest do end end + describe "receive_data/2 - request ended by the connection mid-request" do + @describetag timeout: 5_000 + + setup do + {:ok, stream_response_pid} = + GRPC.Client.Adapters.Mint.StreamResponseProcess.start_link(build(:client_stream), true) + + stream = + build(:client_stream, + payload: %{ + stream_response_pid: stream_response_pid, + response: {:ok, %{request_ref: make_ref()}} + } + ) + + %{stream: stream, stream_response_pid: stream_response_pid} + end + + test "returns the UNAVAILABLE error the connection process produced", %{ + stream: stream, + stream_response_pid: pid + } do + error = GRPC.RPCError.exception(GRPC.Status.unavailable(), "the connection is closed") + + :ok = GRPC.Client.Adapters.Mint.StreamResponseProcess.consume(pid, :error, error) + :ok = GRPC.Client.Adapters.Mint.StreamResponseProcess.done(pid) + + assert {:error, ^error} = Mint.receive_data(stream, []) + end + + test "normalises a raw transport error into an UNAVAILABLE GRPC.RPCError", %{ + stream: stream, + stream_response_pid: pid + } do + error = %TransportError{reason: :closed} + + :ok = GRPC.Client.Adapters.Mint.StreamResponseProcess.consume(pid, :error, error) + :ok = GRPC.Client.Adapters.Mint.StreamResponseProcess.done(pid) + + assert {:error, %GRPC.RPCError{status: status, message: message}} = + Mint.receive_data(stream, []) + + assert status == GRPC.Status.unavailable() + assert message == Exception.message(error) + end + end + describe "receive_data/2 - deadline" do setup do {:ok, stream_response_pid} =