Add test for internal transactions fetcher race condition

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

@@ -363,7 +363,7 @@ defmodule Indexer.Fetcher.InternalTransaction do
 
   defp invalidate_block_from_error(_error_data), do: :ok
 
-  defp defaults do
+  def defaults do
     [
       poll: false,
       flush_interval: :timer.seconds(3),

apps/indexer/test/indexer/fetcher/internal_transaction_test.exs

@@ -5,8 +5,10 @@ defmodule Indexer.Fetcher.InternalTransactionTest do
   import ExUnit.CaptureLog
   import Mox
 
-  alias Explorer.Chain
-  alias Explorer.Chain.PendingBlockOperation
+  alias Ecto.Multi
+  alias Explorer.{Chain, Repo}
+  alias Explorer.Chain.{Block, PendingBlockOperation}
+  alias Explorer.Chain.Import.Runner.Blocks
   alias Indexer.Fetcher.{CoinBalance, InternalTransaction, PendingTransaction}
 
   # MUST use global mode because we aren't guaranteed to get PendingTransactionFetcher's pid back fast enough to `allow`
@@ -466,4 +468,39 @@ defmodule Indexer.Fetcher.InternalTransactionTest do
       assert logs =~ "foreign_key_violation on internal transactions import, foreign transactions hashes:"
     end
   end
+
+  test "doesn't delete pending block operations after block import if no async process was requested", %{
+    json_rpc_named_arguments: json_rpc_named_arguments
+  } do
+    fetcher_options =
+      Keyword.merge([poll: true, json_rpc_named_arguments: json_rpc_named_arguments], InternalTransaction.defaults())
+
+    if fetcher_options[:poll] do
+      expect(EthereumJSONRPC.Mox, :json_rpc, fn [%{id: id}], _options ->
+        {:ok, [%{id: id, result: []}]}
+      end)
+    end
+
+    InternalTransaction.Supervisor.Case.start_supervised!(fetcher_options)
+
+    %Ecto.Changeset{valid?: true, changes: block_changes} =
+      Block.changeset(%Block{}, params_for(:block, miner_hash: insert(:address).hash, number: 1))
+
+    changes_list = [block_changes]
+    timestamp = DateTime.utc_now()
+    options = %{timestamps: %{inserted_at: timestamp, updated_at: timestamp}}
+
+    assert [] = Repo.all(PendingBlockOperation)
+
+    {:ok, %{blocks: [%{number: block_number, hash: block_hash}]}} =
+      Multi.new()
+      |> Blocks.run(changes_list, options)
+      |> Repo.transaction()
+
+    assert %{block_number: ^block_number, block_hash: ^block_hash} = Repo.one(PendingBlockOperation)
+
+    Process.sleep(4000)
+
+    assert %{block_number: ^block_number, block_hash: ^block_hash} = Repo.one(PendingBlockOperation)
+  end
 end

apps/indexer/test/support/indexer/fetcher/internal_transaction_supervisor_case.ex

@@ -4,11 +4,13 @@ defmodule Indexer.Fetcher.InternalTransaction.Supervisor.Case do
   def start_supervised!(fetcher_arguments \\ []) when is_list(fetcher_arguments) do
     merged_fetcher_arguments =
       Keyword.merge(
-        fetcher_arguments,
-        flush_interval: 50,
-        max_batch_size: 1,
-        max_concurrency: 1,
-        poll: false
+        [
+          flush_interval: 50,
+          max_batch_size: 1,
+          max_concurrency: 1,
+          poll: false
+        ],
+        fetcher_arguments
       )
 
     [merged_fetcher_arguments]