Merge branch 'main' into add-ssz_ex-validate-schema

.fork_version

@@ -1 +1 @@
-capella
+deneb

lib/lambda_ethereum_consensus/beacon/pending_blocks.ex

@@ -9,15 +9,18 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
   require Logger
 
   alias LambdaEthereumConsensus.ForkChoice
+  alias LambdaEthereumConsensus.P2P.BlobDownloader
   alias LambdaEthereumConsensus.P2P.BlockDownloader
+  alias LambdaEthereumConsensus.Store.BlobDb
   alias LambdaEthereumConsensus.Store.Blocks
   alias Types.SignedBeaconBlock
 
   use HardForkAliasInjection
 
-  @type block_status :: :pending | :invalid | :processing | :download | :unknown
+  @type block_status :: :pending | :invalid | :processing | :download | :download_blobs | :unknown
   @type block_info ::
-          {SignedBeaconBlock.t(), :pending} | {nil, :invalid | :processing | :download}
+          {SignedBeaconBlock.t(), :pending | :download_blobs}
+          | {nil, :invalid | :processing | :download}
   @type state :: %{Types.root() => block_info()}
 
   ##########################
@@ -43,6 +46,10 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
     schedule_blocks_processing()
     schedule_blocks_download()
 
+    HardForkAliasInjection.on_deneb do
+      schedule_blobs_download()
+    end
+
     {:ok, Map.new()}
   end
 
@@ -94,7 +101,7 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
           state |> Map.put(block_root, {nil, :invalid})
 
         # If parent isn't processed, block is pending
-        parent_status in [:processing, :pending, :download] ->
+        parent_status in [:processing, :pending, :download, :download_blobs] ->
           state
 
         # If parent is not in fork choice, download parent
@@ -135,13 +142,53 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
       downloaded_blocks
       |> Enum.reduce(state, fn signed_block, state ->
         block_root = Ssz.hash_tree_root!(signed_block.message)
-        state |> Map.put(block_root, {signed_block, :pending})
+        new_block_state = HardForkAliasInjection.on_deneb(do: :download_blobs, else: :pending)
+        state |> Map.put(block_root, {signed_block, new_block_state})
       end)
 
     schedule_blocks_download()
     {:noreply, new_state}
   end
 
+  @impl true
+  def handle_info(:download_blobs, state) do
+    blocks_with_blobs =
+      Enum.filter(state, fn {_, {_, s}} -> s == :download_blobs end)
+      |> Enum.map(fn {root, {block, _}} -> {root, block} end)
+
+    blobs_to_download =
+      blocks_with_blobs
+      |> Enum.take(16)
+      |> blocks_to_missing_blobs()
+
+    downloaded_blobs =
+      blobs_to_download
+      |> BlobDownloader.request_blobs_by_root()
+      |> case do
+        {:ok, blobs} ->
+          blobs
+
+        {:error, reason} ->
+          Logger.debug("Blob download failed: '#{reason}'")
+          []
+      end
+
+    Enum.each(downloaded_blobs, &BlobDb.store_blob/1)
+
+    new_state =
+      if length(downloaded_blobs) == length(blobs_to_download) do
+        blocks_with_blobs
+        |> Enum.reduce(state, fn {block_root, signed_block}, state ->
+          state |> Map.put(block_root, {signed_block, :pending})
+        end)
+      else
+        state
+      end
+
+    schedule_blobs_download()
+    {:noreply, new_state}
+  end
+
   ##########################
   ### Private Functions
   ##########################
@@ -151,10 +198,23 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
     state |> Map.get(block_root, {nil, :unknown}) |> elem(1)
   end
 
+  defp blocks_to_missing_blobs(blocks) do
+    Enum.flat_map(blocks, fn {block_root,
+                              %{message: %{body: %{blob_kzg_commitments: commitments}}}} ->
+      0..(length(commitments) - 1)
+      |> Enum.reject(&match?({:ok, _}, BlobDb.get_blob(block_root, &1)))
+      |> Enum.map(&%Types.BlobIdentifier{block_root: block_root, index: &1})
+    end)
+  end
+
   def schedule_blocks_processing do
     Process.send_after(__MODULE__, :process_blocks, 3000)
   end
 
+  def schedule_blobs_download do
+    Process.send_after(__MODULE__, :download_blobs, 500)
+  end
+
   def schedule_blocks_download do
     Process.send_after(__MODULE__, :download_blocks, 1000)
   end

lib/lambda_ethereum_consensus/fork_choice/handlers.ex

@@ -189,7 +189,7 @@ defmodule LambdaEthereumConsensus.ForkChoice.Handlers do
     # Make it a task so it runs concurrently with the state transition
     payload_verification_task =
       Task.async(fn ->
-        if HardForkAliasInjection.deneb?() do
+        HardForkAliasInjection.on_deneb do
           versioned_hashes =
             block.body.blob_kzg_commitments
             |> Enum.map(&Misc.kzg_commitment_to_versioned_hash/1)

lib/lambda_ethereum_consensus/hard_fork_alias_injection.ex

@@ -34,16 +34,27 @@ defmodule HardForkAliasInjection do
 
   ## Examples
 
-      iex> HardForkAliasInjection.on_deneb(true, false)
+      iex> HardForkAliasInjection.on_deneb(do: true, else: false)
       #{is_deneb}
+
+      iex> HardForkAliasInjection.on_deneb(do: true)
+      #{if is_deneb, do: true, else: nil}
   """
   if is_deneb do
-    defmacro on_deneb(code, _default) do
-      code
+    defmacro on_deneb(do: do_clause) do
+      do_clause
+    end
+
+    defmacro on_deneb(do: do_clause, else: _else_clause) do
+      do_clause
     end
   else
-    defmacro on_deneb(_code, default) do
-      default
+    defmacro on_deneb(do: _do_clause) do
+      nil
+    end
+
+    defmacro on_deneb(do: _do_clause, else: else_clause) do
+      else_clause
     end
   end
 end

lib/lambda_ethereum_consensus/state_transition/accessors.ex

@@ -392,7 +392,9 @@ defmodule LambdaEthereumConsensus.StateTransition.Accessors do
   end
 
   defp compute_target_indices(is_matching_target, inclusion_delay) do
-    if HardForkAliasInjection.deneb?() do
+    HardForkAliasInjection.on_deneb do
+      _ = inclusion_delay
+
       if is_matching_target,
         do: [Constants.timely_target_flag_index()],
         else: []

lib/lambda_ethereum_consensus/state_transition/epoch_processing.ex

@@ -142,9 +142,10 @@ defmodule LambdaEthereumConsensus.StateTransition.EpochProcessing do
     activation_exit_epoch = Misc.compute_activation_exit_epoch(current_epoch)
 
     churn_limit =
-      if HardForkAliasInjection.deneb?(),
+      HardForkAliasInjection.on_deneb(
         do: Accessors.get_validator_activation_churn_limit(state),
         else: Accessors.get_validator_churn_limit(state)
+      )
 
     result =
       validators

lib/lambda_ethereum_consensus/state_transition/operations.ex

@@ -235,7 +235,7 @@ defmodule LambdaEthereumConsensus.StateTransition.Operations do
       payload.timestamp != Misc.compute_timestamp_at_slot(state, state.slot) ->
         {:error, "Timestamp verification failed"}
 
-      HardForkAliasInjection.on_deneb(body.blob_kzg_commitments |> length(), -1) >
+      HardForkAliasInjection.on_deneb(do: body.blob_kzg_commitments |> length(), else: -1) >
           ChainSpec.get("MAX_BLOBS_PER_BLOCK") ->
         {:error, "Too many commitments"}
 
@@ -270,12 +270,13 @@ defmodule LambdaEthereumConsensus.StateTransition.Operations do
               transactions_root: transactions_root,
               withdrawals_root: withdrawals_root
             ] ++
-              if HardForkAliasInjection.deneb?(),
+              HardForkAliasInjection.on_deneb(
                 do: [
                   blob_gas_used: payload.blob_gas_used,
                   excess_blob_gas: payload.excess_blob_gas
                 ],
                 else: []
+              )
 
           header = struct!(ExecutionPayloadHeader, fields)
 
@@ -566,7 +567,7 @@ defmodule LambdaEthereumConsensus.StateTransition.Operations do
       current_epoch < validator.activation_epoch + ChainSpec.get("SHARD_COMMITTEE_PERIOD") ->
         {:error, "validator cannot exit yet"}
 
-      not ((if HardForkAliasInjection.deneb?() do
+      not ((HardForkAliasInjection.on_deneb do
               Misc.compute_domain(
                 Constants.domain_voluntary_exit(),
                 fork_version: ChainSpec.get("CAPELLA_FORK_VERSION"),
@@ -846,7 +847,7 @@ defmodule LambdaEthereumConsensus.StateTransition.Operations do
   end
 
   defp check_valid_slot_range(data, state) do
-    if HardForkAliasInjection.deneb?() do
+    HardForkAliasInjection.on_deneb do
       if data.slot + ChainSpec.get("MIN_ATTESTATION_INCLUSION_DELAY") <= state.slot do
         :ok
       else

lib/types/beacon_chain/beacon_block_body_deneb.ex

@@ -62,7 +62,7 @@ defmodule Types.BeaconBlockBodyDeneb do
       {:sync_aggregate, Types.SyncAggregate},
       {:execution_payload, Types.ExecutionPayloadDeneb},
       {:bls_to_execution_changes,
-       {:list, Types.BLSToExecutionChange, ChainSpec.get("MAX_BLS_TO_EXECUTION_CHANGES")}},
+       {:list, Types.SignedBLSToExecutionChange, ChainSpec.get("MAX_BLS_TO_EXECUTION_CHANGES")}},
       {:blob_kzg_commitments,
        {:list, TypeAliases.kzg_commitment(), ChainSpec.get("MAX_BLOB_COMMITMENTS_PER_BLOCK")}}
     ]