Merge pull request #1269 from poanetwork/gsf-block-rewards-for-older-blocks

get rewards of blocks already indexed without rewards

apps/explorer/lib/explorer/chain.ex

@@ -936,6 +936,24 @@ defmodule Explorer.Chain do
     Repo.all(query)
   end
 
+  @doc """
+  Finds blocks without a reward associated, up to the specified limit
+  """
+  def get_blocks_without_reward(limit \\ 250) do
+    Block.get_blocks_without_reward()
+    |> limit(^limit)
+    |> Repo.all()
+  end
+
+  @doc """
+  Finds all transactions of a certain block number
+  """
+  def get_transactions_of_block_number(block_number) do
+    block_number
+    |> Transaction.transactions_with_block_number()
+    |> Repo.all()
+  end
+
   @doc """
   Finds all Blocks validated by the address given.
 

apps/explorer/lib/explorer/chain/block.ex

@@ -100,6 +100,15 @@ defmodule Explorer.Chain.Block do
     |> unique_constraint(:hash, name: :blocks_pkey)
   end
 
+  def get_blocks_without_reward(query \\ __MODULE__) do
+    from(
+      b in query,
+      left_join: r in Reward,
+      on: [block_hash: b.hash],
+      where: is_nil(r.block_hash)
+    )
+  end
+
   @doc """
   Adds to the given block's query a `where` with conditions to filter by the type of block;
   `Uncle`, `Reorg`, or `Block`.

apps/explorer/lib/explorer/chain/transaction.ex

@@ -596,6 +596,16 @@ defmodule Explorer.Chain.Transaction do
     )
   end
 
+  @doc """
+  Builds an `Ecto.Query` to fetch transactions with the specified block_number
+  """
+  def transactions_with_block_number(block_number) do
+    from(
+      t in Transaction,
+      where: t.block_number == ^block_number
+    )
+  end
+
   @doc """
   Builds an `Ecto.Query` to fetch the last nonce from the given address hash.
 

apps/explorer/lib/explorer/chain/wei.ex

@@ -146,6 +146,22 @@ defmodule Explorer.Chain.Wei do
     |> from(:wei)
   end
 
+  @doc """
+  Multiplies two Wei values.
+
+  ## Example
+
+      iex> first = %Explorer.Chain.Wei{value: Decimal.new(10)}
+      iex> second = %Explorer.Chain.Wei{value: Decimal.new(5)}
+      iex> Explorer.Chain.Wei.mult(first, second)
+      %Explorer.Chain.Wei{value: Decimal.new(50)}
+  """
+  def mult(%Wei{value: wei_1}, %Wei{value: wei_2}) do
+    wei_1
+    |> Decimal.mult(wei_2)
+    |> from(:wei)
+  end
+
   @doc """
   Converts `Decimal` representations of various wei denominations (wei, Gwei, ether) to
   a wei base unit.

apps/explorer/test/explorer/chain/block_test.exs

@@ -42,4 +42,20 @@ defmodule Explorer.Chain.BlockTest do
                |> Repo.insert()
     end
   end
+
+  describe "get_blocks_without_reward/1" do
+    test "finds only blocks without rewards" do
+      rewarded_block = insert(:block)
+      insert(:reward, address_hash: insert(:address).hash, block_hash: rewarded_block.hash)
+      unrewarded_block = insert(:block)
+
+      results =
+        Block.get_blocks_without_reward()
+        |> Repo.all()
+        |> Enum.map(& &1.hash)
+
+      refute Enum.member?(results, rewarded_block.hash)
+      assert Enum.member?(results, unrewarded_block.hash)
+    end
+  end
 end

apps/explorer/test/explorer/chain/transaction_test.exs

@@ -178,6 +178,33 @@ defmodule Explorer.Chain.TransactionTest do
     end
   end
 
+  describe "transaction_hash_to_block_number/1" do
+    test "returns only transactions with the specified block number" do
+      target_block = insert(:block, number: 1_000_000)
+
+      :transaction
+      |> insert()
+      |> with_block(target_block)
+
+      :transaction
+      |> insert()
+      |> with_block(target_block)
+
+      :transaction
+      |> insert()
+      |> with_block(insert(:block, number: 1_001_101))
+
+      result =
+        1_000_000
+        |> Transaction.transactions_with_block_number()
+        |> Repo.all()
+        |> Enum.map(& &1.block_number)
+
+      refute Enum.any?(result, fn block_number -> 1_001_101 == block_number end)
+      assert Enum.all?(result, fn block_number -> 1_000_000 == block_number end)
+    end
+  end
+
   describe "last_nonce_by_address_query/1" do
     test "returns the nonce value from the last block" do
       address = insert(:address)

apps/explorer/test/explorer/chain/wei_test.exs

@@ -100,4 +100,34 @@ defmodule Explorer.Chain.WeiTest do
       assert Explorer.Chain.Wei.sub(first, second) == %Explorer.Chain.Wei{value: Decimal.new(-77)}
     end
   end
+
+  describe "mult/1" do
+    test "with one negative parameter return a negative value" do
+      first = %Explorer.Chain.Wei{value: Decimal.new(123)}
+      second = %Explorer.Chain.Wei{value: Decimal.new(-1)}
+
+      assert Explorer.Chain.Wei.mult(first, second) == %Explorer.Chain.Wei{value: Decimal.new(-123)}
+    end
+
+    test "with two negative parameter return positive number" do
+      first = %Explorer.Chain.Wei{value: Decimal.new(-123)}
+      second = %Explorer.Chain.Wei{value: Decimal.new(-100)}
+
+      assert Explorer.Chain.Wei.mult(first, second) == %Explorer.Chain.Wei{value: Decimal.new(12300)}
+    end
+
+    test "with two positive parameters return a positive number" do
+      first = %Explorer.Chain.Wei{value: Decimal.new(123)}
+      second = %Explorer.Chain.Wei{value: Decimal.new(100)}
+
+      assert Explorer.Chain.Wei.mult(first, second) == %Explorer.Chain.Wei{value: Decimal.new(12300)}
+    end
+
+    test "the order of the paramete matters not" do
+      first = %Explorer.Chain.Wei{value: Decimal.new(123)}
+      second = %Explorer.Chain.Wei{value: Decimal.new(-10)}
+
+      assert Explorer.Chain.Wei.mult(first, second) == Explorer.Chain.Wei.mult(second, first)
+    end
+  end
 end

apps/indexer/lib/indexer/block/supervisor.ex

@@ -4,7 +4,7 @@ defmodule Indexer.Block.Supervisor do
   """
 
   alias Indexer.Block
-  alias Indexer.Block.{Catchup, InvalidConsensus, Realtime, Uncle}
+  alias Indexer.Block.{Catchup, InvalidConsensus, Realtime, UncatalogedRewards, Uncle}
 
   use Supervisor
 
@@ -34,7 +34,8 @@ defmodule Indexer.Block.Supervisor do
            %{block_fetcher: block_fetcher, subscribe_named_arguments: subscribe_named_arguments},
            [name: Realtime.Supervisor]
          ]},
-        {Uncle.Supervisor, [[block_fetcher: block_fetcher, memory_monitor: memory_monitor], [name: Uncle.Supervisor]]}
+        {Uncle.Supervisor, [[block_fetcher: block_fetcher, memory_monitor: memory_monitor], [name: Uncle.Supervisor]]},
+        UncatalogedRewards.Processor
       ],
       strategy: :one_for_one
     )

apps/indexer/lib/indexer/block/uncataloged_rewards/importer.ex

@@ -0,0 +1,112 @@
+defmodule Indexer.Block.UncatalogedRewards.Importer do
+  @moduledoc """
+  a module to fetch and import the rewards for blocks that were indexed without the reward
+  """
+
+  alias Ecto.Multi
+  alias EthereumJSONRPC.FetchedBeneficiaries
+  alias Explorer.Chain
+  alias Explorer.Chain.{Block.Reward, Wei}
+
+  # max number of blocks in a single request
+  # higher numbers may cause the requests to time out
+  # lower numbers will generate more requests
+  @chunk_size 10
+
+  @doc """
+  receives a list of blocks and tries to fetch and insert rewards for them
+  """
+  def fetch_and_import_rewards(blocks_batch) do
+    result =
+      blocks_batch
+      |> break_into_chunks_of_block_numbers()
+      |> Enum.reduce([], fn chunk, acc ->
+        chunk
+        |> fetch_beneficiaries()
+        |> fetch_block_rewards()
+        |> insert_reward_group()
+        |> case do
+          :empty -> acc
+          insert -> [insert | acc]
+        end
+      end)
+
+    {:ok, result}
+  rescue
+    e in RuntimeError -> {:error, %{exception: e}}
+  end
+
+  defp fetch_beneficiaries(chunk) do
+    {chunk_start, chunk_end} = Enum.min_max(chunk)
+
+    {:ok, %FetchedBeneficiaries{params_set: result}} =
+      with :ignore <- EthereumJSONRPC.fetch_beneficiaries(chunk_start..chunk_end, json_rpc_named_arguments()) do
+        {:ok, %FetchedBeneficiaries{params_set: MapSet.new()}}
+      end
+
+    result
+  end
+
+  defp fetch_block_rewards(beneficiaries) do
+    Enum.map(beneficiaries, fn beneficiary ->
+      beneficiary_changes =
+        case beneficiary.address_type do
+          :validator ->
+            validation_reward = fetch_validation_reward(beneficiary)
+
+            {:ok, reward} = Wei.cast(beneficiary.reward)
+
+            %{beneficiary | reward: Wei.sum(reward, validation_reward)}
+
+          _ ->
+            beneficiary
+        end
+
+      Reward.changeset(%Reward{}, beneficiary_changes)
+    end)
+  end
+
+  defp fetch_validation_reward(beneficiary) do
+    {:ok, accumulator} = Wei.cast(0)
+
+    beneficiary.block_number
+    |> Chain.get_transactions_of_block_number()
+    |> Enum.reduce(accumulator, fn t, acc ->
+      {:ok, price_as_wei} = Wei.cast(t.gas_used)
+      price_as_wei |> Wei.mult(t.gas_price) |> Wei.sum(acc)
+    end)
+  end
+
+  defp break_into_chunks_of_block_numbers(blocks) do
+    Enum.chunk_while(
+      blocks,
+      [],
+      fn block, acc ->
+        if (acc == [] || hd(acc) + 1 == block.number) && length(acc) < @chunk_size do
+          {:cont, [block.number | acc]}
+        else
+          {:cont, acc, [block.number]}
+        end
+      end,
+      fn
+        [] -> {:cont, []}
+        acc -> {:cont, acc, []}
+      end
+    )
+  end
+
+  defp insert_reward_group([]), do: :empty
+
+  defp insert_reward_group(rewards) do
+    rewards
+    |> Enum.reduce({Multi.new(), 0}, fn changeset, {multi, index} ->
+      {Multi.insert(multi, "insert_#{index}", changeset), index + 1}
+    end)
+    |> elem(0)
+    |> Explorer.Repo.transaction()
+  end
+
+  defp json_rpc_named_arguments do
+    Application.get_env(:explorer, :json_rpc_named_arguments)
+  end
+end

apps/indexer/lib/indexer/block/uncataloged_rewards/processor.ex

@@ -0,0 +1,40 @@
+defmodule Indexer.Block.UncatalogedRewards.Processor do
+  @moduledoc """
+  genserver to find blocks without rewards and fetch their rewards in batches
+  """
+
+  use GenServer
+
+  alias Explorer.Chain
+  alias Indexer.Block.UncatalogedRewards.Importer
+
+  @max_batch_size 150
+  @default_cooldown 300
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, :ok, name: __MODULE__)
+  end
+
+  @impl true
+  def init(args) do
+    send(self(), :import_batch)
+    {:ok, args}
+  end
+
+  @impl true
+  def handle_info(:import_batch, state) do
+    @max_batch_size
+    |> Chain.get_blocks_without_reward()
+    |> import_or_try_later
+
+    {:noreply, state}
+  end
+
+  defp import_or_try_later(batch) do
+    import_results = Importer.fetch_and_import_rewards(batch)
+
+    wait_time = if import_results == {:ok, []}, do: :timer.hours(24), else: @default_cooldown
+
+    Process.send_after(self(), :import_batch, wait_time)
+  end
+end

apps/indexer/test/indexer/block/uncataloged_rewards/importer_test.exs

@@ -0,0 +1,64 @@
+defmodule Indexer.Block.UncatalogedRewards.ImporterTest do
+  use EthereumJSONRPC.Case, async: false
+  use Explorer.DataCase
+
+  import Mox
+  import EthereumJSONRPC, only: [integer_to_quantity: 1]
+  import EthereumJSONRPC.Case
+
+  alias Explorer.Chain
+  alias Indexer.Block.UncatalogedRewards.Importer
+
+  describe "fetch_and_import_rewards/1" do
+    test "return `{:ok, []}` when receiving an empty list" do
+      assert Importer.fetch_and_import_rewards([]) == {:ok, []}
+    end
+
+    @tag :no_geth
+    test "return `{:ok, [transactions executed]}`" do
+      address = insert(:address)
+      block = insert(:block, number: 1234, miner: address)
+
+      expect(EthereumJSONRPC.Mox, :json_rpc, fn [%{id: id, method: "trace_block", params: _params}], _options ->
+        {:ok,
+         [
+           %{
+             id: id,
+             result: [
+               %{
+                 "action" => %{
+                   "author" => to_string(address.hash),
+                   "rewardType" => "external",
+                   "value" => "0xde0b6b3a7640000"
+                 },
+                 "blockHash" => to_string(block.hash),
+                 "blockNumber" => 1234,
+                 "result" => nil,
+                 "subtraces" => 0,
+                 "traceAddress" => [],
+                 "transactionHash" => nil,
+                 "transactionPosition" => nil,
+                 "type" => "reward"
+               }
+             ]
+           }
+         ]}
+      end)
+
+      expected =
+        {:ok,
+         [
+           ok: %{
+             "insert_0" => %Explorer.Chain.Block.Reward{
+               address_hash: address.hash,
+               block_hash: block.hash,
+               address_type: :validator
+             }
+           }
+         ]}
+
+      result = Importer.fetch_and_import_rewards([block])
+      assert result = expected
+    end
+  end
+end