Initial test of sync aggregates without publishing

Author: rodrigo-o

lib/constants.ex

@@ -36,6 +36,9 @@ defmodule Constants do
   @spec target_aggregators_per_committee() :: non_neg_integer()
   def target_aggregators_per_committee(), do: 16
 
+  @spec target_aggregators_per_sync_subcommittee() :: non_neg_integer()
+  def target_aggregators_per_sync_subcommittee(), do: 16
+
   ### Withdrawal prefixes
 
   @spec bls_withdrawal_prefix() :: Types.bytes1()

lib/lambda_ethereum_consensus/p2p/gossip/sync_committee.ex

@@ -0,0 +1,26 @@
+defmodule LambdaEthereumConsensus.P2P.Gossip.SyncCommittee do
+  @moduledoc """
+  This module handles sync committee from specific gossip subnets.
+  Used by validators to fulfill aggregation duties.
+  """
+  alias LambdaEthereumConsensus.ForkChoice
+
+  require Logger
+
+  @spec publish(Types.SyncCommitteeMessage.t(), [non_neg_integer()]) :: :ok
+  def publish(%Types.SyncCommitteeMessage{} = message, subnet_ids) do
+    Enum.each(subnet_ids, fn subnet_id ->
+      topic = topic(subnet_id)
+
+      Logger.info(
+        "[SyncCommittee] Publishing attestation, topic: #{topic}, message #{inspect(message, pretty: true)}"
+      )
+    end)
+  end
+
+  defp topic(subnet_id) do
+    # TODO: this doesn't take into account fork digest changes
+    fork_context = ForkChoice.get_fork_digest() |> Base.encode16(case: :lower)
+    "/eth2/#{fork_context}/sync_committee_#{subnet_id}/ssz_snappy"
+  end
+end

lib/lambda_ethereum_consensus/state_transition/misc.ex

@@ -281,6 +281,15 @@ defmodule LambdaEthereumConsensus.StateTransition.Misc do
     {committee_start, committee_end}
   end
 
+  @doc """
+  Compute the sync committee period for the given ``epoch``. This is used to determine the
+  period in which a validator is assigned to the sync committee.
+  """
+  @spec compute_sync_committee_period(Types.epoch()) :: Types.uint64()
+  def compute_sync_committee_period(epoch) do
+    div(epoch, ChainSpec.get("EPOCHS_PER_SYNC_COMMITTEE_PERIOD"))
+  end
+
   @doc """
   Return the 32-byte fork data root for the ``current_version`` and ``genesis_validators_root``.
   This is used primarily in signature domains to avoid collisions across forks/chains.

lib/lambda_ethereum_consensus/validator/duties.ex

@@ -22,16 +22,26 @@ defmodule LambdaEthereumConsensus.Validator.Duties do
           index_in_committee: Types.uint64()
         }
 
-  @type proposer_duty :: Types.slot()
+  @type proposer_duty :: Types.validator_index()
+
+  @type sync_committee_duty :: %{
+          last_slot_broadcasted: Types.slot(),
+          subnet_ids: [Types.uint64()],
+          validator_index: Types.validator_index()
+        }
 
   @type attester_duties :: [attester_duty()]
   @type proposer_duties :: [proposer_duty()]
+  @type sync_committee_duties :: [sync_committee_duty()]
 
   @type attester_duties_per_slot :: %{Types.slot() => attester_duties()}
   @type proposer_duties_per_slot :: %{Types.slot() => proposer_duties()}
 
-  @type kind :: :proposers | :attesters
-  @type duties :: %{kind() => attester_duties_per_slot() | proposer_duties_per_slot()}
+  @type kind :: :proposers | :attesters | :sync_committees
+  @type duties :: %{
+          kind() =>
+            attester_duties_per_slot() | proposer_duties_per_slot() | sync_committee_duties()
+        }
 
   ############################
   # Accessors
@@ -40,6 +50,15 @@ defmodule LambdaEthereumConsensus.Validator.Duties do
   def current_proposer(duties, epoch, slot),
     do: get_in(duties, [epoch, :proposers, slot])
 
+  @spec current_sync_committee(duties(), Types.epoch(), Types.slot()) ::
+          sync_committee_duties()
+  def current_sync_committee(duties, epoch, slot) do
+    for %{last_slot_broadcasted: last_slot} = duty <- sync_committee(duties, epoch),
+        last_slot < slot do
+      duty
+    end
+  end
+
   @spec current_attesters(duties(), Types.epoch(), Types.slot()) :: attester_duties()
   def current_attesters(duties, epoch, slot) do
     for %{attested?: false} = duty <- attesters(duties, epoch, slot) do
@@ -54,6 +73,7 @@ defmodule LambdaEthereumConsensus.Validator.Duties do
     end
   end
 
+  defp sync_committee(duties, epoch), do: get_in(duties, [epoch, :sync_committees]) || []
   defp attesters(duties, epoch, slot), do: get_in(duties, [epoch, :attesters, slot]) || []
 
   ############################
@@ -75,6 +95,9 @@ defmodule LambdaEthereumConsensus.Validator.Duties do
   @spec aggregated(attester_duty()) :: attester_duty()
   def aggregated(duty), do: Map.put(duty, :should_aggregate?, false)
 
+  @spec sync_committee_broadcasted(sync_committee_duty(), Types.slot()) :: sync_committee_duty()
+  def sync_committee_broadcasted(duty, slot), do: Map.put(duty, :last_slot_broadcasted, slot)
+
   ############################
   # Main functions
 
@@ -92,6 +115,19 @@ defmodule LambdaEthereumConsensus.Validator.Duties do
     end
   end
 
+  @spec compute_current_sync_committees(BeaconState.t(), ValidatorSet.validators()) ::
+          sync_committee_duties()
+  def compute_current_sync_committees(%BeaconState{} = state, validators) do
+    for validator_index <- Map.keys(validators),
+        subnet_ids = Utils.compute_subnets_for_sync_committee(state, validator_index) do
+      %{
+        last_slot_broadcasted: -1,
+        subnet_ids: subnet_ids,
+        validator_index: validator_index
+      }
+    end
+  end
+
   @spec compute_attesters_for_epoch(BeaconState.t(), Types.epoch(), ValidatorSet.validators()) ::
           attester_duties_per_slot()
   def compute_attesters_for_epoch(%BeaconState{} = state, epoch, validators) do

lib/lambda_ethereum_consensus/validator/utils.ex

@@ -52,4 +52,67 @@ defmodule LambdaEthereumConsensus.Validator.Utils do
     |> :binary.decode_unsigned(:little)
     |> rem(modulo) == 0
   end
+
+  @spec compute_subnets_for_sync_committee(BeaconState.t(), Types.validator_index()) :: [
+          Types.uint64()
+        ]
+  def compute_subnets_for_sync_committee(%BeaconState{} = state, validator_index) do
+    target_pubkey = state.validators[validator_index].pubkey
+    current_epoch = Accessors.get_current_epoch(state)
+    next_slot_epoch = Misc.compute_epoch_at_slot(state.slot + 1)
+    current_sync_committee_period = Misc.compute_sync_committee_period(current_epoch)
+    next_slot_sync_committee_period = Misc.compute_sync_committee_period(next_slot_epoch)
+
+    sync_committee =
+      if current_sync_committee_period == next_slot_sync_committee_period,
+        do: state.current_sync_committee,
+        else: state.next_sync_committee
+
+    sync_committee_subnet_size =
+      div(ChainSpec.get("SYNC_COMMITTEE_SIZE"), Constants.sync_committee_subnet_count())
+
+    for {pubkey, index} <- Enum.with_index(sync_committee.pubkeys),
+        pubkey == target_pubkey do
+      div(index, sync_committee_subnet_size)
+    end
+    |> Enum.dedup()
+  end
+
+  # `is_assigned_to_sync_committee` equivalent
+  @spec assigned_to_sync_committee?(BeaconState.t(), Types.epoch(), Types.validator_index()) ::
+          boolean()
+  def assigned_to_sync_committee?(%BeaconState{} = state, epoch, validator_index) do
+    sync_committee_period = Misc.compute_sync_committee_period(epoch)
+    current_epoch = Accessors.get_current_epoch(state)
+    current_sync_committee_period = Misc.compute_sync_committee_period(current_epoch)
+    next_sync_committee_period = current_sync_committee_period + 1
+
+    pubkey = state.validators[validator_index].pubkey
+
+    case sync_committee_period do
+      ^current_sync_committee_period ->
+        Enum.member?(state.current_sync_committee.pubkeys, pubkey)
+
+      ^next_sync_committee_period ->
+        Enum.member?(state.next_sync_committee.pubkeys, pubkey)
+
+      _ ->
+        raise ArgumentError,
+              "Invalid epoch #{epoch}, should be in the current or next sync committee period"
+    end
+  end
+
+  @spec sync_committee_aggregator?(Types.bls_signature()) :: boolean()
+  def sync_committee_aggregator?(signature) do
+    modulo =
+      ChainSpec.get("SYNC_COMMITTEE_SIZE")
+      |> div(Constants.sync_committee_subnet_count())
+      |> div(Constants.target_aggregators_per_sync_subcommittee())
+      |> max(1)
+
+    SszEx.hash(signature)
+    |> binary_part(0, 8)
+    |> :binary.decode_unsigned(:little)
+    |> rem(modulo) == 0
+  end
 end

lib/lambda_ethereum_consensus/validator/validator.ex

@@ -228,6 +228,54 @@ defmodule LambdaEthereumConsensus.Validator do
     Enum.find_index(beacon.validators, &(&1.pubkey == pubkey))
   end
 
+  ################################
+  # Sync Committee
+
+  @spec sync_committee_message_broadcast(
+          t(),
+          Duties.sync_committee_duty(),
+          Types.slot(),
+          Types.root()
+        ) ::
+          :ok
+  def sync_committee_message_broadcast(
+        %{index: validator_index, keystore: keystore},
+        current_duty,
+        slot,
+        head_root
+      ) do
+    %{subnet_ids: subnet_ids} = current_duty
+
+    head_state = BlockStates.get_state_info!(head_root).beacon_state |> go_to_slot(slot)
+    log_debug(validator_index, "broadcasting sync committee message", slot: slot)
+
+    head_state
+    |> get_sync_committee_message(head_root, validator_index, keystore.privkey)
+    |> Gossip.SyncCommittee.publish(subnet_ids)
+    |> log_info_result(validator_index, "published sync committee message", slot: slot)
+  end
+
+  @spec get_sync_committee_message(
+          Types.BeaconState.t(),
+          Types.root(),
+          Types.validator_index(),
+          Bls.privkey()
+        ) ::
+          Types.SyncCommitteeMessage.t()
+  def get_sync_committee_message(head_state, head_root, validator_index, privkey) do
+    epoch = Accessors.get_current_epoch(head_state)
+    domain = Accessors.get_domain(head_state, Constants.domain_sync_committee(), epoch)
+    signing_root = Misc.compute_signing_root(head_root, domain)
+    {:ok, signature} = Bls.sign(privkey, signing_root)
+
+    %Types.SyncCommitteeMessage{
+      slot: head_state.slot,
+      beacon_block_root: head_root,
+      validator_index: validator_index,
+      signature: signature
+    }
+  end
+
   ################################
   # Payload building and proposing
 

lib/lambda_ethereum_consensus/validator/validator_set.ex

@@ -80,6 +80,7 @@ defmodule LambdaEthereumConsensus.ValidatorSet do
     |> update_state(epoch, slot, head_root)
     |> attests(epoch, slot, head_root)
     |> build_payload(slot + 1, head_root)
+    |> sync_committee_broadcasts(epoch, slot, head_root)
   end
 
   @doc """
@@ -103,6 +104,7 @@ defmodule LambdaEthereumConsensus.ValidatorSet do
     set
     |> attests(epoch, slot, head_root)
     |> build_payload(slot + 1, head_root)
+    |> sync_committee_broadcasts(epoch, slot, head_root)
   end
 
   defp process_tick(set, epoch, {slot, :last_third}) do
@@ -140,7 +142,8 @@ defmodule LambdaEthereumConsensus.ValidatorSet do
 
     duties = %{
       proposers: Duties.compute_proposers_for_epoch(beacon, epoch, set.validators),
-      attesters: Duties.compute_attesters_for_epoch(beacon, epoch, set.validators)
+      attesters: Duties.compute_attesters_for_epoch(beacon, epoch, set.validators),
+      sync_committees: Duties.compute_current_sync_committees(beacon, set.validators)
     }
 
     Duties.log_duties_for_epoch(duties, epoch)
@@ -188,6 +191,29 @@ defmodule LambdaEthereumConsensus.ValidatorSet do
 
   defp update_validators(new_validators, set), do: %{set | validators: new_validators}
 
+  ##############################
+  # Sync committee
+
+  defp sync_committee_broadcasts(set, epoch, slot, head_root) do
+    case Duties.current_sync_committee(set.duties, epoch, slot) do
+      [] ->
+        set
+
+      sync_committee_duties ->
+        sync_committee_duties
+        |> Enum.map(&sync_committee_broadcast(&1, slot, head_root, set.validators))
+        |> update_duties(set, epoch, :sync_committees, slot)
+    end
+  end
+
+  defp sync_committee_broadcast(duty, slot, head_root, validators) do
+    validators
+    |> Map.get(duty.validator_index)
+    |> Validator.sync_committee_message_broadcast(duty, slot, head_root)
+
+    Duties.sync_committee_broadcasted(duty, slot)
+  end
+
   ##############################
   # Attestation