Merge remote-tracking branch 'upstream/master'

Author: sleipnir

.github/workflows/ci.yml

@@ -52,7 +52,7 @@ jobs:
       - name: Install Dependencies
         run: mix deps.get 1>/dev/null
       - name: Run Tests
-        run: mix test
+        run: mix test --warnings-as-errors
   interop-tests:
     runs-on: ubuntu-latest
     name: Interop tests

lib/google/api/annotations.pb.ex

@@ -155,24 +155,59 @@ defmodule GRPC.Client.Adapters.Gun do
       ) do
     %{channel: %{adapter_payload: adapter_payload}, payload: payload} = stream
 
-    with {:ok, headers, is_fin} <- recv_headers(adapter_payload, payload, opts) do
-      response = response_stream(is_fin, stream, opts)
+    case recv_headers(adapter_payload, payload, opts) do
+      {:ok, headers, :fin} ->
+        handle_fin_response(headers, opts)
 
-      if(opts[:return_headers]) do
-        {:ok, response, %{headers: headers}}
-      else
-        {:ok, response}
-      end
+      {:ok, headers, :nofin} ->
+        handle_streaming_nofin_response(stream, headers, opts)
+
+      {:error, _} = error ->
+        error
     end
   end
 
   def receive_data(stream, opts) do
     %{payload: payload, channel: %{adapter_payload: adapter_payload}} = stream
 
-    with {:ok, headers, _is_fin} <- recv_headers(adapter_payload, payload, opts),
-         {:ok, body, trailers} <- recv_body(adapter_payload, payload, opts),
+    case recv_headers(adapter_payload, payload, opts) do
+      {:ok, headers, :fin} ->
+        handle_fin_response(headers, opts)
+
+      {:ok, headers, :nofin} ->
+        handle_nofin_response(adapter_payload, payload, stream, headers, opts)
+
+      {:error, _} = error ->
+        error
+    end
+  end
+
+  defp handle_fin_response(headers, opts) do
+    # Trailers-only response: headers contain trailers, check status
+    with :ok <- parse_trailers(headers) do
+      if opts[:return_headers] do
+        {:ok, [], %{headers: headers}}
+      else
+        {:ok, []}
+      end
+    end
+  end
+
+  defp handle_streaming_nofin_response(stream, headers, opts) do
+    response = response_stream(:nofin, stream, opts)
+
+    if opts[:return_headers] do
+      {:ok, response, %{headers: headers}}
+    else
+      {:ok, response}
+    end
+  end
+
+  defp handle_nofin_response(adapter_payload, payload, stream, headers, opts) do
+    # Regular response: fetch body and trailers
+    with {:ok, body, trailers} <- recv_body(adapter_payload, payload, opts),
          {:ok, response} <- parse_response(stream, headers, body, trailers) do
-      if(opts[:return_headers]) do
+      if opts[:return_headers] do
         {:ok, response, %{headers: headers, trailers: trailers}}
       else
         {:ok, response}
@@ -230,25 +265,7 @@ defmodule GRPC.Client.Adapters.Gun do
       {:response, :fin, status, headers} ->
         if status == 200 do
           headers = GRPC.Transport.HTTP2.decode_headers(headers)
-
-          case headers["grpc-status"] do
-            nil ->
-              {:error,
-               GRPC.RPCError.exception(
-                 GRPC.Status.internal(),
-                 "shouldn't finish when getting headers"
-               )}
-
-            "0" ->
-              {:response, headers, :fin}
-
-            _ ->
-              {:error,
-               GRPC.RPCError.exception(
-                 String.to_integer(headers["grpc-status"]),
-                 headers["grpc-message"]
-               )}
-          end
+          {:response, headers, :fin}
         else
           {:error,
            GRPC.RPCError.exception(
@@ -260,16 +277,7 @@ defmodule GRPC.Client.Adapters.Gun do
       {:response, :nofin, status, headers} ->
         if status == 200 do
           headers = GRPC.Transport.HTTP2.decode_headers(headers)
-
-          if headers["grpc-status"] && headers["grpc-status"] != "0" do
-            {:error,
-             GRPC.RPCError.exception(
-               String.to_integer(headers["grpc-status"]),
-               headers["grpc-message"]
-             )}
-          else
-            {:response, headers, :nofin}
-          end
+          {:response, headers, :nofin}
         else
           {:error,
            GRPC.RPCError.exception(
@@ -349,8 +357,6 @@ defmodule GRPC.Client.Adapters.Gun do
     end
   end
 
-  defp response_stream(:fin, _stream, _opts), do: []
-
   defp response_stream(
          :nofin,
          %{
@@ -453,7 +459,14 @@ defmodule GRPC.Client.Adapters.Gun do
     if status == GRPC.Status.ok() do
       :ok
     else
-      {:error, %GRPC.RPCError{status: status, message: trailers["grpc-message"]}}
+      rpc_error =
+        GRPC.RPCError.from_grpc_status_details_bin(%{
+          status: status,
+          message: trailers["grpc-message"],
+          encoded_details_bin: trailers["grpc-status-details-bin"]
+        })
+
+      {:error, rpc_error}
     end
   end
 

lib/google/api/http.pb.ex

@@ -182,7 +182,13 @@ defmodule GRPC.Client.Adapters.Mint.StreamResponseProcess do
     if status == GRPC.Status.ok() do
       {type, decoded_trailers}
     else
-      rpc_error = %GRPC.RPCError{status: status, message: decoded_trailers["grpc-message"]}
+      rpc_error =
+        GRPC.RPCError.from_grpc_status_details_bin(%{
+          status: status,
+          message: decoded_trailers["grpc-message"],
+          encoded_details_bin: decoded_trailers["grpc-status-details-bin"]
+        })
+
       {:error, rpc_error}
     end
   end

lib/grpc/client/adapters/gun.ex

@@ -0,0 +1,28 @@
+defmodule GRPC.Google.RPC do
+  @moduledoc false
+
+  @spec encode_status(Google.Rpc.Status.t()) :: String.t()
+  def encode_status(%Google.Rpc.Status{} = status) do
+    status
+    |> Google.Rpc.Status.encode()
+    |> Base.encode64(padding: true)
+  end
+
+  @spec decode_status(String.t()) :: {:ok, Google.Rpc.Status.t()} | {:error, term()}
+  def decode_status(encoded_details_bin) when is_binary(encoded_details_bin) do
+    {:ok,
+     encoded_details_bin
+     |> decode64()
+     |> Google.Rpc.Status.decode()}
+  rescue
+    e -> {:error, e}
+  end
+
+  defp decode64(str) when rem(byte_size(str), 4) == 0 do
+    Base.decode64!(str, padding: true)
+  end
+
+  defp decode64(str) do
+    Base.decode64!(str, padding: false)
+  end
+end

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

@@ -51,11 +51,15 @@ defmodule GRPC.RPCError do
   See `GRPC.Status` for more details on possible statuses.
   """
 
-  defexception [:status, :message]
+  defexception [:status, :message, :details]
 
   defguard is_rpc_error(e, status) when is_struct(e, __MODULE__) and e.status == status
 
-  @type t :: %__MODULE__{status: GRPC.Status.t(), message: String.t()}
+  @type t :: %__MODULE__{
+          status: GRPC.Status.t(),
+          message: String.t(),
+          details: [Google.Protobuf.Any.t()]
+        }
 
   alias GRPC.Status
 
@@ -68,7 +72,11 @@ defmodule GRPC.RPCError do
   def exception(args) when is_list(args) do
     error = parse_args(args, %__MODULE__{})
 
-    %{error | message: error.message || Status.status_message(error.status)}
+    %{
+      error
+      | message: error.message || Status.status_message(error.status),
+        details: error.details
+    }
   end
 
   defp parse_args([], acc), do: acc
@@ -88,6 +96,11 @@ defmodule GRPC.RPCError do
     parse_args(t, acc)
   end
 
+  defp parse_args([{:details, details} | t], acc) when is_list(details) do
+    acc = %{acc | details: details}
+    parse_args(t, acc)
+  end
+
   @spec exception(status :: Status.t() | atom(), message :: String.t()) :: t()
   def exception(status, message) when is_atom(status) do
     %GRPC.RPCError{status: apply(GRPC.Status, status, []), message: message}
@@ -96,4 +109,28 @@ defmodule GRPC.RPCError do
   def exception(status, message) when is_integer(status) do
     %GRPC.RPCError{status: status, message: message}
   end
+
+  @doc false
+  def from_grpc_status_details_bin(%{
+        status: status,
+        message: message,
+        encoded_details_bin: encoded_details_bin
+      })
+      when is_binary(encoded_details_bin) do
+    case GRPC.Google.RPC.decode_status(encoded_details_bin) do
+      {:ok, rpc_status} ->
+        %__MODULE__{
+          status: status,
+          message: rpc_status.message,
+          details: rpc_status.details
+        }
+
+      {:error, _} ->
+        %__MODULE__{status: status, message: message}
+    end
+  end
+
+  def from_grpc_status_details_bin(%{status: status, message: message}) do
+    %__MODULE__{status: status, message: message}
+  end
 end

lib/grpc/google/rpc.ex

@@ -464,6 +464,10 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
     error = %RPCError{status: GRPC.Status.deadline_exceeded(), message: "Deadline expired"}
     req = send_error(req, error, state, :timeout)
 
+    [req: req]
+    |> ReportException.new(error)
+    |> log_error()
+
     {:stop, req, state}
   end
 
@@ -660,7 +664,7 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
   defp preflight?(_), do: false
 
   defp send_error(req, error, state, reason) do
-    trailers = HTTP2.server_trailers(error.status, error.message)
+    trailers = HTTP2.server_trailers(error.status, error.message, error.details)
 
     status =
       if state.access_mode == :http_transcoding,
@@ -694,7 +698,7 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
     {:wait, ref}
   end
 
-  defp log_error(%ReportException{kind: kind} = exception, stacktrace) do
+  defp log_error(%ReportException{kind: kind} = exception, stacktrace \\ []) do
     crash_reason = GRPC.Logger.crash_reason(kind, exception, stacktrace)
 
     kind

lib/grpc/rpc_error.ex

@@ -309,7 +309,11 @@ defmodule GRPC.Stub do
       end
 
     compressor = Keyword.get(opts, :compressor, channel.compressor)
-    accepted_compressors = Keyword.get(opts, :accepted_compressors, [])
+    accepted_compressors = Keyword.get(opts, :accepted_compressors, channel.accepted_compressors)
+
+    if not is_list(accepted_compressors) do
+      raise ArgumentError, "accepted_compressors is not a list"
+    end
 
     accepted_compressors =
       if compressor do
@@ -321,7 +325,7 @@ defmodule GRPC.Stub do
     stream = %{
       stream
       | codec: Keyword.get(opts, :codec, channel.codec),
-        compressor: Keyword.get(opts, :compressor, channel.compressor),
+        compressor: compressor,
         accepted_compressors: accepted_compressors
     }