ValidatorsManager Genserver removal

Author: rodrigo-o

config/runtime.exs

@@ -171,7 +171,7 @@ if keystore_pass_dir != nil and not File.dir?(keystore_pass_dir) do
   System.halt(2)
 end
 
-config :lambda_ethereum_consensus, LambdaEthereumConsensus.Validator.ValidatorManager,
+config :lambda_ethereum_consensus, LambdaEthereumConsensus.Validator.Setup,
   keystore_dir: keystore_dir,
   keystore_pass_dir: keystore_pass_dir
 

lib/lambda_ethereum_consensus/beacon/beacon_node.ex

@@ -8,7 +8,7 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
   alias LambdaEthereumConsensus.ForkChoice
   alias LambdaEthereumConsensus.StateTransition.Cache
   alias LambdaEthereumConsensus.Store.BlockStates
-  alias LambdaEthereumConsensus.Validator.ValidatorManager
+  alias LambdaEthereumConsensus.Validator
   alias Types.BeaconState
 
   def start_link(opts) do
@@ -24,43 +24,36 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
 
     Cache.initialize_cache()
 
-    libp2p_args = get_libp2p_args()
-
     time = :os.system_time(:second)
 
     ForkChoice.init_store(store, time)
 
-    validator_manager =
-      get_validator_manager(
-        deposit_tree_snapshot,
-        store.head_slot,
-        store.head_root
-      )
+    init_execution_chain(deposit_tree_snapshot, store.head_root)
+
+    validators = Validator.Setup.init(store.head_slot, store.head_root)
+
+    libp2p_args = [genesis_time: store.genesis_time, validators: validators] ++ get_libp2p_args()
 
     children =
       [
         {LambdaEthereumConsensus.Beacon.Ticker, [LambdaEthereumConsensus.Libp2pPort]},
-        {LambdaEthereumConsensus.Libp2pPort, [{:genesis_time, store.genesis_time} | libp2p_args]},
+        {LambdaEthereumConsensus.Libp2pPort, libp2p_args},
         {Task.Supervisor, name: PruneStatesSupervisor},
         {Task.Supervisor, name: PruneBlocksSupervisor},
         {Task.Supervisor, name: PruneBlobsSupervisor}
-      ] ++ validator_manager
+      ]
 
     Supervisor.init(children, strategy: :one_for_all)
   end
 
-  defp get_validator_manager(nil, _, _) do
+  defp init_execution_chain(nil, _) do
     Logger.warning("Deposit data not found. Validator will be disabled.")
     []
   end
 
-  defp get_validator_manager(snapshot, slot, head_root) do
+  defp init_execution_chain(snapshot, head_root) do
     %BeaconState{eth1_data_votes: votes} = BlockStates.get_state_info!(head_root).beacon_state
     LambdaEthereumConsensus.Execution.ExecutionChain.init(snapshot, votes)
-    # TODO: move checkpoint sync outside and move this to application.ex
-    [
-      {ValidatorManager, {slot, head_root}}
-    ]
   end
 
   defp get_libp2p_args() do

lib/lambda_ethereum_consensus/fork_choice/fork_choice.ex

@@ -4,11 +4,10 @@ defmodule LambdaEthereumConsensus.ForkChoice do
   """
 
   require Logger
-
-  # alias LambdaEthereumConsensus.Beacon.Clock
   alias LambdaEthereumConsensus.Execution.ExecutionChain
   alias LambdaEthereumConsensus.ForkChoice.Handlers
   alias LambdaEthereumConsensus.ForkChoice.Head
+  alias LambdaEthereumConsensus.Libp2pPort
   alias LambdaEthereumConsensus.Metrics
   alias LambdaEthereumConsensus.P2P.Gossip.OperationsCollector
   alias LambdaEthereumConsensus.StateTransition.Misc
@@ -18,7 +17,6 @@ defmodule LambdaEthereumConsensus.ForkChoice do
   alias LambdaEthereumConsensus.Store.CheckpointStates
   alias LambdaEthereumConsensus.Store.StateDb
   alias LambdaEthereumConsensus.Store.StoreDb
-  alias LambdaEthereumConsensus.Validator.ValidatorManager
   alias Types.Attestation
   alias Types.BlockInfo
   alias Types.Checkpoint
@@ -271,7 +269,7 @@ defmodule LambdaEthereumConsensus.ForkChoice do
       store.finalized_checkpoint
     )
 
-    ValidatorManager.notify_new_block(slot, head_root)
+    Libp2pPort.notify_new_block({slot, head_root})
 
     Logger.info("[Fork choice] Updated fork choice cache", slot: slot)
 

lib/lambda_ethereum_consensus/validator/validator_manager.ex

@@ -1,21 +1,13 @@
-defmodule LambdaEthereumConsensus.Validator.ValidatorManager do
+defmodule LambdaEthereumConsensus.Validator.Setup do
   @moduledoc """
-  Module that manage the validators state
+  Module that setups the initial validators state
   """
-  use GenServer
 
   require Logger
-  alias LambdaEthereumConsensus.Beacon.Clock
   alias LambdaEthereumConsensus.Validator
 
-  @spec start_link({Types.slot(), Types.root()}) :: :ignore | {:error, any} | {:ok, pid}
-  def start_link({slot, head_root}) do
-    GenServer.start_link(__MODULE__, {slot, head_root}, name: __MODULE__)
-  end
-
-  @spec init({Types.slot(), Types.root()}) ::
-          {:ok, %{Bls.pubkey() => Validator.state()}} | {:stop, any}
-  def init({slot, head_root}) do
+  @spec init(Types.slot(), Types.root()) :: %{Bls.pubkey() => Validator.state()}
+  def init(slot, head_root) do
     config = Application.get_env(:lambda_ethereum_consensus, __MODULE__, [])
     keystore_dir = Keyword.get(config, :keystore_dir)
     keystore_pass_dir = Keyword.get(config, :keystore_pass_dir)
@@ -26,10 +18,10 @@ defmodule LambdaEthereumConsensus.Validator.ValidatorManager do
   defp setup_validators(_s, _r, keystore_dir, keystore_pass_dir)
        when is_nil(keystore_dir) or is_nil(keystore_pass_dir) do
     Logger.warning(
-      "[Validator Manager] No keystore_dir or keystore_pass_dir provided. Validator will not start."
+      "[Validator] No keystore_dir or keystore_pass_dir provided. Validator will not start."
     )
 
-    {:ok, []}
+    %{}
   end
 
   defp setup_validators(slot, head_root, keystore_dir, keystore_pass_dir) do
@@ -42,63 +34,11 @@ defmodule LambdaEthereumConsensus.Validator.ValidatorManager do
       end)
       |> Map.new()
 
-    Logger.info("[Validator Manager] Initialized #{Enum.count(validators)} validators")
-
-    {:ok, validators}
-  end
-
-  @spec notify_new_block({Types.slot(), Types.root()}) :: :ok
-  def notify_new_block({slot, head_root}) do
-    # Making this alone a cast sometimes solves the issue for a while
-    # GenServer.cast(__MODULE__, {:notify_all, {:new_block, slot, head_root}})
-    notify_validators({:new_block, slot, head_root})
-  end
-
-  @spec notify_tick(Clock.logical_time()) :: :ok
-  def notify_tick(logical_time) do
-    # Making this a cast alone doesn't solve the issue
-    # GenServer.cast(__MODULE__, {:notify_all, {:on_tick, logical_time}})
-    notify_validators({:on_tick, logical_time})
-  end
-
-  # TODO: The use of a Genserver and cast is still needed to avoid locking at the clock level.
-  # This is a temporary solution and will be taken off in a future PR.
-  defp notify_validators(msg), do: GenServer.call(__MODULE__, {:notify_all, msg}, 20_000)
+    Logger.info("[Validator] Initialized #{Enum.count(validators)} validators")
 
-  def handle_cast({:notify_all, msg}, validators) do
-    validators = notify_all(validators, msg)
-
-    {:noreply, validators}
+    validators
   end
 
-  def handle_call({:notify_all, msg}, _from, validators) do
-    validators = notify_all(validators, msg)
-
-    {:reply, :ok, validators}
-  end
-
-  defp notify_all(validators, msg) do
-    start_time = System.monotonic_time(:millisecond)
-
-    Logger.info("[Validator Manager] Notifying all Validators with message: #{inspect(msg)}")
-
-    updated_validators = Enum.map(validators, &notify_validator(&1, msg))
-
-    end_time = System.monotonic_time(:millisecond)
-
-    Logger.debug(
-      "[Validator Manager] #{inspect(msg)} notified to all Validators after #{end_time - start_time} ms"
-    )
-
-    updated_validators
-  end
-
-  defp notify_validator({pubkey, validator}, {:on_tick, logical_time}),
-    do: {pubkey, Validator.handle_tick(logical_time, validator)}
-
-  defp notify_validator({pubkey, validator}, {:new_block, slot, head_root}),
-    do: {pubkey, Validator.handle_new_block(slot, head_root, validator)}
-
   @doc """
     Get validator keys from the keystore directory.
     This function expects two files for each validator:
@@ -119,7 +59,7 @@ defmodule LambdaEthereumConsensus.Validator.ValidatorManager do
 
         {keystore_file, keystore_pass_file}
       else
-        Logger.warning("[Validator Manager] Skipping file: #{filename}. Not a keystore file.")
+        Logger.warning("[Validator] Skipping file: #{filename}. Not a keystore file.")
         nil
       end
     end)
@@ -131,7 +71,7 @@ defmodule LambdaEthereumConsensus.Validator.ValidatorManager do
       rescue
         error ->
           Logger.error(
-            "[Validator Manager] Failed to decode keystore file: #{keystore_file}. Pass file: #{keystore_pass_file} Error: #{inspect(error)}"
+            "[Validator] Failed to decode keystore file: #{keystore_file}. Pass file: #{keystore_pass_file} Error: #{inspect(error)}"
           )
 
           nil

lib/libp2p_port.ex

@@ -9,7 +9,6 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
 
   use GenServer
 
-  alias LambdaEthereumConsensus.Validator.ValidatorManager
   alias LambdaEthereumConsensus.Beacon.PendingBlocks
   alias LambdaEthereumConsensus.Beacon.SyncBlocks
   alias LambdaEthereumConsensus.ForkChoice
@@ -22,6 +21,7 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
   alias LambdaEthereumConsensus.P2p.Requests
   alias LambdaEthereumConsensus.StateTransition.Misc
   alias LambdaEthereumConsensus.Utils.BitVector
+  alias LambdaEthereumConsensus.Validator
   alias Libp2pProto.AddPeer
   alias Libp2pProto.Command
   alias Libp2pProto.Enr
@@ -63,6 +63,7 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
 
   @type init_arg ::
           {:genesis_time, Types.uint64()}
+          | {:validators, %{}}
           | {:listen_addr, [String.t()]}
           | {:enable_discovery, boolean()}
           | {:discovery_addr, String.t()}
@@ -113,6 +114,14 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
     GenServer.cast(__MODULE__, {:on_tick, time})
   end
 
+  @spec notify_new_block({Types.slot(), Types.root()}) :: :ok
+  def notify_new_block(data) do
+    # TODO: This is quick workarround to notify the libp2p port about new blocks from within
+    # the ForkChoice.recompute_head/1 without moving the validators to the store this
+    # allows to deferr that move until we simplify the state and remove duplicates.
+    send(self(), {:new_block, data})
+  end
+
   @doc """
   Retrieves identity info from the underlying LibP2P node.
   """
@@ -354,6 +363,7 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
   @impl GenServer
   def init(args) do
     {genesis_time, args} = Keyword.pop!(args, :genesis_time)
+    {validators, args} = Keyword.pop(args, :validators, %{})
     {join_init_topics, args} = Keyword.pop(args, :join_init_topics, false)
     {enable_request_handlers, args} = Keyword.pop(args, :enable_request_handlers, false)
 
@@ -379,6 +389,7 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
     {:ok,
      %{
        genesis_time: genesis_time,
+       validators: validators,
        slot_data: nil,
        port: port,
        subscribers: %{},
@@ -408,7 +419,10 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
   end
 
   @impl GenServer
-  def handle_cast({:on_tick, time}, %{genesis_time: genesis_time} = state) when time < genesis_time, do: {:noreply, state}
+  def handle_cast({:on_tick, time}, %{genesis_time: genesis_time} = state)
+      when time < genesis_time,
+      do: {:noreply, state}
+
   def handle_cast({:on_tick, time}, %{genesis_time: genesis_time, slot_data: slot_data} = state) do
     # TODO: we probably want to remove this from here, but we keep it here to have this serialized
     # with respect to the other fork choice store modifications.
@@ -417,13 +431,11 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
 
     new_slot_data = compute_slot(genesis_time, time)
 
-    if slot_data != new_slot_data do
-      ValidatorManager.notify_tick(slot_data)
-    end
+    updated_state = maybe_tick_validators(slot_data != new_slot_data, new_slot_data, state)
 
-    log_new_slot(new_slot_data)
+    log_new_slot(slot_data, new_slot_data)
 
-    {:noreply, %{state | slot_data: new_slot_data}}
+    {:noreply, updated_state}
   end
 
   def handle_cast(
@@ -486,6 +498,13 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
     {:noreply, new_state}
   end
 
+  @impl GenServer
+  def handle_info({:new_block, data}, %{validators: validators} = state) do
+    updated_validators = notify_validators(validators, {:new_block, data})
+
+    {:noreply, %{state | validators: updated_validators}}
+  end
+
   @impl GenServer
   def handle_info({_port, {:data, data}}, state) do
     %Notification{n: {_, payload}} = Notification.decode(data)
@@ -707,6 +726,38 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
     end)
   end
 
+  # Validator related functions
+
+  defp maybe_tick_validators(false = _slot_data_changed, _slot_data, state), do: state
+
+  defp maybe_tick_validators(true, slot_data, %{validators: validators} = state) do
+    updated_validators = notify_validators(validators, {:on_tick, slot_data})
+
+    %{state | slot_data: slot_data, validators: updated_validators}
+  end
+
+  defp notify_validators(validators, msg) do
+    start_time = System.monotonic_time(:millisecond)
+
+    Logger.info("[Libp2p] Notifying all Validators with message: #{inspect(msg)}")
+
+    updated_validators = Enum.map(validators, &notify_validator(&1, msg))
+
+    end_time = System.monotonic_time(:millisecond)
+
+    Logger.debug(
+      "[Validator Manager] #{inspect(msg)} notified to all Validators after #{end_time - start_time} ms"
+    )
+
+    updated_validators
+  end
+
+  defp notify_validator({pubkey, validator}, {:on_tick, slot_data}),
+    do: {pubkey, Validator.handle_tick(slot_data, validator)}
+
+  defp notify_validator({pubkey, validator}, {:new_block, {slot, head_root}}),
+    do: {pubkey, Validator.handle_new_block(slot, head_root, validator)}
+
   @spec compute_slot(Types.uint64(), Types.uint64()) :: slot_data()
   defp compute_slot(genesis_time, time) do
     # TODO: This was copied as it is from the Clock to convert it into just a Ticker,
@@ -726,12 +777,14 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
     {slot, slot_third}
   end
 
-  defp log_new_slot({slot, :first_third}) do
+  defp log_new_slot({slot, _third}, {slot, _third}), do: :ok
+
+  defp log_new_slot({_prev_slot, _thrid}, {slot, :first_third}) do
     # TODO: as with the previous function, this was copied from the Clock module.
-    # is use :sync, :store as the slot event, probably something to look into.
+    # It use :sync, :store as the slot event, probably something to look into.
     :telemetry.execute([:sync, :store], %{slot: slot})
     Logger.info("[Libp2p] Slot transition", slot: slot)
   end
 
-  defp log_new_slot(_), do: :ok
+  defp log_new_slot(_, _), do: :ok
 end

test/unit/libp2p_port_test.exs

@@ -17,7 +17,9 @@ defmodule Unit.Libp2pPortTest do
   end
 
   defp start_port(name \\ Libp2pPort, init_args \\ []) do
-    start_link_supervised!({Libp2pPort, [opts: [name: name], genesis_time: 42] ++ init_args}, id: name)
+    start_link_supervised!({Libp2pPort, [opts: [name: name], genesis_time: 42] ++ init_args},
+      id: name
+    )
   end
 
   @tag :tmp_dir