Add internal_transactions_indexed_at to transactions schema for catching up on deferred internal transactions on boot.pull/218/head
Chris McCord
7 years ago
committed by
Luke Imhoff
parent
5fe8abd863
commit
e813045564
@ -0,0 +1,162 @@ |
||||
defmodule Explorer.BufferedTask do |
||||
@moduledoc """ |
||||
TODO |
||||
""" |
||||
use GenServer |
||||
require Logger |
||||
|
||||
@callback init(initial :: term, reducer :: function) :: |
||||
{:ok, accumulated_results :: term | initial :: term} | {:error, reason :: term} |
||||
|
||||
@callback run(entries :: list) :: :ok | {:retry, reason :: term} | {:halt, reason :: term} |
||||
|
||||
@flush_interval :timer.seconds(3) |
||||
|
||||
def buffer(server, entry) do |
||||
GenServer.call(server, {:buffer, entry}) |
||||
end |
||||
|
||||
def start_link({module, base_opts}) do |
||||
default_opts = Application.fetch_env!(:explorer, :indexer) |
||||
opts = Keyword.merge(default_opts, base_opts) |
||||
|
||||
GenServer.start_link(__MODULE__, {module, opts}, name: opts[:name]) |
||||
end |
||||
|
||||
def init({callback_module, opts}) do |
||||
send(self(), :initial_stream) |
||||
|
||||
state = %{ |
||||
callback_module: callback_module, |
||||
debug_logs: Keyword.get(opts, :debug_logs, false), |
||||
flush_timer: nil, |
||||
flush_interval: Keyword.get(opts, :flush_interval, @flush_interval), |
||||
max_batch_size: Keyword.fetch!(opts, :max_batch_size), |
||||
max_concurrency: Keyword.fetch!(opts, :max_concurrency), |
||||
buffer: :queue.new(), |
||||
tasks: %{} |
||||
} |
||||
|
||||
{:ok, state} |
||||
end |
||||
|
||||
def handle_info(:initial_stream, state) do |
||||
{:noreply, do_initial_stream(state)} |
||||
end |
||||
|
||||
def handle_info(:flush, state) do |
||||
{:noreply, state |> spawn_next_batch([]) |> schedule_next_buffer_flush()} |
||||
end |
||||
|
||||
def handle_info({:async_perform, entries}, state) do |
||||
{:noreply, spawn_next_batch(state, entries)} |
||||
end |
||||
|
||||
def handle_info({ref, {:performed, :ok}}, state) do |
||||
{:noreply, drop_task(state, ref)} |
||||
end |
||||
|
||||
def handle_info({ref, {:performed, {:retry, _reason}}}, state) do |
||||
{:noreply, drop_task_and_retry(state, ref)} |
||||
end |
||||
|
||||
def handle_info({ref, {:performed, {:halt, _reason}}}, state) do |
||||
{:noreply, drop_task(state, ref)} |
||||
end |
||||
|
||||
def handle_info({:DOWN, _ref, :process, _pid, :normal}, state) do |
||||
{:noreply, state} |
||||
end |
||||
|
||||
def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do |
||||
{:noreply, drop_task_and_retry(state, ref)} |
||||
end |
||||
|
||||
def handle_call({:buffer, entries}, _from, state) do |
||||
{:reply, :ok, buffer_entries(state, entries)} |
||||
end |
||||
|
||||
defp drop_task(state, ref) do |
||||
schedule_async_perform([]) |
||||
%{state | tasks: Map.delete(state.tasks, ref)} |
||||
end |
||||
|
||||
defp drop_task_and_retry(state, ref) do |
||||
batch = Map.fetch!(state.tasks, ref) |
||||
|
||||
state |
||||
|> drop_task(ref) |
||||
|> buffer_entries(batch) |
||||
end |
||||
|
||||
defp buffer_entries(state, entries) do |
||||
%{state | buffer: :queue.join(state.buffer, :queue.from_list(entries))} |
||||
end |
||||
|
||||
defp do_initial_stream(state) do |
||||
state.buffer |
||||
|> state.callback_module.init(fn entry, acc -> |
||||
batch = :queue.in(entry, acc) |
||||
|
||||
if :queue.len(batch) >= state.max_batch_size do |
||||
schedule_async_perform(:queue.to_list(batch)) |
||||
:queue.new() |
||||
else |
||||
batch |
||||
end |
||||
end) |
||||
|> catchup_remaining() |
||||
|
||||
schedule_next_buffer_flush(state) |
||||
end |
||||
|
||||
defp catchup_remaining({:ok, batch}) do |
||||
if :queue.len(batch) > 0 do |
||||
schedule_async_perform(:queue.to_list(batch)) |
||||
end |
||||
|
||||
:ok |
||||
end |
||||
|
||||
defp take_batch(state) do |
||||
{entries, remaining_queue} = |
||||
Enum.reduce_while(1..state.max_batch_size, {[], state.buffer}, fn _, {entries, queue_acc} -> |
||||
case :queue.out(queue_acc) do |
||||
{{:value, entry}, new_queue} -> {:cont, {[entry | entries], new_queue}} |
||||
{:empty, new_queue} -> {:halt, {entries, new_queue}} |
||||
end |
||||
end) |
||||
|
||||
{Enum.reverse(entries), remaining_queue} |
||||
end |
||||
|
||||
defp schedule_async_perform(entries, after_ms \\ 0) do |
||||
Process.send_after(self(), {:async_perform, entries}, after_ms) |
||||
end |
||||
|
||||
defp schedule_next_buffer_flush(state) do |
||||
timer = Process.send_after(self(), :flush, state.flush_interval) |
||||
%{state | flush_timer: timer} |
||||
end |
||||
|
||||
defp spawn_next_batch(state, entries) do |
||||
state = buffer_entries(state, entries) |
||||
|
||||
if Enum.count(state.tasks) < state.max_concurrency and :queue.len(state.buffer) > 0 do |
||||
{batch, new_queue} = take_batch(state) |
||||
|
||||
task = |
||||
Task.Supervisor.async_nolink(Explorer.TaskSupervisor, fn -> |
||||
debug(state, fn -> "processing #{Enum.count(batch)} entries for #{inspect(state.callback_module)}" end) |
||||
{:performed, state.callback_module.run(batch)} |
||||
end) |
||||
|
||||
%{state | tasks: Map.put(state.tasks, task.ref, batch), buffer: new_queue} |
||||
else |
||||
state |
||||
end |
||||
end |
||||
|
||||
defp debug(%{debug_logs: true}, func), do: Logger.debug(func) |
||||
defp debug(%{debug_logs: false}, _func), do: :noop |
||||
end |
||||
@ -1,157 +1,28 @@ |
||||
defmodule Explorer.Indexer.AddressBalanceFetcher do |
||||
@moduledoc """ |
||||
Fetches and indexes `t:Explorer.Chain.Address.t/0` balances. |
||||
Fetches `t:Explorer.Chain.Address.t/0` `fetched_balance`. |
||||
""" |
||||
use GenServer |
||||
require Logger |
||||
|
||||
alias EthereumJSONRPC |
||||
alias Explorer.Chain |
||||
alias Explorer.Chain.{Address, Hash} |
||||
alias Explorer.{BufferedTask, Chain} |
||||
alias Explorer.Chain.{Hash, Address} |
||||
|
||||
@fetch_interval :timer.seconds(3) |
||||
@max_batch_size 100 |
||||
@max_concurrency 2 |
||||
@behaviour BufferedTask |
||||
|
||||
def async_fetch_balances(address_hashes) do |
||||
GenServer.cast(__MODULE__, {:buffer_addresses, address_hashes}) |
||||
end |
||||
|
||||
def start_link(opts) do |
||||
GenServer.start_link(__MODULE__, opts, name: __MODULE__) |
||||
end |
||||
|
||||
def init(opts) do |
||||
opts = Keyword.merge(Application.fetch_env!(:explorer, :indexer), opts) |
||||
send(self(), :fetch_unfetched_addresses) |
||||
|
||||
state = %{ |
||||
debug_logs: Keyword.get(opts, :debug_logs, false), |
||||
flush_timer: nil, |
||||
fetch_interval: Keyword.get(opts, :fetch_interval, @fetch_interval), |
||||
max_batch_size: Keyword.get(opts, :max_batch_size, @max_batch_size), |
||||
buffer: :queue.new(), |
||||
tasks: %{} |
||||
} |
||||
|
||||
{:ok, state} |
||||
end |
||||
|
||||
def handle_info(:fetch_unfetched_addresses, state) do |
||||
{:noreply, stream_unfetched_addresses(state)} |
||||
end |
||||
|
||||
def handle_info(:flush, state) do |
||||
{:noreply, state |> fetch_next_batch([]) |> schedule_next_buffer_flush()} |
||||
end |
||||
|
||||
def handle_info({:async_fetch, hashes}, state) do |
||||
{:noreply, fetch_next_batch(state, hashes)} |
||||
end |
||||
|
||||
def handle_info({ref, {:fetched_balances, results}}, state) do |
||||
:ok = Chain.update_balances(results) |
||||
{:noreply, drop_task(state, ref)} |
||||
end |
||||
|
||||
def handle_info({:DOWN, _ref, :process, _pid, :normal}, state) do |
||||
{:noreply, state} |
||||
end |
||||
|
||||
def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do |
||||
batch = Map.fetch!(state.tasks, ref) |
||||
|
||||
new_state = |
||||
state |
||||
|> drop_task(ref) |
||||
|> buffer_addresses(batch) |
||||
|
||||
{:noreply, new_state} |
||||
end |
||||
|
||||
def handle_cast({:buffer_addresses, address_hashes}, state) do |
||||
string_hashes = for hash <- address_hashes, do: Hash.to_string(hash) |
||||
{:noreply, buffer_addresses(state, string_hashes)} |
||||
BufferedTask.buffer(__MODULE__, string_hashes) |
||||
end |
||||
|
||||
defp drop_task(state, ref) do |
||||
schedule_async_fetch([]) |
||||
%{state | tasks: Map.delete(state.tasks, ref)} |
||||
end |
||||
|
||||
defp buffer_addresses(state, string_hashes) do |
||||
%{state | buffer: :queue.join(state.buffer, :queue.from_list(string_hashes))} |
||||
end |
||||
|
||||
defp stream_unfetched_addresses(state) do |
||||
state.buffer |
||||
|> Chain.stream_unfetched_addresses(fn %Address{hash: hash}, batch -> |
||||
batch = :queue.in(Hash.to_string(hash), batch) |
||||
|
||||
if :queue.len(batch) >= state.max_batch_size do |
||||
schedule_async_fetch(:queue.to_list(batch)) |
||||
:queue.new() |
||||
else |
||||
batch |
||||
end |
||||
def init(acc, reducer) do |
||||
Chain.stream_unfetched_addresses(acc, fn %Address{hash: hash}, acc -> |
||||
reducer.(Hash.to_string(hash), acc) |
||||
end) |
||||
|> fetch_remaining() |
||||
|
||||
schedule_next_buffer_flush(state) |
||||
end |
||||
|
||||
defp fetch_remaining({:ok, batch}) do |
||||
if :queue.len(batch) > 0 do |
||||
schedule_async_fetch(:queue.to_list(batch)) |
||||
end |
||||
def run(string_hashes) do |
||||
{:ok, results} = EthereumJSONRPC.fetch_balances_by_hash(string_hashes) |
||||
:ok = Chain.update_balances(results) |
||||
|
||||
:ok |
||||
end |
||||
|
||||
defp do_fetch_addresses(address_hashes) do |
||||
EthereumJSONRPC.fetch_balances_by_hash(address_hashes) |
||||
end |
||||
|
||||
defp take_batch(queue) do |
||||
{hashes, remaining_queue} = |
||||
Enum.reduce_while(1..@max_batch_size, {[], queue}, fn _, {hashes, queue_acc} -> |
||||
case :queue.out(queue_acc) do |
||||
{{:value, hash}, new_queue} -> {:cont, {[hash | hashes], new_queue}} |
||||
{:empty, new_queue} -> {:halt, {hashes, new_queue}} |
||||
end |
||||
end) |
||||
|
||||
{Enum.reverse(hashes), remaining_queue} |
||||
end |
||||
|
||||
defp schedule_async_fetch(hashes, after_ms \\ 0) do |
||||
Process.send_after(self(), {:async_fetch, hashes}, after_ms) |
||||
end |
||||
|
||||
defp schedule_next_buffer_flush(state) do |
||||
timer = Process.send_after(self(), :flush, state.fetch_interval) |
||||
%{state | flush_timer: timer} |
||||
end |
||||
|
||||
defp fetch_next_batch(state, hashes) do |
||||
state = buffer_addresses(state, hashes) |
||||
|
||||
if Enum.count(state.tasks) < @max_concurrency and :queue.len(state.buffer) > 0 do |
||||
{batch, new_queue} = take_batch(state.buffer) |
||||
|
||||
task = |
||||
Task.Supervisor.async_nolink(Explorer.Indexer.TaskSupervisor, fn -> |
||||
debug(state, fn -> "fetching #{Enum.count(batch)} balances" end) |
||||
{:ok, balances} = do_fetch_addresses(batch) |
||||
{:fetched_balances, balances} |
||||
end) |
||||
|
||||
%{state | tasks: Map.put(state.tasks, task.ref, batch), buffer: new_queue} |
||||
else |
||||
buffer_addresses(state, hashes) |
||||
end |
||||
end |
||||
|
||||
defp debug(%{debug_logs: true}, func), do: Logger.debug(func) |
||||
defp debug(%{debug_logs: false}, _func), do: :noop |
||||
end |
||||
|
||||
@ -0,0 +1,37 @@ |
||||
defmodule Explorer.Indexer.InternalTransactionFetcher do |
||||
@moduledoc """ |
||||
Fetches and indexes `t:Explorer.Chain.InternalTransaction.t/0`. |
||||
""" |
||||
|
||||
alias Explorer.{BufferedTask, Chain} |
||||
alias Explorer.Chain.{Hash, Transaction} |
||||
|
||||
@behaviour BufferedTask |
||||
|
||||
def async_fetch(transactions) do |
||||
string_hashes = |
||||
Enum.map(transactions, fn %Transaction{hash: hash} -> |
||||
Hash.to_string(hash) |
||||
end) |
||||
|
||||
BufferedTask.buffer(__MODULE__, string_hashes) |
||||
end |
||||
|
||||
def init(acc, reducer) do |
||||
Chain.stream_transactions_with_unfetched_internal_transactions(acc, fn %Transaction{hash: hash}, acc -> |
||||
reducer.(Hash.to_string(hash), acc) |
||||
end) |
||||
end |
||||
|
||||
def run(transaction_hashes) do |
||||
case EthereumJSONRPC.fetch_internal_transactions(transaction_hashes) do |
||||
{:ok, internal_params} -> |
||||
{:ok, _} = Chain.import_internal_transactions(internal_params) |
||||
|
||||
:ok |
||||
|
||||
{:error, reason} -> |
||||
{:retry, reason} |
||||
end |
||||
end |
||||
end |
||||
@ -0,0 +1,98 @@ |
||||
defmodule Explorer.BufferedTaskTest do |
||||
use ExUnit.Case, async: true |
||||
|
||||
alias Explorer.BufferedTask |
||||
|
||||
@max_batch_size 2 |
||||
|
||||
defp start_buffer(callback_module) do |
||||
start_supervised( |
||||
{BufferedTask, {callback_module, flush_interval: 50, max_batch_size: @max_batch_size, max_concurrency: 2}} |
||||
) |
||||
end |
||||
|
||||
defmodule CounterTask do |
||||
@behaviour BufferedTask |
||||
|
||||
def initial_collection, do: for(i <- 1..11, do: "#{i}") |
||||
|
||||
def init(acc, reducer) do |
||||
{:ok, Enum.reduce(initial_collection(), acc, fn item, acc -> reducer.(item, acc) end)} |
||||
end |
||||
|
||||
def run(batch) do |
||||
send(__MODULE__, {:run, batch}) |
||||
:ok |
||||
end |
||||
end |
||||
|
||||
defmodule FunTask do |
||||
@behaviour BufferedTask |
||||
|
||||
def init(acc, _reducer) do |
||||
{:ok, acc} |
||||
end |
||||
|
||||
def run([agent, func]) when is_function(func) do |
||||
count = Agent.get_and_update(agent, &{&1, &1 + 1}) |
||||
send(__MODULE__, {:run, count}) |
||||
func.(count) |
||||
end |
||||
|
||||
def run(batch) do |
||||
send(__MODULE__, {:run, batch}) |
||||
:ok |
||||
end |
||||
end |
||||
|
||||
test "init allows buffer to be loaded up with initial entries" do |
||||
Process.register(self(), CounterTask) |
||||
{:ok, buffer} = start_buffer(CounterTask) |
||||
|
||||
CounterTask.initial_collection() |
||||
|> Enum.chunk_every(@max_batch_size) |
||||
|> Enum.each(fn batch -> |
||||
assert_receive {:run, ^batch} |
||||
end) |
||||
|
||||
refute_receive _ |
||||
|
||||
BufferedTask.buffer(buffer, ~w(12 13 14 15 16)) |
||||
assert_receive {:run, ~w(12 13)} |
||||
assert_receive {:run, ~w(14 15)} |
||||
assert_receive {:run, ~w(16)} |
||||
refute_receive _ |
||||
end |
||||
|
||||
test "init with zero entries schedules future buffer flushes" do |
||||
Process.register(self(), FunTask) |
||||
{:ok, buffer} = start_buffer(FunTask) |
||||
refute_receive _ |
||||
|
||||
BufferedTask.buffer(buffer, ~w(some more entries)) |
||||
|
||||
assert_receive {:run, ~w(some more)} |
||||
assert_receive {:run, ~w(entries)} |
||||
refute_receive _ |
||||
end |
||||
|
||||
test "run/1 allows tasks to be programmatically retried" do |
||||
Process.register(self(), FunTask) |
||||
{:ok, buffer} = start_buffer(FunTask) |
||||
{:ok, count} = Agent.start_link(fn -> 1 end) |
||||
|
||||
BufferedTask.buffer(buffer, [ |
||||
count, |
||||
fn |
||||
1 -> {:retry, :because_reasons} |
||||
2 -> {:retry, :because_reasons} |
||||
3 -> :ok |
||||
end |
||||
]) |
||||
|
||||
assert_receive {:run, 1} |
||||
assert_receive {:run, 2} |
||||
assert_receive {:run, 3} |
||||
refute_receive _ |
||||
end |
||||
end |
||||
@ -0,0 +1,15 @@ |
||||
defmodule Explorer.Indexer.AddressBalanceFetcherCase do |
||||
alias Explorer.BufferedTask |
||||
alias Explorer.Indexer.AddressBalanceFetcher |
||||
|
||||
def start_supervised!(options \\ []) when is_list(options) do |
||||
ExUnit.Callbacks.start_supervised!( |
||||
{BufferedTask, |
||||
{AddressBalanceFetcher, |
||||
Keyword.merge( |
||||
[debug_logs: false, fetch_interval: 1, max_batch_size: 1, max_concurrency: 1, name: AddressBalanceFetcher], |
||||
options |
||||
)}} |
||||
) |
||||
end |
||||
end |
||||
Loading…
Reference in new issue