Monitor gun to prevent sending messages to dead processes

grpc/lib/grpc/client/adapters/gun.ex

@@ -319,9 +319,19 @@ if Code.ensure_loaded?(:gun) do
              "timeout when waiting for server"
            )}
 
-        {:error, {reason, msg}} when reason in [:stream_error, :connection_error] ->
+        # Connection-level failures are UNAVAILABLE per the gRPC status spec:
+        # the RPC never completed on a live connection, so callers can safely
+        # retry (deadline errors above stay DEADLINE_EXCEEDED).
+        {:error, {:connection_error, msg}} ->
           {:error,
-           GRPC.RPCError.exception(GRPC.Status.internal(), "#{inspect(reason)}: #{inspect(msg)}")}
+           GRPC.RPCError.exception(
+             GRPC.Status.unavailable(),
+             "connection_error: #{inspect(msg)}"
+           )}
+
+        {:error, {:stream_error, msg}} ->
+          {:error,
+           GRPC.RPCError.exception(GRPC.Status.internal(), "stream_error: #{inspect(msg)}")}
 
         {:error, {reason, msg}} ->
           {:error,

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

@@ -85,6 +85,9 @@ defmodule GRPC.Client.Adapters.Gun.ConnectionProcess do
       {:ok, gun_pid} ->
         case :gun.await_up(gun_pid, await_timeout) do
           {:ok, :http2} ->
+            # Monitor Gun so we don't keep casting into a dead pid if Gun
+            # exhausts its reconnect retries (or crashes)
+            Process.monitor(gun_pid)
             {:ok, %{gun_pid: gun_pid, response_processes: %{}}}
 
           {:ok, proto} ->
@@ -158,6 +161,16 @@ defmodule GRPC.Client.Adapters.Gun.ConnectionProcess do
     {:noreply, new_state}
   end
 
+  # Gun is gone for good (reconnect retries exhausted or a crash). Fail all
+  # in-flight streams and stop
+  def handle_info({:DOWN, _monitor_ref, :process, gun_pid, reason}, %{gun_pid: gun_pid} = state) do
+    Enum.each(state.response_processes, fn {_stream_ref, {response_pid, _monitor_ref}} ->
+      send(response_pid, {:connection_down, reason})
+    end)
+
+    {:stop, {:shutdown, {:gun_down, reason}}, state}
+  end
+
   def handle_info({:DOWN, monitor_ref, :process, _pid, _reason}, state) do
     {:noreply, drop_response_pid_by_monitor(state, monitor_ref)}
   end

grpc/test/grpc/adapters/gun_test.exs

@@ -171,19 +171,19 @@ defmodule GRPC.Client.Adapters.GunTest do
   end
 
   describe "receive_data/2" do
-    test "maps connection-level gun errors to internal RPC errors" do
+    test "maps connection-level gun errors to unavailable RPC errors" do
       {:ok, response_pid} = GRPC.Client.Adapters.Gun.StreamResponseProcess.start_link()
 
       reason = {:protocol_error, :"The preface was not received in a reasonable amount of time."}
       send(response_pid, {:gun_error, self(), reason})
 
       stream = %GRPC.Client.Stream{payload: %{response_pid: response_pid}, server_stream: false}
-      internal = GRPC.Status.internal()
+      unavailable = GRPC.Status.unavailable()
 
-      assert {:error, %GRPC.RPCError{status: ^internal, message: message}} =
+      assert {:error, %GRPC.RPCError{status: ^unavailable, message: message}} =
                Gun.receive_data(stream, timeout: 100)
 
-      assert message =~ ":connection_error"
+      assert message =~ "connection_error"
       assert message =~ "preface"
     end
   end

grpc/test/grpc/integration/server_test.exs

@@ -487,7 +487,7 @@ defmodule GRPC.Integration.ServerTest do
           client_stream = Routeguide.RouteGuide.Stub.route_chat(channel)
           assert %GRPC.Client.Stream{} = client_stream
           {:ok, ex_stream} = GRPC.Stub.recv(client_stream, timeout: :infinity)
-          assert [{:error, %GRPC.RPCError{status: 13}}] = Enum.into(ex_stream, [])
+          assert [{:error, %GRPC.RPCError{status: 14}}] = Enum.into(ex_stream, [])
         end)
       end)