Merge branch 'main' into validator-state-management-refactor

Author: rodrigo-o

lib/lambda_ethereum_consensus/fork_choice/fork_choice.ex

@@ -259,7 +259,7 @@ defmodule LambdaEthereumConsensus.ForkChoice do
     %{slot: slot, body: body} = head_block
 
     OperationsCollector.notify_new_block(head_block)
-    Libp2pPort.notify_new_block(slot, head_root)
+    Libp2pPort.notify_new_head(slot, head_root)
     ExecutionChain.notify_new_block(slot, body.eth1_data, body.execution_payload)
 
     update_fork_choice_data(

lib/lambda_ethereum_consensus/validator/setup.ex

@@ -18,7 +18,7 @@ defmodule LambdaEthereumConsensus.Validator.Setup 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] No keystore_dir or keystore_pass_dir provided. Validators won't start."
+      "[Validator] No keystore_dir or keystore_pass_dir provided. Validator will not start."
     )
 
     []
@@ -46,7 +46,9 @@ defmodule LambdaEthereumConsensus.Validator.Setup do
       when is_binary(keystore_dir) and is_binary(keystore_pass_dir) do
     keystore_dir
     |> File.ls!()
-    |> map_rejecting_nils(&paths_from_filename(keystore_dir, keystore_pass_dir, &1, Path.extname(&1)))
+    |> map_rejecting_nils(
+      &paths_from_filename(keystore_dir, keystore_pass_dir, &1, Path.extname(&1))
+    )
     |> map_rejecting_nils(&decode_key/1)
   end
 
@@ -78,6 +80,29 @@ defmodule LambdaEthereumConsensus.Validator.Setup do
     nil
   end
 
+  @spec notify_validators([Validator.state()], tuple()) :: [Validator.state()]
+  def notify_validators(validators, msg) do
+    start_time = System.monotonic_time(:millisecond)
+
+    Logger.debug("[Validator] 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] #{inspect(msg)} notified to all Validators after #{end_time - start_time} ms"
+    )
+
+    updated_validators
+  end
+
+  defp notify_validator(validator, {:on_tick, slot_data}),
+    do: Validator.handle_tick(slot_data, validator)
+
+  defp notify_validator(validator, {:new_head, slot, head_root}),
+    do: Validator.handle_new_head(slot, head_root, validator)
+
   defp map_rejecting_nils(enumerable, fun) do
     Enum.reduce(enumerable, [], fn elem, acc ->
       case fun.(elem) do

lib/lambda_ethereum_consensus/validator/validator.ex

@@ -95,8 +95,8 @@ defmodule LambdaEthereumConsensus.Validator do
     end
   end
 
-  @spec handle_new_block(Types.slot(), Types.root(), state) :: state
-  def handle_new_block(slot, head_root, %{validator: %{index: nil}} = state) do
+  @spec handle_new_head(Types.slot(), Types.root(), state) :: state
+  def handle_new_head(slot, head_root, %{validator: %{index: nil}} = state) do
     log_error("-1", "setup validator", "index not present handle block",
       slot: slot,
       root: head_root
@@ -105,8 +105,8 @@ defmodule LambdaEthereumConsensus.Validator do
     state
   end
 
-  def handle_new_block(slot, head_root, state) do
-    log_debug(state.validator.index, "recieved new block", slot: slot, root: head_root)
+  def handle_new_head(slot, head_root, state) do
+    log_debug(state.validator.index, "recieved new head", slot: slot, root: head_root)
 
     # TODO: this doesn't take into account reorgs
     state

lib/libp2p_port.ex

@@ -111,12 +111,14 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
     GenServer.start_link(__MODULE__, args, opts)
   end
 
-  @spec notify_new_block(Types.slot(), Types.root()) :: :ok
-  def notify_new_block(slot, head_root) do
-    # TODO: This is quick workarround to notify the libp2p port about new blocks from within
+  @spec notify_new_head(Types.slot(), Types.root()) :: :ok
+  def notify_new_head(slot, head_root) do
+    # TODO: This is quick workarround to notify the libp2p port about new heads 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, slot, head_root})
+    # THIS IS NEEDED BECAUSE FORKCHOICE IS CURRENTLY RUNNING ON LIBP2P PORT.
+    # It could be a simple cast in the future if that's not the case anymore.
+    send(self(), {:new_head, slot, head_root})
   end
 
   @doc """
@@ -503,8 +505,9 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
   end
 
   @impl GenServer
-  def handle_info({:new_block, slot, head_root}, %{validators: validators} = state) do
-    updated_validators = notify_validators(validators, {:new_block, slot, head_root})
+  def handle_info({:new_head, slot, head_root}, %{validators: validators} = state) do
+    updated_validators =
+      Validator.Setup.notify_validators(validators, {:new_head, slot, head_root})
 
     {:noreply, %{state | validators: updated_validators}}
   end
@@ -740,43 +743,21 @@ defmodule LambdaEthereumConsensus.Libp2pPort do
 
     new_slot_data = compute_slot(genesis_time, time)
 
-    updated_state = maybe_tick_validators(slot_data != new_slot_data, new_slot_data, state)
+    updated_state =
+      if slot_data == new_slot_data do
+        state
+      else
+        updated_validators =
+          Validator.Setup.notify_validators(state.validators, {:on_tick, new_slot_data})
+
+        %{state | slot_data: new_slot_data, validators: updated_validators}
+      end
 
     maybe_log_new_slot(slot_data, new_slot_data)
 
     updated_state
   end
 
-  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.debug("[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(validator, {:on_tick, slot_data}),
-    do: Validator.handle_tick(slot_data, validator)
-
-  defp notify_validator(validator, {:new_block, slot, head_root}),
-    do: Validator.handle_new_block(slot, head_root, validator)
-
   defp schedule_next_tick() do
     # For millisecond precision
     time_to_next_tick = @tick_time - rem(:os.system_time(:millisecond), @tick_time)

test/unit/beacon_api/beacon_api_v1_test.exs

@@ -159,7 +159,7 @@ defmodule Unit.BeaconApiTest.V1 do
     alias LambdaEthereumConsensus.P2P.Metadata
     patch(ForkChoice, :get_fork_version, fn -> ChainSpec.get("DENEB_FORK_VERSION") end)
 
-    start_link_supervised!({Libp2pPort, genesis_time: 42})
+    start_link_supervised!({Libp2pPort, genesis_time: :os.system_time(:second)})
     Metadata.init()
     identity = Libp2pPort.get_node_identity()
     metadata = Metadata.get_metadata()

test/unit/libp2p_port_test.exs

@@ -17,7 +17,7 @@ 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},
+    start_link_supervised!({Libp2pPort, [opts: [name: name], genesis_time: :os.system_time(:second)] ++ init_args},
       id: name
     )
   end