@ -14,19 +14,20 @@ defmodule Indexer.Block.Realtime.Fetcher do
import Indexer.Block.Fetcher ,
only : [
async_import_block_rewards : 1 ,
async_import_created_contract_codes : 1 ,
async_import_internal_transactions : 2 ,
async_import_replaced_transactions : 1 ,
async_import_tokens : 1 ,
async_import_token_balances : 1 ,
async_import_uncles : 1 ,
fetch_and_import_range : 2 ,
async_import_replaced_transactions : 1
fetch_and_import_range : 2
]
alias ABI.TypeDecoder
alias Ecto.Changeset
alias EthereumJSONRPC . { FetchedBalances , Subscription }
alias Explorer.Chain
alias Explorer.Chain.TokenTransfer
alias Explorer.Counters.AverageBlockTime
alias Indexer . { Block , TokenBalances , T racer }
alias Indexer . { Block , Tracer }
alias Indexer.Block.Realtime.TaskSupervisor
alias Indexer.Transform.Addresses
alias Timex.Duration
@ -160,42 +161,21 @@ defmodule Indexer.Block.Realtime.Fetcher do
@impl Block.Fetcher
def import (
block_fetcher ,
% Block.Fetcher { json_rpc_named_arguments : json_rpc_named_arguments } = block_fetcher ,
%{
address_coin_balances : %{ params : address_coin_balances_params } ,
address_hash_to_fetched_balance_block_number : address_hash_to_block_number ,
address_token_balances : %{ params : address_token_balances_params } ,
addresses : %{ params : addresses_params } ,
blocks : %{ params : blocks_params } ,
block_rewards : block_rewards ,
transactions : %{ params : transactions_params } ,
token_transfers : %{ params : token_transfers_params }
block_rewards : block_rewards
} = options
) do
with { :internal_transactions ,
{ :ok ,
%{
addresses_params : internal_transactions_addresses_params ,
internal_transactions_params : internal_transactions_params ,
internal_transactions_indexed_at_blocks_params : internal_transactions_indexed_at_blocks_params
} } } <-
{ :internal_transactions ,
internal_transactions ( block_fetcher , %{
addresses_params : addresses_params ,
blocks_params : blocks_params ,
token_transfers_params : token_transfers_params ,
transactions_params : transactions_params
} ) } ,
{ :balances , { :ok , %{ addresses_params : balances_addresses_params , balances_params : balances_params } } } <-
with { :balances , { :ok , %{ addresses_params : balances_addresses_params , balances_params : balances_params } } } <-
{ :balances ,
balances ( block_fetcher , %{
address_hash_to_block_number : address_hash_to_block_number ,
addresses_params : internal_transactions_ addresses_params,
addresses_params : addresses_params ,
balances_params : address_coin_balances_params
} ) } ,
{ :address_token_balances , { :ok , address_token_balances } } <-
{ :address_token_balances , fetch_token_balances ( address_token_balances_params ) } ,
address_current_token_balances = TokenBalances . to_address_current_token_balances ( address_token_balances ) ,
{ block_reward_errors , chain_import_block_rewards } = Map . pop ( block_rewards , :errors ) ,
chain_import_options =
options
@ -203,16 +183,14 @@ defmodule Indexer.Block.Realtime.Fetcher do
|> put_in ( [ :addresses , :params ] , balances_addresses_params )
|> put_in ( [ :blocks , :params , Access . all ( ) , :consensus ] , true )
|> put_in ( [ :block_rewards ] , chain_import_block_rewards )
|> put_in ( [ Access . key ( :address_coin_balances , %{ } ) , :params ] , balances_params )
|> put_in ( [ Access . key ( :address_current_token_balances , %{ } ) , :params ] , address_current_token_balances )
|> put_in ( [ Access . key ( :address_token_balances ) , :params ] , address_token_balances )
|> put_in ( [ Access . key ( :internal_transactions , %{ } ) , :params ] , internal_transactions_params )
|> put_in ( [ :internal_transactions_indexed_at_blocks ] , %{
params : internal_transactions_indexed_at_blocks_params ,
with : :number_only_changeset
} ) ,
|> put_in ( [ Access . key ( :address_coin_balances , %{ } ) , :params ] , balances_params ) ,
{ :import , { :ok , imported } = ok } <- { :import , Chain . import ( chain_import_options ) } do
async_import_remaining_block_data ( imported , %{ block_rewards : %{ errors : block_reward_errors } } )
async_import_remaining_block_data (
imported ,
%{ block_rewards : %{ errors : block_reward_errors } } ,
json_rpc_named_arguments
)
ok
end
end
@ -355,147 +333,20 @@ defmodule Indexer.Block.Realtime.Fetcher do
Enum . any? ( changesets , & ( Map . get ( &1 , :message ) == " Unknown block number " ) )
end
defp async_import_remaining_block_data ( imported , %{ block_rewards : %{ errors : block_reward_errors } } ) do
defp async_import_remaining_block_data (
imported ,
%{ block_rewards : %{ errors : block_reward_errors } } ,
json_rpc_named_arguments
) do
async_import_block_rewards ( block_reward_errors )
async_import_created_contract_codes ( imported )
async_import_internal_transactions ( imported , Keyword . get ( json_rpc_named_arguments , :variant ) )
async_import_tokens ( imported )
async_import_token_balances ( imported )
async_import_uncles ( imported )
async_import_replaced_transactions ( imported )
end
defp internal_transactions (
% Block.Fetcher { json_rpc_named_arguments : json_rpc_named_arguments } ,
%{
addresses_params : addresses_params ,
blocks_params : blocks_params ,
token_transfers_params : token_transfers_params ,
transactions_params : transactions_params
}
) do
variant = Keyword . fetch! ( json_rpc_named_arguments , :variant )
internal_transactions_indexed_at_blocks_params =
case variant do
EthereumJSONRPC.Parity -> blocks_params
_ -> [ ]
end
variant
|> case do
EthereumJSONRPC.Parity ->
blocks_params
|> Enum . map ( fn %{ number : block_number } -> block_number end )
|> EthereumJSONRPC . fetch_block_internal_transactions ( json_rpc_named_arguments )
_ ->
transactions_params
|> transactions_params_to_fetch_internal_transactions_params ( token_transfers_params , json_rpc_named_arguments )
|> EthereumJSONRPC . fetch_internal_transactions ( json_rpc_named_arguments )
end
|> case do
{ :ok , internal_transactions_params } ->
merged_addresses_params =
%{ internal_transactions : internal_transactions_params }
|> Addresses . extract_addresses ( )
|> Kernel . ++ ( addresses_params )
|> Addresses . merge_addresses ( )
{ :ok ,
%{
addresses_params : merged_addresses_params ,
internal_transactions_params : internal_transactions_params ,
internal_transactions_indexed_at_blocks_params : internal_transactions_indexed_at_blocks_params
} }
:ignore ->
{ :ok ,
%{
addresses_params : addresses_params ,
internal_transactions_params : [ ] ,
internal_transactions_indexed_at_blocks_params : [ ]
} }
{ :error , _reason } = error ->
error
end
end
defp transactions_params_to_fetch_internal_transactions_params (
transactions_params ,
token_transfers_params ,
json_rpc_named_arguments
) do
token_transfer_transaction_hash_set = MapSet . new ( token_transfers_params , & &1 . transaction_hash )
Enum . flat_map (
transactions_params ,
& transaction_params_to_fetch_internal_transaction_params_list (
&1 ,
token_transfer_transaction_hash_set ,
json_rpc_named_arguments
)
)
end
defp transaction_params_to_fetch_internal_transaction_params_list (
%{ block_number : block_number , transaction_index : transaction_index , hash : hash } = transaction_params ,
token_transfer_transaction_hash_set ,
json_rpc_named_arguments
)
when is_integer ( block_number ) and is_integer ( transaction_index ) and is_binary ( hash ) do
token_transfer? = hash in token_transfer_transaction_hash_set
if fetch_internal_transactions? ( transaction_params , token_transfer? , json_rpc_named_arguments ) do
[ %{ block_number : block_number , transaction_index : transaction_index , hash_data : hash } ]
else
[ ]
end
end
# 0xa9059cbb - signature of the transfer(address,uint256) function from the ERC-20 token specification.
# Although transaction input data can be faked we use this heuristics to filter simple token transfer internal transactions from indexing because they slow down realtime fetcher
defp fetch_internal_transactions? (
%{
status : :ok ,
created_contract_address_hash : nil ,
input : unquote ( TokenTransfer . transfer_function_signature ( ) ) <> params ,
value : 0
} ,
_ ,
_
) do
types = [ :address , { :uint , 256 } ]
try do
[ _address , _value ] =
params
|> Base . decode16! ( case : :mixed )
|> TypeDecoder . decode_raw ( types )
false
rescue
_ -> true
end
end
defp fetch_internal_transactions? (
%{
status : :ok ,
created_contract_address_hash : nil ,
input : " 0x " ,
to_address_hash : to_address_hash ,
block_number : block_number
} ,
_ ,
json_rpc_named_arguments
) do
Chain . contract_address? ( to_address_hash , block_number , json_rpc_named_arguments )
end
# Token transfers not transferred during contract creation don't need internal transactions as the token transfers
# derive completely from the logs.
defp fetch_internal_transactions? ( %{ status : :ok , created_contract_address_hash : nil } , true , _ ) , do : false
defp fetch_internal_transactions? ( _ , _ , _ ) , do : true
defp balances (
% Block.Fetcher { json_rpc_named_arguments : json_rpc_named_arguments } ,
%{ addresses_params : addresses_params } = options
@ -558,10 +409,4 @@ defmodule Indexer.Block.Realtime.Fetcher do
%{ hash_data : address_hash , block_quantity : integer_to_quantity ( block_number ) }
end )
end
defp fetch_token_balances ( address_token_balances_params ) do
address_token_balances_params
|> MapSet . to_list ( )
|> TokenBalances . fetch_token_balances_from_blockchain ( )
end
end
Ayrat Badykov
6 years ago
committed by
GitHub