feat: persist block with processing status (#1127)

Author: Arkenan

bench/block_processing.exs

@@ -3,6 +3,7 @@ alias LambdaEthereumConsensus.ForkChoice.Handlers
 alias LambdaEthereumConsensus.StateTransition.Cache
 alias LambdaEthereumConsensus.Store
 alias LambdaEthereumConsensus.Store.BlockDb
+alias LambdaEthereumConsensus.Store.BlockDb.BlockInfo
 alias LambdaEthereumConsensus.Store.StateDb
 alias Types.BeaconState
 alias Types.SignedBeaconBlock
@@ -19,8 +20,8 @@ slot = 4_213_280
 
 IO.puts("fetching blocks...")
 {:ok, %BeaconState{} = state} = StateDb.get_state_by_slot(slot)
-{:ok, %SignedBeaconBlock{} = block} = BlockDb.get_block_by_slot(slot)
-{:ok, %SignedBeaconBlock{} = new_block} = BlockDb.get_block_by_slot(slot + 1)
+{:ok, %BlockInfo{signed_block: block}} = BlockDb.get_block_info_by_slot(slot)
+{:ok, %BlockInfo{signed_block: new_block}} = BlockDb.get_block_info_by_slot(slot + 1)
 
 IO.puts("initializing store...")
 {:ok, store} = Types.Store.get_forkchoice_store(state, block)

bench/multiple_blocks_processing.exs

@@ -3,6 +3,7 @@ alias LambdaEthereumConsensus.ForkChoice.Handlers
 alias LambdaEthereumConsensus.StateTransition.Cache
 alias LambdaEthereumConsensus.Store
 alias LambdaEthereumConsensus.Store.BlockDb
+alias LambdaEthereumConsensus.Store.BlockDb.BlockInfo
 alias LambdaEthereumConsensus.Store.StateDb
 alias Types.BeaconState
 alias Types.SignedBeaconBlock
@@ -21,14 +22,14 @@ end_slot = start_slot + count
 
 IO.puts("fetching blocks...")
 {:ok, %BeaconState{} = state} = StateDb.get_state_by_slot(start_slot)
-{:ok, %SignedBeaconBlock{} = signed_block} = BlockDb.get_block_by_slot(state.slot)
+{:ok, %BlockInfo{signed_block: signed_block}} = BlockDb.get_block_info_by_slot(state.slot)
 
 blocks =
   (start_slot + 1)..end_slot
   # NOTE: we have to consider empty slots
   |> Enum.flat_map(fn slot ->
-    case BlockDb.get_block_by_slot(slot) do
-      {:ok, block} -> [block]
+    case BlockDb.get_block_info_by_slot(slot) do
+      {:ok, %BlockInfo{signed_block: block}} -> [block]
       :not_found -> []
     end
   end)

lib/beacon_api/controllers/v2/beacon_controller.ex

@@ -5,6 +5,7 @@ defmodule BeaconApi.V2.BeaconController do
   alias BeaconApi.ErrorController
   alias BeaconApi.Utils
   alias LambdaEthereumConsensus.Store.BlockDb
+  alias LambdaEthereumConsensus.Store.BlockDb.BlockInfo
   alias LambdaEthereumConsensus.Store.Blocks
   alias Types
 
@@ -36,8 +37,8 @@ defmodule BeaconApi.V2.BeaconController do
 
   def get_block(conn, %{block_id: "0x" <> hex_block_id}) do
     with {:ok, block_root} <- Base.decode16(hex_block_id, case: :mixed),
-         %{} = block <- Blocks.get_signed_block(block_root) do
-      conn |> block_response(block)
+         %{} = block_info <- Blocks.get_block_info(block_root) do
+      conn |> block_response(block_info.signed_block)
     else
       nil -> conn |> block_not_found()
       _ -> conn |> ErrorController.bad_request("Invalid block ID: 0x#{hex_block_id}")
@@ -46,8 +47,8 @@ defmodule BeaconApi.V2.BeaconController do
 
   def get_block(conn, %{block_id: block_id}) do
     with {slot, ""} when slot >= 0 <- Integer.parse(block_id),
-         {:ok, block} <- BlockDb.get_block_by_slot(slot) do
-      conn |> block_response(block)
+         {:ok, %BlockInfo{signed_block: signed_block}} <- BlockDb.get_block_info_by_slot(slot) do
+      conn |> block_response(signed_block)
     else
       :not_found ->
         conn |> block_not_found()

lib/beacon_api/helpers.ex

@@ -100,8 +100,8 @@ defmodule BeaconApi.Helpers do
 
   def state_root_by_state_id(id) do
     with {:ok, {block_root, optimistic, finalized}} <- block_root_by_block_id(id),
-         {:ok, block} <- BlockDb.get_block(block_root) do
-      %{message: %{state_root: state_root}} = block
+         {:ok, block_info} <- BlockDb.get_block_info(block_root) do
+      state_root = block_info.signed_block.message.state_root
       {:ok, {state_root, optimistic, finalized}}
     end
   end
@@ -114,8 +114,8 @@ defmodule BeaconApi.Helpers do
           | :invalid_id
   def block_by_block_id(block_id) do
     with {:ok, {root, optimistic, finalized}} <- block_root_by_block_id(block_id),
-         {:ok, block} <- BlockDb.get_block(root) do
-      {:ok, {block, optimistic, finalized}}
+         {:ok, block_info} <- BlockDb.get_block_info(root) do
+      {:ok, {block_info.signed_block, optimistic, finalized}}
     end
   end
 

lib/lambda_ethereum_consensus/beacon/pending_blocks.ex

@@ -12,6 +12,7 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
   alias LambdaEthereumConsensus.P2P.BlobDownloader
   alias LambdaEthereumConsensus.P2P.BlockDownloader
   alias LambdaEthereumConsensus.Store.BlobDb
+  alias LambdaEthereumConsensus.Store.BlockDb.BlockInfo
   alias LambdaEthereumConsensus.Store.Blocks
   alias Types.SignedBeaconBlock
 
@@ -167,6 +168,8 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
     |> Enum.map(fn {root, {block, _}} -> {root, block} end)
     |> Enum.sort_by(fn {_, signed_block} -> signed_block.message.slot end)
     |> Enum.reduce(state, fn {block_root, signed_block}, state ->
+      block_info = BlockInfo.from_block(signed_block, block_root, :pending)
+
       parent_root = signed_block.message.parent_root
       parent_status = get_block_status(state, parent_root)
 
@@ -185,7 +188,7 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
 
         # If all the other conditions are false, add block to fork choice
         true ->
-          ForkChoice.on_block(signed_block, block_root)
+          ForkChoice.on_block(block_info)
           state |> Map.put(block_root, {signed_block, :processing})
       end
     end)

lib/lambda_ethereum_consensus/fork_choice/fork_choice.ex

@@ -12,6 +12,7 @@ defmodule LambdaEthereumConsensus.ForkChoice do
   alias LambdaEthereumConsensus.ForkChoice.Head
   alias LambdaEthereumConsensus.P2P.Gossip.OperationsCollector
   alias LambdaEthereumConsensus.Store.BlockDb
+  alias LambdaEthereumConsensus.Store.BlockDb.BlockInfo
   alias LambdaEthereumConsensus.Store.Blocks
   alias LambdaEthereumConsensus.Store.StateDb
   alias LambdaEthereumConsensus.Store.StoreDb
@@ -36,9 +37,9 @@ defmodule LambdaEthereumConsensus.ForkChoice do
     GenServer.cast(__MODULE__, {:on_tick, time})
   end
 
-  @spec on_block(SignedBeaconBlock.t(), Types.root()) :: :ok | :error
-  def on_block(signed_block, block_root) do
-    GenServer.cast(__MODULE__, {:on_block, block_root, signed_block, self()})
+  @spec on_block(BlockInfo.t()) :: :ok | :error
+  def on_block(block_info) do
+    GenServer.cast(__MODULE__, {:on_block, block_info, self()})
   end
 
   @spec on_attestation(Types.Attestation.t()) :: :ok
@@ -69,16 +70,17 @@ defmodule LambdaEthereumConsensus.ForkChoice do
   end
 
   @impl GenServer
-  def handle_cast({:on_block, block_root, %SignedBeaconBlock{} = signed_block, from}, store) do
-    slot = signed_block.message.slot
+  def handle_cast({:on_block, %BlockInfo{} = block_info, from}, store) do
+    slot = block_info.signed_block.message.slot
+    block_root = block_info.root
 
-    Logger.info("[Fork choice] Adding new block", root: block_root, slot: slot)
+    Logger.info("[Fork choice] Adding new block", root: block_info.root, slot: slot)
 
     %Store{finalized_checkpoint: last_finalized_checkpoint} = store
 
     result =
       :telemetry.span([:sync, :on_block], %{}, fn ->
-        {process_block(block_root, signed_block, store), %{}}
+        {process_block(block_info, store), %{}}
       end)
 
     case result do
@@ -171,8 +173,9 @@ defmodule LambdaEthereumConsensus.ForkChoice do
     end)
   end
 
-  defp process_block(_block_root, %SignedBeaconBlock{} = signed_block, store) do
-    with {:ok, new_store} <- Handlers.on_block(store, signed_block),
+  @spec process_block(BlockInfo.t(), Store.t()) :: Store.t()
+  defp process_block(%BlockInfo{signed_block: signed_block} = block_info, store) do
+    with {:ok, new_store} <- Handlers.on_block(store, block_info),
          # process block attestations
          {:ok, new_store} <-
            signed_block.message.body.attestations

lib/lambda_ethereum_consensus/fork_choice/handlers.ex

@@ -11,6 +11,7 @@ defmodule LambdaEthereumConsensus.ForkChoice.Handlers do
   alias LambdaEthereumConsensus.StateTransition.Misc
   alias LambdaEthereumConsensus.StateTransition.Predicates
   alias LambdaEthereumConsensus.Store.BlobDb
+  alias LambdaEthereumConsensus.Store.BlockDb.BlockInfo
   alias LambdaEthereumConsensus.Store.Blocks
   alias LambdaEthereumConsensus.Store.BlockStates
 
@@ -22,7 +23,6 @@ defmodule LambdaEthereumConsensus.ForkChoice.Handlers do
   alias Types.Checkpoint
   alias Types.IndexedAttestation
   alias Types.NewPayloadRequest
-  alias Types.SignedBeaconBlock
   alias Types.Store
 
   import LambdaEthereumConsensus.Utils, only: [if_then_update: 3, map_ok: 2]
@@ -54,8 +54,9 @@ defmodule LambdaEthereumConsensus.ForkChoice.Handlers do
   A block that is asserted as invalid due to unavailable PoW block may be valid at a later time,
   consider scheduling it for later processing in such case.
   """
-  @spec on_block(Store.t(), SignedBeaconBlock.t()) :: {:ok, Store.t()} | {:error, String.t()}
-  def on_block(%Store{} = store, %SignedBeaconBlock{message: block} = signed_block) do
+  @spec on_block(Store.t(), BlockInfo.t()) :: {:ok, Store.t()} | {:error, String.t()}
+  def on_block(%Store{} = store, %BlockInfo{} = block_info) do
+    block = block_info.signed_block.message
     %{epoch: finalized_epoch, root: finalized_root} = store.finalized_checkpoint
     finalized_slot = Misc.compute_start_slot_at_epoch(finalized_epoch)
 
@@ -80,11 +81,11 @@ defmodule LambdaEthereumConsensus.ForkChoice.Handlers do
       finalized_root != Store.get_checkpoint_block(store, block.parent_root, finalized_epoch) ->
         {:error, "block isn't descendant of latest finalized block"}
 
-      not (Ssz.hash_tree_root!(block) |> data_available?(block.body.blob_kzg_commitments)) ->
+      not (block_info.root |> data_available?(block.body.blob_kzg_commitments)) ->
         {:error, "blob data not available"}
 
       true ->
-        compute_post_state(store, signed_block, base_state)
+        compute_post_state(store, block_info, base_state)
     end
   end
 
@@ -188,8 +189,8 @@ defmodule LambdaEthereumConsensus.ForkChoice.Handlers do
   end
 
   # Check the block is valid and compute the post-state.
-  def compute_post_state(%Store{} = store, %SignedBeaconBlock{} = signed_block, state) do
-    %{message: block} = signed_block
+  def compute_post_state(%Store{} = store, %BlockInfo{} = block_info, state) do
+    block = block_info.signed_block.message
 
     payload = block.body.execution_payload
     parent_beacon_block_root = block.parent_root
@@ -210,33 +211,31 @@ defmodule LambdaEthereumConsensus.ForkChoice.Handlers do
         |> handle_verify_payload_result()
       end)
 
-    with {:ok, state} <- StateTransition.state_transition(state, signed_block, true),
+    with {:ok, state} <- StateTransition.state_transition(state, block_info.signed_block, true),
          {:ok, _execution_status} <- Task.await(payload_verification_task) do
       seconds_per_slot = ChainSpec.get("SECONDS_PER_SLOT")
       intervals_per_slot = Constants.intervals_per_slot()
       # Add proposer score boost if the block is timely
       time_into_slot = rem(store.time - store.genesis_time, seconds_per_slot)
       is_before_attesting_interval = time_into_slot < div(seconds_per_slot, intervals_per_slot)
 
-      block_root = Ssz.hash_tree_root!(block)
-
       # Add new block and state to the store
-      BlockStates.store_state(block_root, state)
+      BlockStates.store_state(block_info.root, state)
 
       is_first_block = store.proposer_boost_root == <<0::256>>
       # TODO: store block timeliness data?
       is_timely = Store.get_current_slot(store) == block.slot and is_before_attesting_interval
 
       store
-      |> Store.store_block(block_root, signed_block)
+      |> Store.store_block_info(block_info)
       |> if_then_update(
         is_timely and is_first_block,
-        &%Store{&1 | proposer_boost_root: block_root}
+        &%Store{&1 | proposer_boost_root: block_info.root}
       )
       # Update checkpoints in store if necessary
       |> update_checkpoints(state.current_justified_checkpoint, state.finalized_checkpoint)
       # Eagerly compute unrealized justification and finality
-      |> compute_pulled_up_tip(block_root, signed_block.message, state)
+      |> compute_pulled_up_tip(block_info.root, block_info.signed_block.message, state)
     end
   end
 

lib/lambda_ethereum_consensus/p2p/incoming_requests/handler.ex

@@ -84,7 +84,7 @@ defmodule LambdaEthereumConsensus.P2P.IncomingRequests.Handler do
       # TODO: extend cache to support slots as keys
       response_chunk =
         start_slot..end_slot
-        |> Enum.map(&BlockDb.get_block_by_slot/1)
+        |> Enum.map(&BlockDb.get_block_info_by_slot/1)
         |> Enum.map(&map_block_result/1)
         |> Enum.reject(&(&1 == :skip))
         |> ReqResp.encode_response()
@@ -103,7 +103,7 @@ defmodule LambdaEthereumConsensus.P2P.IncomingRequests.Handler do
       response_chunk =
         roots
         |> Enum.take(truncated_count)
-        |> Enum.map(&Blocks.get_signed_block/1)
+        |> Enum.map(&Blocks.get_block_info/1)
         |> Enum.map(&map_block_result/1)
         |> Enum.reject(&(&1 == :skip))
         |> ReqResp.encode_response()
@@ -124,6 +124,11 @@ defmodule LambdaEthereumConsensus.P2P.IncomingRequests.Handler do
   defp map_block_result({:ok, block}), do: map_block_result(block)
   defp map_block_result({:error, _}), do: {:error, {2, "Server Error"}}
 
-  defp map_block_result(block),
-    do: {:ok, {block, BeaconChain.get_fork_digest_for_slot(block.message.slot)}}
+  alias LambdaEthereumConsensus.Store.BlockDb.BlockInfo
+
+  defp map_block_result(%BlockInfo{} = block_info),
+    do:
+      {:ok,
+       {block_info.signed_block,
+        BeaconChain.get_fork_digest_for_slot(block_info.signed_block.message.slot)}}
 end