Merge pull request #3330 from poanetwork/vb-exact-addr-txs-counter

Caching of address transactions counter, remove query 10_000 rows limit

CHANGELOG.md

@@ -1,6 +1,7 @@
 ## Current
 
 ### Features
+- [#3330](https://github.com/poanetwork/blockscout/pull/3330) - Caching of address transactions counter, remove query 10_000 rows limit
 
 
 ### Fixes

apps/block_scout_web/assets/js/pages/address.js

@@ -94,7 +94,7 @@ const elements = {
     render ($el, state, oldState) {
       if (state.countersFetched && state.transactionCount) {
         if (oldState.transactionCount === state.transactionCount) return
-        $el.empty().append('>= ' + numeral(state.transactionCount).format() + ' Transactions')
+        $el.empty().append(numeral(state.transactionCount).format() + ' Transactions')
         $el.show()
         $el.parent('.address-detail-item').removeAttr('style')
       } else {

apps/block_scout_web/lib/block_scout_web/controllers/address_controller.ex

@@ -4,6 +4,7 @@ defmodule BlockScoutWeb.AddressController do
   import BlockScoutWeb.Chain, only: [paging_options: 1, next_page_params: 3, split_list_by_page: 1]
 
   alias BlockScoutWeb.{AccessHelpers, AddressView}
+  alias Explorer.Counters.AddressTransactionsCounter
   alias Explorer.{Chain, Market}
   alias Explorer.ExchangeRates.Token
   alias Phoenix.View
@@ -109,25 +110,11 @@ defmodule BlockScoutWeb.AddressController do
     |> List.to_tuple()
   end
 
-  defp transaction_count(address) do
-    if contract?(address) do
-      incoming_transaction_count = Chain.address_to_incoming_transaction_count(address.hash)
-
-      if incoming_transaction_count == 0 do
-        Chain.total_transactions_sent_by_address(address.hash)
-      else
-        incoming_transaction_count
-      end
-    else
-      Chain.total_transactions_sent_by_address(address.hash)
-    end
+  def transaction_count(address) do
+    AddressTransactionsCounter.fetch(address)
   end
 
   defp validation_count(address) do
     Chain.address_to_validation_count(address.hash)
   end
-
-  defp contract?(%{contract_code: nil}), do: false
-
-  defp contract?(%{contract_code: _}), do: true
 end

apps/block_scout_web/test/block_scout_web/controllers/address_controller_test.exs

@@ -6,7 +6,7 @@ defmodule BlockScoutWeb.AddressControllerTest do
   import Mox
 
   alias Explorer.Chain.Address
-  alias Explorer.Counters.AddressesCounter
+  alias Explorer.Counters.{AddressesCounter, AddressTransactionsCounter}
 
   describe "GET index/2" do
     setup :set_mox_global

apps/explorer/config/config.exs

@@ -88,6 +88,17 @@ config :explorer, Explorer.Counters.AddressesCounter,
   enable_consolidation: true,
   update_interval_in_seconds: balances_update_interval || 30 * 60
 
+address_transactions_counter_cache_period =
+  case Integer.parse(System.get_env("ADDRESS_TRANSACTIONS_COUNTER_CACHE_PERIOD", "")) do
+    {secs, ""} -> :timer.seconds(secs)
+    _ -> :timer.hours(1)
+  end
+
+config :explorer, Explorer.Counters.AddressTransactionsCounter,
+  enabled: true,
+  enable_consolidation: true,
+  period: address_transactions_counter_cache_period
+
 bridge_market_cap_update_interval =
   if System.get_env("BRIDGE_MARKET_CAP_UPDATE_INTERVAL") do
     case Integer.parse(System.get_env("BRIDGE_MARKET_CAP_UPDATE_INTERVAL")) do

apps/explorer/lib/explorer/application.ex

@@ -78,6 +78,7 @@ defmodule Explorer.Application do
       configure(Explorer.Chain.Events.Listener),
       configure(Explorer.Counters.AddressesWithBalanceCounter),
       configure(Explorer.Counters.AddressesCounter),
+      configure(Explorer.Counters.AddressTransactionsCounter),
       configure(Explorer.Counters.AverageBlockTime),
       configure(Explorer.Counters.Bridge),
       configure(Explorer.Validator.MetadataProcessor),

apps/explorer/lib/explorer/chain.ex

@@ -714,15 +714,11 @@ defmodule Explorer.Chain do
 
   @spec address_to_incoming_transaction_count(Hash.Address.t()) :: non_neg_integer()
   def address_to_incoming_transaction_count(address_hash) do
-    paging_options = %PagingOptions{page_size: @max_incoming_transactions_count}
-
-    base_query =
-      paging_options
-      |> fetch_transactions()
-
     to_address_query =
-      base_query
-      |> where([t], t.to_address_hash == ^address_hash)
+      from(
+        transaction in Transaction,
+        where: transaction.to_address_hash == ^address_hash
+      )
 
     Repo.aggregate(to_address_query, :count, :hash, timeout: :infinity)
   end
@@ -1949,6 +1945,25 @@ defmodule Explorer.Chain do
     Repo.one(query)
   end
 
+  @spec address_to_transaction_count(Address.t()) :: non_neg_integer()
+  def address_to_transaction_count(address) do
+    if contract?(address) do
+      incoming_transaction_count = address_to_incoming_transaction_count(address.hash)
+
+      if incoming_transaction_count == 0 do
+        total_transactions_sent_by_address(address.hash)
+      else
+        incoming_transaction_count
+      end
+    else
+      total_transactions_sent_by_address(address.hash)
+    end
+  end
+
+  defp contract?(%{contract_code: nil}), do: false
+
+  defp contract?(%{contract_code: _}), do: true
+
   @doc """
   Returns a stream of unfetched `t:Explorer.Chain.Address.CoinBalance.t/0`.
 

apps/explorer/lib/explorer/counters/address_transactions_counter.ex

@@ -0,0 +1,112 @@
+defmodule Explorer.Counters.AddressTransactionsCounter do
+  @moduledoc """
+  Caches Address transactions counter.
+  """
+  use GenServer
+
+  alias Explorer.Chain
+
+  @cache_name :address_transactions_counter
+  @last_update_key "last_update"
+  @cache_period Application.get_env(:explorer, __MODULE__)[:period]
+
+  @ets_opts [
+    :set,
+    :named_table,
+    :public,
+    read_concurrency: true
+  ]
+
+  config = Application.get_env(:explorer, Explorer.Counters.AddressesCounter)
+  @enable_consolidation Keyword.get(config, :enable_consolidation)
+
+  @spec start_link(term()) :: GenServer.on_start()
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, :ok, name: __MODULE__)
+  end
+
+  @impl true
+  def init(_args) do
+    create_cache_table()
+
+    {:ok, %{consolidate?: enable_consolidation?()}, {:continue, :ok}}
+  end
+
+  @impl true
+  def handle_continue(:ok, %{consolidate?: true} = state) do
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_continue(:ok, state) do
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_info(:consolidate, state) do
+    {:noreply, state}
+  end
+
+  def fetch(address) do
+    if cache_expired?(address) do
+      Task.start_link(fn ->
+        update_cache(address)
+      end)
+    end
+
+    address_hash_string = get_address_hash_string(address)
+    fetch_from_cache("hash_#{address_hash_string}")
+  end
+
+  def cache_name, do: @cache_name
+
+  defp cache_expired?(address) do
+    address_hash_string = get_address_hash_string(address)
+    updated_at = fetch_from_cache("hash_#{address_hash_string}_#{@last_update_key}")
+
+    cond do
+      is_nil(updated_at) -> true
+      current_time() - updated_at > @cache_period -> true
+      true -> false
+    end
+  end
+
+  defp update_cache(address) do
+    address_hash_string = get_address_hash_string(address)
+    put_into_cache("hash_#{address_hash_string}_#{@last_update_key}", current_time())
+    new_data = Chain.address_to_transaction_count(address)
+    put_into_cache("hash_#{address_hash_string}", new_data)
+  end
+
+  defp fetch_from_cache(key) do
+    case :ets.lookup(@cache_name, key) do
+      [{_, value}] ->
+        value
+
+      [] ->
+        0
+    end
+  end
+
+  defp put_into_cache(key, value) do
+    :ets.insert(@cache_name, {key, value})
+  end
+
+  defp get_address_hash_string(address) do
+    Base.encode16(address.hash.bytes, case: :lower)
+  end
+
+  defp current_time do
+    utc_now = DateTime.utc_now()
+
+    DateTime.to_unix(utc_now, :millisecond)
+  end
+
+  def create_cache_table do
+    if :ets.whereis(@cache_name) == :undefined do
+      :ets.new(@cache_name, @ets_opts)
+    end
+  end
+
+  def enable_consolidation?, do: @enable_consolidation
+end