@ -2,23 +2,26 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.HistoricalMessagesOnL2 do
@moduledoc """ @moduledoc """
Handles the discovery and processing of historical messages between Layer 1 ( L1 ) and Layer 2 ( L2 ) within an Arbitrum rollup . Handles the discovery and processing of historical messages between Layer 1 ( L1 ) and Layer 2 ( L2 ) within an Arbitrum rollup .
L1 - to - L2 messages are discovered by requesting rollup transactions through RPC . ## L1-to-L2 Messages
This is necessary because some Arbitrum - specific fields are not included in the L1 - to - L2 messages are discovered by first inspecting the database to identify
already indexed transactions within the database . potentially missed messages . Then , rollup transactions are requested through RPC
to fetch the necessary data . This is required because some Arbitrum - specific fields ,
such as the ` requestId ` , are not included in the already indexed transactions within
the database .
## L2-to-L1 Messages
L2 - to - L1 messages are discovered by analyzing the logs of already indexed rollup L2 - to - L1 messages are discovered by analyzing the logs of already indexed rollup
transactions . transactions .
""" """
import Indexer.Fetcher.Arbitrum.Utils.Logging , only : [ log_warning : 1 , log_info : 1 ] import Indexer.Fetcher.Arbitrum.Utils.Logging , only : [ log_warning : 1 , log_info : 1 , log_debug : 1 ]
alias EthereumJSONRPC.Block.ByNumber , as : BlockByNumber
alias EthereumJSONRPC.Transaction , as : TransactionByRPC alias EthereumJSONRPC.Transaction , as : TransactionByRPC
alias Explorer.Chain alias Explorer.Chain
alias Indexer.Fetcher.Arbitrum.Messaging alias Indexer.Fetcher.Arbitrum.Messaging
alias Indexer.Fetcher.Arbitrum.Utils . { Db , Logging , Rpc } alias Indexer.Fetcher.Arbitrum.Utils . { Db , Rpc }
require Logger require Logger
@ -34,25 +37,31 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.HistoricalMessagesOnL2 do
## Parameters ## Parameters
- ` end_block ` : The ending block number up to which the discovery should occur . - ` end_block ` : The ending block number up to which the discovery should occur .
If ` nil ` or lesser than the indexer first block , the function If ` nil ` or less than the indexer ' s first block, the function returns with no
returns with no action taken . action taken .
- ` state ` : Contains the operational configuration , including the depth of - ` state ` : Contains the operational configuration , including the depth of
blocks to consider for the starting point of message discovery . blocks to consider for the starting point of message discovery and the
first block of the rollup chain .
## Returns ## Returns
- ` { :ok , nil } ` : If ` end_block ` is ` nil ` , indicating no discovery action was required . - ` { :ok , nil } ` : If ` end_block ` is ` nil ` , indicating no discovery action was
- ` { :ok , rollup_first_block } ` : If ` end_block ` is lesser than the indexer first block , required .
indicating that the " genesis " of the block chain was reached . - ` { :ok , rollup_first_block } ` : If ` end_block ` is less than the indexer ' s first
block , indicating that the " genesis " of the blockchain was reached .
- ` { :ok , start_block } ` : Upon successful discovery of historical messages , where - ` { :ok , start_block } ` : Upon successful discovery of historical messages , where
` start_block ` indicates the necessity to consider another block range in the next ` start_block ` indicates the necessity to consider another block range in the
iteration of message discovery . next iteration of message discovery .
- ` { :ok , end_block + 1 } ` : If the required block range is not fully indexed , - ` { :ok , end_block + 1 } ` : If the required block range is not fully indexed ,
indicating that the next iteration of message discovery should start with the same indicating that the next iteration of message discovery should start with the
block range . same block range .
""" """
@spec discover_historical_messages_from_l2 ( nil | integer ( ) , %{ @spec discover_historical_messages_from_l2 ( nil | integer ( ) , %{
:config = > %{ :config = > %{
:messages_to_l2_blocks_depth = > non_neg_integer ( ) , :missed_messages_blocks_depth = > non_neg_integer ( ) ,
:rollup_rpc = > %{
:first_block = > non_neg_integer ( ) ,
optional ( any ( ) ) = > any ( )
} ,
optional ( any ( ) ) = > any ( ) optional ( any ( ) ) = > any ( )
} , } ,
optional ( any ( ) ) = > any ( ) optional ( any ( ) ) = > any ( )
@ -72,13 +81,13 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.HistoricalMessagesOnL2 do
end_block , end_block ,
%{ %{
config : %{ config : %{
messages_from_l2 _blocks_depth : messages_from_l2 _blocks_depth , missed_m essages_blocks_depth : missed_m essages_blocks_depth ,
rollup_rpc : %{ first_block : rollup_first_block } rollup_rpc : %{ first_block : rollup_first_block }
} }
} = _state } = _state
) )
when is_integer ( end_block ) do when is_integer ( end_block ) do
start_block = max ( rollup_first_block , end_block - messages_from_l2 _blocks_depth + 1 ) start_block = max ( rollup_first_block , end_block - missed_m essages_blocks_depth + 1 )
if Db . indexed_blocks? ( start_block , end_block ) do if Db . indexed_blocks? ( start_block , end_block ) do
do_discover_historical_messages_from_l2 ( start_block , end_block ) do_discover_historical_messages_from_l2 ( start_block , end_block )
@ -107,10 +116,11 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.HistoricalMessagesOnL2 do
# ## Returns # ## Returns
# - `{:ok, start_block}`: A tuple indicating successful processing, returning the initial # - `{:ok, start_block}`: A tuple indicating successful processing, returning the initial
# starting block number. # starting block number.
@spec do_discover_historical_messages_from_l2 ( non_neg_integer ( ) , non_neg_integer ( ) ) :: { :ok , non_neg_integer ( ) }
defp do_discover_historical_messages_from_l2 ( start_block , end_block ) do defp do_discover_historical_messages_from_l2 ( start_block , end_block ) do
log_info ( " Block range for discovery historical messages from L2: #{ start_block } .. #{ end_block } " ) log_info ( " Block range for discovery historical messages from L2: #{ start_block } .. #{ end_block } " )
logs = Db . l2_to_l1_logs ( start_block , end_block ) logs = Db . logs_for_missed_messages_from_l2 ( start_block , end_block )
unless logs == [ ] do unless logs == [ ] do
messages = messages =
@ -126,35 +136,40 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.HistoricalMessagesOnL2 do
@doc """ @doc """
Initiates the discovery of historical messages sent from L1 to L2 up to a specified block number . Initiates the discovery of historical messages sent from L1 to L2 up to a specified block number .
This function orchestrates the process of discovering historical L1 - to - L2 messages within This function orchestrates the process of discovering historical L1 - to - L2
a given rollup block range , based on the existence of the ` requestId ` field in the rollup messages within a given rollup block range , based on the existence of the
transaction body . Transactions are requested through RPC because already indexed ` requestId ` field in the rollup transaction body . The initial list of
transactions from the database cannot be utilized ; the ` requestId ` field is not included transactions that could contain the messages is received from the database , and
in the transaction model . The function ensures that the block range has been indexed then their bodies are re - requested through RPC because already indexed
before proceeding with message discovery and import . The imported messages are marked as transactions from the database cannot be utilized ; the ` requestId ` field is not
` :relayed ` , as they represent completed actions from L1 to L2 . included in the transaction model . The function ensures that the block range
has been indexed before proceeding with message discovery and import . The
imported messages are marked as ` :relayed ` , as they represent completed actions
from L1 to L2 .
## Parameters ## Parameters
- ` end_block ` : The ending block number for the discovery operation . - ` end_block ` : The ending block number for the discovery operation . If ` nil ` or
If ` nil ` or lesser than the indexer first block , the function less than the indexer ' s first block, the function returns with no action
returns with no action taken . taken .
- ` state ` : The current state of the operation , containing configuration parameters - ` state ` : The current state of the operation , containing configuration
including ` messages_to_l2_blocks_depth ` , ` chunk_size ` , and JSON RPC connection parameters including the depth of blocks to consider for the starting point
settings . of message discovery , size of chunk to make request to RPC , and JSON RPC
connection settings .
## Returns ## Returns
- ` { :ok , nil } ` : If ` end_block ` is ` nil ` , indicating no action was necessary . - ` { :ok , nil } ` : If ` end_block ` is ` nil ` , indicating no action was necessary .
- ` { :ok , rollup_first_block } ` : If ` end_block ` is lesser than the indexer first block , - ` { :ok , rollup_first_block } ` : If ` end_block ` is less than the indexer ' s first
indicating that the " genesis " of the block chain was reached . block , indicating that the " genesis " of the blockchain was reached .
- ` { :ok , start_block } ` : On successful completion of historical message discovery , where - ` { :ok , start_block } ` : On successful completion of historical message
` start_block ` indicates the necessity to consider another block range in the next discovery , where ` start_block ` indicates the necessity to consider another
iteration of message discovery . block range in the next iteration of message discovery .
- ` { :ok , end_block + 1 } ` : If the required block range is not fully indexed , indicating - ` { :ok , end_block + 1 } ` : If the required block range is not fully indexed ,
that the next iteration of message discovery should start with the same block range . indicating that the next iteration of message discovery should start with the
same block range .
""" """
@spec discover_historical_messages_to_l2 ( nil | integer ( ) , %{ @spec discover_historical_messages_to_l2 ( nil | integer ( ) , %{
:config = > %{ :config = > %{
:messages_to_l2 _blocks_depth = > non_neg_integer ( ) , :missed_m essages_blocks_depth = > non_neg_integer ( ) ,
:rollup_rpc = > %{ :rollup_rpc = > %{
:chunk_size = > non_neg_integer ( ) , :chunk_size = > non_neg_integer ( ) ,
:json_rpc_named_arguments = > EthereumJSONRPC . json_rpc_named_arguments ( ) , :json_rpc_named_arguments = > EthereumJSONRPC . json_rpc_named_arguments ( ) ,
@ -177,10 +192,10 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.HistoricalMessagesOnL2 do
def discover_historical_messages_to_l2 ( def discover_historical_messages_to_l2 (
end_block , end_block ,
%{ config : %{ messages_to_l2 _blocks_depth : _ , rollup_rpc : %{ first_block : _ } } = config } = _state %{ config : %{ missed_m essages_blocks_depth : _ , rollup_rpc : %{ first_block : _ } } = config } = _state
) )
when is_integer ( end_block ) do when is_integer ( end_block ) do
start_block = max ( config . rollup_rpc . first_block , end_block - config . messages_to_l2 _blocks_depth + 1 ) start_block = max ( config . rollup_rpc . first_block , end_block - config . missed_m essages_blocks_depth + 1 )
# Although indexing blocks is not necessary to determine the completion of L1-to-L2 messages, # Although indexing blocks is not necessary to determine the completion of L1-to-L2 messages,
# for database consistency, it is preferable to delay marking these messages as completed. # for database consistency, it is preferable to delay marking these messages as completed.
@ -195,22 +210,36 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.HistoricalMessagesOnL2 do
end end
end end
# The function iterates through the block range in chunks, making RPC calls to fetch rollup block # Discovers and processes historical messages sent from L1 to L2 within a
# data and extract transactions. Each transaction is filtered for L1-to-L2 messages based on # specified rollup block range.
# existence of `requestId` field in the transaction body, and then imported into the database. #
# The imported messages are marked as `:relayed` as they represent completed actions from L1 to L2. # This function identifies which of already indexed transactions within the
# block range contains L1-to-L2 messages and makes RPC calls to fetch
# transaction data. These transactions are then processed to construct proper
# message structures, which are imported into the database. The imported
# messages are marked as `:relayed` as they represent completed actions from L1
# to L2.
# #
# Already indexed transactions from the database cannot be used because the `requestId` field is # Note: Already indexed transactions from the database cannot be used because
# not included in the transaction model. # the `requestId` field is not included in the transaction model.
# #
# ## Parameters # ## Parameters
# - `start_block`: The starting block number for the discovery range. # - `start_block`: The starting block number for the discovery range.
# - `end_block`: The ending block number for the discovery range. # - `end_block`: The ending block number for the discovery range.
# - `config`: The configuration map containing settings for RPC communication and chunk size. # - `config`: The configuration map containing settings for RPC communication
# and chunk size.
# #
# ## Returns # ## Returns
# - `{:ok, start_block}`: A tuple indicating successful processing, returning the initial # - `{:ok, start_block}`: A tuple indicating successful processing, returning
# starting block number. # the initial starting block number.
@spec do_discover_historical_messages_to_l2 ( non_neg_integer ( ) , non_neg_integer ( ) , %{
:rollup_rpc = > %{
:chunk_size = > non_neg_integer ( ) ,
:json_rpc_named_arguments = > EthereumJSONRPC . json_rpc_named_arguments ( ) ,
optional ( any ( ) ) = > any ( )
} ,
optional ( any ( ) ) = > any ( )
} ) :: { :ok , non_neg_integer ( ) }
defp do_discover_historical_messages_to_l2 ( defp do_discover_historical_messages_to_l2 (
start_block , start_block ,
end_block , end_block ,
@ -218,68 +247,56 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.HistoricalMessagesOnL2 do
) do ) do
log_info ( " Block range for discovery historical messages to L2: #{ start_block } .. #{ end_block } " ) log_info ( " Block range for discovery historical messages to L2: #{ start_block } .. #{ end_block } " )
{ messages , _ } = transactions = Db . transactions_for_missed_messages_to_l2 ( start_block , end_block )
start_block . . end_block transactions_length = length ( transactions )
|> Enum . chunk_every ( chunk_size )
|> Enum . reduce ( { [ ] , 0 } , fn chunk , { messages_acc , chunks_counter } -> if transactions_length > 0 do
Logging . log_details_chunk_handling ( log_debug ( " #{ transactions_length } historical messages to L2 discovered " )
" Collecting rollup data " ,
{ " block " , " blocks " } ,
chunk ,
chunks_counter ,
end_block - start_block + 1
)
messages =
transactions
|> Enum . chunk_every ( chunk_size )
|> Enum . reduce ( [ ] , fn chunk , messages_acc ->
# Since DB does not contain the field RequestId specific to Arbitrum # Since DB does not contain the field RequestId specific to Arbitrum
# all transactions will be requested from the rollup RPC endpoint. # all transactions will be requested from the rollup RPC endpoint.
# The catchup process intended to be run once and only for the BS instance # The catchup process intended to be run once and only for the BS instance
# which are already exist, so it does not make sense to introduce # which are already exist, so it does not make sense to introduce
# the new field in DB # the new field in DB
requests = build_block_by_number _requests ( chunk ) requests = build_transaction _requests ( chunk )
messages = messages =
requests requests
|> Rpc . make_chunked_request ( json_rpc_named_arguments , " eth_getBlockByNumber " ) |> Rpc . make_chunked_request ( json_rpc_named_arguments , " eth_getTransactionByHash " )
|> get_transactions ( ) |> Enum . map ( & transaction_json_to_map / 1 )
|> Enum . map ( fn tx ->
tx
|> TransactionByRPC . to_elixir ( )
|> TransactionByRPC . elixir_to_params ( )
end )
|> Messaging . filter_l1_to_l2_messages ( false ) |> Messaging . filter_l1_to_l2_messages ( false )
{ messages ++ messages_acc , chunks_counter + length ( chunk ) } messages ++ messages_acc
end ) end )
unless messages == [ ] do
log_info ( " #{ length ( messages ) } completions of L1-to-L2 messages will be imported " ) log_info ( " #{ length ( messages ) } completions of L1-to-L2 messages will be imported " )
end
import_to_db ( messages ) import_to_db ( messages )
end
{ :ok , start_block } { :ok , start_block }
end end
# Constructs a list of `eth_getBlockByNumber` requests for a given list of block number s. # Constructs a list of `eth_getTransactionByHash` requests for a given list of transaction hashe s.
defp build_block_by_number_requests ( block_number s ) do defp build_transaction_requests ( tx_hashe s ) do
block_number s tx_hashe s
|> Enum . reduce ( [ ] , fn block_num , requests_list -> |> Enum . reduce ( [ ] , fn tx_hash , requests_list ->
[ [
BlockByNumber . request ( %{ Rpc . transaction_by_hash_request ( %{ id : 0 , hash : tx_hash } )
id : block_num ,
number : block_num
} )
| requests_list | requests_list
] ]
end ) end )
end end
# Aggregates transactions from a list of blocks, combining them into a single list . # Transforms a JSON transaction object into a map .
defp get_transactions ( blocks_by_rpc ) do @spec transaction_json_to_map ( %{ String . t ( ) = > any ( ) } ) :: map ( )
blocks_by_rpc defp transaction_json_to_map ( transaction_json ) do
|> Enum . reduce ( [ ] , fn block_by_rpc , txs -> transaction_json
block_by_rpc [ " transactions " ] ++ txs |> TransactionByRPC . to_elixir ( )
end ) |> TransactionByRPC . elixir_to_params ( )
end end
# Imports a list of messages into the database. # Imports a list of messages into the database.
Alexander Kolotov
4 months ago
committed by
GitHub