remove nonconsensus blocks from pending blocks

apps/explorer/lib/explorer/chain.ex

@@ -38,6 +38,7 @@ defmodule Explorer.Chain do
     Import,
     InternalTransaction,
     Log,
+    PendingBlockOperation,
     SmartContract,
     StakingPool,
     Token,
@@ -1660,6 +1661,21 @@ defmodule Explorer.Chain do
     Repo.stream_reduce(query, initial, reducer)
   end
 
+  @spec remove_nonconsensus_blocks_from_pending_ops() :: :ok
+  def remove_nonconsensus_blocks_from_pending_ops do
+    query =
+      from(
+        po in PendingBlockOperation,
+        inner_join: block in Block,
+        on: block.hash == po.block_hash,
+        where: block.consensus == false
+      )
+
+    {_, _} = Repo.delete_all(query)
+
+    :ok
+  end
+
   @spec stream_transactions_with_unfetched_created_contract_codes(
           fields :: [
             :block_hash

apps/explorer/test/explorer/chain_test.exs

@@ -18,6 +18,7 @@ defmodule Explorer.ChainTest do
     Hash,
     InternalTransaction,
     Log,
+    PendingBlockOperation,
     Token,
     TokenTransfer,
     Transaction,
@@ -35,6 +36,21 @@ defmodule Explorer.ChainTest do
 
   setup :verify_on_exit!
 
+  describe "remove_nonconsensus_blocks_from_pending_ops/0" do
+    test "removes pending ops for nonconsensus blocks" do
+      block = insert(:block)
+      insert(:pending_block_operation, block: block, fetch_internal_transactions: true)
+
+      nonconsensus_block = insert(:block, consensus: false)
+      insert(:pending_block_operation, block: nonconsensus_block, fetch_internal_transactions: true)
+
+      :ok = Chain.remove_nonconsensus_blocks_from_pending_ops()
+
+      assert Repo.get(PendingBlockOperation, block.hash)
+      assert is_nil(Repo.get(PendingBlockOperation, nonconsensus_block.hash))
+    end
+  end
+
   describe "count_addresses_with_balance_from_cache/0" do
     test "returns the number of addresses with fetched_coin_balance > 0" do
       insert(:address, fetched_coin_balance: 0)

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

@@ -76,6 +76,8 @@ defmodule Indexer.Fetcher.InternalTransaction do
         reducer.(block_number, acc)
       end)
 
+    :ok = Chain.remove_nonconsensus_blocks_from_pending_ops()
+
     final
   end