feat: add support for multiple validators.

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,10 @@ 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)
+config :lambda_ethereum_consensus, LambdaEthereumConsensus.Validator.Supervisor,
+  keystore_dir: keystore_dir,
+  keystore_pass_dir: keystore_pass_dir
 
-  config :lambda_ethereum_consensus, LambdaEthereumConsensus.Validator,
-    pubkey: pubkey,
-    privkey: privkey
-end
-
-# Metrics
-
-# Configures metrics
 # TODO: we should set this dynamically
 block_time_ms =
   case network do
@@ -175,6 +168,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_node.ex

@@ -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
+    [
+      {Validator.Supervisor, {slot, head_root}},
+      {LambdaEthereumConsensus.Execution.ExecutionChain, {genesis_time, snapshot, votes}}
+    ]
   end
 
   defp get_libp2p_args() do

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
 

lib/lambda_ethereum_consensus/validator/supervisor.ex

@@ -0,0 +1,66 @@
+defmodule LambdaEthereumConsensus.Validator.Supervisor do
+  @moduledoc false
+
+  use Supervisor
+
+  require Logger
+  alias LambdaEthereumConsensus.Validator
+
+  def start_link(opts) do
+    Supervisor.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @impl true
+  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)
+
+    if keystore_dir == nil or keystore_pass_dir == nil do
+      Logger.warning(
+        "[Validator] No keystore_dir or keystore_pass_dir provided. Validator will not start."
+      )
+
+      :ignore
+    else
+      validator_keys = get_validator_keys(keystore_dir, keystore_pass_dir)
+
+      children =
+        validator_keys
+        |> Enum.map(fn {pubkey, privkey} ->
+          Supervisor.child_spec({Validator, {slot, head_root, {pubkey, privkey}}},
+            id: pubkey |> Base.encode16(case: :lower) |> String.to_atom()
+          )
+        end)
+
+      Supervisor.init(children, strategy: :one_for_one)
+    end
+  end
+
+  @spec get_validator_keys(binary(), binary()) :: list({Bls.pubkey(), Bls.privkey()})
+  defp get_validator_keys(keystore_dir, keystore_pass_dir) do
+    keystore_files = File.ls!(keystore_dir) |> Enum.sort()
+    keystore_pass_files = File.ls!(keystore_pass_dir) |> Enum.sort()
+
+    Enum.zip(keystore_files, keystore_pass_files)
+    |> Enum.map(fn {keystore_file, keystore_pass_file} ->
+      keystore_file = Path.join(keystore_dir, keystore_file)
+      keystore_pass_file = Path.join(keystore_pass_dir, keystore_pass_file)
+
+      # TODO: remove `try` and handle errors properly
+      # TODO: match keystore file and pass file based on name
+      try do
+        Keystore.decode_from_files!(keystore_file, keystore_pass_file)
+      rescue
+        error ->
+          Logger.error(
+            "[Validator] Failed to decode keystore file: #{keystore_file}. Pass file: #{keystore_pass_file} Error: #{inspect(error)}"
+          )
+
+          nil
+      end
+    end)
+    |> Enum.filter(&is_tuple/1)
+  end
+end

lib/lambda_ethereum_consensus/validator/validator.ex

@@ -26,7 +26,11 @@ defmodule LambdaEthereumConsensus.Validator do
   ### Public API
   ##########################
 
-  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  def start_link({_, _, {pubkey, _}} = opts) do
+    # TODO: if possible, use validator index instead of pubkey
+    name = Atom.to_string(__MODULE__) <> "_" <> Base.encode16(pubkey, case: :lower)
+    GenServer.start_link(__MODULE__, opts, name: String.to_atom(name))
+  end
 
   def notify_new_block(slot, head_root),
     do: GenServer.cast(__MODULE__, {:new_block, slot, head_root})
@@ -35,29 +39,39 @@ defmodule LambdaEthereumConsensus.Validator do
   ### GenServer Callbacks
   ##########################
 
-  @impl true
-  def init({slot, head_root}) do
-    config = Application.get_env(:lambda_ethereum_consensus, __MODULE__, [])
-
-    validator =
-      case {Keyword.get(config, :pubkey), Keyword.get(config, :privkey)} do
-        {nil, nil} -> nil
-        {pubkey, privkey} -> %{index: nil, privkey: privkey, pubkey: pubkey}
-      end
+  @type state :: %{
+          slot: Types.slot(),
+          root: Types.root(),
+          duties: %{
+            attester: list(:not_computed),
+            proposer: :not_computed | list(Types.slot())
+          },
+          validator: any(),
+          payload_builder: {Types.slot(), Types.root(), BlockBuilder.payload_id()} | nil
+        }
 
+  @impl true
+  @spec init({Types.slot(), Types.root(), {Bls.pubkey(), Bls.privkey()}}) ::
+          {:ok, state, {:continue, any}}
+  def init({head_slot, head_root, {pubkey, privkey}}) do
     state = %{
-      slot: slot,
+      slot: head_slot,
       root: head_root,
       duties: empty_duties(),
-      validator: validator
+      validator: %{
+        pubkey: pubkey,
+        privkey: privkey,
+        index: nil
+      },
+      payload_builder: nil
     }
 
     {:ok, state, {:continue, nil}}
   end
 
-  @impl true
-  def handle_continue(nil, %{validator: nil} = state), do: {:noreply, state}
+  @spec handle_continue(nil, state) :: {:noreply, state}
 
+  @impl true
   def handle_continue(nil, %{slot: slot, root: root} = state) do
     case try_setup_validator(state, slot, root) do
       nil ->
@@ -69,6 +83,7 @@ defmodule LambdaEthereumConsensus.Validator do
     end
   end
 
+  @spec try_setup_validator(state, Types.slot(), Types.root()) :: state | nil
   defp try_setup_validator(state, slot, root) do
     epoch = Misc.compute_epoch_at_slot(slot)
     beacon = fetch_target_state(epoch, root)
@@ -87,6 +102,8 @@ defmodule LambdaEthereumConsensus.Validator do
     end
   end
 
+  @spec handle_cast(any, state) :: {:noreply, state}
+
   @impl true
   def handle_cast(_, %{validator: nil} = state), do: {:noreply, state}
 
@@ -186,6 +203,7 @@ defmodule LambdaEthereumConsensus.Validator do
     %{state | slot: slot, root: head_root, duties: new_duties}
   end
 
+  @spec fetch_target_state(Types.epoch(), Types.root()) :: Types.BeaconState.t()
   defp fetch_target_state(epoch, root) do
     {:ok, state} = Handlers.compute_target_checkpoint_state(epoch, root)
     state
@@ -458,6 +476,8 @@ defmodule LambdaEthereumConsensus.Validator do
     Map.put(duty, :subnet_id, subnet_id)
   end
 
+  @spec fetch_validator_index(Types.BeaconState.t(), %{index: nil, pubkey: Bls.pubkey()}) ::
+          non_neg_integer() | nil
   defp fetch_validator_index(beacon, %{index: nil, pubkey: pk}) do
     Enum.find_index(beacon.validators, &(&1.pubkey == pk))
   end
@@ -501,7 +521,7 @@ defmodule LambdaEthereumConsensus.Validator do
   end
 
   defp propose(%{root: head_root, validator: validator} = state, proposed_slot) do
-    {^proposed_slot, ^head_root, payload_id} = state.block_builder
+    {^proposed_slot, ^head_root, payload_id} = state.payload_builder
 
     build_result =
       BlockBuilder.build_block(