From 51a6f10a13bb3cc631a9bfe104f467b1d13624d6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?C=CE=BBstor?= Date: Fri, 14 Aug 2026 14:42:53 +0200 Subject: [PATCH] fix(gun): Monitor gun to prevent sending messages to dead processes --- grpc/lib/grpc/client/adapters/gun.ex | 14 ++++++++++++-- .../grpc/client/adapters/gun/connection_process.ex | 13 +++++++++++++ grpc/test/grpc/adapters/gun_test.exs | 8 ++++---- grpc/test/grpc/integration/server_test.exs | 2 +- 4 files changed, 30 insertions(+), 7 deletions(-) diff --git a/grpc/lib/grpc/client/adapters/gun.ex b/grpc/lib/grpc/client/adapters/gun.ex index 7e53a9db7..22ea2712e 100644 --- a/grpc/lib/grpc/client/adapters/gun.ex +++ b/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, diff --git a/grpc/lib/grpc/client/adapters/gun/connection_process.ex b/grpc/lib/grpc/client/adapters/gun/connection_process.ex index e3de391a1..ca69a2943 100644 --- a/grpc/lib/grpc/client/adapters/gun/connection_process.ex +++ b/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 diff --git a/grpc/test/grpc/adapters/gun_test.exs b/grpc/test/grpc/adapters/gun_test.exs index 8aca4912e..bfa1311fc 100644 --- a/grpc/test/grpc/adapters/gun_test.exs +++ b/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 diff --git a/grpc/test/grpc/integration/server_test.exs b/grpc/test/grpc/integration/server_test.exs index e6b479fd6..9c093daed 100644 --- a/grpc/test/grpc/integration/server_test.exs +++ b/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)