refactor: remove IncomingRequests GenServer (#1195)

Co-authored-by: Tomás Arjovsky <[email protected]>

Author: avilagaston9

lib/lambda_ethereum_consensus/beacon/beacon_node.ex

@@ -42,7 +42,6 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
       [
         {LambdaEthereumConsensus.Beacon.Clock, {store.genesis_time, time}},
         {LambdaEthereumConsensus.Libp2pPort, libp2p_args},
-        LambdaEthereumConsensus.P2P.IncomingRequests,
         LambdaEthereumConsensus.Beacon.SyncBlocks,
         {Task.Supervisor, name: PruneStatesSupervisor},
         {Task.Supervisor, name: PruneBlocksSupervisor},
@@ -83,7 +82,8 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
       enable_discovery: true,
       discovery_addr: "0.0.0.0:#{port}",
       bootnodes: bootnodes,
-      join_init_topics: true
+      join_init_topics: true,
+      enable_request_handlers: true
     ]
   end
 

lib/lambda_ethereum_consensus/p2p/incoming_requests/incoming_requests.ex

@@ -1,55 +1,71 @@
-defmodule LambdaEthereumConsensus.P2P.IncomingRequests.Handler do
+defmodule LambdaEthereumConsensus.P2P.IncomingRequestsHandler do
   @moduledoc """
   This module handles Req/Resp domain requests.
   """
-  require Logger
 
   alias LambdaEthereumConsensus.ForkChoice
-  alias LambdaEthereumConsensus.Libp2pPort
   alias LambdaEthereumConsensus.P2P.Metadata
   alias LambdaEthereumConsensus.P2P.ReqResp
   alias LambdaEthereumConsensus.Store.BlockDb
   alias LambdaEthereumConsensus.Store.Blocks
 
   require Logger
 
-  @spec handle(String.t(), String.t(), binary()) :: any()
-  def handle(name, message_id, message) do
-    case handle_req(name, message_id, message) do
-      :ok -> :ok
-      {:error, error} -> Logger.error("[#{name}] Request error: #{inspect(error)}")
+  @request_prefix "/eth2/beacon_chain/req/"
+  @request_names [
+    "status/1",
+    "goodbye/1",
+    "ping/1",
+    "beacon_blocks_by_range/2",
+    "beacon_blocks_by_root/2",
+    "metadata/2"
+  ]
+
+  @spec protocol_ids() :: list(String.t())
+  def protocol_ids() do
+    @request_names |> Enum.map(&Enum.join([@request_prefix, &1, "/ssz_snappy"]))
+  end
+
+  @spec handle(String.t(), String.t(), binary()) :: {:ok, any()} | {:error, String.t()}
+  def handle(@request_prefix <> name, message_id, message) do
+    Logger.debug("'#{name}' request received")
+
+    result =
+      :telemetry.span([:port, :request], %{}, fn ->
+        {handle_req(name, message_id, message), %{module: "handler", request: inspect(name)}}
+      end)
+
+    case result do
+      {:error, error} -> {:error, "[#{name}] Request error: #{inspect(error)}"}
+      result -> result
     end
   end
 
   @spec handle_req(String.t(), String.t(), binary()) ::
-          :ok | {:error, String.t()}
+          {:ok, any()} | {:error, String.t()}
   defp handle_req(protocol_name, message_id, message)
 
   defp handle_req("status/1/ssz_snappy", message_id, message) do
     with {:ok, request} <- ReqResp.decode_request(message, Types.StatusMessage) do
       Logger.debug("[Status] '#{inspect(request)}'")
       payload = ForkChoice.get_current_status_message() |> ReqResp.encode_ok()
-      Libp2pPort.send_response(message_id, payload)
+      {:ok, {message_id, payload}}
     end
   end
 
   defp handle_req("goodbye/1/ssz_snappy", _, "") do
     # ignore empty messages
-    Logger.debug("[Goodbye] empty message")
+    {:error, "Empty message"}
   end
 
   defp handle_req("goodbye/1/ssz_snappy", message_id, message) do
     case ReqResp.decode_request(message, TypeAliases.uint64()) do
       {:ok, goodbye_reason} ->
         Logger.debug("[Goodbye] reason: #{goodbye_reason}")
         payload = ReqResp.encode_ok({0, TypeAliases.uint64()})
-        Libp2pPort.send_response(message_id, payload)
+        {:ok, {message_id, payload}}
 
       # Ignore read errors, since some peers eagerly disconnect.
-      {:error, "failed to read"} ->
-        Logger.debug("[Goodbye] failed to read")
-        :ok
-
       err ->
         err
     end
@@ -61,14 +77,14 @@ defmodule LambdaEthereumConsensus.P2P.IncomingRequests.Handler do
       Logger.debug("[Ping] seq_number: #{seq_num}")
       seq_number = Metadata.get_seq_number()
       payload = ReqResp.encode_ok({seq_number, TypeAliases.uint64()})
-      Libp2pPort.send_response(message_id, payload)
+      {:ok, {message_id, payload}}
     end
   end
 
   defp handle_req("metadata/2/ssz_snappy", message_id, _message) do
     # NOTE: there's no request content so we just ignore it
     payload = Metadata.get_metadata() |> ReqResp.encode_ok()
-    Libp2pPort.send_response(message_id, payload)
+    {:ok, {message_id, payload}}
   end
 
   defp handle_req("beacon_blocks_by_range/2/ssz_snappy", message_id, message) do
@@ -89,7 +105,7 @@ defmodule LambdaEthereumConsensus.P2P.IncomingRequests.Handler do
         |> Enum.reject(&(&1 == :skip))
         |> ReqResp.encode_response()
 
-      Libp2pPort.send_response(message_id, response_chunk)
+      {:ok, {message_id, response_chunk}}
     end
   end
 
@@ -108,14 +124,13 @@ defmodule LambdaEthereumConsensus.P2P.IncomingRequests.Handler do
         |> Enum.reject(&(&1 == :skip))
         |> ReqResp.encode_response()
 
-      Libp2pPort.send_response(message_id, response_chunk)
+      {:ok, {message_id, response_chunk}}
     end
   end
 
   defp handle_req(protocol, _message_id, _message) do
     # This should never happen, since Libp2p only accepts registered protocols
-    Logger.error("Unsupported protocol: #{protocol}")
-    :ok
+    {:error, "Unsupported protocol: #{protocol}"}
   end
 
   defp map_block_result(:not_found), do: map_block_result(nil)

lib/lambda_ethereum_consensus/p2p/incoming_requests/receiver.ex

@@ -85,6 +85,14 @@ defmodule LambdaEthereumConsensus.Telemetry do
       counter("network.pubsub_topics_un_deliverable_message.count", tags: [:topic]),
       counter("network.pubsub_topics_validate_message.count", tags: [:topic]),
       counter("port.message.count", tags: [:function, :direction]),
+      last_value("port.request.stop.duration",
+        unit: {:native, :millisecond},
+        tags: [:module, :request]
+      ),
+      last_value("port.request.exception.duration",
+        unit: {:native, :millisecond},
+        tags: [:module, :request]
+      ),
       sum("network.request.blocks", tags: [:result, :type, :reason]),
 
       # Sync metrics

lib/lambda_ethereum_consensus/p2p/incoming_requests/handler.ex renamed to lib/lambda_ethereum_consensus/p2p/incoming_requests_handler.ex

@@ -15,6 +15,7 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
   alias LambdaEthereumConsensus.P2P.Gossip.BeaconBlock
   alias LambdaEthereumConsensus.P2P.Gossip.BlobSideCar
   alias LambdaEthereumConsensus.P2P.Gossip.OperationsCollector
+  alias LambdaEthereumConsensus.P2P.IncomingRequestsHandler
   alias LambdaEthereumConsensus.P2P.Peerbook
   alias LambdaEthereumConsensus.P2p.Requests
   alias LambdaEthereumConsensus.StateTransition.Misc
@@ -62,6 +63,7 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
           | {:discovery_addr, String.t()}
           | {:bootnodes, [String.t()]}
           | {:join_init_topics, boolean()}
+          | {:enable_request_handlers, boolean()}
 
   @type node_identity() :: %{
           peer_id: binary(),
@@ -116,15 +118,15 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
     |> Map.take([:peer_id, :pretty_peer_id, :enr, :p2p_addresses, :discovery_addresses])
   end
 
-  @doc """
-  Sets a Req/Resp handler for the given protocol ID. After this call,
-  peer requests are sent to the current process' mailbox. To handle them,
-  use `handle_request/0`.
-  """
-  @spec set_handler(GenServer.server(), String.t()) :: :ok | {:error, String.t()}
-  def set_handler(pid \\ __MODULE__, protocol_id) do
+  # Sets libp2pport as the Req/Resp handler for the given protocol ID.
+  @spec set_handler(String.t(), port()) :: boolean()
+  defp set_handler(protocol_id, port) do
     :telemetry.execute([:port, :message], %{}, %{function: "set_handler", direction: "elixir->"})
-    call_command(pid, {:set_handler, %SetHandler{protocol_id: protocol_id}})
+
+    c = {:set_handler, %SetHandler{protocol_id: protocol_id}}
+    data = Command.encode(%Command{c: c})
+
+    send_data(port, data)
   end
 
   @doc """
@@ -166,27 +168,15 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
     GenServer.cast(pid, {:send_request, peer_id, protocol_id, message, handler})
   end
 
-  @doc """
-  Returns the next request received by the server for registered handlers
-  on the current process. If there are no requests, it waits for one.
-  """
-  @spec handle_request() :: {String.t(), String.t(), binary()}
-  def handle_request() do
-    receive do
-      {:request, {_protocol_id, _message_id, _message} = request} -> request
-    end
-  end
-
-  @doc """
-  Sends a response for the request with the given message ID.
-  """
-  @spec send_response(GenServer.server(), String.t(), binary()) ::
-          :ok | {:error, String.t()}
-  def send_response(pid \\ __MODULE__, request_id, response) do
+  # Sends a response for the request with the given message ID.
+  @spec send_response({String.t(), binary()}, port()) :: boolean()
+  defp send_response({request_id, response}, port) do
     :telemetry.execute([:port, :message], %{}, %{function: "send_response", direction: "elixir->"})
 
-    c = %SendResponse{request_id: request_id, message: response}
-    call_command(pid, {:send_response, c})
+    c = {:send_response, %SendResponse{request_id: request_id, message: response}}
+    data = Command.encode(%Command{c: c})
+
+    send_data(port, data)
   end
 
   @doc """
@@ -299,6 +289,12 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
     OperationsCollector.init()
   end
 
+  @spec enable_request_handlers(port()) :: :ok | {:error, String.t()}
+  defp enable_request_handlers(port) do
+    IncomingRequestsHandler.protocol_ids()
+    |> Enum.each(fn protocol_id -> set_handler(protocol_id, port) end)
+  end
+
   def add_block(pid \\ __MODULE__, block), do: GenServer.cast(pid, {:add_block, block})
 
   ########################
@@ -308,6 +304,7 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
   @impl GenServer
   def init(args) do
     {join_init_topics, args} = Keyword.pop(args, :join_init_topics, false)
+    {enable_request_handlers, args} = Keyword.pop(args, :enable_request_handlers, false)
 
     port = Port.open({:spawn, @port_name}, [:binary, {:packet, 4}, :exit_status])
 
@@ -319,6 +316,7 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
     |> then(&send_data(port, &1))
 
     if join_init_topics, do: join_init_topics(port)
+    if enable_request_handlers, do: enable_request_handlers(port)
 
     Peerbook.init()
 
@@ -412,15 +410,21 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
   defp handle_notification(
          %Request{
            protocol_id: protocol_id,
-           handler: handler,
            request_id: request_id,
            message: message
          },
-         state
+         %{port: port} = state
        ) do
     :telemetry.execute([:port, :message], %{}, %{function: "request", direction: "->elixir"})
-    handler_pid = :erlang.binary_to_term(handler)
-    send(handler_pid, {:request, {protocol_id, request_id, message}})
+
+    case IncomingRequestsHandler.handle(protocol_id, request_id, message) do
+      {:ok, response} ->
+        send_response(response, port)
+
+      {:error, reason} ->
+        Logger.error("[Libp2pPort] Error handling request. Reason: #{inspect(reason)}")
+    end
+
     state
   end
 

lib/lambda_ethereum_consensus/telemetry.ex

@@ -123,8 +123,8 @@ func NewPeerNotification(id []byte) proto_defs.Notification {
 	return proto_defs.Notification{N: &proto_defs.Notification_NewPeer{NewPeer: newPeerNotification}}
 }
 
-func RequestNotification(protocolId string, handler []byte, requestId string, message []byte) proto_defs.Notification {
-	requestNotification := &proto_defs.Request{ProtocolId: []byte(protocolId), Handler: handler, RequestId: []byte(requestId), Message: message}
+func RequestNotification(protocolId string, requestId string, message []byte) proto_defs.Notification {
+	requestNotification := &proto_defs.Request{ProtocolId: []byte(protocolId), RequestId: []byte(requestId), Message: message}
 	return proto_defs.Notification{N: &proto_defs.Notification_Request{Request: requestNotification}}
 }