[WIP] Improved ExWire Block Sync

apps/blockchain/lib/blockchain/block.ex

@@ -286,7 +286,7 @@ defmodule Blockchain.Block do
       iex> trie = MerklePatriciaTree.Trie.new(MerklePatriciaTree.Test.random_ets_db())
       iex> {updated_block, _new_trie} = Blockchain.Block.put_receipt(%Blockchain.Block{}, 6, %Blockchain.Transaction.Receipt{state: <<1, 2, 3>>, cumulative_gas: 10, bloom_filter: <<2, 3, 4>>, logs: []}, trie)
       iex> {updated_block, _new_trie} = Blockchain.Block.put_receipt(updated_block, 7, %Blockchain.Transaction.Receipt{state: <<4, 5, 6>>, cumulative_gas: 11, bloom_filter: <<5, 6, 7>>, logs: []}, trie)
-      iex> Blockchain.Block.get_receipt(updated_block, 6, trie.db)
+      iex> Blockchain.Block.x(updated_block, 6, trie.db)
       %Blockchain.Transaction.Receipt{state: <<1, 2, 3>>, cumulative_gas: 10, bloom_filter: <<2, 3, 4>>, logs: []}
 
       iex> trie = MerklePatriciaTree.Trie.new(MerklePatriciaTree.Test.random_ets_db())

apps/cli/lib/mix/tasks/mana.ex

@@ -36,6 +36,7 @@ defmodule Mix.Tasks.Mana do
          sync: sync,
          bootnodes: bootnodes,
          warp: warp,
+         fast: fast,
          debug: debug
        }} ->
         :ok = Logger.warn("Starting mana chain #{Atom.to_string(chain_name)}...")
@@ -48,13 +49,14 @@ defmodule Mix.Tasks.Mana do
             sync: sync,
             discovery: discovery,
             bootnodes: bootnodes,
-            warp: warp
+            warp: warp,
+            fast: fast
           )
 
         {:ok, _} = Application.ensure_all_started(:ex_wire)
 
-        # No Halt
-        Process.sleep(:infinity)
+      # No Halt
+      # Process.sleep(:infinity)
 
       {:error, error} ->
         _ = Logger.error("Error: #{error}")

apps/cli/lib/parser/mana_parser.ex

@@ -8,6 +8,7 @@ defmodule CLI.Parser.ManaParser do
   @default_no_sync false
   @default_bootnodes "from_chain"
   @default_warp false
+  @default_fast false
   @default_debug false
 
   @doc """
@@ -19,6 +20,7 @@ defmodule CLI.Parser.ManaParser do
     * `--no-sync` - Perform syncing (default: false)
     * `--bootnodes` - Comma separated list of bootnodes (default: from_chain)
     * `--warp` - Perform warp sync (default: false)
+    * `--fast` - Perform fast sync (default: false)
     * `--debug` - Add remote debugging (default: false)
 
   ## Examples
@@ -30,16 +32,18 @@ defmodule CLI.Parser.ManaParser do
         sync: false,
         bootnodes: :from_chain,
         warp: false,
+        fast: false,
         debug: false
       }}
 
-      iex> CLI.Parser.ManaParser.mana_args(["--chain", "foundation", "--bootnodes", "enode://google.com,enode://apple.com", "--warp", "--debug"])
+      iex> CLI.Parser.ManaParser.mana_args(["--chain", "foundation", "--bootnodes", "enode://google.com,enode://apple.com", "--warp", "--fast", "--debug"])
       {:ok, %{
         chain_name: :foundation,
         discovery: true,
         sync: true,
         bootnodes: ["enode://google.com", "enode://apple.com"],
         warp: true,
+        fast: true,
         debug: true
       }}
 
@@ -50,6 +54,7 @@ defmodule CLI.Parser.ManaParser do
         sync: true,
         bootnodes: :from_chain,
         warp: false,
+        fast: false,
         debug: false
       }}
 
@@ -64,6 +69,7 @@ defmodule CLI.Parser.ManaParser do
              sync: boolean(),
              bootnodes: :from_chain | list(String.t()),
              warp: boolean(),
+             fast: boolean(),
              debug: boolean()
            }}
           | {:error, String.t()}
@@ -76,6 +82,7 @@ defmodule CLI.Parser.ManaParser do
           no_sync: :boolean,
           bootnodes: :string,
           warp: :boolean,
+          fast: :boolean,
           debug: :boolean
         ]
       )
@@ -85,6 +92,7 @@ defmodule CLI.Parser.ManaParser do
          {:ok, sync} <- get_sync(kw_args),
          {:ok, bootnodes} <- get_bootnodes(kw_args),
          {:ok, warp} <- get_warp(kw_args),
+         {:ok, fast} <- get_fast(kw_args),
          {:ok, debug} <- get_debug(kw_args) do
       {:ok,
        %{
@@ -93,6 +101,7 @@ defmodule CLI.Parser.ManaParser do
          sync: sync,
          bootnodes: bootnodes,
          warp: warp,
+         fast: fast,
          debug: debug
        }}
     end
@@ -158,6 +167,15 @@ defmodule CLI.Parser.ManaParser do
     {:ok, given_warp}
   end
 
+  @spec get_fast(fast: boolean()) :: {:ok, boolean()} | {:error, String.t()}
+  defp get_fast(kw_args) do
+    given_fast =
+      kw_args
+      |> Keyword.get(:fast, @default_fast)
+
+    {:ok, given_fast}
+  end
+
   @spec get_debug(debug: boolean()) :: {:ok, boolean()} | {:error, String.t()}
   defp get_debug(kw_args) do
     given_debug =

apps/ex_wire/lib/ex_wire.ex

@@ -13,8 +13,8 @@ defmodule ExWire do
   alias ExWire.NodeDiscoverySupervisor
   alias ExWire.PeerSupervisor
   alias ExWire.Sync
+  alias ExWire.Sync.{BlockState, WarpState}
   alias ExWire.Sync.WarpProcessor.PowProcessor
-  alias ExWire.Sync.WarpState
   alias ExWire.TCPListeningSupervisor
   alias MerklePatriciaTree.{CachingTrie, DB.RocksDB, Trie}
 
@@ -39,6 +39,8 @@ defmodule ExWire do
       |> Trie.new()
       |> CachingTrie.new()
 
+    block_queue = BlockState.load_block_queue(db)
+
     warp_queue =
       if warp do
         WarpState.load_warp_queue(db)
@@ -64,8 +66,11 @@ defmodule ExWire do
             # Peer supervisor maintains a pool of outbound peers
             child_spec({PeerSupervisor, start_nodes}, []),
 
+            # Processes blocks
+            {ExWire.Sync.BlockProcessor, {trie}},
+
             # Sync coordinates asking peers for new blocks
-            child_spec({Sync, {trie, chain, warp, warp_queue}}, [])
+            child_spec({Sync, {trie, chain, block_queue, warp, warp_queue}}, [])
           ]
       else
         []

apps/ex_wire/lib/ex_wire/config.ex

@@ -28,6 +28,7 @@ defmodule ExWire.Config do
           | :public_ip
           | :sync
           | :warp
+          | :fast
 
   @doc """
   Allows application to configure ExWire before it starts.
@@ -156,6 +157,11 @@ defmodule ExWire.Config do
     get_env(given_params, :warp)
   end
 
+  @spec fast?(Keyword.t()) :: boolean()
+  def fast?(given_params \\ []) do
+    get_env(given_params, :fast)
+  end
+
   @spec bootnodes(Keyword.t()) :: [String.t()]
   def bootnodes(given_params \\ []) do
     if conf_ip = System.get_env("BOOTNODES") do

apps/ex_wire/lib/ex_wire/dev_p2p/session.ex

@@ -65,7 +65,7 @@ defmodule ExWire.DEVp2p.Session do
   """
   @spec disconnect(t) :: t
   def disconnect(session = %__MODULE__{}) do
-    %{session | hello_sent: nil, hello_received: nil, packet_id_map: PacketIdMap.default_map()}
+    %{session | hello_sent: nil, hello_received: nil, packet_id_map: nil}
   end
 
   @doc """
@@ -100,6 +100,6 @@ defmodule ExWire.DEVp2p.Session do
   """
   @spec compatible_capabilities?(t) :: boolean()
   def compatible_capabilities?(%__MODULE__{packet_id_map: packet_id_map}) do
-    packet_id_map != PacketIdMap.default_map()
+    Map.has_key?(packet_id_map.ids_to_modules, 0x10)
   end
 end

apps/ex_wire/lib/ex_wire/framing/frame.ex

@@ -16,6 +16,7 @@ defmodule ExWire.Framing.Frame do
   @type frame :: binary()
 
   @padding_size 16
+  @min_frame_size 16 + 16
 
   @spec frame(integer(), ExRLP.t(), Secrets.t()) :: {frame, Secrets.t()}
   def frame(
@@ -97,7 +98,11 @@ defmodule ExWire.Framing.Frame do
   end
 
   @spec unframe(binary(), Secrets.t()) ::
-          {:ok, integer(), binary(), binary(), Secrets.t()} | {:error, String.t()}
+          {:ok, integer(), binary(), binary(), Secrets.t()}
+          | {:error,
+             :insufficient_data
+             | :failed_to_match_header_ingress_mac
+             | :failed_to_match_frame_ingress_mac}
   def unframe(
         frame,
         frame_secrets = %Secrets{
@@ -107,80 +112,84 @@ defmodule ExWire.Framing.Frame do
           mac_secret: mac_secret
         }
       ) do
-    <<
-      # is header always 128 bits?
-      header_enc::binary-size(16),
-      header_mac::binary-size(16),
-      frame_rest::binary()
-    >> = frame
-
-    # verify header mac
-    ingress_mac = update_mac(ingress_mac, mac_encoder, mac_secret, header_enc)
-    expected_header_mac = ingress_mac |> MAC.final() |> Binary.take(16)
-
-    if expected_header_mac != header_mac do
-      :ok =
-        Logger.error(
-          "Failed to match ingress header mac, expected: #{inspect(expected_header_mac)}, got #{
-            inspect(header_mac)
-          }"
-        )
-
-      {:error, "Failed to match header ingress mac"}
+    if byte_size(frame) < @min_frame_size do
+      {:error, :insufficient_data}
     else
-      {decoder_stream, header} = AES.stream_decrypt(header_enc, decoder_stream)
-
       <<
-        frame_size::integer-size(24),
-        _header_data_and_padding::binary()
-      >> = header
-
-      # TODO: We should read the header? But, it's unused by all clients.
-      # header_rlp = header_data_and_padding |> ExRLP.decode
-      # protocol_id = Enum.at(header_rlp, 0) |> ExRLP.decode
-
-      frame_padding_bytes = padding_size(frame_size, @padding_size)
-
-      if byte_size(frame_rest) < frame_size + frame_padding_bytes + 16 do
-        {:error, "Insufficent data"}
+        # is header always 128 bits?
+        header_enc::binary-size(16),
+        header_mac::binary-size(16),
+        frame_rest::binary()
+      >> = frame
+
+      # verify header mac
+      ingress_mac = update_mac(ingress_mac, mac_encoder, mac_secret, header_enc)
+      expected_header_mac = ingress_mac |> MAC.final() |> Binary.take(16)
+
+      if expected_header_mac != header_mac do
+        :ok =
+          Logger.error(
+            "Failed to match ingress header mac, expected: #{inspect(expected_header_mac)}, got #{
+              inspect(header_mac)
+            }"
+          )
+
+        {:error, :failed_to_match_header_ingress_mac}
       else
-        # let's go and ignore the entire header data....
+        {decoder_stream, header} = AES.stream_decrypt(header_enc, decoder_stream)
+
         <<
-          frame_enc::binary-size(frame_size),
-          frame_padding::binary-size(frame_padding_bytes),
-          frame_mac::binary-size(16),
-          frame_rest::binary()
-        >> = frame_rest
+          frame_size::integer-size(24),
+          _header_data_and_padding::binary()
+        >> = header
 
-        frame_enc_with_padding = frame_enc <> frame_padding
+        # TODO: We should read the header? But, it's unused by all clients.
+        # header_rlp = header_data_and_padding |> ExRLP.decode
+        # protocol_id = Enum.at(header_rlp, 0) |> ExRLP.decode
 
-        ingress_mac = MAC.update(ingress_mac, frame_enc_with_padding)
-        ingress_mac = update_mac(ingress_mac, mac_encoder, mac_secret, nil)
-        expected_frame_mac = ingress_mac |> MAC.final() |> Binary.take(16)
+        frame_padding_bytes = padding_size(frame_size, @padding_size)
 
-        if expected_frame_mac != frame_mac do
-          {:error, "Failed to match frame ingress mac"}
+        if byte_size(frame_rest) < frame_size + frame_padding_bytes + 16 do
+          {:error, :insufficient_data}
         else
-          {decoder_stream, frame_with_padding} =
-            AES.stream_decrypt(frame_enc_with_padding, decoder_stream)
-
-          <<
-            frame::binary-size(frame_size),
-            _frame_padding::binary()
-          >> = frame_with_padding
-
+          # let's go and ignore the entire header data....
           <<
-            packet_type_rlp::binary-size(1),
-            packet_data_rlp::binary()
-          >> = frame
-
-          {
-            :ok,
-            packet_type_rlp |> ExRLP.decode() |> :binary.decode_unsigned(),
-            packet_data_rlp |> ExRLP.decode(),
-            frame_rest,
-            %{frame_secrets | ingress_mac: ingress_mac, decoder_stream: decoder_stream}
-          }
+            frame_enc::binary-size(frame_size),
+            frame_padding::binary-size(frame_padding_bytes),
+            frame_mac::binary-size(16),
+            frame_rest::binary()
+          >> = frame_rest
+
+          frame_enc_with_padding = frame_enc <> frame_padding
+
+          ingress_mac = MAC.update(ingress_mac, frame_enc_with_padding)
+          ingress_mac = update_mac(ingress_mac, mac_encoder, mac_secret, nil)
+          expected_frame_mac = ingress_mac |> MAC.final() |> Binary.take(16)
+
+          if expected_frame_mac != frame_mac do
+            {:error, :failed_to_match_frame_ingress_mac}
+          else
+            {decoder_stream, frame_with_padding} =
+              AES.stream_decrypt(frame_enc_with_padding, decoder_stream)
+
+            <<
+              frame::binary-size(frame_size),
+              _frame_padding::binary()
+            >> = frame_with_padding
+
+            <<
+              packet_type_rlp::binary-size(1),
+              packet_data_rlp::binary()
+            >> = frame
+
+            {
+              :ok,
+              packet_type_rlp |> ExRLP.decode() |> :binary.decode_unsigned(),
+              packet_data_rlp |> ExRLP.decode(),
+              frame_rest,
+              %{frame_secrets | ingress_mac: ingress_mac, decoder_stream: decoder_stream}
+            }
+          end
         end
       end
     end

apps/ex_wire/lib/ex_wire/p2p/manager.ex

@@ -133,13 +133,15 @@ defmodule ExWire.P2P.Manager do
         # TOOD: How does this work exactly? Is this for multiple frames?
         handle_packet_data(frame_rest, conn_after_handle)
 
-      {:error, "Insufficent data"} ->
+      {:error, :insufficient_data} ->
         %{conn | queued_data: total_data}
 
       {:error, reason} ->
         _ =
           Logger.error(
-            "[Network] [#{peer}] Failed to read incoming packet from #{peer.host_name} `#{reason}`)"
+            "[Network] [#{peer}] Failed to read incoming packet from #{peer.host_name} `#{
+              to_string(reason)
+            }`)"
           )
 
         %{conn | last_error: reason}