fix: Transactions and token transfers block_consensus (#10285)

apps/explorer/config/config.exs

@@ -123,6 +123,8 @@ config :explorer, Explorer.Migrator.SanitizeMissingBlockRanges, enabled: true
 config :explorer, Explorer.Migrator.SanitizeIncorrectNFTTokenTransfers, enabled: true
 config :explorer, Explorer.Migrator.TokenTransferTokenType, enabled: true
 config :explorer, Explorer.Migrator.SanitizeIncorrectWETHTokenTransfers, enabled: true
+config :explorer, Explorer.Migrator.TransactionBlockConsensus, enabled: true
+config :explorer, Explorer.Migrator.TokenTransferBlockConsensus, enabled: true
 
 config :explorer, Explorer.Chain.Fetcher.CheckBytecodeMatchingOnDemand, enabled: true
 

apps/explorer/config/runtime/test.exs

@@ -45,6 +45,8 @@ config :explorer, Explorer.Migrator.SanitizeMissingBlockRanges, enabled: false
 config :explorer, Explorer.Migrator.SanitizeIncorrectNFTTokenTransfers, enabled: false
 config :explorer, Explorer.Migrator.TokenTransferTokenType, enabled: false
 config :explorer, Explorer.Migrator.SanitizeIncorrectWETHTokenTransfers, enabled: false
+config :explorer, Explorer.Migrator.TransactionBlockConsensus, enabled: false
+config :explorer, Explorer.Migrator.TokenTransferBlockConsensus, enabled: false
 
 config :explorer,
   realtime_events_sender: Explorer.Chain.Events.SimpleSender

apps/explorer/lib/explorer/application.ex

@@ -138,6 +138,8 @@ defmodule Explorer.Application do
         configure(Explorer.Migrator.SanitizeIncorrectNFTTokenTransfers),
         configure(Explorer.Migrator.TokenTransferTokenType),
         configure(Explorer.Migrator.SanitizeIncorrectWETHTokenTransfers),
+        configure(Explorer.Migrator.TransactionBlockConsensus),
+        configure(Explorer.Migrator.TokenTransferBlockConsensus),
         configure_chain_type_dependent_process(Explorer.Chain.Cache.StabilityValidatorsCounters, :stability)
       ]
       |> List.flatten()

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

@@ -12,7 +12,7 @@ defmodule Explorer.Chain.Import do
   require Logger
 
   @stages [
-    Import.Stage.AddressesBlocksCoinBalances,
+    Import.Stage.BlockRelated,
     Import.Stage.BlockReferencing,
     Import.Stage.BlockFollowing,
     Import.Stage.BlockPending

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

@@ -1,7 +1,7 @@
 defmodule Explorer.Chain.Import.Stage.BlockFollowing do
   @moduledoc """
   Imports any tables that follows and cannot be imported at the same time as
-  those imported by `Explorer.Chain.Import.Stage.AddressesBlocksCoinBalances` and `Explorer.Chain.Import.Stage.BlockReferencing`
+  those imported by `Explorer.Chain.Import.Stage.BlockRelated` and `Explorer.Chain.Import.Stage.BlockReferencing`
   """
 
   alias Explorer.Chain.Import.{Runner, Stage}

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

@@ -2,7 +2,7 @@ defmodule Explorer.Chain.Import.Stage.BlockPending do
   @moduledoc """
   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.AddressesBlocksCoinBalances` and `Explorer.Chain.Import.Stage.BlockReferencing`
+  `Explorer.Chain.Import.Stage.BlockRelated` and `Explorer.Chain.Import.Stage.BlockReferencing`
   """
 
   alias Explorer.Chain.Import.{Runner, Stage}

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

@@ -1,18 +1,16 @@
 defmodule Explorer.Chain.Import.Stage.BlockReferencing do
   @moduledoc """
   Imports any tables that reference `t:Explorer.Chain.Block.t/0` and that were
-  imported by `Explorer.Chain.Import.Stage.AddressesBlocksCoinBalances`.
+  imported by `Explorer.Chain.Import.Stage.BlockRelated`.
   """
 
   alias Explorer.Chain.Import.{Runner, Stage}
 
   @behaviour Stage
   @default_runners [
-    Runner.Transactions,
     Runner.Transaction.Forks,
     Runner.Logs,
     Runner.Tokens,
-    Runner.TokenTransfers,
     Runner.TokenInstances,
     Runner.Address.TokenBalances,
     Runner.TransactionActions,

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

@@ -1,7 +1,6 @@
-defmodule Explorer.Chain.Import.Stage.AddressesBlocksCoinBalances do
+defmodule Explorer.Chain.Import.Stage.BlockRelated do
   @moduledoc """
-  Import addresses, blocks and balances.
-  No tables have foreign key to addresses anymore, so it's possible to import addresses along with them.
+  Import blocks along with block related entities.
   """
 
   alias Explorer.Chain.Import.{Runner, Stage}
@@ -13,7 +12,9 @@ defmodule Explorer.Chain.Import.Stage.AddressesBlocksCoinBalances do
   @rest_runners [
     Runner.Address.CoinBalances,
     Runner.Blocks,
-    Runner.Address.CoinBalancesDaily
+    Runner.Address.CoinBalancesDaily,
+    Runner.Transactions,
+    Runner.TokenTransfers
   ]
 
   @impl Stage

apps/explorer/lib/explorer/migrator/token_transfer_block_consensus.ex

@@ -0,0 +1,57 @@
+defmodule Explorer.Migrator.TokenTransferBlockConsensus do
+  @moduledoc """
+  Fixes token transfers block_consensus field
+  """
+
+  use Explorer.Migrator.FillingMigration
+
+  import Ecto.Query
+
+  alias Explorer.Chain.TokenTransfer
+  alias Explorer.Migrator.FillingMigration
+  alias Explorer.Repo
+
+  @migration_name "token_transfers_block_consensus"
+
+  @impl FillingMigration
+  def migration_name, do: @migration_name
+
+  @impl FillingMigration
+  def last_unprocessed_identifiers do
+    limit = batch_size() * concurrency()
+
+    unprocessed_data_query()
+    |> select([tt], {tt.transaction_hash, tt.block_hash, tt.log_index})
+    |> limit(^limit)
+    |> Repo.all(timeout: :infinity)
+  end
+
+  @impl FillingMigration
+  def unprocessed_data_query do
+    from(
+      tt in TokenTransfer,
+      join: block in assoc(tt, :block),
+      where: tt.block_consensus != block.consensus
+    )
+  end
+
+  @impl FillingMigration
+  def update_batch(token_transfer_ids) do
+    token_transfer_ids
+    |> build_update_query()
+    |> Repo.query!([], timeout: :infinity)
+  end
+
+  @impl FillingMigration
+  def update_cache, do: :ok
+
+  defp build_update_query(token_transfer_ids) do
+    """
+    UPDATE token_transfers tt
+    SET block_consensus = b.consensus
+    FROM blocks b
+    WHERE tt.block_hash = b.hash
+      AND (tt.transaction_hash, tt.block_hash, tt.log_index) IN #{TokenTransfer.encode_token_transfer_ids(token_transfer_ids)};
+    """
+  end
+end

apps/explorer/lib/explorer/migrator/transaction_block_consensus.ex

@@ -0,0 +1,52 @@
+defmodule Explorer.Migrator.TransactionBlockConsensus do
+  @moduledoc """
+  Fixes transactions block_consensus field
+  """
+
+  use Explorer.Migrator.FillingMigration
+
+  import Ecto.Query
+
+  alias Explorer.Chain.Transaction
+  alias Explorer.Migrator.FillingMigration
+  alias Explorer.Repo
+
+  @migration_name "transactions_block_consensus"
+
+  @impl FillingMigration
+  def migration_name, do: @migration_name
+
+  @impl FillingMigration
+  def last_unprocessed_identifiers do
+    limit = batch_size() * concurrency()
+
+    unprocessed_data_query()
+    |> select([t], t.hash)
+    |> limit(^limit)
+    |> Repo.all(timeout: :infinity)
+  end
+
+  @impl FillingMigration
+  def unprocessed_data_query do
+    from(
+      transaction in Transaction,
+      join: block in assoc(transaction, :block),
+      where: transaction.block_consensus != block.consensus
+    )
+  end
+
+  @impl FillingMigration
+  def update_batch(transaction_hashes) do
+    query =
+      from(transaction in Transaction,
+        join: block in assoc(transaction, :block),
+        where: transaction.hash in ^transaction_hashes,
+        update: [set: [block_consensus: block.consensus]]
+      )
+
+    Repo.update_all(query, [], timeout: :infinity)
+  end
+
+  @impl FillingMigration
+  def update_cache, do: :ok
+end

apps/explorer/test/explorer/chain/import_test.exs

@@ -375,12 +375,12 @@ defmodule Explorer.Chain.ImportTest do
       not_existing_block_hash = "0xf6b4b8c88df3ebd252ec476328334dc026cf66606a84fb769b3d3cbccc8471db"
 
       incorrect_data =
-        update_in(@import_data, [:transactions, :params], fn params ->
+        update_in(@import_data, [:logs, :params], fn params ->
           [params |> Enum.at(0) |> Map.put(:block_hash, not_existing_block_hash)]
         end)
 
       assert_raise(Postgrex.Error, fn -> Import.all(incorrect_data) end)
-      assert [] = Repo.all(Transaction)
+      assert [] = Repo.all(Log)
       assert %{consensus: true, refetch_needed: true} = Repo.one(Block)
     end