fix(mint): fail with UNAVAILABLE when the connection drops

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
 

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{

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(

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

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} =