feat: add support for multiple validators. (#1080)

Author: mpaulucci

config/runtime.exs

@@ -21,8 +21,8 @@ switches = [
   listen_address: [:string, :keep],
   discovery_port: :integer,
   boot_nodes: :string,
-  keystore_file: :string,
-  keystore_password_file: :string
+  keystore_dir: :string,
+  keystore_pass_dir: :string
 ]
 
 is_testing = Config.config_env() == :test
@@ -50,8 +50,8 @@ enable_beacon_api = Keyword.get(args, :beacon_api, not is_nil(beacon_api_port))
 listen_addresses = Keyword.get_values(args, :listen_address)
 discovery_port = Keyword.get(args, :discovery_port, 9000)
 cli_bootnodes = Keyword.get(args, :boot_nodes, "")
-keystore = Keyword.get(args, :keystore_file)
-keystore_pass = Keyword.get(args, :keystore_password_file)
+keystore_dir = Keyword.get(args, :keystore_dir)
+keystore_pass_dir = Keyword.get(args, :keystore_pass_dir)
 
 if not is_nil(testnet_dir) and not is_nil(checkpoint_sync_url) do
   IO.puts("Both checkpoint sync and testnet url specified (only one should be specified).")
@@ -153,17 +153,28 @@ config :lambda_ethereum_consensus, BeaconApi.Endpoint,
     layout: false
   ]
 
-if is_binary(keystore) and is_binary(keystore_pass) do
-  {pubkey, privkey} = Keystore.decode_from_files!(keystore, keystore_pass)
+# Validator setup
 
-  config :lambda_ethereum_consensus, LambdaEthereumConsensus.Validator,
-    pubkey: pubkey,
-    privkey: privkey
+if (keystore_dir != nil and keystore_pass_dir == nil) or
+     (keystore_pass_dir !== nil and keystore_dir == nil) do
+  IO.puts("Both keystore_dir and keystore_pass_dir must be provided.")
+  System.halt(2)
 end
 
-# Metrics
+if keystore_dir != nil and not File.dir?(keystore_dir) do
+  IO.puts("Keystore directory not found: #{keystore_dir}")
+  System.halt(2)
+end
+
+if keystore_pass_dir != nil and not File.dir?(keystore_pass_dir) do
+  IO.puts("Keystore password directory not found: #{keystore_pass_dir}")
+  System.halt(2)
+end
+
+config :lambda_ethereum_consensus, LambdaEthereumConsensus.Validator.ValidatorManager,
+  keystore_dir: keystore_dir,
+  keystore_pass_dir: keystore_pass_dir
 
-# Configures metrics
 # TODO: we should set this dynamically
 block_time_ms =
   case network do
@@ -175,6 +186,8 @@ block_time_ms =
     _ -> 12_000
   end
 
+# Metrics
+
 config :lambda_ethereum_consensus, LambdaEthereumConsensus.Telemetry,
   block_processing_buckets: [0.5, 1.0, 1.5, 2, 4, 6, 8] |> Enum.map(&(&1 * block_time_ms)),
   port: metrics_port

lib/lambda_ethereum_consensus/beacon/beacon_chain.ex

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

lib/lambda_ethereum_consensus/beacon/beacon_node.ex

@@ -9,7 +9,7 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
   alias LambdaEthereumConsensus.StateTransition.Cache
   alias LambdaEthereumConsensus.Store.Blocks
   alias LambdaEthereumConsensus.Store.BlockStates
-  alias LambdaEthereumConsensus.Validator
+  alias LambdaEthereumConsensus.Validator.ValidatorManager
   alias Types.BeaconState
 
   def start_link(opts) do
@@ -65,17 +65,13 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
     []
   end
 
-  defp get_validator_children(snapshot, head_slot, head_root, genesis_time) do
-    if is_nil(Application.get_env(:lambda_ethereum_consensus, Validator)) do
-      []
-    else
-      %BeaconState{eth1_data_votes: votes} = BlockStates.get_state!(head_root)
-      # TODO: move checkpoint sync outside and move this to application.ex
-      [
-        {Validator, {head_slot, head_root}},
-        {LambdaEthereumConsensus.Execution.ExecutionChain, {genesis_time, snapshot, votes}}
-      ]
-    end
+  defp get_validator_children(snapshot, slot, head_root, genesis_time) do
+    %BeaconState{eth1_data_votes: votes} = BlockStates.get_state!(head_root)
+    # TODO: move checkpoint sync outside and move this to application.ex
+    [
+      {ValidatorManager, {slot, head_root}},
+      {LambdaEthereumConsensus.Execution.ExecutionChain, {genesis_time, snapshot, votes}}
+    ]
   end
 
   defp get_libp2p_args() do

lib/lambda_ethereum_consensus/execution/execution_client.ex

@@ -64,8 +64,8 @@ defmodule LambdaEthereumConsensus.Execution.ExecutionClient do
   """
   def notify_forkchoice_updated(fork_choice_state, payload_attributes) do
     case EngineApi.forkchoice_updated(fork_choice_state, payload_attributes) do
-      {:ok, %{"payload_id" => nil, "payload_status" => %{"status" => status}}} ->
-        {:error, "No payload id, status is #{parse_status(status)}"}
+      {:ok, %{"payload_id" => nil, "payload_status" => payload_status}} ->
+        {:error, "No payload id, status is #{inspect(payload_status)}"}
 
       {:ok, %{"payload_id" => payload_id, "payload_status" => %{"status" => "VALID"}}} ->
         {:ok, payload_id}

lib/lambda_ethereum_consensus/fork_choice/fork_choice.ex

@@ -14,7 +14,7 @@ defmodule LambdaEthereumConsensus.ForkChoice do
   alias LambdaEthereumConsensus.Store.Blocks
   alias LambdaEthereumConsensus.Store.StateDb
   alias LambdaEthereumConsensus.Store.StoreDb
-  alias LambdaEthereumConsensus.Validator
+  alias LambdaEthereumConsensus.Validator.ValidatorManager
   alias Types.Attestation
   alias Types.BeaconState
   alias Types.SignedBeaconBlock
@@ -194,7 +194,7 @@ defmodule LambdaEthereumConsensus.ForkChoice do
     %{slot: slot, body: body} = head_block
 
     OperationsCollector.notify_new_block(head_block)
-    Validator.notify_new_block(slot, head_root)
+    ValidatorManager.notify_new_block(slot, head_root)
     ExecutionChain.notify_new_block(slot, body.eth1_data, body.execution_payload)
 
     BeaconChain.update_fork_choice_cache(

lib/lambda_ethereum_consensus/logger/console_logger.ex

@@ -2,6 +2,7 @@ defmodule ConsoleLogger do
   @moduledoc """
   Custom logger formatter for console output.
   """
+  alias LambdaEthereumConsensus.Utils
 
   @pattern Logger.Formatter.compile(" $time $message ")
 
@@ -42,9 +43,7 @@ defmodule ConsoleLogger do
   end
 
   def format_metadata_value(:root, root) do
-    encoded = root |> Base.encode16(case: :lower)
-    # get the first 3 and last 4 characters
-    "0x#{String.slice(encoded, 0, 3)}..#{String.slice(encoded, -4, 4)}"
+    Utils.format_shorten_binary(root)
   end
 
   def format_metadata_value(:slot, slot) do

lib/lambda_ethereum_consensus/utils.ex

@@ -38,4 +38,13 @@ defmodule LambdaEthereumConsensus.Utils do
   @spec map_err(any() | {:error, String.t()}, String.t()) :: any() | {:error, String.t()}
   def map_err({:error, _}, reason), do: {:error, reason}
   def map_err(v, _), do: v
+
+  @doc """
+  Format a binary to a shortened hexadecimal representation.
+  """
+  @spec format_shorten_binary(binary) :: String.t()
+  def format_shorten_binary(binary) do
+    encoded = binary |> Base.encode16(case: :lower)
+    "0x#{String.slice(encoded, 0, 3)}..#{String.slice(encoded, -4, 4)}"
+  end
 end

lib/lambda_ethereum_consensus/validator/block_builder.ex

@@ -202,8 +202,16 @@ defmodule LambdaEthereumConsensus.Validator.BlockBuilder do
   end
 
   defp get_finalized_block_hash(state) do
-    finalized_block = Blocks.get_block!(state.finalized_checkpoint.root)
-    finalized_hash = finalized_block.body.execution_payload.block_hash
+    finalized_root = state.finalized_checkpoint.root
+
+    finalized_hash =
+      if finalized_root == <<0::256>> do
+        <<0::256>>
+      else
+        finalized_block = Blocks.get_block!(state.finalized_checkpoint.root)
+        finalized_block.body.execution_payload.block_hash
+      end
+
     {:ok, finalized_hash}
   end