Exclude pending transactions for unfetched internal transactions

Parity doesn't support asking for the internal transactions of pending
transactions.

apps/explorer/lib/explorer/chain.ex

@@ -1109,7 +1109,34 @@ defmodule Explorer.Chain do
   end
 
   @doc """
-  Returns a stream of all transactions with unfetched internal transactions.
+  Returns a stream of all collated transactions with unfetched internal transactions.
+
+  Only transactions that have been collated into a block are returned; pending transactions not in a block are filtered
+  out.
+
+      iex> pending = insert(:transaction)
+      iex> unfetched_collated =
+      ...>   :transaction |>
+      ...>   insert() |>
+      ...>   with_block()
+      iex> fetched_collated =
+      ...>   :transaction |>
+      ...>   insert() |>
+      ...>   with_block(internal_transactions_indexed_at: DateTime.utc_now())
+      iex> {:ok, hash_set} = Explorer.Chain.stream_transactions_with_unfetched_internal_transactions(
+      ...>   [:hash],
+      ...>   MapSet.new(),
+      ...>   fn %Explorer.Chain.Transaction{hash: hash}, acc ->
+      ...>     MapSet.put(acc, hash)
+      ...>   end
+      ...> )
+      iex> pending.hash in hash_set
+      false
+      iex> unfetched_collated.hash in hash_set
+      true
+      iex> fetched_collated.hash in hash_set
+      false
+
   """
   @spec stream_transactions_with_unfetched_internal_transactions(
           fields :: [
@@ -1137,7 +1164,13 @@ defmodule Explorer.Chain do
   def stream_transactions_with_unfetched_internal_transactions(fields, initial, reducer) when is_function(reducer, 2) do
     Repo.transaction(
       fn ->
-        query = from(t in Transaction, where: is_nil(t.internal_transactions_indexed_at), select: ^fields)
+        query =
+          from(
+            t in Transaction,
+            # exclude pending transactions
+            where: not is_nil(t.block_hash) and is_nil(t.internal_transactions_indexed_at),
+            select: ^fields
+          )
 
         query
         |> Repo.stream(timeout: :infinity)

apps/explorer/lib/explorer/chain/transaction.ex

@@ -7,8 +7,8 @@ defmodule Explorer.Chain.Transaction do
   alias Explorer.Chain.{Address, Block, Data, Gas, Hash, InternalTransaction, Log, Wei}
   alias Explorer.Chain.Transaction.Status
 
-  @optional_attrs ~w(block_hash block_number cumulative_gas_used from_address_hash gas_used index status
+  @optional_attrs ~w(block_hash block_number cumulative_gas_used from_address_hash gas_used index
-                     to_address_hash)a
+                     internal_transactions_indexed_at status to_address_hash)a
   @required_attrs ~w(gas gas_price hash input nonce public_key r s standard_v v value)a
 
   @typedoc """

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

@@ -0,0 +1,32 @@
+defmodule Explorer.Indexer.InternalTransactionFetcherTest do
+  use Explorer.DataCase, async: true
+
+  alias Explorer.Indexer.InternalTransactionFetcher
+
+  describe "init/2" do
+    test "does not buffer pending transactions" do
+      insert(:transaction)
+
+      assert InternalTransactionFetcher.init([], fn hash_string, acc -> [hash_string | acc] end) == []
+    end
+
+    test "buffers collated transactions with unfetched internal transactions" do
+      collated_unfetched_transaction =
+        :transaction
+        |> insert()
+        |> with_block()
+
+      assert InternalTransactionFetcher.init([], fn hash_string, acc -> [hash_string | acc] end) == [
+               to_string(collated_unfetched_transaction.hash)
+             ]
+    end
+
+    test "does not buffer collated transactions with fetched internal transactions" do
+      :transaction
+      |> insert()
+      |> with_block(internal_transactions_indexed_at: DateTime.utc_now())
+
+      assert InternalTransactionFetcher.init([], fn hash_string, acc -> [hash_string | acc] end) == []
+    end
+  end
+end

apps/explorer/test/support/factory.ex

@@ -95,6 +95,7 @@ defmodule Explorer.Factory do
 
     cumulative_gas_used = collated_params[:cumulative_gas_used] || Enum.random(21_000..100_000)
     gas_used = collated_params[:gas_used] || Enum.random(21_000..100_000)
+    internal_transactions_indexed_at = collated_params[:internal_transactions_indexed_at]
     status = collated_params[:status] || Enum.random(0..1)
 
     transaction
@@ -104,6 +105,7 @@ defmodule Explorer.Factory do
       cumulative_gas_used: cumulative_gas_used,
       gas_used: gas_used,
       index: next_transaction_index,
+      internal_transactions_indexed_at: internal_transactions_indexed_at,
       status: status
     })
     |> Repo.update!()

coveralls.json

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