feat: make checkpoint sync asynchronous (#1219)

Author: Arkenan

lib/lambda_ethereum_consensus/beacon/beacon_node.ex

@@ -41,7 +41,6 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
       [
         {LambdaEthereumConsensus.Beacon.Clock, {store.genesis_time, time}},
         {LambdaEthereumConsensus.Libp2pPort, libp2p_args},
-        LambdaEthereumConsensus.Beacon.SyncBlocks,
         {Task.Supervisor, name: PruneStatesSupervisor},
         {Task.Supervisor, name: PruneBlocksSupervisor},
         {Task.Supervisor, name: PruneBlobsSupervisor}

lib/lambda_ethereum_consensus/beacon/pending_blocks.ex

@@ -7,7 +7,6 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
   require Logger
 
   alias LambdaEthereumConsensus.ForkChoice
-  alias LambdaEthereumConsensus.Libp2pPort
   alias LambdaEthereumConsensus.P2P.BlockDownloader
 
   alias LambdaEthereumConsensus.Metrics
@@ -140,7 +139,7 @@ defmodule LambdaEthereumConsensus.Beacon.PendingBlocks do
   end
 
   defp process_downloaded_block({:ok, [block]}) do
-    Libp2pPort.add_block(block)
+    add_block(block)
   end
 
   defp process_downloaded_block({:error, reason}) do

lib/lambda_ethereum_consensus/beacon/sync_blocks.ex

@@ -3,128 +3,69 @@ defmodule LambdaEthereumConsensus.Beacon.SyncBlocks do
     Performs an optimistic block sync from the finalized checkpoint to the current slot.
   """
 
-  use Task
-
   require Logger
 
   alias LambdaEthereumConsensus.ForkChoice
   alias LambdaEthereumConsensus.Libp2pPort
   alias LambdaEthereumConsensus.P2P.BlockDownloader
-  alias LambdaEthereumConsensus.P2P.Gossip
   alias LambdaEthereumConsensus.StateTransition.Misc
-  alias Types.SignedBeaconBlock
 
   @blocks_per_chunk 16
+  @retries 50
 
-  @type chunk :: %{from: Types.slot(), count: integer()}
-
-  def start_link(opts) do
-    Task.start_link(__MODULE__, :run, [opts])
-  end
+  @doc """
+  Calculates how which blocks need to be downloaded to be up to date., and launches the download
+  requests. Returns the amount of blocks that need to be downloaded.
 
-  def run(_opts) do
+  If N blocks should be downloaded, N/16 range requests are performed. When each of those
+  finish, each block of those responses will be sent to libp2p port module individually using
+  Libp2pPort.add_block/1.
+  """
+  @spec run() :: non_neg_integer()
+  def run() do
     # Initial sleep for faster app start
-    Process.sleep(1000)
     checkpoint = ForkChoice.get_finalized_checkpoint()
     initial_slot = Misc.compute_start_slot_at_epoch(checkpoint.epoch) + 1
     last_slot = ForkChoice.get_current_chain_slot()
 
     # If we're around genesis, we consider ourselves synced
-    if last_slot > 0 do
-      perform_sync(initial_slot, last_slot)
+    if last_slot <= 0 do
+      Logger.info("[Optimistic sync] At genesis. No block sync will be needed.")
+      0
     else
-      start_subscriptions()
-    end
-  end
-
-  @spec perform_sync(integer(), integer()) :: :ok
-  def perform_sync(initial_slot, last_slot) do
-    Enum.chunk_every(initial_slot..last_slot, @blocks_per_chunk)
-    |> Enum.map(fn chunk ->
-      first_slot = List.first(chunk)
-      last_slot = List.last(chunk)
-      count = last_slot - first_slot + 1
-      %{from: first_slot, count: count}
-    end)
-    |> perform_sync()
-  end
-
-  @spec perform_sync([chunk()]) :: :ok
-  def perform_sync(chunks) do
-    remaining = chunks |> Stream.map(fn %{count: c} -> c end) |> Enum.sum()
-    Logger.info("[Optimistic Sync] Blocks remaining: #{remaining}")
-
-    results =
-      chunks
-      |> Task.async_stream(
-        fn chunk -> fetch_blocks_by_slot(chunk.from, chunk.count) end,
-        max_concurrency: 4,
-        timeout: 20_000,
-        on_timeout: :kill_task
+      Logger.info(
+        "[Optimistic sync] Performing optimistic sync between slots #{initial_slot} and #{last_slot}, for a total of #{last_slot - initial_slot + 1} slots."
       )
-      |> Enum.map(fn
-        {:ok, result} -> result
-        {:error, error} -> {:error, error}
-        {:exit, :timeout} -> {:error, "timeout"}
-      end)
-
-    results
-    |> Enum.flat_map(fn
-      {:ok, blocks} -> blocks
-      _other -> []
-    end)
-    |> tap(fn blocks ->
-      Logger.info("[Optimistic Sync] Downloaded #{length(blocks)} blocks successfully.")
-    end)
-    |> Enum.each(&Libp2pPort.add_block/1)
-
-    remaining_chunks =
-      Enum.zip(chunks, results)
-      |> Enum.flat_map(fn
-        {chunk, {:error, reason}} ->
-          if not String.contains?(inspect(reason), "failed to dial") do
-            Logger.debug(
-              "[Optimistic Sync] Failed downloading the chunk #{inspect(chunk)}. Reason: #{inspect(reason)}"
-            )
-          end
 
-          [chunk]
-
-        _other ->
-          []
+      initial_slot..last_slot
+      |> Enum.chunk_every(@blocks_per_chunk)
+      |> Enum.map(fn chunk ->
+        first_slot = List.first(chunk)
+        last_slot = List.last(chunk)
+        count = last_slot - first_slot + 1
+
+        Logger.info(
+          "[Optimistic sync] Sending request for slots #{first_slot} to #{last_slot} (request size = #{count})."
+        )
+
+        BlockDownloader.request_blocks_by_range(
+          first_slot,
+          count,
+          &on_chunk_downloaded/1,
+          @retries
+        )
+
+        count
       end)
-
-    if Enum.empty?(chunks) do
-      Logger.info("[Optimistic Sync] Sync completed")
-      start_subscriptions()
-    else
-      Process.sleep(1000)
-      perform_sync(remaining_chunks)
+      |> Enum.sum()
     end
   end
 
-  # TODO: handle subscription failures.
-  defp start_subscriptions() do
-    Gossip.BeaconBlock.subscribe_to_topic()
-    Gossip.BlobSideCar.subscribe_to_topics()
-    Gossip.OperationsCollector.subscribe_to_topics()
+  defp on_chunk_downloaded({:ok, range, blocks}) do
+    Libp2pPort.notify_blocks_downloaded(range, blocks)
   end
 
-  @spec fetch_blocks_by_slot(Types.slot(), non_neg_integer()) ::
-          {:ok, [SignedBeaconBlock.t()]} | {:error, String.t()}
-  def fetch_blocks_by_slot(from, count) do
-    case BlockDownloader.request_blocks_by_range_sync(from, count, 0) do
-      {:ok, blocks} ->
-        {:ok, blocks}
-
-      {:error, error} ->
-        if not String.contains?(inspect(error), "failed to dial") do
-          Logger.debug(
-            "Blocks download failed for slot #{from} count #{count} Error: #{inspect(error)}"
-          )
-        end
-
-        {:error, error}
-    end
+  defp on_chunk_downloaded({:error, range, reason}) do
+    Libp2pPort.notify_block_download_failed(range, reason)
   end
 end

lib/lambda_ethereum_consensus/p2p/block_downloader.ex

@@ -17,8 +17,10 @@ defmodule LambdaEthereumConsensus.P2P.BlockDownloader do
   # so we want to try again with a different peer
   @default_retries 5
 
+  @type range :: {Types.slot(), Types.slot()}
   @type download_result :: {:ok, [SignedBeaconBlock.t()]} | {:error, any()}
-  @type on_blocks :: (download_result() -> term())
+  @type on_blocks ::
+          ({:ok, range(), [SignedBeaconBlock.t()]} | {:error, range(), any()} -> term())
 
   @doc """
   Requests a series of blocks in batch, and synchronously (the caller will block waiting for the
@@ -73,7 +75,7 @@ defmodule LambdaEthereumConsensus.P2P.BlockDownloader do
          :ok <- verify_batch(blocks, slot, count) do
       tags = %{result: "success", type: "by_slot", reason: "success"}
       :telemetry.execute([:network, :request], %{blocks: count}, tags)
-      on_blocks.({:ok, blocks})
+      on_blocks.({:ok, {slot, slot + count - 1}, blocks})
     else
       {:error, reason} ->
         tags = %{type: "by_slot", reason: parse_reason(reason)}
@@ -85,7 +87,8 @@ defmodule LambdaEthereumConsensus.P2P.BlockDownloader do
           request_blocks_by_range(slot, count, on_blocks, retries - 1)
         else
           :telemetry.execute([:network, :request], %{blocks: 0}, Map.put(tags, :result, "error"))
-          on_blocks.({:error, reason})
+          # TODO: Add block range that failed in the reason
+          on_blocks.({:error, {slot, slot + count - 1}, reason})
           {:error, reason}
         end
     end

lib/lambda_ethereum_consensus/p2p/gossip/beacon_block.ex

@@ -57,6 +57,8 @@ defmodule LambdaEthereumConsensus.P2P.Gossip.BeaconBlock do
     "/eth2/#{fork_context}/beacon_block/ssz_snappy"
   end
 
+  def topics(), do: [topic()]
+
   ##########################
   ### Private functions
   ##########################

lib/lambda_ethereum_consensus/p2p/gossip/operations_collector.ex

@@ -220,7 +220,7 @@ defmodule LambdaEthereumConsensus.P2P.Gossip.OperationsCollector do
     end
   end
 
-  defp topics() do
+  def topics() do
     fork_context = ForkChoice.get_fork_digest() |> Base.encode16(case: :lower)
 
     topics =