Check left blocks in pending block operations in order to decide, if we need to display indexing int tx banner at the top

.tool-versions

@@ -1,3 +1,3 @@
-elixir 1.14.4-otp-25
+elixir 1.14.5-otp-25
 erlang 25.3
 nodejs 18.13.0

CHANGELOG.md

@@ -21,6 +21,7 @@
 
 ### Chore
 
+- [#7576](https://github.com/blockscout/blockscout/pull/7576) - Check left blocks in pending block operations in order to decide, if we need to display indexing int tx banner at the top
 - [#7543](https://github.com/blockscout/blockscout/pull/7543) - Allow hyphen in DB username
 
 <details>

apps/explorer/lib/explorer/application.ex

@@ -18,6 +18,7 @@ defmodule Explorer.Application do
     GasUsage,
     MinMissingBlockNumber,
     NetVersion,
+    PendingBlockOperation,
     Transaction,
     Transactions,
     TransactionsApiV2,
@@ -55,21 +56,22 @@ defmodule Explorer.Application do
       Explorer.SmartContract.VyperDownloader,
       {Registry, keys: :duplicate, name: Registry.ChainEvents, id: Registry.ChainEvents},
       {Admin.Recovery, [[], [name: Admin.Recovery]]},
-      Transaction,
+      Accounts,
       AddressSum,
       AddressSumMinusBurnt,
       Block,
+      BlockNumber,
       Blocks,
       GasPriceOracle,
       GasUsage,
       NetVersion,
-      BlockNumber,
+      PendingBlockOperation,
-      con_cache_child_spec(MarketHistoryCache.cache_name()),
+      Transaction,
-      con_cache_child_spec(RSK.cache_name(), ttl_check_interval: :timer.minutes(1), global_ttl: :timer.minutes(30)),
       Transactions,
       TransactionsApiV2,
-      Accounts,
       Uncles,
+      con_cache_child_spec(MarketHistoryCache.cache_name()),
+      con_cache_child_spec(RSK.cache_name(), ttl_check_interval: :timer.minutes(1), global_ttl: :timer.minutes(30)),
       {Redix, redix_opts()},
       {Explorer.Utility.MissingRangesManipulator, []}
     ]

apps/explorer/lib/explorer/chain.ex

@@ -82,6 +82,7 @@ defmodule Explorer.Chain do
 
   alias Explorer.Chain.Cache.Block, as: BlockCache
   alias Explorer.Chain.Cache.Helper, as: CacheHelper
+  alias Explorer.Chain.Cache.PendingBlockOperation, as: PendingBlockOperationCache
   alias Explorer.Chain.Fetcher.{CheckBytecodeMatchingOnDemand, LookUpSmartContractSourcesOnDemand}
   alias Explorer.Chain.Import.Runner
   alias Explorer.Chain.InternalTransaction.{CallType, Type}
@@ -1344,27 +1345,46 @@ defmodule Explorer.Chain do
       if variant == EthereumJSONRPC.Ganache || variant == EthereumJSONRPC.Arbitrum do
         true
       else
-        with {:transactions_exist, true} <- {:transactions_exist, select_repo(options).exists?(Transaction)},
+        check_left_blocks_to_index_internal_transactions(options)
-             min_block_number when not is_nil(min_block_number) <-
+      end
-               select_repo(options).aggregate(Transaction, :min, :block_number) do
+    end
-          min_block_number =
+  end
-            min_block_number
-            |> Decimal.max(EthereumJSONRPC.first_block_to_fetch(:trace_first_block))
-            |> Decimal.to_integer()
-
-          query =
-            from(
-              block in Block,
-              join: pending_ops in assoc(block, :pending_operations),
-              where: block.consensus and block.number == ^min_block_number
-            )
 
-          !select_repo(options).exists?(query)
+  defp check_left_blocks_to_index_internal_transactions(options) do
-        else
+    with {:transactions_exist, true} <- {:transactions_exist, select_repo(options).exists?(Transaction)},
-          {:transactions_exist, false} -> true
+         min_block_number when not is_nil(min_block_number) <-
-          nil -> false
+           select_repo(options).aggregate(Transaction, :min, :block_number) do
-        end
+      min_block_number =
+        min_block_number
+        |> Decimal.max(EthereumJSONRPC.first_block_to_fetch(:trace_first_block))
+        |> Decimal.to_integer()
+
+      query =
+        from(
+          block in Block,
+          join: pending_ops in assoc(block, :pending_operations),
+          where: block.consensus and block.number == ^min_block_number
+        )
+
+      if select_repo(options).exists?(query) do
+        false
+      else
+        check_indexing_internal_transactions_threshold()
       end
+    else
+      {:transactions_exist, false} -> true
+      nil -> false
+    end
+  end
+
+  defp check_indexing_internal_transactions_threshold do
+    pbo_count = PendingBlockOperationCache.estimated_count()
+
+    if pbo_count <
+         Application.get_env(:indexer, Indexer.Fetcher.InternalTransaction)[:indexing_finished_threshold] do
+      true
+    else
+      false
     end
   end
 
@@ -2297,7 +2317,7 @@ defmodule Explorer.Chain do
     if Application.get_env(:indexer, Indexer.Supervisor)[:enabled] &&
          not Application.get_env(:indexer, Indexer.Fetcher.InternalTransaction.Supervisor)[:disabled?] do
       %{max: max_saved_block_number} = BlockNumber.get_all()
-      pbo_count = Repo.aggregate(PendingBlockOperation, :count, timeout: :infinity)
+      pbo_count = PendingBlockOperationCache.estimated_count()
 
       min_blockchain_trace_block_number = min_block_number_from_config(:trace_first_block)
 

apps/explorer/lib/explorer/chain/cache/pending_block_operation.ex

@@ -0,0 +1,73 @@
+defmodule Explorer.Chain.Cache.PendingBlockOperation do
+  @moduledoc """
+  Cache for estimated `pending_block_operations` count.
+  """
+
+  use Explorer.Chain.MapCache,
+    name: :pending_block_operations_count,
+    key: :count,
+    key: :async_task,
+    global_ttl: Application.get_env(:explorer, __MODULE__)[:global_ttl],
+    ttl_check_interval: :timer.seconds(1),
+    callback: &async_task_on_deletion(&1)
+
+  require Logger
+
+  alias Explorer.Chain.Cache.Helper
+  alias Explorer.Chain.PendingBlockOperation
+  alias Explorer.Repo
+
+  @doc """
+  Estimated count of `t:Explorer.Chain.PendingBlockOperation.t/0`.
+
+  """
+  @spec estimated_count() :: non_neg_integer()
+  def estimated_count do
+    cached_value = __MODULE__.get_count()
+
+    if is_nil(cached_value) do
+      count = Helper.estimated_count_from("pending_block_operations")
+
+      max(count, 0)
+    else
+      cached_value
+    end
+  end
+
+  defp handle_fallback(:count) do
+    # This will get the task PID if one exists and launch a new task if not
+    # See next `handle_fallback` definition
+    get_async_task()
+
+    {:return, nil}
+  end
+
+  defp handle_fallback(:async_task) do
+    # If this gets called it means an async task was requested, but none exists
+    # so a new one needs to be launched
+    {:ok, task} =
+      Task.start(fn ->
+        try do
+          result = Repo.aggregate(PendingBlockOperation, :count, timeout: :infinity)
+
+          set_count(result)
+        rescue
+          e ->
+            Logger.debug([
+              "Couldn't update pending_block_operations count: ",
+              Exception.format(:error, e, __STACKTRACE__)
+            ])
+        end
+
+        set_async_task(nil)
+      end)
+
+    {:update, task}
+  end
+
+  # By setting this as a `callback` an async task will be started each time the
+  # `count` expires (unless there is one already running)
+  defp async_task_on_deletion({:delete, _, :count}), do: get_async_task()
+
+  defp async_task_on_deletion(_data), do: nil
+end

apps/explorer/test/explorer/chain_test.exs

@@ -30,6 +30,7 @@ defmodule Explorer.ChainTest do
   alias Explorer.{Chain, Etherscan}
   alias Explorer.Chain.Cache.Block, as: BlockCache
   alias Explorer.Chain.Cache.Transaction, as: TransactionCache
+  alias Explorer.Chain.Cache.PendingBlockOperation, as: PendingBlockOperationCache
   alias Explorer.Chain.InternalTransaction.Type
 
   alias Explorer.Chain.Supply.ProofOfAuthority
@@ -1207,6 +1208,12 @@ defmodule Explorer.ChainTest do
   end
 
   describe "finished_indexing_internal_transactions?/0" do
+    setup do
+      Supervisor.terminate_child(Explorer.Supervisor, PendingBlockOperationCache.child_id())
+      Supervisor.restart_child(Explorer.Supervisor, PendingBlockOperationCache.child_id())
+      on_exit(fn -> Supervisor.terminate_child(Explorer.Supervisor, PendingBlockOperationCache.child_id()) end)
+    end
+
     test "finished indexing" do
       block = insert(:block, number: 1)
 
@@ -1538,8 +1545,12 @@ defmodule Explorer.ChainTest do
 
   describe "indexed_ratio_internal_transactions/0" do
     setup do
+      Supervisor.terminate_child(Explorer.Supervisor, PendingBlockOperationCache.child_id())
+      Supervisor.restart_child(Explorer.Supervisor, PendingBlockOperationCache.child_id())
+
       on_exit(fn ->
         Application.put_env(:indexer, :trace_first_block, "")
+        Supervisor.terminate_child(Explorer.Supervisor, PendingBlockOperationCache.child_id())
       end)
     end
 
@@ -1552,6 +1563,8 @@ defmodule Explorer.ChainTest do
         end
       end
 
+      Chain.indexed_ratio_internal_transactions()
+
       assert Decimal.compare(Chain.indexed_ratio_internal_transactions(), Decimal.from_float(0.7)) == :eq
     end
 

config/runtime.exs

@@ -216,6 +216,9 @@ config :explorer, Explorer.Chain.Cache.Block,
 config :explorer, Explorer.Chain.Cache.Transaction,
   global_ttl: ConfigHelper.parse_time_env_var("CACHE_TXS_COUNT_PERIOD", "2h")
 
+config :explorer, Explorer.Chain.Cache.PendingBlockOperation,
+  global_ttl: ConfigHelper.parse_time_env_var("CACHE_PBO_COUNT_PERIOD", "20m")
+
 config :explorer, Explorer.Chain.Cache.GasPriceOracle,
   global_ttl: ConfigHelper.parse_time_env_var("GAS_PRICE_ORACLE_CACHE_PERIOD", "30s"),
   num_of_blocks: ConfigHelper.parse_integer_env_var("GAS_PRICE_ORACLE_NUM_OF_BLOCKS", 200),
@@ -323,6 +326,10 @@ config :explorer, Explorer.Chain.Cache.Uncles,
   ttl_check_interval: ConfigHelper.cache_ttl_check_interval(disable_indexer?),
   global_ttl: ConfigHelper.cache_global_ttl(disable_indexer?)
 
+config :explorer, Explorer.Chain.Cache.Uncles,
+  ttl_check_interval: ConfigHelper.cache_ttl_check_interval(disable_indexer?),
+  global_ttl: ConfigHelper.cache_global_ttl(disable_indexer?)
+
 config :explorer, Explorer.ThirdPartyIntegrations.Sourcify,
   server_url: System.get_env("SOURCIFY_SERVER_URL") || "https://sourcify.dev/server",
   enabled: ConfigHelper.parse_bool_env_var("SOURCIFY_INTEGRATION_ENABLED"),
@@ -494,7 +501,9 @@ config :indexer, Indexer.Fetcher.TokenInstance.Sanitize,
 
 config :indexer, Indexer.Fetcher.InternalTransaction,
   batch_size: ConfigHelper.parse_integer_env_var("INDEXER_INTERNAL_TRANSACTIONS_BATCH_SIZE", 10),
-  concurrency: ConfigHelper.parse_integer_env_var("INDEXER_INTERNAL_TRANSACTIONS_CONCURRENCY", 4)
+  concurrency: ConfigHelper.parse_integer_env_var("INDEXER_INTERNAL_TRANSACTIONS_CONCURRENCY", 4),
+  indexing_finished_threshold:
+    ConfigHelper.parse_integer_env_var("INDEXER_INTERNAL_TRANSACTIONS_INDEXING_FINISHED_THRESHOLD", 1000)
 
 config :indexer, Indexer.Fetcher.CoinBalance,
   batch_size: ConfigHelper.parse_integer_env_var("INDEXER_COIN_BALANCES_BATCH_SIZE", 500),