Merge branch 'main' of github.com:lambdaclass/lambda_ethereum_consensus into refactor-pending-blocks

Author: mpaulucci

.github/workflows/ci.yml

@@ -169,7 +169,7 @@ jobs:
       run: mix compile --warnings-as-errors
     - name: Run the node
       # NOTE: this starts and then stops the application. It should catch simple runtime errors
-      run: mix run -- --checkpoint-sync https://sync-mainnet.beaconcha.in/
+      run: mix run -- --checkpoint-sync-url https://sync-mainnet.beaconcha.in/
 
   test:
     name: Test

Makefile

@@ -116,11 +116,11 @@ iex: compile-all
 
 #▶️ checkpoint-sync: @ Run an interactive terminal using checkpoint sync.
 checkpoint-sync: compile-all
-	iex -S mix run -- --checkpoint-sync https://sync-mainnet.beaconcha.in/
+	iex -S mix run -- --checkpoint-sync-url https://sync-mainnet.beaconcha.in/
 
 #▶️ sepolia: @ Run an interactive terminal using sepolia network
 sepolia: compile-all
-	iex -S mix run -- --checkpoint-sync https://sepolia.beaconstate.info --network sepolia
+	iex -S mix run -- --checkpoint-sync-url https://sepolia.beaconstate.info --network sepolia
 
 #🔴 test: @ Run tests
 test: compile-all

README.md

@@ -78,10 +78,10 @@ The iex terminal can be closed by pressing ctrl+c two times.
 ### Checkpoint Sync
 
 You can also sync from a checkpoint given by a trusted third-party.
-For that, get the URL that serves the checkpoint, and pass it to the node with the "--checkpoint-sync" flag:
+You can specify a URL to fetch it from with the "--checkpoint-sync-url" flag:
 
 ```shell
-iex -S mix run -- --checkpoint-sync <your_url_here>
+iex -S mix run -- --checkpoint-sync-url <your_url_here>
 ```
 
 Some public endpoints can be found in [eth-clients.github.io/checkpoint-sync-endpoints](https://eth-clients.github.io/checkpoint-sync-endpoints/).

bench/ssz.exs

@@ -35,7 +35,7 @@ Benchee.run(
   time: 5
 )
 
-## Benchmark Merkleization 
+## Benchmark Merkleization
 
 list = Stream.cycle([65_535]) |> Enum.take(316)
 schema = {:list, {:int, 16}, 1024}

config/runtime.exs

@@ -2,7 +2,7 @@ import Config
 
 switches = [
   network: :string,
-  checkpoint_sync: :string,
+  checkpoint_sync_url: :string,
   execution_endpoint: :string,
   execution_jwt: :string,
   mock_execution: :boolean,
@@ -23,12 +23,12 @@ if not is_testing and not Enum.empty?(remaining_args) do
 end
 
 network = Keyword.get(args, :network, "mainnet")
-checkpoint_sync = Keyword.get(args, :checkpoint_sync)
+checkpoint_sync_url = Keyword.get(args, :checkpoint_sync_url)
 execution_endpoint = Keyword.get(args, :execution_endpoint, "http://localhost:8551")
 jwt_path = Keyword.get(args, :execution_jwt)
 
 config :lambda_ethereum_consensus, LambdaEthereumConsensus.ForkChoice,
-  checkpoint_sync: checkpoint_sync
+  checkpoint_sync_url: checkpoint_sync_url
 
 configs_per_network = %{
   "minimal" => MinimalConfig,

lib/lambda_ethereum_consensus/application.ex

@@ -49,7 +49,7 @@ defmodule LambdaEthereumConsensus.Application do
 
   def checkpoint_sync_url do
     Application.fetch_env!(:lambda_ethereum_consensus, LambdaEthereumConsensus.ForkChoice)
-    |> Keyword.fetch!(:checkpoint_sync)
+    |> Keyword.fetch!(:checkpoint_sync_url)
   end
 
   defp get_operation_mode do

lib/lambda_ethereum_consensus/beacon/beacon_node.ex

@@ -26,7 +26,7 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
 
       :not_found ->
         Logger.error(
-          "[Sync] No initial state or block found. Please specify the URL to fetch them from via the --checkpoint-sync flag"
+          "[Sync] No initial state or block found. Please specify the URL to fetch them from via the --checkpoint-sync-url flag"
         )
 
         System.stop(1)

lib/lambda_ethereum_consensus/beacon/checkpoint_sync.ex

@@ -10,7 +10,7 @@ defmodule LambdaEthereumConsensus.Beacon.CheckpointSync do
   @doc """
   Safely retrieves the last finalized state and block
   """
-  @spec get_state(String.t()) ::
+  @spec get_finalized_block_and_state(String.t()) ::
           {:ok, {Types.BeaconState.t(), Types.SignedBeaconBlock.t()}} | {:error, any()}
   def get_finalized_block_and_state(url) do
     tasks = [Task.async(__MODULE__, :get_state, [url]), Task.async(__MODULE__, :get_block, [url])]

lib/lambda_ethereum_consensus/execution/engine_api/mocked.ex

@@ -20,6 +20,6 @@ defmodule LambdaEthereumConsensus.Execution.EngineApi.Mocked do
 
   @spec forkchoice_updated(map, map | any) :: {:ok, any} | {:error, any}
   def forkchoice_updated(_forkchoice_state, _payload_attributes) do
-    {:ok, %{"payload_id" => nil, payload_status: %{"status" => "SYNCING"}}}
+    {:ok, %{"payload_id" => nil, "payload_status" => %{"status" => "SYNCING"}}}
   end
 end

lib/lambda_ethereum_consensus/fork_choice/fork_choice.ex

@@ -7,7 +7,6 @@ defmodule LambdaEthereumConsensus.ForkChoice do
   require Logger
 
   alias LambdaEthereumConsensus.Beacon.BeaconChain
-  alias LambdaEthereumConsensus.Execution.ExecutionClient
   alias LambdaEthereumConsensus.ForkChoice.{Handlers, Helpers}
   alias LambdaEthereumConsensus.Store.Blocks
   alias Types.Attestation
@@ -166,21 +165,11 @@ defmodule LambdaEthereumConsensus.ForkChoice do
   @spec recompute_head(Store.t()) :: :ok
   def recompute_head(store) do
     {:ok, head_root} = Helpers.get_head(store)
-
     head_block = Blocks.get_block!(head_root)
-    head_execution_hash = head_block.body.execution_payload.block_hash
+
+    Handlers.notify_forkchoice_update(store, head_block)
 
     finalized_checkpoint = store.finalized_checkpoint
-    finalized_block = Blocks.get_block!(store.finalized_checkpoint.root)
-    finalized_execution_hash = finalized_block.body.execution_payload.block_hash
-
-    # TODO: do someting with the result from the execution client
-    # TODO: compute safe block hash
-    ExecutionClient.notify_forkchoice_updated(
-      head_execution_hash,
-      finalized_execution_hash,
-      finalized_execution_hash
-    )
 
     BeaconChain.update_fork_choice_cache(
       head_root,

lib/lambda_ethereum_consensus/fork_choice/handlers.ex

@@ -3,6 +3,7 @@ defmodule LambdaEthereumConsensus.ForkChoice.Handlers do
   Handlers that update the fork choice store.
   """
 
+  alias LambdaEthereumConsensus.Execution.ExecutionClient
   alias LambdaEthereumConsensus.StateTransition
   alias LambdaEthereumConsensus.StateTransition.{Accessors, EpochProcessing, Misc, Predicates}
   alias LambdaEthereumConsensus.Store.Blocks
@@ -184,6 +185,21 @@ defmodule LambdaEthereumConsensus.ForkChoice.Handlers do
     end
   end
 
+  def notify_forkchoice_update(store, head_block) do
+    head_execution_hash = head_block.body.execution_payload.block_hash
+
+    finalized_block = Blocks.get_block!(store.finalized_checkpoint.root)
+    finalized_execution_hash = finalized_block.body.execution_payload.block_hash
+
+    # TODO: do someting with the result from the execution client
+    # TODO: compute safe block hash
+    ExecutionClient.notify_forkchoice_updated(
+      head_execution_hash,
+      finalized_execution_hash,
+      finalized_execution_hash
+    )
+  end
+
   ### Private functions ###
 
   # Update checkpoints in store if necessary

lib/lambda_ethereum_consensus/logger/console_logger.ex

@@ -19,7 +19,7 @@ defmodule ConsoleLogger do
   defp level_color(:info), do: :green
   defp level_color(:warning), do: :yellow
   defp level_color(:error), do: :red
-  defp level_color(_), do: :default
+  defp level_color(_), do: :default_color
 
   defp format_level(level) do
     upcased = level |> Atom.to_string() |> String.upcase()

lib/lambda_ethereum_consensus/state_transition/misc.ex

@@ -170,7 +170,7 @@ defmodule LambdaEthereumConsensus.StateTransition.Misc do
     first_8_bytes
   end
 
-  @spec uint_to_bytes(non_neg_integer(), 8 | 32 | 64) :: binary()
+  @spec uint_to_bytes(non_neg_integer(), 8 | 16 | 32 | 64) :: binary()
   def uint_to_bytes(value, size) do
     # Converts an unsigned integer value to a bytes value
     <<value::unsigned-integer-little-size(size)>>

lib/snappy_ex.ex

@@ -0,0 +1,126 @@
+defmodule SnappyEx do
+  @moduledoc """
+    Encoder/decoder implementation for the [Snappy framing format](https://github.com/google/snappy/blob/main/framing_format.txt)
+  """
+  import Bitwise
+
+  @bit_mask_32 2 ** 32 - 1
+  @chunk_size_limit 65_540
+
+  @stream_identifier "sNaPpY"
+
+  @id_compressed_data 0x00
+  @id_uncompressed_data 0x01
+  @id_padding 0xFE
+  @id_stream_identifier 0xFF
+
+  @ids_payload_chunks [@id_compressed_data, @id_uncompressed_data]
+  @ids_reserved_unskippable_chunks 0x02..0x7F
+  @ids_reserved_skippable_chunks 0x80..0xFD
+
+  ##########################
+  ### Public API
+  ##########################
+
+  @doc """
+  Compresses the given data.
+  Returns the compressed data.
+
+  ## Examples
+
+      iex> SnappyEx.compress("")
+      <<0xFF, 6::little-size(24)>> <> "sNaPpY"
+  """
+  @spec compress(binary()) :: binary()
+  def compress(data) when is_binary(data) do
+    # TODO: implement
+    <<@id_stream_identifier, 6::little-size(24)>> <> @stream_identifier
+  end
+
+  @doc """
+  Uncompresses a given stream.
+  Returns a result tuple with the uncompressed data or an error message.
+
+  ## Examples
+
+      iex> SnappyEx.decompress(<<0xFF, 6::little-size(24)>> <> "sNaPpY")
+      {:ok, ""}
+  """
+  @spec decompress(nonempty_binary()) :: {:ok, binary()} | {:error, String.t()}
+  def decompress(<<@id_stream_identifier>> <> _ = chunks), do: decompress_frames(chunks, <<>>)
+
+  @spec decompress(<<>>) :: {:error, String.t()}
+  def decompress(chunks) when is_binary(chunks), do: {:error, "no stream identifier at beginning"}
+
+  @spec compute_checksum(binary()) :: Types.uint32()
+  def compute_checksum(data) when is_binary(data) do
+    checksum = Crc32c.calc!(data)
+
+    # the crc32c checksum of the uncompressed data is masked before inserted into the
+    # frame using masked_checksum = ((checksum >> 15) | (checksum << 17)) + 0xa282ead8
+    (checksum >>> 15 ||| checksum <<< 17) + 0xA282EAD8 &&& @bit_mask_32
+  end
+
+  ##########################
+  ### Private Functions
+  ##########################
+
+  defp decompress_frames("", acc), do: {:ok, acc}
+
+  defp decompress_frames(chunks, acc) do
+    with {:ok, {id, data, remaining_chunks}} <- process_chunk_metadata(chunks),
+         {:ok, new_acc} <- parse_chunk(acc, id, data) do
+      decompress_frames(remaining_chunks, new_acc)
+    end
+  end
+
+  # chunk layout: 1-byte chunk_id, 3-bytes chunk_size, remaining chunks if any data present.
+  defp process_chunk_metadata(chunks) when byte_size(chunks) < 4,
+    do: {:error, "header too small"}
+
+  defp process_chunk_metadata(<<_id::size(8), size::little-size(24), rest::binary>>)
+       when byte_size(rest) < size,
+       do: {:error, "missing data in chunk. expected: #{byte_size(rest)}. got: #{size}"}
+
+  defp process_chunk_metadata(<<id::size(8), size::little-size(24), rest::binary>>) do
+    <<chunk::binary-size(size), remaining_chunks::binary>> = rest
+    {:ok, {id, chunk, remaining_chunks}}
+  end
+
+  # Stream identifier
+  # NOTE: it can appear more than once, and must be validated each time
+  defp parse_chunk(acc, @id_stream_identifier, @stream_identifier), do: {:ok, acc}
+  defp parse_chunk(_, @id_stream_identifier, _), do: {:error, "invalid stream identifier"}
+
+  # Data-carrying chunks (compressed or uncompressed)
+  defp parse_chunk(_acc, id, data)
+       when id in @ids_payload_chunks and byte_size(data) > @chunk_size_limit,
+       do: {:error, "chunk is bigger than limit"}
+
+  defp parse_chunk(acc, id, data) when id in @ids_payload_chunks do
+    <<checksum::little-size(32), compressed_data::binary>> = data
+
+    with {:ok, uncompressed_data} <- decompress_payload(id, compressed_data),
+         :ok <- verify_checksum(uncompressed_data, checksum) do
+      {:ok, <<acc::binary, uncompressed_data::binary>>}
+    end
+  end
+
+  # Skippable chunks (padding or reserved)
+  defp parse_chunk(acc, id, _data)
+       when id == @id_padding or id in @ids_reserved_skippable_chunks,
+       do: {:ok, acc}
+
+  # Reserved unskippable chunks
+  defp parse_chunk(_acc, id, _data) when id in @ids_reserved_unskippable_chunks,
+    do: {:error, "unskippable chunk of type: #{id}"}
+
+  defp decompress_payload(@id_compressed_data, data), do: :snappyer.decompress(data)
+  defp decompress_payload(@id_uncompressed_data, data), do: {:ok, data}
+
+  defp verify_checksum(data, checksum) do
+    if checksum == compute_checksum(data),
+      do: :ok,
+      else: {:error, "invalid checksum"}
+  end
+end