test fix, comments and small refactorings

apps/block_scout_web/test/block_scout_web/controllers/address_read_contract_controller_test.exs

@@ -44,13 +44,15 @@ defmodule BlockScoutWeb.AddressReadContractControllerTest do
     test "returns not found for an unverified contract", %{conn: conn} do
       contract_address = insert(:contract_address)
 
-      transaction = insert(:transaction, from_address: contract_address)
+      transaction = insert(:transaction, from_address: contract_address) |> with_block()
 
       insert(
         :internal_transaction_create,
         index: 0,
         transaction: transaction,
-        created_contract_address: contract_address
+        created_contract_address: contract_address,
+        block_hash: transaction.block_hash,
+        block_index: 0
       )
 
       conn = get(conn, address_read_contract_path(BlockScoutWeb.Endpoint, :index, contract_address.hash))

apps/block_scout_web/test/block_scout_web/features/viewing_app_test.exs

@@ -7,7 +7,7 @@ defmodule BlockScoutWeb.ViewingAppTest do
   alias BlockScoutWeb.Counters.BlocksIndexedCounter
   alias Explorer.Counters.AddressesCounter
   alias Explorer.{Repo}
-  alias Explorer.Chain.{PendingBlockOperation, Transaction}
+  alias Explorer.Chain.PendingBlockOperation
 
   setup do
     start_supervised!(AddressesCounter)

apps/explorer/lib/explorer/chain.ex

@@ -724,29 +724,20 @@ defmodule Explorer.Chain do
   """
   @spec finished_indexing?() :: boolean()
   def finished_indexing? do
-    transaction_exists =
-      Transaction
-      |> limit(1)
-      |> Repo.one()
-
-    min_block_number_transaction = Repo.aggregate(Transaction, :min, :block_number)
-
-    if transaction_exists do
-      if min_block_number_transaction do
-        query =
-          from(
-            b in Block,
-            join: pending_ops in assoc(b, :pending_operations),
-            where: pending_ops.fetch_internal_transactions,
-            where: b.consensus and b.number == ^min_block_number_transaction
-          )
+    with {:transactions_exist, true} <- {:transactions_exist, Repo.exists?(Transaction)},
+         min_block_number when not is_nil(min_block_number) <- Repo.aggregate(Transaction, :min, :block_number) do
+      query =
+        from(
+          b in Block,
+          join: pending_ops in assoc(b, :pending_operations),
+          where: pending_ops.fetch_internal_transactions,
+          where: b.consensus and b.number == ^min_block_number
+        )
 
-        !Repo.exists?(query)
-      else
-        false
-      end
+      !Repo.exists?(query)
     else
-      true
+      {:transactions_exist, false} -> true
+      nil -> false
     end
   end
 
@@ -1338,10 +1329,7 @@ defmodule Explorer.Chain do
   @doc """
   The number of `t:Explorer.Chain.InternalTransaction.t/0`.
 
-      iex> transaction =
-      ...>   :transaction |>
-      ...>   insert() |>
-      ...>   with_block()
+      iex> transaction = :transaction |> insert() |> with_block()
       iex> insert(:internal_transaction, index: 0, transaction: transaction, block_hash: transaction.block_hash, block_index: 0)
       iex> Explorer.Chain.internal_transaction_count()
       1
@@ -1353,8 +1341,7 @@ defmodule Explorer.Chain do
 
   """
   def internal_transaction_count do
-    InternalTransaction.where_nonpending_block()
-    |> Repo.aggregate(:count, :transaction_hash)
+    Repo.aggregate(InternalTransaction.where_nonpending_block(), :count, :transaction_hash)
   end
 
   @doc """
@@ -1630,7 +1617,8 @@ defmodule Explorer.Chain do
   end
 
   @doc """
-  Returns a stream of all blocks with unfetched internal transactions.
+  Returns a stream of all blocks with unfetched internal transactions, using
+  the `pending_block_operation` table.
 
   Only blocks with consensus are returned.
 

apps/explorer/lib/explorer/chain/import/runner/blocks.ex

@@ -317,7 +317,6 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
       nonconsensus_hashes
       |> MapSet.new()
       |> MapSet.union(MapSet.new(hashes))
-      |> MapSet.to_list()
       |> Enum.sort()
       |> Enum.map(fn hash ->
         %{block_hash: hash, fetch_internal_transactions: true}

apps/explorer/lib/explorer/chain/import/runner/internal_transactions.ex

@@ -18,10 +18,7 @@ defmodule Explorer.Chain.Import.Runner.InternalTransactions do
   # milliseconds
   @timeout 60_000
 
-  @type imported :: [
-          %{required(:index) => non_neg_integer(), required(:transaction_hash) => Hash.Full.t()}
-          | %{required(:empty_block_number) => non_neg_integer()}
-        ]
+  @type imported :: [InternalTransaction.t()]
 
   @impl Runner
   def ecto_schema_module, do: InternalTransaction
@@ -50,7 +47,8 @@ defmodule Explorer.Chain.Import.Runner.InternalTransactions do
 
     update_transactions_options = %{timeout: transactions_timeout, timestamps: timestamps}
 
-    internal_transactions_changes = Enum.filter(changes_list, &Map.has_key?(&1, :type))
+    # filter out params with just `block_number` (indicating blocks without internal transactions)
+    internal_transactions_params = Enum.filter(changes_list, &Map.has_key?(&1, :type))
 
     # Enforce ShareLocks tables order (see docs: sharelocks.md)
     multi
@@ -64,14 +62,14 @@ defmodule Explorer.Chain.Import.Runner.InternalTransactions do
       acquire_transactions(repo, pending_block_hashes)
     end)
     |> Multi.run(:invalid_block_numbers, fn _, %{acquire_transactions: transactions} ->
-      invalid_block_numbers(transactions, internal_transactions_changes)
+      invalid_block_numbers(transactions, internal_transactions_params)
     end)
     |> Multi.run(:valid_internal_transactions, fn _,
                                                   %{
                                                     acquire_transactions: transactions,
                                                     invalid_block_numbers: invalid_block_numbers
                                                   } ->
-      valid_internal_transactions(transactions, internal_transactions_changes, invalid_block_numbers)
+      valid_internal_transactions(transactions, internal_transactions_params, invalid_block_numbers)
     end)
     |> Multi.run(:internal_transactions, fn repo, %{valid_internal_transactions: valid_internal_transactions} ->
       insert(repo, valid_internal_transactions, insert_options)
@@ -201,6 +199,7 @@ defmodule Explorer.Chain.Import.Runner.InternalTransactions do
       from(
         pending_ops in PendingBlockOperation,
         where: pending_ops.block_hash in ^block_hashes,
+        where: pending_ops.fetch_internal_transactions,
         select: pending_ops.block_hash,
         # Enforce PendingBlockOperation ShareLocks order (see docs: sharelocks.md)
         order_by: [asc: pending_ops.block_hash],
@@ -224,10 +223,17 @@ defmodule Explorer.Chain.Import.Runner.InternalTransactions do
     {:ok, repo.all(query)}
   end
 
-  defp invalid_block_numbers(transactions, internal_transactions_changes) do
+  defp invalid_block_numbers(transactions, internal_transactions_params) do
+    # Finds all mistmatches between transactions and internal transactions
+    # for a block number:
+    # - there are no internal txs for some transactions
+    # - there are no transactions for some internal transactions
+    # - there are internal txs with a different block number than their transactions
+    # Returns block numbers where any of these issues is found
+
     required_tuples = MapSet.new(transactions, &{&1.hash, &1.block_number})
 
-    candidate_tuples = MapSet.new(internal_transactions_changes, &{&1.transaction_hash, &1.block_number})
+    candidate_tuples = MapSet.new(internal_transactions_params, &{&1.transaction_hash, &1.block_number})
 
     all_tuples = MapSet.union(required_tuples, candidate_tuples)
 
@@ -242,11 +248,11 @@ defmodule Explorer.Chain.Import.Runner.InternalTransactions do
     {:ok, invalid_numbers}
   end
 
-  defp valid_internal_transactions(transactions, internal_transactions_changes, invalid_block_numbers) do
+  defp valid_internal_transactions(transactions, internal_transactions_params, invalid_block_numbers) do
     blocks_map = Map.new(transactions, &{&1.block_number, &1.block_hash})
 
     valid_internal_txs =
-      internal_transactions_changes
+      internal_transactions_params
       |> Enum.group_by(& &1.block_number)
       |> Map.drop(invalid_block_numbers)
       |> Enum.flat_map(fn {block_number, entries} ->
@@ -266,13 +272,17 @@ defmodule Explorer.Chain.Import.Runner.InternalTransactions do
   end
 
   def defer_internal_transactions_primary_key(repo) do
-    # allows internal_transactions primary key to skips being checked during the
-    # DB transactions and instead be checked at the end of it.
-    # This allows us to use a more efficient upserting logic
+    # Allows internal_transactions primary key to not be checked during the
+    # DB transactions and instead be checked only at the end of it.
+    # This allows us to use a more efficient upserting logic, while keeping the
+    # uniqueness valid.
     SQL.query(repo, "SET CONSTRAINTS internal_transactions_pkey DEFERRED")
   end
 
   def remove_left_over_internal_transactions(repo, valid_internal_transactions) do
+    # Removes internal transactions that were part of a block before a refetch
+    # and have not been upserted with new ones (if any exist).
+
     case valid_internal_transactions do
       [] ->
         {:ok, []}

apps/explorer/lib/explorer/chain/import/runner/transactions.ex

@@ -78,10 +78,8 @@ defmodule Explorer.Chain.Import.Runner.Transactions do
        when is_list(changes_list) do
     on_conflict = Map.get_lazy(options, :on_conflict, &default_on_conflict/0)
 
-    ordered_changes_list =
-      changes_list
-      # Enforce Transaction ShareLocks order (see docs: sharelocks.md)
-      |> Enum.sort_by(& &1.hash)
+    # Enforce Transaction ShareLocks order (see docs: sharelocks.md)
+    ordered_changes_list = Enum.sort_by(changes_list, & &1.hash)
 
     Import.insert_changes_list(
       repo,

apps/explorer/lib/explorer/chain/import/stage/block_pending.ex

@@ -1,7 +1,8 @@
 defmodule Explorer.Chain.Import.Stage.BlockPending do
   @moduledoc """
-  Imports any tables that follows and cannot be imported at the same time as
-  those imported by `Explorer.Chain.Import.Stage.Addresses`,
+  Imports any tables that uses `Explorer.Chain.PendingBlockOperation` to track
+  progress and cannot be imported at the same time as those imported by
+  `Explorer.Chain.Import.Stage.Addresses`,
   `Explorer.Chain.Import.Stage.AddressReferencing` and
   `Explorer.Chain.Import.Stage.BlockReferencing`
   """

apps/explorer/lib/explorer/chain/internal_transaction.ex

@@ -22,11 +22,15 @@ defmodule Explorer.Chain.InternalTransaction do
    * `to_address` - the sink of the `value`
    * `to_address_hash` - hash of the sink of the `value`
    * `trace_address` - list of traces
-   * `transaction` - transaction in which this transaction occurred
+   * `transaction` - transaction in which this internal transaction occurred
    * `transaction_hash` - foreign key for `transaction`
    * `transaction_index` - the `t:Explorer.Chain.Transaction.t/0` `index` of `transaction` in `block_number`.
    * `type` - type of internal transaction
    * `value` - value of transferred from `from_address` to `to_address`
+   * `block` - block in which this internal transaction occurred
+   * `block_hash` - foreign key for `block`
+   * `block_index` - the index of this internal transaction inside the `block`
+   * `pending_block` - `nil` if `block` has all its internal transactions fetched
   """
   @type t :: %__MODULE__{
           block_number: Explorer.Chain.Block.block_number() | nil,
@@ -409,17 +413,12 @@ defmodule Explorer.Chain.InternalTransaction do
   on its own or use empty types to know that a block has no internal transactions.
   """
   def blockless_changeset(%__MODULE__{} = internal_transaction, attrs \\ %{}) do
-    changeset =
-      internal_transaction
-      |> cast(attrs, ~w(type block_number)a)
+    changeset = cast(internal_transaction, attrs, ~w(type block_number)a)
 
-    if is_nil(get_field(changeset, :type)) do
-      changeset
-      |> validate_required(~w(block_number)a)
+    if validate_required(changeset, ~w(type)a).valid? do
+      type_changeset(changeset, attrs)
     else
-      changeset
-      |> validate_required(~w(type)a)
-      |> type_changeset(attrs)
+      validate_required(changeset, ~w(block_number)a)
     end
   end
 

apps/indexer/lib/indexer/fetcher/internal_transaction.ex

@@ -145,7 +145,6 @@ defmodule Indexer.Fetcher.InternalTransaction do
   end
 
   defp import_internal_transaction(internal_transactions_params, unique_numbers) do
-    unique_numbers_count = Enum.count(unique_numbers)
     internal_transactions_params_without_failed_creations = remove_failed_creations(internal_transactions_params)
 
     addresses_params =
@@ -192,7 +191,7 @@ defmodule Indexer.Fetcher.InternalTransaction do
             ]
           end,
           step: step,
-          error_count: unique_numbers_count
+          error_count: Enum.count(unique_numbers)
         )
 
         # re-queue the de-duped entries