fix(gun): map connection-level gun errors to internal status

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

@@ -89,6 +89,11 @@ defmodule GRPC.Client.Adapters.Gun.StreamResponseProcess do
     |> push_message({:error, {:stream_error, reason}}, true)
   end
 
+  def handle_info({:gun_error, _conn_pid, reason}, state) do
+    state
+    |> push_message({:error, {:connection_error, reason}}, true)
+  end
+
   def handle_info({:connection_down, reason}, state) do
     push_message(state, {:error, {:connection_error, reason}}, true)
   end
@@ -98,6 +103,10 @@ defmodule GRPC.Client.Adapters.Gun.StreamResponseProcess do
     push_message(state, {:error, {:unexpected_message, inspect(msg)}}, true)
   end
 
+  defp push_message(%{done: true} = state, _message, _terminal?) do
+    {:noreply, state}
+  end
+
   defp push_message(%{waiter: {from, timeout_ref}} = state, message, terminal?) do
     cancel_timeout(timeout_ref)
     GenServer.reply(from, message)

grpc/test/grpc/adapters/gun/stream_response_process_test.exs

@@ -26,6 +26,44 @@ defmodule GRPC.Client.Adapters.Gun.StreamResponseProcessTest do
     assert_receive {:DOWN, ^monitor_ref, :process, ^pid, :normal}, 500
   end
 
+  test "maps connection-level gun errors to connection errors" do
+    {:ok, pid} = StreamResponseProcess.start_link()
+    monitor_ref = Process.monitor(pid)
+
+    reason = {:protocol_error, :"The preface was not received in a reasonable amount of time."}
+    send(pid, {:gun_error, self(), reason})
+
+    assert {:error, {:connection_error, ^reason}} = StreamResponseProcess.await(pid, 100)
+    assert_receive {:DOWN, ^monitor_ref, :process, ^pid, :normal}, 500
+  end
+
+  test "returns connection-level gun errors to an awaiting caller" do
+    {:ok, pid} = StreamResponseProcess.start_link()
+    monitor_ref = Process.monitor(pid)
+
+    task = Task.async(fn -> StreamResponseProcess.await(pid, 1_000) end)
+
+    waiter =
+      Enum.find_value(1..50, fn _ ->
+        case :sys.get_state(pid) do
+          %{waiter: nil} ->
+            Process.sleep(10)
+            nil
+
+          %{waiter: waiter} ->
+            waiter
+        end
+      end)
+
+    assert {_from, _timeout_ref} = waiter
+
+    reason = {:protocol_error, :"The preface was not received in a reasonable amount of time."}
+    send(pid, {:gun_error, self(), reason})
+
+    assert {:error, {:connection_error, ^reason}} = Task.await(task)
+    assert_receive {:DOWN, ^monitor_ref, :process, ^pid, :normal}, 500
+  end
+
   test "keeps queued shutdown errors after non-final headers" do
     {:ok, pid} = StreamResponseProcess.start_link()
     monitor_ref = Process.monitor(pid)
@@ -38,4 +76,26 @@ defmodule GRPC.Client.Adapters.Gun.StreamResponseProcessTest do
     assert_receive {:DOWN, ^monitor_ref, :process, ^pid, :normal}, 500
     assert {:error, {:connection_error, :closed}} = StreamResponseProcess.await(pid, 100)
   end
+
+  test "ignores connection errors after terminal trailers are already queued" do
+    {:ok, pid} = StreamResponseProcess.start_link()
+    monitor_ref = Process.monitor(pid)
+
+    send(pid, {:gun_trailers, self(), make_ref(), [{"grpc-status", "0"}]})
+    send(pid, {:connection_down, :shutdown})
+
+    assert {:trailers, [{"grpc-status", "0"}]} = StreamResponseProcess.await(pid, 100)
+    assert_receive {:DOWN, ^monitor_ref, :process, ^pid, :normal}, 500
+  end
+
+  test "ignores duplicate terminal connection errors" do
+    {:ok, pid} = StreamResponseProcess.start_link()
+    monitor_ref = Process.monitor(pid)
+
+    send(pid, {:connection_down, :first})
+    send(pid, {:connection_down, :second})
+
+    assert {:error, {:connection_error, :first}} = StreamResponseProcess.await(pid, 100)
+    assert_receive {:DOWN, ^monitor_ref, :process, ^pid, :normal}, 500
+  end
 end

grpc/test/grpc/adapters/gun_test.exs

@@ -122,4 +122,22 @@ defmodule GRPC.Client.Adapters.GunTest do
       assert %{conn_pid: nil} = disconnected_again.adapter_payload
     end
   end
+
+  describe "receive_data/2" do
+    test "maps connection-level gun errors to internal 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()
+
+      assert {:error, %GRPC.RPCError{status: ^internal, message: message}} =
+               Gun.receive_data(stream, timeout: 100)
+
+      assert message =~ ":connection_error"
+      assert message =~ "preface"
+    end
+  end
 end