Merge pull request #917 from poanetwork/888

Optimize entry memory usage for work queues

apps/explorer/lib/explorer/chain/hash.ex

@@ -48,6 +48,9 @@ defmodule Explorer.Chain.Hash do
       %__MODULE__{byte_count: ^byte_count, bytes: <<_::big-integer-size(byte_count)-unit(@bits_per_byte)>>} = cast ->
         {:ok, cast}
 
+      <<_::big-integer-size(byte_count)-unit(@bits_per_byte)>> ->
+        {:ok, %__MODULE__{byte_count: byte_count, bytes: term}}
+
       <<"0x", hexadecimal_digits::binary>> ->
         cast_hexadecimal_digits(hexadecimal_digits, byte_count)
 

apps/explorer/lib/explorer/exchange_rates/exchange_rates.ex

@@ -26,12 +26,8 @@ defmodule Explorer.ExchangeRates do
   # Callback for successful fetch
   @impl GenServer
   def handle_info({_ref, {:ok, tokens}}, state) do
-    records =
-      for %Token{symbol: symbol} = token <- tokens do
-        {symbol, token}
-      end
-
     if store() == :ets do
+      records = Enum.map(tokens, &Token.to_tuple/1)
       :ets.insert(table_name(), records)
     end
 
@@ -95,7 +91,7 @@ defmodule Explorer.ExchangeRates do
   def lookup(symbol) do
     if store() == :ets do
       case :ets.lookup(table_name(), symbol) do
-        [{_key, token} | _] -> token
+        [tuple | _] when is_tuple(tuple) -> Token.from_tuple(tuple)
         _ -> nil
       end
     end
@@ -134,7 +130,7 @@ defmodule Explorer.ExchangeRates do
   defp list_from_store(:ets) do
     table_name()
     |> :ets.tab2list()
-    |> Enum.map(fn {_, rate} -> rate end)
+    |> Enum.map(&Token.from_tuple/1)
     |> Enum.sort_by(fn %Token{symbol: symbol} -> symbol end)
   end
 

apps/explorer/lib/explorer/exchange_rates/token.ex

@@ -28,7 +28,50 @@ defmodule Explorer.ExchangeRates.Token do
           volume_24h_usd: Decimal.t()
         }
 
+  @enforce_keys ~w(available_supply btc_value id last_updated market_cap_usd name symbol usd_value volume_24h_usd)a
   defstruct ~w(available_supply btc_value id last_updated market_cap_usd name symbol usd_value volume_24h_usd)a
 
-  def null, do: %__MODULE__{}
+  def null,
+    do: %__MODULE__{
+      symbol: nil,
+      id: nil,
+      name: nil,
+      available_supply: nil,
+      usd_value: nil,
+      volume_24h_usd: nil,
+      market_cap_usd: nil,
+      btc_value: nil,
+      last_updated: nil
+    }
+
+  def to_tuple(%__MODULE__{
+        symbol: symbol,
+        id: id,
+        name: name,
+        available_supply: available_supply,
+        usd_value: usd_value,
+        volume_24h_usd: volume_24h_usd,
+        market_cap_usd: market_cap_usd,
+        btc_value: btc_value,
+        last_updated: last_updated
+      }) do
+    # symbol is first because it is the key used for lookup in `Explorer.ExchangeRates`'s ETS table
+    {symbol, id, name, available_supply, usd_value, volume_24h_usd, market_cap_usd, btc_value, last_updated}
+  end
+
+  def from_tuple(
+        {symbol, id, name, available_supply, usd_value, volume_24h_usd, market_cap_usd, btc_value, last_updated}
+      ) do
+    %__MODULE__{
+      symbol: symbol,
+      id: id,
+      name: name,
+      available_supply: available_supply,
+      usd_value: usd_value,
+      volume_24h_usd: volume_24h_usd,
+      market_cap_usd: market_cap_usd,
+      btc_value: btc_value,
+      last_updated: last_updated
+    }
+  end
 end

apps/explorer/test/explorer/chain/hash/full_test.exs

@@ -1,5 +1,17 @@
 defmodule Explorer.Chain.Hash.FullTest do
   use ExUnit.Case, async: true
 
-  doctest Explorer.Chain.Hash.Full
+  alias Explorer.Chain.Hash
+
+  doctest Hash.Full
+
+  describe "cast" do
+    test ~S|is not confused by big integer that starts with <<48, 120>> which is "0x"| do
+      assert {:ok, _} =
+               Hash.Full.cast(
+                 <<48, 120, 238, 242, 122, 170, 157, 194, 106, 180, 42, 65, 178, 64, 202, 214, 148, 99, 171, 74, 64, 18,
+                   14, 163, 47, 7, 39, 180, 235, 9, 98, 158>>
+               )
+    end
+  end
 end

apps/explorer/test/explorer/exchange_rates/exchange_rates_test.exs

@@ -22,7 +22,7 @@ defmodule Explorer.ExchangeRatesTest do
   end
 
   test "start_link" do
-    stub(TestSource, :fetch_exchange_rates, fn -> {:ok, [%Token{}]} end)
+    stub(TestSource, :fetch_exchange_rates, fn -> {:ok, [Token.null()]} end)
     set_mox_global()
 
     assert {:ok, _} = ExchangeRates.start_link([])
@@ -46,7 +46,7 @@ defmodule Explorer.ExchangeRatesTest do
     ExchangeRates.init([])
     state = %{}
 
-    expect(TestSource, :fetch_exchange_rates, fn -> {:ok, [%Token{}]} end)
+    expect(TestSource, :fetch_exchange_rates, fn -> {:ok, [Token.null()]} end)
     set_mox_global()
 
     assert {:noreply, ^state} = ExchangeRates.handle_info(:update, state)
@@ -63,28 +63,29 @@ defmodule Explorer.ExchangeRatesTest do
       expected_token = %Token{
         available_supply: Decimal.new("1000000.0"),
         btc_value: Decimal.new("1.000"),
-        id: "test",
+        id: "test_id",
         last_updated: DateTime.utc_now(),
         market_cap_usd: Decimal.new("1000000.0"),
-        name: "test",
-        symbol: "test",
+        name: "test_name",
+        symbol: "test_symbol",
         usd_value: Decimal.new("1.0"),
         volume_24h_usd: Decimal.new("1000.0")
       }
 
-      expected_id = expected_token.id
+      expected_symbol = expected_token.symbol
+      expected_tuple = Token.to_tuple(expected_token)
 
       state = %{}
 
       assert {:noreply, ^state} = ExchangeRates.handle_info({nil, {:ok, [expected_token]}}, state)
 
-      assert [{^expected_id, ^expected_token}] = :ets.lookup(ExchangeRates.table_name(), expected_id)
+      assert [^expected_tuple] = :ets.lookup(ExchangeRates.table_name(), expected_symbol)
     end
 
     test "with failed fetch" do
       state = %{}
 
-      expect(TestSource, :fetch_exchange_rates, fn -> {:ok, [%Token{}]} end)
+      expect(TestSource, :fetch_exchange_rates, fn -> {:ok, [Token.null()]} end)
       set_mox_global()
 
       assert {:noreply, ^state} = ExchangeRates.handle_info({nil, {:error, "some error"}}, state)
@@ -97,12 +98,12 @@ defmodule Explorer.ExchangeRatesTest do
     ExchangeRates.init([])
 
     rates = [
-      %Token{symbol: "z"},
-      %Token{symbol: "a"}
+      %Token{Token.null() | symbol: "z"},
+      %Token{Token.null() | symbol: "a"}
     ]
 
     expected_rates = Enum.reverse(rates)
-    for rate <- rates, do: :ets.insert(ExchangeRates.table_name(), {rate.symbol, rate})
+    for rate <- rates, do: :ets.insert(ExchangeRates.table_name(), Token.to_tuple(rate))
 
     assert expected_rates == ExchangeRates.list()
   end
@@ -110,11 +111,11 @@ defmodule Explorer.ExchangeRatesTest do
   test "lookup/1" do
     ExchangeRates.init([])
 
-    z = %Token{symbol: "z"}
+    z = %Token{Token.null() | symbol: "z"}
 
-    rates = [z, %Token{symbol: "a"}]
+    rates = [z, %Token{Token.null() | symbol: "a"}]
 
-    for rate <- rates, do: :ets.insert(ExchangeRates.table_name(), {rate.symbol, rate})
+    for rate <- rates, do: :ets.insert(ExchangeRates.table_name(), Token.to_tuple(rate))
 
     assert z == ExchangeRates.lookup("z")
     assert nil == ExchangeRates.lookup("nope")

apps/indexer/lib/indexer/block/fetcher.ex

@@ -5,7 +5,7 @@ defmodule Indexer.Block.Fetcher do
 
   require Logger
 
-  alias Explorer.Chain.{Block, Import}
+  alias Explorer.Chain.{Address, Block, Import}
   alias Indexer.{CoinBalance, AddressExtraction, Token, TokenTransfers}
   alias Indexer.Address.{CoinBalances, TokenBalances}
   alias Indexer.Block.Fetcher.Receipts
@@ -164,7 +164,7 @@ defmodule Indexer.Block.Fetcher do
         address_hash_to_fetched_balance_block_number: address_hash_to_block_number
       }) do
     addresses
-    |> Enum.map(fn address_hash ->
+    |> Enum.map(fn %Address{hash: address_hash} ->
       block_number = Map.fetch!(address_hash_to_block_number, to_string(address_hash))
       %{address_hash: address_hash, block_number: block_number}
     end)

apps/indexer/lib/indexer/coin_balance/fetcher.ex

@@ -29,9 +29,9 @@ defmodule Indexer.CoinBalance.Fetcher do
           %{required(:address_hash) => Hash.Address.t(), required(:block_number) => Block.block_number()}
         ]) :: :ok
   def async_fetch_balances(balance_fields) when is_list(balance_fields) do
-    params_list = Enum.map(balance_fields, &balance_fields_to_params/1)
+    entries = Enum.map(balance_fields, &entry/1)
 
-    BufferedTask.buffer(__MODULE__, params_list)
+    BufferedTask.buffer(__MODULE__, entries)
   end
 
   @doc false
@@ -57,7 +57,7 @@ defmodule Indexer.CoinBalance.Fetcher do
     {:ok, final} =
       Chain.stream_unfetched_balances(initial, fn address_fields, acc ->
         address_fields
-        |> balance_fields_to_params()
+        |> entry()
         |> reducer.(acc)
       end)
 
@@ -65,14 +65,17 @@ defmodule Indexer.CoinBalance.Fetcher do
   end
 
   @impl BufferedTask
-  def run(params_list, _retries, json_rpc_named_arguments) do
+  def run(entries, _retries, json_rpc_named_arguments) do
     # the same address may be used more than once in the same block, but we only want one `Balance` for a given
     # `{address, block}`, so take unique params only
-    unique_params_list = Enum.uniq(params_list)
+    unique_entries = Enum.uniq(entries)
 
-    Logger.debug(fn -> "fetching #{length(unique_params_list)} balances" end)
+    Logger.debug(fn -> "fetching #{length(unique_entries)} balances" end)
 
-    case EthereumJSONRPC.fetch_balances(unique_params_list, json_rpc_named_arguments) do
+    unique_entries
+    |> Enum.map(&entry_to_params/1)
+    |> EthereumJSONRPC.fetch_balances(json_rpc_named_arguments)
+    |> case do
       {:ok, balances_params} ->
         value_fetched_at = DateTime.utc_now()
 
@@ -89,16 +92,20 @@ defmodule Indexer.CoinBalance.Fetcher do
         :ok
 
       {:error, reason} ->
-        Logger.debug(fn -> "failed to fetch #{length(unique_params_list)} balances, #{inspect(reason)}" end)
-        {:retry, unique_params_list}
+        Logger.debug(fn -> "failed to fetch #{length(unique_entries)} balances, #{inspect(reason)}" end)
+        {:retry, unique_entries}
     end
   end
 
-  defp balance_fields_to_params(%{address_hash: address_hash, block_number: block_number})
-       when is_integer(block_number) do
+  defp entry_to_params({address_hash_bytes, block_number}) when is_integer(block_number) do
+    {:ok, address_hash} = Hash.Address.cast(address_hash_bytes)
     %{block_quantity: integer_to_quantity(block_number), hash_data: to_string(address_hash)}
   end
 
+  defp entry(%{address_hash: %Hash{bytes: address_hash_bytes}, block_number: block_number}) do
+    {address_hash_bytes, block_number}
+  end
+
   # We want to record all historical balances for an address, but have the address itself have balance from the
   # `Balance` with the greatest block_number for that address.
   def balances_params_to_address_params(balances_params) do

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

@@ -43,9 +43,9 @@ defmodule Indexer.InternalTransaction.Fetcher do
   """
   @spec async_fetch([%{required(:block_number) => Block.block_number(), required(:hash) => Hash.Full.t()}]) :: :ok
   def async_fetch(transactions_fields, timeout \\ 5000) when is_list(transactions_fields) do
-    params_list = Enum.map(transactions_fields, &transaction_fields_to_params/1)
+    entries = Enum.map(transactions_fields, &entry/1)
 
-    BufferedTask.buffer(__MODULE__, params_list, timeout)
+    BufferedTask.buffer(__MODULE__, entries, timeout)
   end
 
   @doc false
@@ -74,7 +74,7 @@ defmodule Indexer.InternalTransaction.Fetcher do
         initial,
         fn transaction_fields, acc ->
           transaction_fields
-          |> transaction_fields_to_params()
+          |> entry()
           |> reducer.(acc)
         end
       )
@@ -82,17 +82,25 @@ defmodule Indexer.InternalTransaction.Fetcher do
     final
   end
 
-  defp transaction_fields_to_params(%{block_number: block_number, hash: hash}) when is_integer(block_number) do
+  defp entry(%{block_number: block_number, hash: %Hash{bytes: bytes}}) when is_integer(block_number) do
+    {block_number, bytes}
+  end
+
+  defp params({block_number, hash_bytes}) when is_integer(block_number) do
+    {:ok, hash} = Hash.Full.cast(hash_bytes)
     %{block_number: block_number, hash_data: to_string(hash)}
   end
 
   @impl BufferedTask
-  def run(transactions_params, _retries, json_rpc_named_arguments) do
-    unique_transactions_params = unique_transactions_params(transactions_params)
+  def run(entries, _retries, json_rpc_named_arguments) do
+    unique_entries = unique_entries(entries)
 
-    Logger.debug(fn -> "fetching internal transactions for #{length(unique_transactions_params)} transactions" end)
+    Logger.debug(fn -> "fetching internal transactions for #{length(unique_entries)} transactions" end)
 
-    case EthereumJSONRPC.fetch_internal_transactions(unique_transactions_params, json_rpc_named_arguments) do
+    unique_entries
+    |> Enum.map(&params/1)
+    |> EthereumJSONRPC.fetch_internal_transactions(json_rpc_named_arguments)
+    |> case do
       {:ok, internal_transactions_params} ->
         addresses_params = AddressExtraction.extract_addresses(%{internal_transactions: internal_transactions_params})
 
@@ -115,7 +123,7 @@ defmodule Indexer.InternalTransaction.Fetcher do
             Logger.error(fn ->
               [
                 "failed to import internal transactions for ",
-                to_string(length(transactions_params)),
+                to_string(length(entries)),
                 " transactions at ",
                 to_string(step),
                 ": ",
@@ -123,17 +131,17 @@ defmodule Indexer.InternalTransaction.Fetcher do
               ]
             end)
 
-            # re-queue the de-duped transactions_params
-            {:retry, unique_transactions_params}
+            # re-queue the de-duped entries
+            {:retry, unique_entries}
         end
 
       {:error, reason} ->
         Logger.error(fn ->
-          "failed to fetch internal transactions for #{length(transactions_params)} transactions: #{inspect(reason)}"
+          "failed to fetch internal transactions for #{length(entries)} transactions: #{inspect(reason)}"
         end)
 
-        # re-queue the de-duped transactions_params
-        {:retry, unique_transactions_params}
+        # re-queue the de-duped entries
+        {:retry, unique_entries}
 
       :ignore ->
         :ok
@@ -141,29 +149,29 @@ defmodule Indexer.InternalTransaction.Fetcher do
   end
 
   # Protection and improved reporting for https://github.com/poanetwork/blockscout/issues/289
-  defp unique_transactions_params(transactions_params) do
-    transactions_params_by_hash_data = Enum.group_by(transactions_params, fn %{hash_data: hash_data} -> hash_data end)
+  defp unique_entries(entries) do
+    entries_by_hash_bytes = Enum.group_by(entries, &elem(&1, 1))
 
-    if map_size(transactions_params_by_hash_data) < length(transactions_params) do
-      {unique_transactions_params, duplicate_transactions_params} =
-        transactions_params_by_hash_data
+    if map_size(entries_by_hash_bytes) < length(entries) do
+      {unique_entries, duplicate_entries} =
+        entries_by_hash_bytes
         |> Map.values()
         |> uniques_and_duplicates()
 
       Logger.error(fn ->
-        duplicate_transactions_params
+        duplicate_entries
         |> Stream.with_index()
         |> Enum.reduce(
-          ["Duplicate transaction params being used to fetch internal transactions:\n"],
-          fn {transaction_params, index}, acc ->
-            [acc, "  ", to_string(index + 1), ". ", inspect(transaction_params), "\n"]
+          ["Duplicate entries being used to fetch internal transactions:\n"],
+          fn {entry, index}, acc ->
+            [acc, "  ", to_string(index + 1), ". ", inspect(entry), "\n"]
           end
         )
       end)
 
-      unique_transactions_params
+      unique_entries
     else
-      transactions_params
+      entries
     end
   end
 

apps/indexer/lib/indexer/sequence.ex

@@ -182,8 +182,8 @@ defmodule Indexer.Sequence do
   def handle_call(:pop, _from, %__MODULE__{queue: queue, current: current, step: step} = state) do
     {reply, new_state} =
       case {current, :queue.out(queue)} do
-        {_, {{:value, range}, new_queue}} ->
-          {range, %__MODULE__{state | queue: new_queue}}
+        {_, {{:value, {first, last}}, new_queue}} ->
+          {first..last, %__MODULE__{state | queue: new_queue}}
 
         {nil, {:empty, new_queue}} ->
           {:halt, %__MODULE__{state | queue: new_queue}}
@@ -251,8 +251,8 @@ defmodule Indexer.Sequence do
     {:error, "Range (#{inspect(range)}) direction is opposite step (#{step}) direction"}
   end
 
-  defp reduce_chunked_range(_.._ = range, count, step, initial, reducer) when count <= abs(step) do
-    {:ok, reducer.(range, initial)}
+  defp reduce_chunked_range(first..last, count, step, initial, reducer) when count <= abs(step) do
+    {:ok, reducer.({first, last}, initial)}
   end
 
   defp reduce_chunked_range(first..last, _, step, initial, reducer) do
@@ -277,7 +277,7 @@ defmodule Indexer.Sequence do
             {:cont, full_chunk_last}
           end
 
-        {action, reducer.(chunk_first..chunk_last, acc)}
+        {action, reducer.({chunk_first, chunk_last}, acc)}
       end)
 
     {:ok, final}

apps/indexer/lib/indexer/token_balance/fetcher.ex

@@ -20,8 +20,9 @@ defmodule Indexer.TokenBalance.Fetcher do
   ]
 
   @spec async_fetch([%TokenBalance{}]) :: :ok
-  def async_fetch(token_balances_params) do
-    BufferedTask.buffer(__MODULE__, token_balances_params, :infinity)
+  def async_fetch(token_balances) do
+    formatted_params = Enum.map(token_balances, &entry/1)
+    BufferedTask.buffer(__MODULE__, formatted_params, :infinity)
   end
 
   @doc false
@@ -45,34 +46,34 @@ defmodule Indexer.TokenBalance.Fetcher do
   @impl BufferedTask
   def init(initial, reducer, _) do
     {:ok, final} =
-      Chain.stream_unfetched_token_balances(initial, fn token_balances_params, acc ->
-        reducer.(token_balances_params, acc)
+      Chain.stream_unfetched_token_balances(initial, fn token_balance, acc ->
+        token_balance
+        |> entry()
+        |> reducer.(acc)
       end)
 
     final
   end
 
   @impl BufferedTask
-  def run(token_balances, _retries, _json_rpc_named_arguments) do
-    Logger.debug(fn -> "fetching #{length(token_balances)} token balances" end)
+  def run(entries, _retries, _json_rpc_named_arguments) do
+    Logger.debug(fn -> "fetching #{length(entries)} token balances" end)
 
     result =
-      token_balances
-      |> fetch_from_blockchain
-      |> import_token_balances
+      entries
+      |> Enum.map(&format_params/1)
+      |> fetch_from_blockchain()
+      |> import_token_balances()
 
     if result == :ok do
       :ok
     else
-      {:retry, token_balances}
+      {:retry, entries}
     end
   end
 
-  def fetch_from_blockchain(token_balances) do
-    {:ok, token_balances} =
-      token_balances
-      |> Stream.map(&format_params/1)
-      |> TokenBalances.fetch_token_balances_from_blockchain()
+  def fetch_from_blockchain(params_list) do
+    {:ok, token_balances} = TokenBalances.fetch_token_balances_from_blockchain(params_list)
 
     TokenBalances.log_fetching_errors(__MODULE__, token_balances)
 
@@ -91,14 +92,21 @@ defmodule Indexer.TokenBalance.Fetcher do
     end
   end
 
-  defp format_params(%TokenBalance{
+  defp entry(%TokenBalance{
          token_contract_address_hash: token_contract_address_hash,
          address_hash: address_hash,
          block_number: block_number
        }) do
+    {address_hash.bytes, token_contract_address_hash.bytes, block_number}
+  end
+
+  defp format_params({address_hash_bytes, token_contract_address_hash_bytes, block_number}) do
+    {:ok, token_contract_address_hash} = Hash.Address.cast(token_contract_address_hash_bytes)
+    {:ok, address_hash} = Hash.Address.cast(address_hash_bytes)
+
     %{
-      token_contract_address_hash: Hash.to_string(token_contract_address_hash),
-      address_hash: Hash.to_string(address_hash),
+      token_contract_address_hash: to_string(token_contract_address_hash),
+      address_hash: to_string(address_hash),
       block_number: block_number
     }
   end

apps/indexer/test/indexer/coin_balance/fetcher_test.exs

@@ -251,9 +251,10 @@ defmodule Indexer.CoinBalance.FetcherTest do
         end)
       end
 
-      params_list = Enum.map(block_quantities, &%{block_quantity: &1, hash_data: hash_data})
+      {:ok, %Hash{bytes: address_hash_bytes}} = Hash.Address.cast(hash_data)
+      entries = Enum.map(block_quantities, &{address_hash_bytes, quantity_to_integer(&1)})
 
-      case CoinBalance.Fetcher.run(params_list, 0, json_rpc_named_arguments) do
+      case CoinBalance.Fetcher.run(entries, 0, json_rpc_named_arguments) do
         :ok ->
           balances = Repo.all(from(balance in Address.CoinBalance, where: balance.address_hash == ^hash_data))
 
@@ -283,33 +284,9 @@ defmodule Indexer.CoinBalance.FetcherTest do
         other ->
           # not all nodes behind the `https://mainnet.infura.io` pool are fully-synced.  Node that aren't fully-synced
           # won't have historical address balances.
-          assert {:retry, ^params_list} = other
+          assert {:retry, ^entries} = other
       end
     end
-
-    test "duplicate params retry unique params", %{json_rpc_named_arguments: json_rpc_named_arguments} do
-      hash_data = "0x000000000000000000000000000000000"
-
-      if json_rpc_named_arguments[:transport] == EthereumJSONRPC.Mox do
-        EthereumJSONRPC.Mox
-        |> expect(:json_rpc, fn [%{id: id, method: "eth_getBalance", params: [^hash_data, "0x1"]}], _options ->
-          {:ok, [%{id: id, error: %{code: 404, message: "Not Found"}}]}
-        end)
-      end
-
-      assert CoinBalance.Fetcher.run(
-               [%{block_quantity: "0x1", hash_data: hash_data}, %{block_quantity: "0x1", hash_data: hash_data}],
-               0,
-               json_rpc_named_arguments
-             ) ==
-               {:retry,
-                [
-                  %{
-                    block_quantity: "0x1",
-                    hash_data: "0x000000000000000000000000000000000"
-                  }
-                ]}
-    end
   end
 
   defp wait(producer) do

apps/indexer/test/indexer/internal_transaction/fetcher_test.exs

@@ -5,6 +5,8 @@ defmodule Indexer.InternalTransaction.FetcherTest do
   import ExUnit.CaptureLog
   import Mox
 
+  alias Explorer.Chain.{Hash, Transaction}
+
   alias Indexer.{CoinBalance, InternalTransaction, PendingTransaction}
 
   # MUST use global mode because we aren't guaranteed to get PendingTransactionFetcher's pid back fast enough to `allow`
@@ -99,9 +101,7 @@ defmodule Indexer.InternalTransaction.FetcherTest do
                [],
                fn hash_string, acc -> [hash_string | acc] end,
                json_rpc_named_arguments
-             ) == [
-               %{block_number: block.number, hash_data: to_string(collated_unfetched_transaction.hash)}
-             ]
+             ) == [{block.number, collated_unfetched_transaction.hash.bytes}]
     end
 
     test "does not buffer collated transactions with fetched internal transactions", %{
@@ -129,14 +129,15 @@ defmodule Indexer.InternalTransaction.FetcherTest do
 
       CoinBalance.Supervisor.Case.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
 
+      %Transaction{hash: %Hash{bytes: bytes}} =
         insert(:transaction, hash: "0x03cd5899a63b6f6222afda8705d059fd5a7d126bcabe962fb654d9736e6bcafa")
 
       log =
         capture_log(fn ->
           InternalTransaction.Fetcher.run(
             [
-              %{block_number: 1, hash_data: "0x03cd5899a63b6f6222afda8705d059fd5a7d126bcabe962fb654d9736e6bcafa"},
-              %{block_number: 1, hash_data: "0x03cd5899a63b6f6222afda8705d059fd5a7d126bcabe962fb654d9736e6bcafa"}
+              {1, bytes},
+              {1, bytes}
             ],
             0,
             json_rpc_named_arguments
@@ -145,9 +146,9 @@ defmodule Indexer.InternalTransaction.FetcherTest do
 
       assert log =~
                """
-               Duplicate transaction params being used to fetch internal transactions:
-                 1. %{block_number: 1, hash_data: \"0x03cd5899a63b6f6222afda8705d059fd5a7d126bcabe962fb654d9736e6bcafa\"}
-                 2. %{block_number: 1, hash_data: \"0x03cd5899a63b6f6222afda8705d059fd5a7d126bcabe962fb654d9736e6bcafa\"}
+               Duplicate entries being used to fetch internal transactions:
+                 1. {1, <<3, 205, 88, 153, 166, 59, 111, 98, 34, 175, 218, 135, 5, 208, 89, 253, 90, 125, 18, 107, 202, 190, 150, 47, 182, 84, 217, 115, 110, 107, 202, 250>>}
+                 2. {1, <<3, 205, 88, 153, 166, 59, 111, 98, 34, 175, 218, 135, 5, 208, 89, 253, 90, 125, 18, 107, 202, 190, 150, 47, 182, 84, 217, 115, 110, 107, 202, 250>>}
                """
     end
 
@@ -160,19 +161,18 @@ defmodule Indexer.InternalTransaction.FetcherTest do
 
       CoinBalance.Supervisor.Case.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
 
-      # not a real transaction hash, so that it fails
+      # not a real transaction hash, so that fetch fails
+      %Transaction{hash: %Hash{bytes: bytes}} =
         insert(:transaction, hash: "0x0000000000000000000000000000000000000000000000000000000000000001")
 
       assert InternalTransaction.Fetcher.run(
                [
-                 %{block_number: 1, hash_data: "0x0000000000000000000000000000000000000000000000000000000000000001"},
-                 %{block_number: 1, hash_data: "0x0000000000000000000000000000000000000000000000000000000000000001"}
+                 {1, bytes},
+                 {1, bytes}
                ],
                0,
                json_rpc_named_arguments
-             ) ==
-               {:retry,
-                [%{block_number: 1, hash_data: "0x0000000000000000000000000000000000000000000000000000000000000001"}]}
+             ) == {:retry, [{1, bytes}]}
     end
   end
 end

apps/indexer/test/indexer/token_balance/fetcher_test.exs

@@ -4,26 +4,37 @@ defmodule Indexer.TokenBalance.FetcherTest do
 
   import Mox
 
-  alias Explorer.Chain.Address
+  alias Explorer.Chain.{Address, Hash}
   alias Indexer.TokenBalance
 
+  @moduletag :capture_log
+
   setup :verify_on_exit!
   setup :set_mox_global
 
   describe "init/3" do
     test "returns unfetched token balances" do
-      %Address.TokenBalance{address_hash: address_hash} =
-        insert(:token_balance, block_number: 1_000, value_fetched_at: nil)
+      %Address.TokenBalance{
+        address_hash: %Hash{bytes: address_hash_bytes},
+        token_contract_address_hash: %Hash{bytes: token_contract_address_hash_bytes},
+        block_number: block_number
+      } = insert(:token_balance, block_number: 1_000, value_fetched_at: nil)
 
       insert(:token_balance, value_fetched_at: DateTime.utc_now())
 
-      assert TokenBalance.Fetcher.init([], &[&1.address_hash | &2], nil) == [address_hash]
+      assert TokenBalance.Fetcher.init([], &[&1 | &2], nil) == [
+               {address_hash_bytes, token_contract_address_hash_bytes, block_number}
+             ]
     end
   end
 
   describe "run/3" do
     test "imports the given token balances" do
-      token_balance = insert(:token_balance, value_fetched_at: nil, value: nil)
+      %Address.TokenBalance{
+        address_hash: %Hash{bytes: address_hash_bytes} = address_hash,
+        token_contract_address_hash: %Hash{bytes: token_contract_address_hash_bytes},
+        block_number: block_number
+      } = insert(:token_balance, value_fetched_at: nil, value: nil)
 
       expect(
         EthereumJSONRPC.Mox,
@@ -40,11 +51,10 @@ defmodule Indexer.TokenBalance.FetcherTest do
         end
       )
 
-      assert TokenBalance.Fetcher.run([token_balance], 0, nil) == :ok
+      assert TokenBalance.Fetcher.run([{address_hash_bytes, token_contract_address_hash_bytes, block_number}], 0, nil) ==
+               :ok
 
-      token_balance_updated =
-        Address.TokenBalance
-        |> Explorer.Repo.get_by(address_hash: token_balance.address_hash)
+      token_balance_updated = Explorer.Repo.get_by(Address.TokenBalance, address_hash: address_hash)
 
       assert token_balance_updated.value == Decimal.new(1_000_000_000_000_000_000_000_000)
       assert token_balance_updated.value_fetched_at != nil