fix: base not found in tree (#1318)

Author: rodrigo-o

.iex.exs

@@ -0,0 +1,21 @@
+alias LambdaEthereumConsensus.ForkChoice
+alias LambdaEthereumConsensus.ForkChoice.Head
+alias LambdaEthereumConsensus.StateTransition.Misc
+alias LambdaEthereumConsensus.Store.Blocks
+alias LambdaEthereumConsensus.Store.StoreDb
+alias LambdaEthereumConsensus.Utils
+
+# Some convenience functions for debugging
+store = fn -> StoreDb.fetch_store() |> elem(1) end
+
+head_root = fn -> store.() |> Head.get_head() |> elem(1) |> Utils.format_binary() end
+head_slot = fn -> store.() |> Head.get_head() |> elem(1) |> Blocks.get_block_info() |> then(& &1.signed_block.message.slot) end
+
+store_calculated_slot = fn -> store.() |> ForkChoice.get_current_slot() end
+
+epoch = fn slot -> slot |> Misc.compute_epoch_at_slot() end
+
+block_info = fn "0x"<>root -> root |> Base.decode16(case: :lower) |> elem(1) |> Blocks.get_block_info() end
+
+blocks_by_status = fn status -> Blocks.get_blocks_with_status(status) |> elem(1) end
+blocks_by_status_count = fn status -> blocks_by_status.(status) |> Enum.count() end

Makefile

@@ -172,35 +172,23 @@ checkpoint-sync: compile-all
 
 #▶️ checkpoint-sync.logfile: @ Run an interactive terminal using checkpoint sync with a log file.
 checkpoint-sync.logfile: compile-all
-	iex -S mix run -- --checkpoint-sync-url https://mainnet-checkpoint-sync.stakely.io/ --metrics --log-file ./logs/mainnet.log
-
-#▶️ checkpoint-sync.logfile: @ Run an interactive terminal using checkpoint sync with a log file.
-checkpoint-sync.logfile: compile-all
-	iex -S mix run -- --checkpoint-sync-url https://mainnet-checkpoint-sync.stakely.io/ --metrics --log-file ./logs/mainnet.log
+	iex -S mix run -- --checkpoint-sync-url https://mainnet-checkpoint-sync.stakely.io/ --metrics --log-file ./logs/mainnet.log --discovery-port $(DYSCOVERY_PORT)
 
 #▶️ sepolia: @ Run an interactive terminal using sepolia network
 sepolia: compile-all
 	iex -S mix run -- --checkpoint-sync-url https://sepolia.beaconstate.info --network sepolia --metrics --discovery-port $(DYSCOVERY_PORT)
 
 #▶️ sepolia.logfile: @ Run an interactive terminal using sepolia network with a log file
 sepolia.logfile: compile-all
-	iex -S mix run -- --checkpoint-sync-url https://sepolia.beaconstate.info --network sepolia --metrics --log-file ./logs/sepolia.log
-
-#▶️ sepolia.logfile: @ Run an interactive terminal using sepolia network with a log file
-sepolia.logfile: compile-all
-	iex -S mix run -- --checkpoint-sync-url https://sepolia.beaconstate.info --network sepolia --metrics --log-file ./logs/sepolia.log
+	iex -S mix run -- --checkpoint-sync-url https://sepolia.beaconstate.info --network sepolia --metrics --log-file ./logs/sepolia.log --discovery-port $(DYSCOVERY_PORT)
 
 #▶️ holesky: @ Run an interactive terminal using holesky network
 holesky: compile-all
 	iex -S mix run -- --checkpoint-sync-url https://checkpoint-sync.holesky.ethpandaops.io --network holesky --discovery-port $(DYSCOVERY_PORT)
 
 #▶️ holesky.logfile: @ Run an interactive terminal using holesky network with a log file
 holesky.logfile: compile-all
-	iex -S mix run -- --checkpoint-sync-url https://checkpoint-sync.holesky.ethpandaops.io --network holesky --log-file ./logs/holesky.log
-
-#▶️ holesky.logfile: @ Run an interactive terminal using holesky network with a log file
-holesky.logfile: compile-all
-	iex -S mix run -- --checkpoint-sync-url https://checkpoint-sync.holesky.ethpandaops.io --network holesky --log-file ./logs/holesky.log
+	iex -S mix run -- --checkpoint-sync-url https://checkpoint-sync.holesky.ethpandaops.io --network holesky --log-file ./logs/holesky.log --discovery-port $(DYSCOVERY_PORT)
 
 #▶️ gnosis: @ Run an interactive terminal using gnosis network
 gnosis: compile-all

lib/lambda_ethereum_consensus/beacon/pending_blocks.ex

@@ -12,11 +12,13 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
   alias LambdaEthereumConsensus.P2P.BlockDownloader
   alias LambdaEthereumConsensus.Store.BlobDb
   alias LambdaEthereumConsensus.Store.Blocks
+  alias LambdaEthereumConsensus.Utils
   alias Types.BlockInfo
   alias Types.SignedBeaconBlock
   alias Types.Store
 
-  @type block_status :: :pending | :invalid | :download | :download_blobs | :unknown
+  @type block_status ::
+          :transitioned | :pending | :invalid | :download | :download_blobs | :unknown
   @type block_info ::
           {SignedBeaconBlock.t(), :pending | :download_blobs}
           | {nil, :invalid | :download}
@@ -40,15 +42,18 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
   def add_block(store, signed_block) do
     block_info = BlockInfo.from_block(signed_block)
     loaded_block = Blocks.get_block_info(block_info.root)
+    log_md = [slot: signed_block.message.slot, root: block_info.root]
 
     # If the block is new or was to be downloaded, we store it.
     if is_nil(loaded_block) or loaded_block.status == :download do
       missing_blobs = missing_blobs(block_info)
 
       if Enum.empty?(missing_blobs) do
+        Logger.debug("[PendingBlocks] No missing blobs for block, process it", log_md)
         Blocks.new_block_info(block_info)
         process_block_and_check_children(store, block_info)
       else
+        Logger.debug("[PendingBlocks] Missing blobs for block, scheduling download", log_md)
         BlobDownloader.request_blobs_by_root(missing_blobs, &process_blobs/2, @download_retries)
 
         block_info
@@ -72,6 +77,7 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
       {:ok, blocks} ->
         blocks
         |> Enum.sort_by(fn %BlockInfo{} = block_info -> block_info.signed_block.message.slot end)
+        # Could we process just one/a small amount of blocks at a time? would it make more sense?
         |> Enum.reduce(store, fn block_info, store ->
           {store, _state} = process_block(store, block_info)
           store
@@ -101,16 +107,28 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
     end
   end
 
-  defp process_block(store, block_info) do
+  defp process_block(store, %BlockInfo{signed_block: %{message: message}} = block_info) do
     if block_info.status != :pending do
-      Logger.error("Called process block for a block that's not ready: #{block_info}")
+      Logger.error(
+        "[PendingBlocks] Called process block for a block that's not ready: #{block_info}"
+      )
     end
 
-    parent_root = block_info.signed_block.message.parent_root
+    log_md = [slot: message.slot, root: block_info.root]
+    parent_root = message.parent_root
+
+    Logger.debug(
+      "[PendingBlocks] Processing block, parent: #{Utils.format_binary(parent_root)}",
+      log_md
+    )
 
     case Blocks.get_block_info(parent_root) do
       nil ->
-        Logger.debug("[PendingBlocks] Add parent to download #{inspect(parent_root)}")
+        Logger.debug(
+          "[PendingBlocks] Add parent with root: #{Utils.format_shorten_binary(parent_root)} to download",
+          log_md
+        )
+
         Blocks.add_block_to_download(parent_root)
 
         BlockDownloader.request_blocks_by_root(
@@ -127,19 +145,25 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
         {store, :download_pending}
 
       %BlockInfo{status: :invalid} ->
+        Logger.warning(
+          "[PendingBlocks] Parent block with root:#{Utils.format_shorten_binary(parent_root)} is invalid, making this block also invalid",
+          log_md
+        )
+
         Blocks.change_status(block_info, :invalid)
         {store, :invalid}
 
       %BlockInfo{status: :transitioned} ->
         case ForkChoice.on_block(store, block_info) do
           {:ok, store} ->
+            Logger.debug("[PendingBlocks] Block transitioned after ForkChoice.on_block/2", log_md)
             Blocks.change_status(block_info, :transitioned)
             {store, :transitioned}
 
           {:error, reason, store} ->
-            Logger.error("[PendingBlocks] Saving block as invalid #{reason}",
-              slot: block_info.signed_block.message.slot,
-              root: block_info.root
+            Logger.error(
+              "[PendingBlocks] Saving block as invalid after ForkChoice.on_block/2 error: #{reason}",
+              log_md
             )
 
             Blocks.change_status(block_info, :invalid)
@@ -157,15 +181,15 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
 
   defp process_downloaded_block(store, {:error, reason}) do
     # We might want to declare a block invalid here.
-    Logger.error("Error downloading block: #{inspect(reason)}")
+    Logger.error("[PendingBlocks] Error downloading block: #{inspect(reason)}")
     {:ok, store}
   end
 
-  defp process_blobs(store, {:ok, blobs}), do: {:ok, add_blobs(store, blobs)}
+  def process_blobs(store, {:ok, blobs}), do: {:ok, add_blobs(store, blobs)}
 
-  defp process_blobs(store, {:error, reason}) do
+  def process_blobs(store, {:error, reason}) do
     # We might want to declare a block invalid here.
-    Logger.error("Error downloading blobs: #{inspect(reason)}")
+    Logger.error("[PendingBlocks] Error downloading blobs: #{inspect(reason)}")
     {:ok, store}
   end
 
@@ -178,19 +202,20 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
     |> Enum.map(&BlobDb.store_blob/1)
     |> Enum.uniq()
     |> Enum.reduce(store, fn root, store ->
-      with %BlockInfo{} = block_info <- Blocks.get_block_info(root),
+      with %BlockInfo{status: :download_blobs} = block_info <- Blocks.get_block_info(root),
            [] <- missing_blobs(block_info) do
         block_info
         |> Blocks.change_status(:pending)
         |> then(&process_block_and_check_children(store, &1))
       else
-        _ -> store
+        _ ->
+          store
       end
     end)
   end
 
   @spec missing_blobs(BlockInfo.t()) :: [Types.BlobIdentifier.t()]
-  defp missing_blobs(%BlockInfo{root: root, signed_block: signed_block}) do
+  def missing_blobs(%BlockInfo{root: root, signed_block: signed_block}) do
     signed_block.message.body.blob_kzg_commitments
     |> Stream.with_index()
     |> Enum.filter(&blob_needs_download?(&1, root))

lib/lambda_ethereum_consensus/fork_choice/fork_choice.ex

@@ -65,6 +65,11 @@ defmodule LambdaEthereumConsensus.ForkChoice do
         |> tap(fn store ->
           StoreDb.persist_store(store)
           Logger.info("[Fork choice] Added new block", slot: slot, root: block_root)
+
+          Logger.info("[Fork choice] Recomputed head",
+            slot: store.head_slot,
+            root: store.head_root
+          )
         end)
         |> then(&{:ok, &1})
 

lib/lambda_ethereum_consensus/p2p/blob_downloader.ex

@@ -122,7 +122,7 @@ defmodule LambdaEthereumConsensus.P2P.BlobDownloader do
   defp get_some_peer() do
     case P2P.Peerbook.get_some_peer() do
       nil ->
-        # TODO: handle no-peers asynchronously
+        # TODO: (#1317) handle no-peers asynchronously
         raise "No peers available to request blobs from."
 
       peer_id ->

lib/lambda_ethereum_consensus/p2p/block_downloader.ex

@@ -186,7 +186,7 @@ defmodule LambdaEthereumConsensus.P2P.BlockDownloader do
   defp get_some_peer() do
     case P2P.Peerbook.get_some_peer() do
       nil ->
-        # TODO: handle no-peers asynchronously
+        # TODO: (#1317) handle no-peers asynchronously
         raise "No peers available to request blocks from."
 
       peer_id ->

lib/lambda_ethereum_consensus/utils.ex

@@ -48,6 +48,14 @@ defmodule LambdaEthereumConsensus.Utils do
     "0x#{String.slice(encoded, 0, 3)}..#{String.slice(encoded, -4, 4)}"
   end
 
+  @doc """
+  Format a binary to its hexadecimal representation.
+  """
+  @spec format_binary(binary) :: String.t()
+  def format_binary(binary) do
+    "0x#{binary |> Base.encode16(case: :lower)}"
+  end
+
   @doc """
   Format a bitstring to a base 2 representation.
   """