Massive blocks fetcher (#9486)
* Massive blocks fetcher * Improve massive blocks fetcher log * Update apps/explorer/lib/explorer/utility/massive_block.ex Co-authored-by: Maxim Filonov <53992153+sl1depengwyn@users.noreply.github.com> * Add low priority queue for MassiveBlocksFetcher --------- Co-authored-by: Maxim Filonov <53992153+sl1depengwyn@users.noreply.github.com>pull/9640/head
Qwerty5Uiop
8 months ago
committed by
GitHub
parent
de905162a0
commit
400b45b145
@ -0,0 +1,42 @@ |
||||
defmodule Explorer.Utility.MassiveBlock do |
||||
@moduledoc """ |
||||
Module is responsible for keeping the block numbers that are too large for regular import |
||||
and need more time to complete. |
||||
""" |
||||
|
||||
use Explorer.Schema |
||||
|
||||
alias Explorer.Repo |
||||
|
||||
@primary_key false |
||||
typed_schema "massive_blocks" do |
||||
field(:number, :integer, primary_key: true) |
||||
|
||||
timestamps() |
||||
end |
||||
|
||||
@doc false |
||||
def changeset(massive_block \\ %__MODULE__{}, params) do |
||||
cast(massive_block, params, [:number]) |
||||
end |
||||
|
||||
def get_last_block_number(except_numbers) do |
||||
__MODULE__ |
||||
|> where([mb], mb.number not in ^except_numbers) |
||||
|> select([mb], max(mb.number)) |
||||
|> Repo.one() |
||||
end |
||||
|
||||
def insert_block_numbers(numbers) do |
||||
now = DateTime.utc_now() |
||||
params = Enum.map(numbers, &%{number: &1, inserted_at: now, updated_at: now}) |
||||
|
||||
Repo.insert_all(__MODULE__, params, on_conflict: {:replace, [:updated_at]}, conflict_target: :number) |
||||
end |
||||
|
||||
def delete_block_number(number) do |
||||
__MODULE__ |
||||
|> where([mb], mb.number == ^number) |
||||
|> Repo.delete_all() |
||||
end |
||||
end |
||||
@ -0,0 +1,11 @@ |
||||
defmodule Explorer.Repo.Migrations.CreateMassiveBlocks do |
||||
use Ecto.Migration |
||||
|
||||
def change do |
||||
create table(:massive_blocks, primary_key: false) do |
||||
add(:number, :bigint, primary_key: true) |
||||
|
||||
timestamps() |
||||
end |
||||
end |
||||
end |
||||
@ -0,0 +1,88 @@ |
||||
defmodule Indexer.Block.Catchup.MassiveBlocksFetcher do |
||||
@moduledoc """ |
||||
Fetches and indexes blocks by numbers from massive_blocks table. |
||||
""" |
||||
|
||||
use GenServer |
||||
|
||||
require Logger |
||||
|
||||
alias Explorer.Utility.MassiveBlock |
||||
alias Indexer.Block.Fetcher |
||||
|
||||
@increased_interval 10000 |
||||
|
||||
@spec start_link(term()) :: GenServer.on_start() |
||||
def start_link(_) do |
||||
GenServer.start_link(__MODULE__, :ok, name: __MODULE__) |
||||
end |
||||
|
||||
@impl true |
||||
def init(_) do |
||||
send_new_task() |
||||
|
||||
{:ok, %{block_fetcher: generate_block_fetcher(), low_priority_blocks: []}} |
||||
end |
||||
|
||||
@impl true |
||||
def handle_info(:task, %{low_priority_blocks: low_priority_blocks} = state) do |
||||
{result, new_low_priority_blocks} = |
||||
case MassiveBlock.get_last_block_number(low_priority_blocks) do |
||||
nil -> |
||||
case low_priority_blocks do |
||||
[number | rest] -> |
||||
failed_blocks = process_block(state.block_fetcher, number) |
||||
{:processed, rest ++ failed_blocks} |
||||
|
||||
[] -> |
||||
{:empty, []} |
||||
end |
||||
|
||||
number -> |
||||
failed_blocks = process_block(state.block_fetcher, number) |
||||
{:processed, low_priority_blocks ++ failed_blocks} |
||||
end |
||||
|
||||
case result do |
||||
:processed -> send_new_task() |
||||
:empty -> send_new_task(@increased_interval) |
||||
end |
||||
|
||||
{:noreply, %{state | low_priority_blocks: new_low_priority_blocks}} |
||||
end |
||||
|
||||
def handle_info(_, state) do |
||||
{:noreply, state} |
||||
end |
||||
|
||||
defp process_block(block_fetcher, number) do |
||||
case Fetcher.fetch_and_import_range(block_fetcher, number..number, %{timeout: :infinity}) do |
||||
{:ok, _result} -> |
||||
Logger.info("MassiveBlockFetcher successfully proceed block #{inspect(number)}") |
||||
MassiveBlock.delete_block_number(number) |
||||
[] |
||||
|
||||
{:error, error} -> |
||||
Logger.error("MassiveBlockFetcher failed: #{inspect(error)}") |
||||
[number] |
||||
end |
||||
end |
||||
|
||||
defp generate_block_fetcher do |
||||
receipts_batch_size = Application.get_env(:indexer, :receipts_batch_size) |
||||
receipts_concurrency = Application.get_env(:indexer, :receipts_concurrency) |
||||
json_rpc_named_arguments = Application.get_env(:indexer, :json_rpc_named_arguments) |
||||
|
||||
%Fetcher{ |
||||
broadcast: :catchup, |
||||
callback_module: Indexer.Block.Catchup.Fetcher, |
||||
json_rpc_named_arguments: json_rpc_named_arguments, |
||||
receipts_batch_size: receipts_batch_size, |
||||
receipts_concurrency: receipts_concurrency |
||||
} |
||||
end |
||||
|
||||
defp send_new_task(interval \\ 0) do |
||||
Process.send_after(self(), :task, interval) |
||||
end |
||||
end |
||||
Loading…
Reference in new issue