refactor: remove GenServer behavior from gossip BeaconBlock (#1113)

Author: avilagaston9

lib/lambda_ethereum_consensus/beacon/beacon_chain.ex

@@ -4,7 +4,6 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconChain do
   use GenServer
 
   alias LambdaEthereumConsensus.Beacon.PendingBlocks
-  alias LambdaEthereumConsensus.P2P.Gossip
   alias LambdaEthereumConsensus.StateTransition.Misc
   alias LambdaEthereumConsensus.Validator.ValidatorManager
   alias Types.BeaconState
@@ -243,11 +242,6 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconChain do
 
   defp notify_subscribers(logical_time) do
     log_new_slot(logical_time)
-
-    Enum.each([Gossip.BeaconBlock], fn subscriber ->
-      GenServer.cast(subscriber, {:on_tick, logical_time})
-    end)
-
     ValidatorManager.notify_tick(logical_time)
   end
 

lib/lambda_ethereum_consensus/beacon/beacon_node.ex

@@ -53,7 +53,6 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
         LambdaEthereumConsensus.Beacon.PendingBlocks,
         LambdaEthereumConsensus.Beacon.SyncBlocks,
         LambdaEthereumConsensus.P2P.Gossip.Attestation,
-        LambdaEthereumConsensus.P2P.Gossip.BeaconBlock,
         LambdaEthereumConsensus.P2P.Gossip.BlobSideCar,
         LambdaEthereumConsensus.P2P.Gossip.OperationsCollector,
         {Task.Supervisor, name: PruneStatesSupervisor},
@@ -93,7 +92,8 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
       listen_addr: listen_addr,
       enable_discovery: true,
       discovery_addr: "0.0.0.0:#{port}",
-      bootnodes: bootnodes
+      bootnodes: bootnodes,
+      join_init_topics: true
     ]
   end
 

lib/lambda_ethereum_consensus/beacon/sync_blocks.ex

@@ -83,8 +83,9 @@ defmodule LambdaEthereumConsensus.Beacon.SyncBlocks do
     end
   end
 
+  # TODO: handle subscription failures.
   defp start_subscriptions() do
-    Gossip.BeaconBlock.start()
+    Gossip.BeaconBlock.subscribe_to_topic()
     Gossip.BlobSideCar.start()
     Gossip.OperationsCollector.start()
   end

lib/lambda_ethereum_consensus/p2p/gossip/beacon_block.ex

@@ -8,63 +8,17 @@ defmodule LambdaEthereumConsensus.P2P.Gossip.BeaconBlock do
   alias LambdaEthereumConsensus.P2P.Gossip.Handler
   alias Types.SignedBeaconBlock
 
-  use GenServer
-
   require Logger
   @behaviour Handler
 
-  @type state :: %{topic: String.t(), slot: Types.slot()}
-
   ##########################
   ### Public API
   ##########################
 
-  def start_link(init_arg) do
-    GenServer.start_link(__MODULE__, init_arg, name: __MODULE__)
-  end
-
-  def start() do
-    GenServer.call(__MODULE__, :start)
-  end
-
-  @spec notify_slot(Types.slot()) :: :ok
-  def notify_slot(slot) do
-    GenServer.cast(__MODULE__, {:slot_transition, slot})
-  end
-
   @impl true
-  def handle_gossip_message(topic, msg_id, message) do
-    GenServer.cast(__MODULE__, {:gossipsub, {topic, msg_id, message}})
-  end
-
-  ##########################
-  ### GenServer Callbacks
-  ##########################
-
-  @impl true
-  @spec init(any()) :: {:ok, state()} | {:stop, any()}
-  def init(_init_arg) do
-    # TODO: this doesn't take into account fork digest changes
-    fork_context = BeaconChain.get_fork_digest() |> Base.encode16(case: :lower)
+  def handle_gossip_message(_topic, msg_id, message) do
     slot = BeaconChain.get_current_slot()
-    topic_name = "/eth2/#{fork_context}/beacon_block/ssz_snappy"
-    Libp2pPort.join_topic(topic_name)
-    {:ok, %{topic: topic_name, slot: slot}}
-  end
-
-  @impl true
-  def handle_call(:start, _from, %{topic: topic_name} = state) do
-    Libp2pPort.subscribe_to_topic(topic_name, __MODULE__)
-    {:reply, :ok, state}
-  end
 
-  @impl true
-  def handle_cast({:on_tick, {slot, _}}, state) do
-    {:noreply, state |> Map.put(:slot, slot)}
-  end
-
-  @impl true
-  def handle_cast({:gossipsub, {_topic, msg_id, message}}, %{slot: slot} = state) do
     with {:ok, uncompressed} <- :snappyer.decompress(message),
          {:ok, signed_block} <- Ssz.from_ssz(uncompressed, SignedBeaconBlock),
          :ok <- validate(signed_block, slot) do
@@ -81,7 +35,37 @@ defmodule LambdaEthereumConsensus.P2P.Gossip.BeaconBlock do
         Libp2pPort.validate_message(msg_id, :reject)
     end
 
-    {:noreply, state}
+    :ok
+  end
+
+  @spec join_topic() :: :ok
+  def join_topic() do
+    # TODO: this doesn't take into account fork digest changes
+    topic_name = topic()
+    Libp2pPort.join_topic(self(), topic_name)
+  end
+
+  @spec subscribe_to_topic() :: :ok | :error
+  def subscribe_to_topic() do
+    topic()
+    |> Libp2pPort.subscribe_to_topic(__MODULE__)
+    |> case do
+      :ok ->
+        :ok
+
+      {:error, reason} ->
+        Logger.error("[Gossip] Subscription failed: '#{reason}'")
+        :error
+    end
+  end
+
+  ##########################
+  ### Private functions
+  ##########################
+
+  defp topic() do
+    fork_context = BeaconChain.get_fork_digest() |> Base.encode16(case: :lower)
+    "/eth2/#{fork_context}/beacon_block/ssz_snappy"
   end
 
   @spec validate(SignedBeaconBlock.t(), Types.slot()) :: :ok | {:error, any}

lib/libp2p_port.ex

@@ -10,6 +10,7 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
   use GenServer
 
   alias LambdaEthereumConsensus.Beacon.BeaconChain
+  alias LambdaEthereumConsensus.P2P.Gossip.BeaconBlock
   alias LambdaEthereumConsensus.StateTransition.Misc
   alias LambdaEthereumConsensus.Utils.BitVector
   alias Types.EnrForkId
@@ -53,6 +54,7 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
           | {:discovery_addr, String.t()}
           | {:bootnodes, [String.t()]}
           | {:new_peer_handler, pid()}
+          | {:join_init_topics, boolean()}
 
   ######################
   ### API
@@ -164,14 +166,14 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
   Joins the given topic.
   This does not subscribe to the topic, use `subscribe_to_topic/2` for that.
   """
-  @spec join_topic(GenServer.server(), String.t()) :: :ok | {:error, String.t()}
+  @spec join_topic(GenServer.server(), String.t()) :: :ok
   def join_topic(pid \\ __MODULE__, topic_name) do
     :telemetry.execute([:port, :message], %{}, %{
       function: "join_topic",
       direction: "elixir->"
     })
 
-    call_command(pid, {:join, %JoinTopic{name: topic_name}})
+    cast_command(pid, {:join, %JoinTopic{name: topic_name}})
   end
 
   @doc """
@@ -264,13 +266,18 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
     cast_command(pid, {:update_enr, enr})
   end
 
+  defp join_init_topics() do
+    BeaconBlock.join_topic()
+  end
+
   ########################
   ### GenServer Callbacks
   ########################
 
   @impl GenServer
   def init(args) do
     {new_peer_handler, args} = Keyword.pop(args, :new_peer_handler, nil)
+    {join_init_topics, args} = Keyword.pop(args, :join_init_topics, false)
 
     port = Port.open({:spawn, @port_name}, [:binary, {:packet, 4}, :exit_status])
 
@@ -281,6 +288,8 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
     |> InitArgs.encode()
     |> then(&send_data(port, &1))
 
+    if join_init_topics, do: join_init_topics()
+
     {:ok, %{port: port, new_peer_handler: new_peer_handler, subscriptors: %{}}}
   end