Add stability validators (#9390)
* Add stability validators * Process review comments * Fix testspull/9537/head
nikitosing
9 months ago
committed by
GitHub
parent
fdbb881a18
commit
546b732ac1
@ -0,0 +1,83 @@ |
||||
defmodule BlockScoutWeb.API.V2.ValidatorController do |
||||
use BlockScoutWeb, :controller |
||||
|
||||
alias Explorer.Chain.Cache.StabilityValidatorsCounters |
||||
alias Explorer.Chain.Stability.Validator, as: ValidatorStability |
||||
|
||||
import BlockScoutWeb.PagingHelper, |
||||
only: [ |
||||
delete_parameters_from_next_page_params: 1, |
||||
stability_validators_state_options: 1, |
||||
validators_stability_sorting: 1 |
||||
] |
||||
|
||||
import BlockScoutWeb.Chain, |
||||
only: [ |
||||
split_list_by_page: 1, |
||||
paging_options: 1, |
||||
next_page_params: 4 |
||||
] |
||||
|
||||
@api_true api?: true |
||||
|
||||
@doc """ |
||||
Function to handle GET requests to `/api/v2/validators/stability` endpoint. |
||||
""" |
||||
@spec stability_validators_list(Plug.Conn.t(), map()) :: Plug.Conn.t() |
||||
def stability_validators_list(conn, params) do |
||||
options = |
||||
[ |
||||
necessity_by_association: %{ |
||||
:address => :optional |
||||
} |
||||
] |
||||
|> Keyword.merge(@api_true) |
||||
|> Keyword.merge(paging_options(params)) |
||||
|> Keyword.merge(validators_stability_sorting(params)) |
||||
|> Keyword.merge(stability_validators_state_options(params)) |
||||
|
||||
{validators, next_page} = options |> ValidatorStability.get_paginated_validators() |> split_list_by_page() |
||||
|
||||
next_page_params = |
||||
next_page |
||||
|> next_page_params( |
||||
validators, |
||||
delete_parameters_from_next_page_params(params), |
||||
&ValidatorStability.next_page_params/1 |
||||
) |
||||
|
||||
conn |
||||
|> render(:stability_validators, %{validators: validators, next_page_params: next_page_params}) |
||||
end |
||||
|
||||
@doc """ |
||||
Function to handle GET requests to `/api/v2/validators/stability/counters` endpoint. |
||||
""" |
||||
@spec stability_validators_counters(Plug.Conn.t(), map()) :: Plug.Conn.t() |
||||
def stability_validators_counters(conn, _params) do |
||||
%{ |
||||
validators_counter: validators_counter, |
||||
new_validators_counter: new_validators_counter, |
||||
active_validators_counter: active_validators_counter |
||||
} = StabilityValidatorsCounters.get_counters(@api_true) |
||||
|
||||
conn |
||||
|> json(%{ |
||||
validators_counter: validators_counter, |
||||
new_validators_counter_24h: new_validators_counter, |
||||
active_validators_counter: active_validators_counter, |
||||
active_validators_percentage: |
||||
calculate_active_validators_percentage(active_validators_counter, validators_counter) |
||||
}) |
||||
end |
||||
|
||||
defp calculate_active_validators_percentage(active_validators_counter, validators_counter) do |
||||
if Decimal.compare(validators_counter, Decimal.new(0)) == :gt do |
||||
active_validators_counter |
||||
|> Decimal.div(validators_counter) |
||||
|> Decimal.mult(100) |
||||
|> Decimal.to_float() |
||||
|> Float.floor(2) |
||||
end |
||||
end |
||||
end |
||||
@ -0,0 +1,17 @@ |
||||
defmodule BlockScoutWeb.API.V2.ValidatorView do |
||||
use BlockScoutWeb, :view |
||||
|
||||
alias BlockScoutWeb.API.V2.Helper |
||||
|
||||
def render("stability_validators.json", %{validators: validators, next_page_params: next_page_params}) do |
||||
%{"items" => Enum.map(validators, &prepare_validator(&1)), "next_page_params" => next_page_params} |
||||
end |
||||
|
||||
defp prepare_validator(validator) do |
||||
%{ |
||||
"address" => Helper.address_with_info(nil, validator.address, validator.address_hash, true), |
||||
"state" => validator.state, |
||||
"blocks_validated_count" => validator.blocks_validated |
||||
} |
||||
end |
||||
end |
||||
@ -0,0 +1,205 @@ |
||||
defmodule BlockScoutWeb.API.V2.ValidatorControllerTest do |
||||
use BlockScoutWeb.ConnCase |
||||
|
||||
alias Explorer.Chain.Address |
||||
alias Explorer.Chain.Stability.Validator, as: ValidatorStability |
||||
alias Explorer.Chain.Cache.StabilityValidatorsCounters |
||||
|
||||
if Application.compile_env(:explorer, :chain_type) == "stability" do |
||||
describe "/validators/stability" do |
||||
test "get paginated list of the validators", %{conn: conn} do |
||||
validators = |
||||
insert_list(51, :validator_stability) |
||||
|> Enum.sort_by( |
||||
fn validator -> |
||||
{Keyword.fetch!(ValidatorStability.state_enum(), validator.state), validator.address_hash.bytes} |
||||
end, |
||||
:desc |
||||
) |
||||
|
||||
request = get(conn, "/api/v2/validators/stability") |
||||
assert response = json_response(request, 200) |
||||
|
||||
request_2nd_page = get(conn, "/api/v2/validators/stability", response["next_page_params"]) |
||||
assert response_2nd_page = json_response(request_2nd_page, 200) |
||||
|
||||
check_paginated_response(response, response_2nd_page, validators) |
||||
end |
||||
|
||||
test "sort by blocks_validated asc", %{conn: conn} do |
||||
validators = |
||||
for _ <- 0..50 do |
||||
validator = insert(:validator_stability) |
||||
blocks_count = Enum.random(0..50) |
||||
|
||||
_ = |
||||
for _ <- 0..blocks_count do |
||||
insert(:block, miner_hash: validator.address_hash, miner: nil) |
||||
end |
||||
|
||||
{validator, blocks_count} |
||||
end |
||||
|> Enum.sort(&compare_default_sorting_for_asc/2) |
||||
|
||||
init_params = %{"sort" => "blocks_validated", "order" => "asc"} |
||||
request = get(conn, "/api/v2/validators/stability", init_params) |
||||
assert response = json_response(request, 200) |
||||
|
||||
request_2nd_page = |
||||
get(conn, "/api/v2/validators/stability", Map.merge(init_params, response["next_page_params"])) |
||||
|
||||
assert response_2nd_page = json_response(request_2nd_page, 200) |
||||
|
||||
check_paginated_response(response, response_2nd_page, validators) |
||||
end |
||||
|
||||
test "sort by blocks_validated desc", %{conn: conn} do |
||||
validators = |
||||
for _ <- 0..50 do |
||||
validator = insert(:validator_stability) |
||||
blocks_count = Enum.random(0..50) |
||||
|
||||
_ = |
||||
for _ <- 0..blocks_count do |
||||
insert(:block, miner_hash: validator.address_hash, miner: nil) |
||||
end |
||||
|
||||
{validator, blocks_count} |
||||
end |
||||
|> Enum.sort(&compare_default_sorting_for_desc/2) |
||||
|
||||
init_params = %{"sort" => "blocks_validated", "order" => "desc"} |
||||
request = get(conn, "/api/v2/validators/stability", init_params) |
||||
assert response = json_response(request, 200) |
||||
|
||||
request_2nd_page = |
||||
get(conn, "/api/v2/validators/stability", Map.merge(init_params, response["next_page_params"])) |
||||
|
||||
assert response_2nd_page = json_response(request_2nd_page, 200) |
||||
|
||||
check_paginated_response(response, response_2nd_page, validators) |
||||
end |
||||
|
||||
test "state_filter=probation", %{conn: conn} do |
||||
insert_list(51, :validator_stability, state: Enum.random([:active, :inactive])) |
||||
|
||||
validators = |
||||
insert_list(51, :validator_stability, state: :probation) |
||||
|> Enum.sort_by( |
||||
fn validator -> |
||||
{Keyword.fetch!(ValidatorStability.state_enum(), validator.state), validator.address_hash.bytes} |
||||
end, |
||||
:desc |
||||
) |
||||
|
||||
init_params = %{"state_filter" => "probation"} |
||||
|
||||
request = get(conn, "/api/v2/validators/stability", init_params) |
||||
assert response = json_response(request, 200) |
||||
|
||||
request_2nd_page = |
||||
get(conn, "/api/v2/validators/stability", Map.merge(init_params, response["next_page_params"])) |
||||
|
||||
assert response_2nd_page = json_response(request_2nd_page, 200) |
||||
|
||||
check_paginated_response(response, response_2nd_page, validators) |
||||
end |
||||
end |
||||
|
||||
describe "/validators/stability/counters" do |
||||
test "get counters", %{conn: conn} do |
||||
_validator_active1 = |
||||
insert(:validator_stability, state: :active, inserted_at: DateTime.add(DateTime.utc_now(), -2, :day)) |
||||
|
||||
_validator_active2 = insert(:validator_stability, state: :active) |
||||
_validator_active3 = insert(:validator_stability, state: :active) |
||||
|
||||
_validator_inactive1 = |
||||
insert(:validator_stability, state: :inactive, inserted_at: DateTime.add(DateTime.utc_now(), -2, :day)) |
||||
|
||||
_validator_inactive2 = insert(:validator_stability, state: :inactive) |
||||
_validator_inactive3 = insert(:validator_stability, state: :inactive) |
||||
|
||||
_validator_probation1 = |
||||
insert(:validator_stability, state: :probation, inserted_at: DateTime.add(DateTime.utc_now(), -2, :day)) |
||||
|
||||
_validator_probation2 = insert(:validator_stability, state: :probation) |
||||
_validator_probation3 = insert(:validator_stability, state: :probation) |
||||
|
||||
StabilityValidatorsCounters.consolidate() |
||||
:timer.sleep(500) |
||||
|
||||
percentage = (3 / 9 * 100) |> Float.floor(2) |
||||
request = get(conn, "/api/v2/validators/stability/counters") |
||||
|
||||
assert %{ |
||||
"active_validators_counter" => "3", |
||||
"active_validators_percentage" => ^percentage, |
||||
"new_validators_counter_24h" => "6", |
||||
"validators_counter" => "9" |
||||
} = json_response(request, 200) |
||||
end |
||||
end |
||||
end |
||||
|
||||
defp compare_item(%ValidatorStability{} = validator, json) do |
||||
assert Address.checksum(validator.address_hash) == json["address"]["hash"] |
||||
assert to_string(validator.state) == json["state"] |
||||
end |
||||
|
||||
defp compare_item({%ValidatorStability{} = validator, count}, json) do |
||||
assert json["blocks_validated_count"] == count + 1 |
||||
assert compare_item(validator, json) |
||||
end |
||||
|
||||
defp check_paginated_response(first_page_resp, second_page_resp, list) do |
||||
assert Enum.count(first_page_resp["items"]) == 50 |
||||
assert first_page_resp["next_page_params"] != nil |
||||
compare_item(Enum.at(list, 50), Enum.at(first_page_resp["items"], 0)) |
||||
compare_item(Enum.at(list, 1), Enum.at(first_page_resp["items"], 49)) |
||||
|
||||
assert Enum.count(second_page_resp["items"]) == 1 |
||||
assert second_page_resp["next_page_params"] == nil |
||||
compare_item(Enum.at(list, 0), Enum.at(second_page_resp["items"], 0)) |
||||
end |
||||
|
||||
defp compare_default_sorting_for_asc({validator_1, blocks_count_1}, {validator_2, blocks_count_2}) do |
||||
case { |
||||
compare(blocks_count_1, blocks_count_2), |
||||
compare( |
||||
Keyword.fetch!(ValidatorStability.state_enum(), validator_1.state), |
||||
Keyword.fetch!(ValidatorStability.state_enum(), validator_2.state) |
||||
), |
||||
compare(validator_1.address_hash.bytes, validator_2.address_hash.bytes) |
||||
} do |
||||
{:lt, _, _} -> false |
||||
{:eq, :lt, _} -> false |
||||
{:eq, :eq, :lt} -> false |
||||
_ -> true |
||||
end |
||||
end |
||||
|
||||
defp compare_default_sorting_for_desc({validator_1, blocks_count_1}, {validator_2, blocks_count_2}) do |
||||
case { |
||||
compare(blocks_count_1, blocks_count_2), |
||||
compare( |
||||
Keyword.fetch!(ValidatorStability.state_enum(), validator_1.state), |
||||
Keyword.fetch!(ValidatorStability.state_enum(), validator_2.state) |
||||
), |
||||
compare(validator_1.address_hash.bytes, validator_2.address_hash.bytes) |
||||
} do |
||||
{:gt, _, _} -> false |
||||
{:eq, :lt, _} -> false |
||||
{:eq, :eq, :lt} -> false |
||||
_ -> true |
||||
end |
||||
end |
||||
|
||||
defp compare(a, b) do |
||||
cond do |
||||
a < b -> :lt |
||||
a > b -> :gt |
||||
true -> :eq |
||||
end |
||||
end |
||||
end |
||||
@ -0,0 +1,105 @@ |
||||
defmodule Explorer.Chain.Cache.StabilityValidatorsCounters do |
||||
@moduledoc """ |
||||
Counts and store counters of validators stability. |
||||
|
||||
It loads the count asynchronously and in a time interval of 30 minutes. |
||||
""" |
||||
|
||||
use GenServer |
||||
|
||||
alias Explorer.Chain |
||||
alias Explorer.Chain.Stability.Validator, as: ValidatorStability |
||||
|
||||
@validators_counter_key "stability_validators_counter" |
||||
@new_validators_counter_key "new_stability_validators_counter" |
||||
@active_validators_counter_key "active_stability_validators_counter" |
||||
|
||||
# It is undesirable to automatically start the consolidation in all environments. |
||||
# Consider the test environment: if the consolidation initiates but does not |
||||
# finish before a test ends, that test will fail. This way, hundreds of |
||||
# tests were failing before disabling the consolidation and the scheduler in |
||||
# the test env. |
||||
config = Application.compile_env(:explorer, __MODULE__) |
||||
@enable_consolidation Keyword.get(config, :enable_consolidation) |
||||
|
||||
@update_interval_in_milliseconds Keyword.get(config, :update_interval_in_milliseconds) |
||||
|
||||
@doc """ |
||||
Starts a process to periodically update validators stability counters |
||||
""" |
||||
@spec start_link(term()) :: GenServer.on_start() |
||||
def start_link(_) do |
||||
GenServer.start_link(__MODULE__, [], name: __MODULE__) |
||||
end |
||||
|
||||
@impl true |
||||
def init(_args) do |
||||
{:ok, %{consolidate?: @enable_consolidation}, {:continue, :ok}} |
||||
end |
||||
|
||||
defp schedule_next_consolidation do |
||||
Process.send_after(self(), :consolidate, @update_interval_in_milliseconds) |
||||
end |
||||
|
||||
@impl true |
||||
def handle_continue(:ok, %{consolidate?: true} = state) do |
||||
consolidate() |
||||
schedule_next_consolidation() |
||||
|
||||
{:noreply, state} |
||||
end |
||||
|
||||
@impl true |
||||
def handle_continue(:ok, state) do |
||||
{:noreply, state} |
||||
end |
||||
|
||||
@impl true |
||||
def handle_info(:consolidate, state) do |
||||
consolidate() |
||||
schedule_next_consolidation() |
||||
|
||||
{:noreply, state} |
||||
end |
||||
|
||||
@doc """ |
||||
Fetches values for a stability validators counters from the `last_fetched_counters` table. |
||||
""" |
||||
@spec get_counters(Keyword.t()) :: map() |
||||
def get_counters(options) do |
||||
%{ |
||||
validators_counter: Chain.get_last_fetched_counter(@validators_counter_key, options), |
||||
new_validators_counter: Chain.get_last_fetched_counter(@new_validators_counter_key, options), |
||||
active_validators_counter: Chain.get_last_fetched_counter(@active_validators_counter_key, options) |
||||
} |
||||
end |
||||
|
||||
@doc """ |
||||
Consolidates the info by populating the `last_fetched_counters` table with the current database information. |
||||
""" |
||||
@spec consolidate() :: any() |
||||
def consolidate do |
||||
tasks = [ |
||||
Task.async(fn -> ValidatorStability.count_validators() end), |
||||
Task.async(fn -> ValidatorStability.count_new_validators() end), |
||||
Task.async(fn -> ValidatorStability.count_active_validators() end) |
||||
] |
||||
|
||||
[validators_counter, new_validators_counter, active_validators_counter] = Task.await_many(tasks, :infinity) |
||||
|
||||
Chain.upsert_last_fetched_counter(%{ |
||||
counter_type: @validators_counter_key, |
||||
value: validators_counter |
||||
}) |
||||
|
||||
Chain.upsert_last_fetched_counter(%{ |
||||
counter_type: @new_validators_counter_key, |
||||
value: new_validators_counter |
||||
}) |
||||
|
||||
Chain.upsert_last_fetched_counter(%{ |
||||
counter_type: @active_validators_counter_key, |
||||
value: active_validators_counter |
||||
}) |
||||
end |
||||
end |
||||
@ -0,0 +1,288 @@ |
||||
defmodule Explorer.Chain.Stability.Validator do |
||||
@moduledoc """ |
||||
Stability validators |
||||
""" |
||||
|
||||
use Explorer.Schema |
||||
|
||||
alias Explorer.Chain.{Address, Import} |
||||
alias Explorer.Chain.Hash.Address, as: HashAddress |
||||
alias Explorer.{Chain, Repo, SortingHelper} |
||||
alias Explorer.SmartContract.Reader |
||||
|
||||
require Logger |
||||
|
||||
@default_sorting [ |
||||
asc: :state, |
||||
asc: :address_hash |
||||
] |
||||
|
||||
@state_enum [active: 0, probation: 1, inactive: 2] |
||||
|
||||
@primary_key false |
||||
typed_schema "validators_stability" do |
||||
field(:address_hash, HashAddress, primary_key: true) |
||||
field(:state, Ecto.Enum, values: @state_enum) |
||||
field(:blocks_validated, :integer, virtual: true) |
||||
|
||||
has_one(:address, Address, foreign_key: :hash, references: :address_hash) |
||||
timestamps() |
||||
end |
||||
|
||||
@required_attrs ~w(address_hash)a |
||||
@optional_attrs ~w(state)a |
||||
def changeset(%__MODULE__{} = validator, attrs) do |
||||
validator |
||||
|> cast(attrs, @required_attrs ++ @optional_attrs) |
||||
|> validate_required(@required_attrs) |
||||
|> unique_constraint(:address_hash) |
||||
end |
||||
|
||||
@doc """ |
||||
Get validators list. |
||||
Keyword could contain: |
||||
- paging_options |
||||
- necessity_by_association |
||||
- sorting (supported by `Explorer.SortingHelper` module) |
||||
- state (one of `@state_enum`) |
||||
""" |
||||
@spec get_paginated_validators(keyword()) :: [t()] |
||||
def get_paginated_validators(options \\ []) do |
||||
paging_options = Keyword.get(options, :paging_options, Chain.default_paging_options()) |
||||
necessity_by_association = Keyword.get(options, :necessity_by_association, %{}) |
||||
sorting = Keyword.get(options, :sorting, []) |
||||
states = Keyword.get(options, :state, []) |
||||
|
||||
__MODULE__ |
||||
|> apply_filter_by_state(states) |
||||
|> select_merge([vs], %{ |
||||
blocks_validated: |
||||
fragment( |
||||
"SELECT count(*) FROM blocks WHERE miner_hash = ?", |
||||
vs.address_hash |
||||
) |
||||
}) |
||||
|> Chain.join_associations(necessity_by_association) |
||||
|> SortingHelper.apply_sorting(sorting, @default_sorting) |
||||
|> SortingHelper.page_with_sorting(paging_options, sorting, @default_sorting) |
||||
|> Chain.select_repo(options).all() |
||||
end |
||||
|
||||
defp apply_filter_by_state(query, []), do: query |
||||
|
||||
defp apply_filter_by_state(query, states) do |
||||
query |
||||
|> where([vs], vs.state in ^states) |
||||
end |
||||
|
||||
@doc """ |
||||
Get all validators |
||||
""" |
||||
@spec get_all_validators(keyword()) :: [t()] |
||||
def get_all_validators(options \\ []) do |
||||
__MODULE__ |
||||
|> Chain.select_repo(options).all() |
||||
end |
||||
|
||||
@get_active_validator_list_abi %{ |
||||
"type" => "function", |
||||
"stateMutability" => "view", |
||||
"payable" => false, |
||||
"outputs" => [%{"type" => "address[]", "name" => "", "internalType" => "address[]"}], |
||||
"name" => "getActiveValidatorList", |
||||
"inputs" => [] |
||||
} |
||||
|
||||
@get_validator_list_abi %{ |
||||
"type" => "function", |
||||
"stateMutability" => "view", |
||||
"payable" => false, |
||||
"outputs" => [%{"type" => "address[]", "name" => "", "internalType" => "address[]"}], |
||||
"name" => "getValidatorList", |
||||
"inputs" => [] |
||||
} |
||||
|
||||
@get_validator_missing_blocks_abi %{ |
||||
"inputs" => [ |
||||
%{ |
||||
"internalType" => "address", |
||||
"name" => "validator", |
||||
"type" => "address" |
||||
} |
||||
], |
||||
"name" => "getValidatorMissingBlocks", |
||||
"outputs" => [ |
||||
%{ |
||||
"internalType" => "uint256", |
||||
"name" => "", |
||||
"type" => "uint256" |
||||
} |
||||
], |
||||
"stateMutability" => "view", |
||||
"type" => "function" |
||||
} |
||||
|
||||
@get_active_validator_list_method_id "a5aa7380" |
||||
@get_validator_list_method_id "e35c0f7d" |
||||
@get_validator_missing_blocks_method_id "41ee9a53" |
||||
|
||||
@stability_validator_controller_contract "0x0000000000000000000000000000000000000805" |
||||
|
||||
@doc """ |
||||
Do batch eth_call of `getValidatorList` and `getActiveValidatorList` methods to `@stability_validator_controller_contract`. |
||||
Returns a map with two lists: `active` and `all`, or nil if error. |
||||
""" |
||||
@spec fetch_validators_lists :: nil | %{active: list(binary()), all: list(binary())} |
||||
def fetch_validators_lists do |
||||
abi = [@get_active_validator_list_abi, @get_validator_list_abi] |
||||
params = %{@get_validator_list_method_id => [], @get_active_validator_list_method_id => []} |
||||
|
||||
case Reader.query_contract(@stability_validator_controller_contract, abi, params, false) do |
||||
%{ |
||||
@get_active_validator_list_method_id => {:ok, [active_validators_list]}, |
||||
@get_validator_list_method_id => {:ok, [validators_list]} |
||||
} -> |
||||
%{active: active_validators_list, all: validators_list} |
||||
|
||||
error -> |
||||
Logger.warn(fn -> ["Error on getting validator lists: #{inspect(error)}"] end) |
||||
nil |
||||
end |
||||
end |
||||
|
||||
@doc """ |
||||
Do batch eth_call of `getValidatorMissingBlocks` method to #{@stability_validator_controller_contract}. |
||||
Accept: list of validator address hashes |
||||
Returns a map: validator_address_hash => missing_blocks_number |
||||
""" |
||||
@spec fetch_missing_blocks_numbers(list(binary())) :: map() |
||||
def fetch_missing_blocks_numbers(validators_address_hashes) do |
||||
validators_address_hashes |
||||
|> Enum.map(&format_request_missing_blocks_number/1) |
||||
|> Reader.query_contracts([@get_validator_missing_blocks_abi]) |
||||
|> Enum.zip_reduce(validators_address_hashes, %{}, fn response, address_hash, acc -> |
||||
result = |
||||
case format_missing_blocks_result(response) do |
||||
{:error, message} -> |
||||
Logger.warn(fn -> ["Error on getValidatorMissingBlocks for #{validators_address_hashes}: #{message}"] end) |
||||
nil |
||||
|
||||
amount -> |
||||
amount |
||||
end |
||||
|
||||
Map.put(acc, address_hash, result) |
||||
end) |
||||
end |
||||
|
||||
defp format_missing_blocks_result({:ok, [amount]}) do |
||||
amount |
||||
end |
||||
|
||||
defp format_missing_blocks_result({:error, error_message}) do |
||||
{:error, error_message} |
||||
end |
||||
|
||||
defp format_request_missing_blocks_number(address_hash) do |
||||
%{ |
||||
contract_address: @stability_validator_controller_contract, |
||||
method_id: @get_validator_missing_blocks_method_id, |
||||
args: [address_hash] |
||||
} |
||||
end |
||||
|
||||
@doc """ |
||||
Convert missing block number to state |
||||
""" |
||||
@spec missing_block_number_to_state(integer()) :: atom() |
||||
def missing_block_number_to_state(integer) when integer > 0, do: :probation |
||||
def missing_block_number_to_state(integer) when integer == 0, do: :active |
||||
def missing_block_number_to_state(_), do: nil |
||||
|
||||
@doc """ |
||||
Delete validators by address hashes |
||||
""" |
||||
@spec delete_validators_by_address_hashes([binary() | HashAddress.t()]) :: {non_neg_integer(), nil | []} | :ignore |
||||
def delete_validators_by_address_hashes(list) when is_list(list) and length(list) > 0 do |
||||
__MODULE__ |
||||
|> where([vs], vs.address_hash in ^list) |
||||
|> Repo.delete_all() |
||||
end |
||||
|
||||
def delete_validators_by_address_hashes(_), do: :ignore |
||||
|
||||
@doc """ |
||||
Insert validators |
||||
""" |
||||
@spec insert_validators([map()]) :: {non_neg_integer(), nil | []} |
||||
def insert_validators(validators) do |
||||
Repo.insert_all(__MODULE__, validators, |
||||
on_conflict: {:replace_all_except, [:inserted_at]}, |
||||
conflict_target: [:address_hash] |
||||
) |
||||
end |
||||
|
||||
@doc """ |
||||
Append timestamps (:inserted_at, :updated_at) |
||||
""" |
||||
@spec append_timestamps(map()) :: map() |
||||
def append_timestamps(validator) do |
||||
Map.merge(validator, Import.timestamps()) |
||||
end |
||||
|
||||
@doc """ |
||||
Derive next page params from %Explorer.Chain.Stability.Validator{} |
||||
""" |
||||
@spec next_page_params(t()) :: map() |
||||
def next_page_params(%__MODULE__{state: state, address_hash: address_hash, blocks_validated: blocks_validated}) do |
||||
%{"state" => state, "address_hash" => address_hash, "blocks_validated" => blocks_validated} |
||||
end |
||||
|
||||
@doc """ |
||||
Returns state enum |
||||
""" |
||||
@spec state_enum() :: Keyword.t() |
||||
def state_enum, do: @state_enum |
||||
|
||||
@doc """ |
||||
Returns dynamic query for validated blocks count. Needed for SortingHelper |
||||
""" |
||||
@spec dynamic_validated_blocks() :: Ecto.Query.dynamic_expr() |
||||
def dynamic_validated_blocks do |
||||
dynamic( |
||||
[vs], |
||||
fragment( |
||||
"SELECT count(*) FROM blocks WHERE miner_hash = ?", |
||||
vs.address_hash |
||||
) |
||||
) |
||||
end |
||||
|
||||
@doc """ |
||||
Returns total count of validators. |
||||
""" |
||||
@spec count_validators() :: integer() |
||||
def count_validators do |
||||
Repo.aggregate(__MODULE__, :count, :address_hash) |
||||
end |
||||
|
||||
@doc """ |
||||
Returns count of new validators (inserted withing last 24h). |
||||
""" |
||||
@spec count_new_validators() :: integer() |
||||
def count_new_validators do |
||||
__MODULE__ |
||||
|> where([vs], vs.inserted_at >= ago(1, "day")) |
||||
|> Repo.aggregate(:count, :address_hash) |
||||
end |
||||
|
||||
@doc """ |
||||
Returns count of active validators. |
||||
""" |
||||
@spec count_active_validators() :: integer() |
||||
def count_active_validators do |
||||
__MODULE__ |
||||
|> where([vs], vs.state == :active) |
||||
|> Repo.aggregate(:count, :address_hash) |
||||
end |
||||
end |
||||
@ -0,0 +1,14 @@ |
||||
defmodule Explorer.Repo.Stability.Migrations.AddStabilityValidators do |
||||
use Ecto.Migration |
||||
|
||||
def change do |
||||
create table(:validators_stability, primary_key: false) do |
||||
add(:address_hash, :bytea, null: false, primary_key: true) |
||||
add(:state, :integer, default: 0) |
||||
|
||||
timestamps() |
||||
end |
||||
|
||||
create_if_not_exists(index(:validators_stability, ["state ASC", "address_hash ASC"])) |
||||
end |
||||
end |
||||
@ -0,0 +1,70 @@ |
||||
defmodule Indexer.Fetcher.Stability.Validator do |
||||
@moduledoc """ |
||||
GenServer responsible for updating the list of stability validators in the database. |
||||
""" |
||||
use GenServer |
||||
|
||||
alias Explorer.Chain.Hash.Address, as: AddressHash |
||||
alias Explorer.Chain.Stability.Validator, as: ValidatorStability |
||||
|
||||
def start_link(_) do |
||||
GenServer.start_link(__MODULE__, [], name: __MODULE__) |
||||
end |
||||
|
||||
def init(state) do |
||||
GenServer.cast(__MODULE__, :update_validators_list) |
||||
|
||||
{:ok, state} |
||||
end |
||||
|
||||
def handle_cast(:update_validators_list, state) do |
||||
validators_from_db = ValidatorStability.get_all_validators() |
||||
|
||||
case ValidatorStability.fetch_validators_lists() do |
||||
%{active: active_validator_addresses_list, all: validator_addresses_list} -> |
||||
validators_map = Enum.reduce(validator_addresses_list, %{}, fn address, map -> Map.put(map, address, true) end) |
||||
|
||||
active_validators_map = |
||||
Enum.reduce(active_validator_addresses_list, %{}, fn address, map -> Map.put(map, address, true) end) |
||||
|
||||
address_hashes_to_drop_from_db = |
||||
Enum.flat_map(validators_from_db, fn validator -> |
||||
(is_nil(validators_map[validator.address_hash.bytes]) && [validator.address_hash]) || [] |
||||
end) |
||||
|
||||
grouped = |
||||
Enum.group_by(validator_addresses_list, fn validator_address -> active_validators_map[validator_address] end) |
||||
|
||||
inactive = |
||||
Enum.map(grouped[nil] || [], fn address_hash -> |
||||
{:ok, address_hash} = AddressHash.load(address_hash) |
||||
|
||||
%{address_hash: address_hash, state: :inactive} |> ValidatorStability.append_timestamps() |
||||
end) |
||||
|
||||
validators_to_missing_blocks_numbers = ValidatorStability.fetch_missing_blocks_numbers(grouped[true] || []) |
||||
|
||||
active = |
||||
Enum.map(grouped[true] || [], fn address_hash_init -> |
||||
{:ok, address_hash} = AddressHash.load(address_hash_init) |
||||
|
||||
%{ |
||||
address_hash: address_hash, |
||||
state: |
||||
ValidatorStability.missing_block_number_to_state( |
||||
validators_to_missing_blocks_numbers[address_hash_init] |
||||
) |
||||
} |
||||
|> ValidatorStability.append_timestamps() |
||||
end) |
||||
|
||||
ValidatorStability.insert_validators(active ++ inactive) |
||||
ValidatorStability.delete_validators_by_address_hashes(address_hashes_to_drop_from_db) |
||||
|
||||
_ -> |
||||
nil |
||||
end |
||||
|
||||
{:noreply, state} |
||||
end |
||||
end |
||||
@ -0,0 +1,270 @@ |
||||
defmodule Indexer.Fetcher.Stability.ValidatorTest do |
||||
use EthereumJSONRPC.Case |
||||
use Explorer.DataCase |
||||
|
||||
import Mox |
||||
|
||||
alias Explorer.Chain.Stability.Validator, as: ValidatorStability |
||||
alias EthereumJSONRPC.Encoder |
||||
|
||||
setup :verify_on_exit! |
||||
setup :set_mox_global |
||||
|
||||
@accepts_list_of_addresses %{ |
||||
"type" => "function", |
||||
"stateMutability" => "view", |
||||
"payable" => false, |
||||
"outputs" => [%{"type" => "address[]", "name" => "", "internalType" => "address[]"}], |
||||
"name" => "getActiveValidatorList", |
||||
"inputs" => [%{"type" => "address[]", "name" => "", "internalType" => "address[]"}] |
||||
} |
||||
|
||||
@accepts_integer %{ |
||||
"type" => "function", |
||||
"stateMutability" => "view", |
||||
"payable" => false, |
||||
"outputs" => [ |
||||
%{ |
||||
"internalType" => "uint256", |
||||
"name" => "", |
||||
"type" => "uint256" |
||||
} |
||||
], |
||||
"name" => "getActiveValidatorList", |
||||
"inputs" => [ |
||||
%{ |
||||
"internalType" => "uint256", |
||||
"name" => "", |
||||
"type" => "uint256" |
||||
} |
||||
] |
||||
} |
||||
|
||||
if Application.compile_env(:explorer, :chain_type) == "stability" do |
||||
describe "check update_validators_list" do |
||||
test "deletes absent validators" do |
||||
_validator = insert(:validator_stability) |
||||
_validator_active = insert(:validator_stability, state: :active) |
||||
_validator_inactive = insert(:validator_stability, state: :inactive) |
||||
_validator_probation = insert(:validator_stability, state: :probation) |
||||
|
||||
start_supervised!({Indexer.Fetcher.Stability.Validator, name: Indexer.Fetcher.Stability.Validator}) |
||||
|
||||
EthereumJSONRPC.Mox |
||||
|> expect(:json_rpc, 1, fn |
||||
[ |
||||
%{ |
||||
id: id_1, |
||||
jsonrpc: "2.0", |
||||
method: "eth_call", |
||||
params: [%{data: "0xa5aa7380", to: "0x0000000000000000000000000000000000000805"}, "latest"] |
||||
}, |
||||
%{ |
||||
id: id_2, |
||||
jsonrpc: "2.0", |
||||
method: "eth_call", |
||||
params: [%{data: "0xe35c0f7d", to: "0x0000000000000000000000000000000000000805"}, "latest"] |
||||
} |
||||
], |
||||
_ -> |
||||
<<"0x", _method_id::binary-size(8), result::binary>> = |
||||
[@accepts_list_of_addresses] |
||||
|> ABI.parse_specification() |
||||
|> Enum.at(0) |
||||
|> Encoder.encode_function_call([[]]) |
||||
|
||||
{:ok, |
||||
[ |
||||
%{ |
||||
id: id_1, |
||||
jsonrpc: "2.0", |
||||
result: "0x" <> result |
||||
}, |
||||
%{ |
||||
id: id_2, |
||||
jsonrpc: "2.0", |
||||
result: "0x" <> result |
||||
} |
||||
]} |
||||
end) |
||||
|
||||
:timer.sleep(100) |
||||
assert ValidatorStability.get_all_validators() == [] |
||||
end |
||||
|
||||
test "updates validators" do |
||||
validator_active1 = insert(:validator_stability, state: :active) |
||||
validator_active2 = insert(:validator_stability, state: :active) |
||||
_validator_active3 = insert(:validator_stability, state: :active) |
||||
|
||||
validator_inactive1 = insert(:validator_stability, state: :inactive) |
||||
validator_inactive2 = insert(:validator_stability, state: :inactive) |
||||
_validator_inactive3 = insert(:validator_stability, state: :inactive) |
||||
|
||||
validator_probation1 = insert(:validator_stability, state: :probation) |
||||
validator_probation2 = insert(:validator_stability, state: :probation) |
||||
_validator_probation3 = insert(:validator_stability, state: :probation) |
||||
|
||||
start_supervised!({Indexer.Fetcher.Stability.Validator, name: Indexer.Fetcher.Stability.Validator}) |
||||
|
||||
EthereumJSONRPC.Mox |
||||
|> expect(:json_rpc, fn |
||||
[ |
||||
%{ |
||||
id: id_1, |
||||
jsonrpc: "2.0", |
||||
method: "eth_call", |
||||
params: [%{data: "0xa5aa7380", to: "0x0000000000000000000000000000000000000805"}, "latest"] |
||||
}, |
||||
%{ |
||||
id: id_2, |
||||
jsonrpc: "2.0", |
||||
method: "eth_call", |
||||
params: [%{data: "0xe35c0f7d", to: "0x0000000000000000000000000000000000000805"}, "latest"] |
||||
} |
||||
], |
||||
_ -> |
||||
<<"0x", _method_id::binary-size(8), result_all::binary>> = |
||||
[@accepts_list_of_addresses] |
||||
|> ABI.parse_specification() |
||||
|> Enum.at(0) |
||||
|> Encoder.encode_function_call([ |
||||
[ |
||||
validator_active1.address_hash.bytes, |
||||
validator_active2.address_hash.bytes, |
||||
validator_inactive1.address_hash.bytes, |
||||
validator_inactive2.address_hash.bytes, |
||||
validator_probation1.address_hash.bytes, |
||||
validator_probation2.address_hash.bytes |
||||
] |
||||
]) |
||||
|
||||
<<"0x", _method_id::binary-size(8), result_active::binary>> = |
||||
[@accepts_list_of_addresses] |
||||
|> ABI.parse_specification() |
||||
|> Enum.at(0) |
||||
|> Encoder.encode_function_call([ |
||||
[ |
||||
validator_active1.address_hash.bytes, |
||||
validator_inactive1.address_hash.bytes, |
||||
validator_probation1.address_hash.bytes |
||||
] |
||||
]) |
||||
|
||||
{:ok, |
||||
[ |
||||
%{ |
||||
id: id_1, |
||||
jsonrpc: "2.0", |
||||
result: "0x" <> result_active |
||||
}, |
||||
%{ |
||||
id: id_2, |
||||
jsonrpc: "2.0", |
||||
result: "0x" <> result_all |
||||
} |
||||
]} |
||||
end) |
||||
|
||||
"0x" <> address_1 = to_string(validator_active1.address_hash) |
||||
"0x" <> address_2 = to_string(validator_inactive1.address_hash) |
||||
"0x" <> address_3 = to_string(validator_probation1.address_hash) |
||||
|
||||
EthereumJSONRPC.Mox |
||||
|> expect(:json_rpc, fn |
||||
[ |
||||
%{ |
||||
id: id_1, |
||||
jsonrpc: "2.0", |
||||
method: "eth_call", |
||||
params: [ |
||||
%{ |
||||
data: "0x41ee9a53000000000000000000000000" <> ^address_1, |
||||
to: "0x0000000000000000000000000000000000000805" |
||||
}, |
||||
"latest" |
||||
] |
||||
}, |
||||
%{ |
||||
id: id_2, |
||||
jsonrpc: "2.0", |
||||
method: "eth_call", |
||||
params: [ |
||||
%{ |
||||
data: "0x41ee9a53000000000000000000000000" <> ^address_2, |
||||
to: "0x0000000000000000000000000000000000000805" |
||||
}, |
||||
"latest" |
||||
] |
||||
}, |
||||
%{ |
||||
id: id_3, |
||||
jsonrpc: "2.0", |
||||
method: "eth_call", |
||||
params: [ |
||||
%{ |
||||
data: "0x41ee9a53000000000000000000000000" <> ^address_3, |
||||
to: "0x0000000000000000000000000000000000000805" |
||||
}, |
||||
"latest" |
||||
] |
||||
} |
||||
], |
||||
_ -> |
||||
<<"0x", _method_id::binary-size(8), result_1::binary>> = |
||||
[@accepts_integer] |
||||
|> ABI.parse_specification() |
||||
|> Enum.at(0) |
||||
|> Encoder.encode_function_call([10]) |
||||
|
||||
<<"0x", _method_id::binary-size(8), result_2::binary>> = |
||||
[@accepts_integer] |
||||
|> ABI.parse_specification() |
||||
|> Enum.at(0) |
||||
|> Encoder.encode_function_call([1]) |
||||
|
||||
<<"0x", _method_id::binary-size(8), result_3::binary>> = |
||||
[@accepts_integer] |
||||
|> ABI.parse_specification() |
||||
|> Enum.at(0) |
||||
|> Encoder.encode_function_call([0]) |
||||
|
||||
{:ok, |
||||
[ |
||||
%{ |
||||
id: id_1, |
||||
jsonrpc: "2.0", |
||||
result: "0x" <> result_1 |
||||
}, |
||||
%{ |
||||
id: id_2, |
||||
jsonrpc: "2.0", |
||||
result: "0x" <> result_2 |
||||
}, |
||||
%{ |
||||
id: id_3, |
||||
jsonrpc: "2.0", |
||||
result: "0x" <> result_3 |
||||
} |
||||
]} |
||||
end) |
||||
|
||||
:timer.sleep(100) |
||||
validators = ValidatorStability.get_all_validators() |
||||
|
||||
assert Enum.count(validators) == 6 |
||||
|
||||
map = |
||||
Enum.reduce(validators, %{}, fn validator, map -> Map.put(map, validator.address_hash.bytes, validator) end) |
||||
|
||||
assert %ValidatorStability{state: :inactive} = map[validator_active2.address_hash.bytes] |
||||
assert %ValidatorStability{state: :inactive} = map[validator_inactive2.address_hash.bytes] |
||||
assert %ValidatorStability{state: :inactive} = map[validator_probation2.address_hash.bytes] |
||||
|
||||
assert %ValidatorStability{state: :probation} = map[validator_active1.address_hash.bytes] |
||||
assert %ValidatorStability{state: :probation} = map[validator_inactive1.address_hash.bytes] |
||||
assert %ValidatorStability{state: :active} = map[validator_probation1.address_hash.bytes] |
||||
end |
||||
end |
||||
end |
||||
end |
||||
Loading…
Reference in new issue