remove nonconsensus internal transactions

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

@@ -368,8 +368,15 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
        ) do
     with {:ok, deleted_token_transfers} <-
            remove_nonconsensus_token_transfers(repo, nonconsensus_block_numbers, insert_options),
-         {:ok, deleted_logs} <- remove_nonconsensus_logs(repo, nonconsensus_block_numbers, insert_options) do
-      {:ok, %{token_transfers: deleted_token_transfers, logs: deleted_logs}}
+         {:ok, deleted_logs} <- remove_nonconsensus_logs(repo, nonconsensus_block_numbers, insert_options),
+         {:ok, deleted_internal_transactions} <-
+           remove_nonconsensus_internal_transactions(repo, nonconsensus_block_numbers, insert_options) do
+      {:ok,
+       %{
+         token_transfers: deleted_token_transfers,
+         logs: deleted_logs,
+         internal_transactions: deleted_internal_transactions
+       }}
     end
   end
 
@@ -406,6 +413,46 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     end
   end
 
+  defp remove_nonconsensus_internal_transactions(repo, nonconsensus_block_numbers, %{timeout: timeout}) do
+    transaction_query =
+      from(transaction in Transaction,
+        where: transaction.block_number in ^nonconsensus_block_numbers,
+        select: map(transaction, [:hash]),
+        order_by: transaction.hash
+      )
+
+    ordered_internal_transactions =
+      from(internal_transaction in InternalTransaction,
+        inner_join: transaction in subquery(transaction_query),
+        on: internal_transaction.transaction_hash == transaction.hash,
+        select: map(internal_transaction, [:transaction_hash, :index]),
+        # Enforce Log ShareLocks order (see docs: sharelocks.md)
+        order_by: [
+          internal_transaction.transaction_hash,
+          internal_transaction.index
+        ],
+        lock: "FOR UPDATE OF i0"
+      )
+
+    query =
+      from(internal_transaction in InternalTransaction,
+        select: map(internal_transaction, [:transaction_hash, :index]),
+        inner_join: ordered_internal_transaction in subquery(ordered_internal_transactions),
+        on:
+          ordered_internal_transaction.transaction_hash == internal_transaction.transaction_hash and
+            ordered_internal_transaction.index == internal_transaction.index
+      )
+
+    try do
+      {_count, deleted_internal_transactions} = repo.delete_all(query, timeout: timeout)
+
+      {:ok, deleted_internal_transactions}
+    rescue
+      postgrex_error in Postgrex.Error ->
+        {:error, %{exception: postgrex_error, block_numbers: nonconsensus_block_numbers}}
+    end
+  end
+
   defp remove_nonconsensus_logs(repo, nonconsensus_block_numbers, %{timeout: timeout}) do
     transaction_query =
       from(transaction in Transaction,

apps/explorer/test/explorer/chain/import/runner/blocks_test.exs

@@ -7,7 +7,7 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
 
   alias Ecto.Multi
   alias Explorer.Chain.Import.Runner.{Blocks, Transactions}
-  alias Explorer.Chain.{Address, Block, Log, Transaction, TokenTransfer}
+  alias Explorer.Chain.{Address, Block, InternalTransaction, Log, Transaction, TokenTransfer}
   alias Explorer.Chain
   alias Explorer.Repo
 
@@ -177,6 +177,30 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
       assert count(Log) == 0
     end
 
+    test "remove_nonconsensus_date deletes nonconsensus internal transactions", %{
+      consensus_block: %{number: block_number} = block,
+      options: options
+    } do
+      old_block = insert(:block, number: block_number, consensus: true)
+      forked_transaction = :transaction |> insert() |> with_block(old_block)
+
+      %InternalTransaction{index: index, transaction_hash: hash} =
+        insert(:internal_transaction, index: 0, transaction: forked_transaction)
+
+      assert count(InternalTransaction) == 1
+
+      assert {:ok,
+              %{
+                remove_nonconsensus_data: %{
+                  internal_transactions: [
+                    %{transaction_hash: ^hash, index: ^index}
+                  ]
+                }
+              }} = run_block_consensus_change(block, true, options)
+
+      assert count(InternalTransaction) == 0
+    end
+
     test "derive_address_current_token_balances inserts rows if there is an address_token_balance left for the rows deleted by delete_address_current_token_balances",
          %{consensus_block: %{number: block_number} = block, options: options} do
       token = insert(:token)