Fetch pending transactions

Use `parity_pendingTransactions` JSONRPC to get the pending transactions
and import them.  If they conflict with pre-existing transactions, then
the previous transactions win, under the assumption that sometimes
pending transactions repository transaction will COMMIT after the the
realtime index has COMMITed that same transaction being validated as the
timeout for transactions is 60 seconds while the block rate is faster
than that.

apps/ethereum_jsonrpc/lib/ethereum_jsonrpc/parity.ex

@@ -6,6 +6,7 @@ defmodule EthereumJSONRPC.Parity do
   import EthereumJSONRPC, only: [config: 1, json_rpc: 2, request: 1]
 
   alias EthereumJSONRPC.Parity.Traces
+  alias EthereumJSONRPC.{Transaction, Transactions}
 
   @doc """
   Fetches the `t:Explorer.Chain.InternalTransaction.changeset/2` params from the Parity trace URL.
@@ -46,6 +47,27 @@ defmodule EthereumJSONRPC.Parity do
     end
   end
 
+  @doc """
+  Fetches the pending transactions from the Parity node.
+
+  *NOTE*: The pending transactions are local to the node that is contacted and may not be consistent across nodes based
+  on the transactions that each node has seen and how each node prioritizes collating transactions into the next block.
+  """
+  @spec fetch_pending_transactions() :: {:ok, [Transaction.params()]} | {:error, reason :: term}
+  def fetch_pending_transactions do
+    with {:ok, transactions} <-
+           %{id: 1, method: "parity_pendingTransactions", params: []}
+           |> request()
+           |> json_rpc(config(:url)) do
+      transactions_params =
+        transactions
+        |> Transactions.to_elixir()
+        |> Transactions.elixir_to_params()
+
+      {:ok, transactions_params}
+    end
+  end
+
   defp response_to_trace(%{"id" => transaction_hash, "result" => %{"trace" => traces}}) when is_list(traces) do
     traces
     |> Stream.with_index()

apps/explorer/lib/explorer/indexer/address_extraction.ex

@@ -43,6 +43,11 @@ defmodule Explorer.Indexer.AddressExtraction do
       }
   """
 
+  @transactions_address_maps [
+    %{from: :from_address_hash, to: :hash},
+    %{from: :to_address_hash, to: :hash}
+  ]
+
   @entity_to_address_map %{
     blocks: [%{from: :miner_hash, to: :hash}],
     internal_transactions: [
@@ -53,10 +58,7 @@ defmodule Explorer.Indexer.AddressExtraction do
         %{from: :created_contract_code, to: :contract_code}
       ]
     ],
-    transactions: [
-      %{from: :from_address_hash, to: :hash},
-      %{from: :to_address_hash, to: :hash}
-    ],
+    transactions: @transactions_address_maps,
     logs: [%{from: :address_hash, to: :hash}]
   }
 
@@ -65,6 +67,22 @@ defmodule Explorer.Indexer.AddressExtraction do
   """
   @type params :: %{required(:hash) => String.t(), optional(:contract_code) => String.t()}
 
+  @doc """
+  Extracts the `from_address_hash` and `to_address_hash` from all the `transactions_params`.
+  """
+  @spec transactions_params_to_addresses_params([
+          %{
+            required(:from_address_hash) => String.t(),
+            optional(:to_address_hash) => String.t()
+          }
+        ]) :: [params]
+  def transactions_params_to_addresses_params(transactions_params) do
+    transactions_params
+    |> extract_addresses_from_collection(@transactions_address_maps)
+    |> List.flatten()
+    |> merge_addresses()
+  end
+
   @doc """
   Extract addresses from block, internal transaction, transaction, and log parameters.
   """

apps/explorer/lib/explorer/indexer/pending_transaction_fetcher.ex

@@ -0,0 +1,101 @@
+defmodule Explorer.Indexer.PendingTransactionFetcher do
+  @moduledoc """
+  Fetches pending transactions and imports them.
+
+  *NOTE*: Pending transactions are imported with with `on_conflict: :nothing`, so that they don't overwrite their own
+  validated version that may make it to the database first.
+  """
+  use GenServer
+
+  require Logger
+
+  import EthereumJSONRPC.Parity, only: [fetch_pending_transactions: 0]
+  import Explorer.Indexer.AddressExtraction, only: [transactions_params_to_addresses_params: 1]
+
+  alias Explorer.{Chain, Indexer}
+  alias Explorer.Indexer.PendingTransactionFetcher
+
+  # milliseconds
+  @default_interval 1_000
+
+  defstruct interval: @default_interval,
+            task_ref: nil,
+            task_pid: nil
+
+  @gen_server_options ~w(debug name spawn_opt timeout)a
+
+  @doc """
+  Starts the pending transaction fetcher.
+
+  ## Options
+
+    * `:debug` - if present, the corresponding function in the [`:sys` module](http://www.erlang.org/doc/man/sys.html)
+      is invoked
+    * `:name` - used for name registration as described in the "Name registration" section of the `GenServer` module
+      documentation
+    * `:pending_transaction_interval` - the millisecond time between checking for pending transactions.  Defaults to
+      `#{@default_interval}` milliseconds.
+    * `:spawn_opt` - if present, its value is passed as options to the underlying process as in `Process.spawn/4`
+    * `:timeout` - if present, the server is allowed to spend the given number of milliseconds initializing or it will
+      be terminated and the start function will return `{:error, :timeout}`
+
+  """
+  def start_link(opts) do
+    GenServer.start_link(__MODULE__, Keyword.drop(opts, @gen_server_options), Keyword.take(opts, @gen_server_options))
+  end
+
+  @impl GenServer
+  def init(opts) do
+    opts =
+      :explorer
+      |> Application.fetch_env!(:indexer)
+      |> Keyword.merge(opts)
+
+    state =
+      %PendingTransactionFetcher{interval: opts[:pending_transaction_interval] || @default_interval}
+      |> schedule_fetch()
+
+    {:ok, state}
+  end
+
+  @impl GenServer
+  def handle_info(:fetch, %PendingTransactionFetcher{} = state) do
+    {:ok, pid, ref} = Indexer.start_monitor(fn -> task(state) end)
+    {:noreply, %PendingTransactionFetcher{state | task_ref: ref, task_pid: pid}}
+  end
+
+  def handle_info({:DOWN, ref, :process, pid, reason}, %PendingTransactionFetcher{task_ref: ref, task_pid: pid} = state) do
+    case reason do
+      :normal ->
+        :ok
+
+      _ ->
+        Logger.error(fn -> "pending transaction fetcher task exited due to #{inspect(reason)}.  Rescheduling." end)
+    end
+
+    new_state =
+      %PendingTransactionFetcher{state | task_ref: nil, task_pid: nil}
+      |> schedule_fetch()
+
+    {:noreply, new_state}
+  end
+
+  defp schedule_fetch(%PendingTransactionFetcher{interval: interval} = state) do
+    Process.send_after(self(), :fetch, interval)
+    state
+  end
+
+  defp task(%PendingTransactionFetcher{} = _state) do
+    {:ok, transactions_params} = fetch_pending_transactions()
+    addresses_params = transactions_params_to_addresses_params(transactions_params)
+
+    # There's no need to queue up fetching the address balance since theses are pending transactions and cannot have
+    # affected the address balance yet since address balance is a balance at a give block and these transactions are
+    # blockless.
+    {:ok, _} =
+      Chain.import_blocks(
+        addresses: [params: addresses_params],
+        transactions: [on_conflict: :nothing, params: transactions_params]
+      )
+  end
+end

apps/explorer/lib/explorer/indexer/supervisor.ex

@@ -5,7 +5,7 @@ defmodule Explorer.Indexer.Supervisor do
 
   use Supervisor
 
-  alias Explorer.Indexer.{AddressBalanceFetcher, BlockFetcher, InternalTransactionFetcher}
+  alias Explorer.Indexer.{AddressBalanceFetcher, BlockFetcher, InternalTransactionFetcher, PendingTransactionFetcher}
 
   def start_link(opts) do
     Supervisor.start_link(__MODULE__, opts, name: __MODULE__)
@@ -16,6 +16,7 @@ defmodule Explorer.Indexer.Supervisor do
     children = [
       {Task.Supervisor, name: Explorer.Indexer.TaskSupervisor},
       {AddressBalanceFetcher, name: AddressBalanceFetcher},
+      {PendingTransactionFetcher, name: PendingTransactionFetcher},
       {InternalTransactionFetcher, name: InternalTransactionFetcher},
       {BlockFetcher, []}
     ]

apps/explorer/test/explorer/indexer/block_fetcher_test.exs

@@ -51,7 +51,7 @@ defmodule Explorer.Indexer.BlockFetcherTest do
       InternalTransactionFetcherCase.start_supervised!()
       start_supervised!(BlockFetcher)
 
-      wait(fn ->
+      wait_for_results(fn ->
         Repo.one!(from(block in Block, where: block.number == ^latest_block_number))
       end)
 
@@ -59,7 +59,7 @@ defmodule Explorer.Indexer.BlockFetcherTest do
 
       previous_batch_block_number = latest_block_number - default_blocks_batch_size
 
-      wait(fn ->
+      wait_for_results(fn ->
         Repo.one!(from(block in Block, where: block.number == ^previous_batch_block_number))
       end)
 
@@ -276,23 +276,4 @@ defmodule Explorer.Indexer.BlockFetcherTest do
       counts.buffer == 0 and counts.tasks == 0
     end)
   end
-
-  defp wait(producer) do
-    producer.()
-  rescue
-    Ecto.NoResultsError ->
-      Process.sleep(100)
-      wait(producer)
-  catch
-    :exit,
-    {:timeout,
-     {GenServer, :call,
-      [
-        _,
-        {:checkout, _, _, _},
-        _
-      ]}} ->
-      Process.sleep(100)
-      wait(producer)
-  end
 end

apps/explorer/test/explorer/indexer/internal_transaction_fetcher_test.exs

@@ -1,7 +1,26 @@
 defmodule Explorer.Indexer.InternalTransactionFetcherTest do
-  use Explorer.DataCase, async: true
+  use Explorer.DataCase, async: false
 
-  alias Explorer.Indexer.InternalTransactionFetcher
+  alias Explorer.Chain.Transaction
+  alias Explorer.Indexer.{AddressBalanceFetcherCase, InternalTransactionFetcher, PendingTransactionFetcher}
+
+  test "does not try to fetch pending transactions from Explorer.Indexer.PendingTransactionFetcher" do
+    start_supervised!({Task.Supervisor, name: Explorer.Indexer.TaskSupervisor})
+    AddressBalanceFetcherCase.start_supervised!()
+    start_supervised!(PendingTransactionFetcher)
+
+    wait_for_results(fn ->
+      Repo.one!(from(transaction in Transaction, where: is_nil(transaction.block_hash), limit: 1))
+    end)
+
+    :transaction
+    |> insert(hash: "0x3a3eb134e6792ce9403ea4188e5e79693de9e4c94e499db132be086400da79e6")
+    |> with_block()
+
+    hash_strings = InternalTransactionFetcher.init([], fn hash_string, acc -> [hash_string | acc] end)
+
+    assert :ok = InternalTransactionFetcher.run(hash_strings, 0)
+  end
 
   describe "init/2" do
     test "does not buffer pending transactions" do

apps/explorer/test/explorer/indexer/pending_transaction_fetcher_test.exs

@@ -0,0 +1,23 @@
+defmodule Explorer.Indexer.PendingTransactionFetcherTest do
+  # `async: false` due to use of named GenServer
+  use Explorer.DataCase, async: false
+
+  alias Explorer.Chain.Transaction
+  alias Explorer.Indexer.PendingTransactionFetcher
+
+  describe "start_link/1" do
+    # this test may fail if Sokol so low volume that the pending transactions are empty for too long
+    test "starts fetching pending transactions" do
+      assert Repo.aggregate(Transaction, :count, :hash) == 0
+
+      start_supervised!({Task.Supervisor, name: Explorer.Indexer.TaskSupervisor})
+      start_supervised!(PendingTransactionFetcher)
+
+      wait_for_results(fn ->
+        Repo.one!(from(transaction in Transaction, where: is_nil(transaction.block_hash), limit: 1))
+      end)
+
+      assert Repo.aggregate(Transaction, :count, :hash) >= 1
+    end
+  end
+end

apps/explorer/test/support/data_case.ex

@@ -39,4 +39,23 @@ defmodule Explorer.DataCase do
 
     :ok
   end
+
+  def wait_for_results(producer) do
+    producer.()
+  rescue
+    Ecto.NoResultsError ->
+      Process.sleep(100)
+      wait_for_results(producer)
+  catch
+    :exit,
+    {:timeout,
+     {GenServer, :call,
+      [
+        _,
+        {:checkout, _, _, _},
+        _
+      ]}} ->
+      Process.sleep(100)
+      wait_for_results(producer)
+  end
 end

coveralls.json

@@ -1,7 +1,7 @@
 {
   "coverage_options": {
     "treat_no_relevant_lines_as_covered": true,
-    "minimum_coverage": 92.9
+    "minimum_coverage": 92.7
   },
   "terminal_options": {
     "file_column_width": 120