Adding fast sync block processing logic

apps/blockchain/lib/blockchain/blocktree.ex

@@ -55,15 +55,21 @@ defmodule Blockchain.Blocktree do
         else: {:valid, trie}
 
     with {:valid, trie} <- validation do
-      {:ok, {block_hash, updated_trie}} = Block.put_block(block, trie, specified_block_hash)
+      add_block_without_validation(blocktree, block, trie, specified_block_hash)
+    end
+  end
 
-      # Cache computed block hash
-      block = %{block | block_hash: block_hash}
+  @spec add_block_without_validation(t, Block.t(), TrieStorage.t(), EVM.hash() | nil) ::
+          {:ok, {t, TrieStorage.t(), EVM.hash()}}
+  def add_block_without_validation(blocktree, block, trie, specified_block_hash \\ nil) do
+    {:ok, {block_hash, updated_trie}} = Block.put_block(block, trie, specified_block_hash)
 
-      updated_blocktree = update_best_block(blocktree, block)
+    # Cache computed block hash
+    block = %{block | block_hash: block_hash}
 
-      {:ok, {updated_blocktree, updated_trie, block_hash}}
-    end
+    updated_blocktree = update_best_block(blocktree, block)
+
+    {:ok, {updated_blocktree, updated_trie, block_hash}}
   end
 
   @spec update_best_block(t, Block.t()) :: t

apps/ex_wire/lib/ex_wire/struct/block_queue.ex

@@ -144,14 +144,10 @@ defmodule ExWire.Struct.BlockQueue do
         block_receipts_to_request: receipts_to_request
     }
 
-    if fast_sync_in_progress do
-      {updated_block_queue, block_tree, trie, should_request_body}
-    else
-      {new_block_queue, new_block_tree, new_trie} =
-        process_block_queue(updated_block_queue, block_tree, chain, trie)
+    {new_block_queue, new_block_tree, new_trie} =
+      process_block_queue(updated_block_queue, block_tree, chain, trie)
 
-      {new_block_queue, new_block_tree, new_trie, should_request_body}
-    end
+    {new_block_queue, new_block_tree, new_trie, should_request_body}
   end
 
   @doc """
@@ -334,6 +330,36 @@ defmodule ExWire.Struct.BlockQueue do
   defp do_process_blocks([], block_queue, block_tree, _chain, trie),
     do: {block_queue, block_tree, trie}
 
+  defp do_process_blocks(
+         [block | rest],
+         block_queue = %__MODULE__{fast_sync_in_progress: true},
+         block_tree,
+         chain,
+         trie
+       ) do
+    {:ok, {updated_blocktree, updated_trie, block_hash}} =
+      Blocktree.add_block_without_validation(block_tree, block, trie)
+
+    :ok =
+      Logger.debug(fn ->
+        "[Block Queue] Added block #{block.header.number} (0x#{
+          Base.encode16(block_hash, case: :lower)
+        }) to new block tree without validation during fast sync."
+      end)
+
+    {backlogged_blocks, new_backlog} = Map.pop(block_queue.backlog, block_hash, [])
+
+    new_block_queue = %{block_queue | backlog: new_backlog}
+
+    do_process_blocks(
+      backlogged_blocks ++ rest,
+      new_block_queue,
+      updated_blocktree,
+      chain,
+      updated_trie
+    )
+  end
+
   defp do_process_blocks([block | rest], block_queue, block_tree, chain, trie) do
     {new_block_tree, new_trie, new_backlog, extra_blocks} =
       case Blocktree.verify_and_add_block(
@@ -464,12 +490,14 @@ defmodule ExWire.Struct.BlockQueue do
       }
   """
   @spec get_complete_blocks(t) :: {t, [Block.t()]}
-  def get_complete_blocks(block_queue = %__MODULE__{queue: queue}) do
+  def get_complete_blocks(
+        block_queue = %__MODULE__{queue: queue, fast_sync_in_progress: fast_syncing}
+      ) do
     {queue, blocks} =
       Enum.reduce(queue, {queue, []}, fn {number, block_map}, {queue, blocks} ->
         {final_block_map, new_blocks} =
           Enum.reduce(block_map, {block_map, []}, fn {hash, block_item}, {block_map, blocks} ->
-            if block_item.ready and
+            if block_item.ready and (not fast_syncing or block_item.receipts_added) and
                  MapSet.size(block_item.commitments) >= ExWire.Config.commitment_count() do
               {Map.delete(block_map, hash), [block_item.block | blocks]}
             else

apps/ex_wire/lib/ex_wire/sync.ex

@@ -130,21 +130,6 @@ defmodule ExWire.Sync do
         {:processed_block_chunk, chunk_hash, processed_blocks, block},
         state = %{warp_queue: warp_queue}
       ) do
-    #
-    #    ######
-    #    ## TODO: REMOVE THIS !!!!
-    #    ######
-    #
-    #    if block do
-    #      block_hashes = [
-    #        Block.hash(block),
-    #        Exth.decode_hex("0x8e9726d8a4fd1ece78e2cac3aa0eb73f22444c99320c24cc879440398304fb60"),
-    #        Exth.decode_hex("0x4a4e2470a6b55b2af27dfdf5562ba1ceafa6f75d92e0859b2db6225184a74f9b")
-    #      ]
-    #      :ok = Logger.debug(fn -> "[Sync] Sending GetReceipts request: #{Exth.encode_hex(Block.hash(block))}, Header: #{Header.to_string(block.header)}" end)
-    #      true = send_with_retry(GetReceipts.new(block_hashes), :random, :request_receipts)
-    #    end
-
     next_state =
       warp_queue
       |> WarpQueue.processed_block_chunk(chunk_hash, block, processed_blocks)
@@ -387,22 +372,38 @@ defmodule ExWire.Sync do
   def handle_receipts(
         %Receipts{receipts: blocks_receipts},
         _peer,
-        state = %{block_queue: block_queue}
+        state = %{
+          trie: trie,
+          chain: chain,
+          block_tree: block_tree,
+          block_queue: block_queue
+        }
       ) do
-    case BlockQueue.add_receipts(block_queue, blocks_receipts) do
-      {updated_block_queue, []} ->
-        :ok = Logger.debug("Processed receipts, no new ones queued to fetch.")
-        %{state | block_queue: updated_block_queue}
+    updated_block_queue =
+      case BlockQueue.add_receipts(block_queue, blocks_receipts) do
+        {updated_block_queue, []} ->
+          :ok = Logger.debug("Processed receipts, no new ones queued to fetch.")
+          updated_block_queue
+
+        {updated_block_queue, hashes_to_request} ->
+          :ok =
+            Logger.debug(fn ->
+              "[Sync] Sending GetReceipts request for [#{Enum.count(hashes_to_request)}] receipts."
+            end)
+
+          _ = send_with_retry(GetReceipts.new(hashes_to_request), :random, :request_receipts)
+          updated_block_queue
+      end
 
-      {updated_block_queue, hashes_to_request} ->
-        :ok =
-          Logger.debug(fn ->
-            "[Sync] Sending GetReceipts request for [#{Enum.count(hashes_to_request)}] receipts."
-          end)
+    {final_queue, final_tree, final_trie} =
+      BlockQueue.process_block_queue(updated_block_queue, block_tree, chain, trie)
 
-        _ = send_with_retry(GetReceipts.new(hashes_to_request), :random, :request_receipts)
-        %{state | block_queue: updated_block_queue}
-    end
+    %{
+      state
+      | trie: final_trie,
+        block_tree: final_tree,
+        block_queue: final_queue
+    }
   end
 
   @doc """