Merge branch 'pp-pending-blocks-ops' into ab-do-no-remove-nonconsensus-log

apps/explorer/lib/explorer/chain.ex

@@ -1664,7 +1664,18 @@ 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(block_hashes) do
+    query =
+      from(
+        po in PendingBlockOperation,
+        where: po.block_hash in ^block_hashes
+      )
+
+    {_, _} = Repo.delete_all(query)
+
+    :ok
+  end
+
   def remove_nonconsensus_blocks_from_pending_ops do
     query =
       from(

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

@@ -8,7 +8,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
   import Ecto.Query, only: [from: 2, subquery: 1]
 
   alias Ecto.{Changeset, Multi, Repo}
-
+  alias Explorer.Chain
   alias Explorer.Chain.{Address, Block, Import, PendingBlockOperation, Transaction}
   alias Explorer.Chain.Block.Reward
   alias Explorer.Chain.Import.Runner
@@ -303,6 +303,8 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
         timeout: timeout
       )
 
+    :ok = Chain.remove_nonconsensus_blocks_from_pending_ops(removed_consensus_block_hashes)
+
     {:ok, removed_consensus_block_hashes}
   rescue
     postgrex_error in Postgrex.Error ->

apps/explorer/test/explorer/chain_test.exs

@@ -49,6 +49,23 @@ defmodule Explorer.ChainTest do
       assert Repo.get(PendingBlockOperation, block.hash)
       assert is_nil(Repo.get(PendingBlockOperation, nonconsensus_block.hash))
     end
+
+    test "removes pending ops for nonconsensus blocks by block hashes" 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)
+
+      nonconsensus_block1 = insert(:block, consensus: false)
+      insert(:pending_block_operation, block: nonconsensus_block1, fetch_internal_transactions: true)
+
+      :ok = Chain.remove_nonconsensus_blocks_from_pending_ops([nonconsensus_block1.hash])
+
+      assert Repo.get(PendingBlockOperation, block.hash)
+      assert Repo.get(PendingBlockOperation, nonconsensus_block.hash)
+      assert is_nil(Repo.get(PendingBlockOperation, nonconsensus_block1.hash))
+    end
   end
 
   describe "count_addresses_with_balance_from_cache/0" do

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

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