From 536960363ac16e5f6a8f4c2f01c435913f39d3af Mon Sep 17 00:00:00 2001 From: Alexander Kolotov Date: Wed, 24 Jul 2024 02:49:30 -0600 Subject: [PATCH] feat: missing Arbitrum batches re-discovery (#10446) * functionality to re-discover missed batches * Proper request for safe block --- .../lib/explorer/chain/arbitrum/reader.ex | 88 ++++++ .../arbitrum/tracking_batches_statuses.ex | 82 ++++- .../lib/indexer/fetcher/arbitrum/utils/db.ex | 128 ++++++++ .../fetcher/arbitrum/workers/new_batches.ex | 282 +++++++++++++++++- config/runtime.exs | 3 +- docker-compose/envs/common-blockscout.env | 1 + 6 files changed, 551 insertions(+), 33 deletions(-) diff --git a/apps/explorer/lib/explorer/chain/arbitrum/reader.ex b/apps/explorer/lib/explorer/chain/arbitrum/reader.ex index 1fd6623f26..62d5ce6810 100644 --- a/apps/explorer/lib/explorer/chain/arbitrum/reader.ex +++ b/apps/explorer/lib/explorer/chain/arbitrum/reader.ex @@ -1061,4 +1061,92 @@ defmodule Explorer.Chain.Arbitrum.Reader do keyset -> {:ok, {keyset.batch_number, keyset.data}} end end + + @doc """ + Retrieves the batch numbers of missing L1 batches within a specified range. + + This function constructs a query to find the batch numbers of L1 batches that + are missing within the given range of batch numbers. It uses a right join with + a generated series to identify batch numbers that do not exist in the + `arbitrum_l1_batches` table. + + ## Parameters + - `start_batch_number`: The starting batch number of the search range. + - `end_batch_number`: The ending batch number of the search range. + + ## Returns + - A list of batch numbers in ascending order that are missing within the specified range. + """ + @spec find_missing_batches(non_neg_integer(), non_neg_integer()) :: [non_neg_integer()] + def find_missing_batches(start_batch_number, end_batch_number) + when is_integer(start_batch_number) and is_integer(end_batch_number) and end_batch_number >= start_batch_number do + query = + from(batch in L1Batch, + right_join: + missing_range in fragment( + """ + ( + SELECT distinct b1.number + FROM generate_series((?)::integer, (?)::integer) AS b1(number) + WHERE NOT EXISTS + (SELECT 1 FROM arbitrum_l1_batches b2 WHERE b2.number=b1.number) + ORDER BY b1.number DESC + ) + """, + ^start_batch_number, + ^end_batch_number + ), + on: batch.number == missing_range.number, + select: missing_range.number, + order_by: missing_range.number, + distinct: missing_range.number + ) + + query + |> Repo.all(timeout: :infinity) + end + + @doc """ + Retrieves L1 block numbers for the given list of batch numbers. + + This function finds the numbers of L1 blocks that include L1 transactions + associated with batches within the specified list of batch numbers. + + ## Parameters + - `batch_numbers`: A list of batch numbers for which to retrieve the L1 block numbers. + + ## Returns + - A map where the keys are batch numbers and the values are corresponding L1 block numbers. + """ + @spec get_l1_blocks_of_batches_by_numbers([non_neg_integer()]) :: %{non_neg_integer() => FullBlock.block_number()} + def get_l1_blocks_of_batches_by_numbers(batch_numbers) when is_list(batch_numbers) do + query = + from(batch in L1Batch, + join: l1tx in assoc(batch, :commitment_transaction), + where: batch.number in ^batch_numbers, + select: {batch.number, l1tx.block_number} + ) + + query + |> Repo.all(timeout: :infinity) + |> Enum.reduce(%{}, fn {batch_number, l1_block_number}, acc -> + Map.put(acc, batch_number, l1_block_number) + end) + end + + @doc """ + Retrieves the minimum and maximum batch numbers of L1 batches. + + ## Returns + - A tuple containing the minimum and maximum batch numbers or `{nil, nil}` if no batches are found. + """ + @spec get_min_max_batch_numbers() :: {non_neg_integer(), non_neg_integer()} | {nil | nil} + def get_min_max_batch_numbers do + query = + from(batch in L1Batch, + select: {min(batch.number), max(batch.number)} + ) + + Repo.one(query) + end end diff --git a/apps/indexer/lib/indexer/fetcher/arbitrum/tracking_batches_statuses.ex b/apps/indexer/lib/indexer/fetcher/arbitrum/tracking_batches_statuses.ex index 05e101696e..2b90b6d523 100644 --- a/apps/indexer/lib/indexer/fetcher/arbitrum/tracking_batches_statuses.ex +++ b/apps/indexer/lib/indexer/fetcher/arbitrum/tracking_batches_statuses.ex @@ -1,11 +1,12 @@ defmodule Indexer.Fetcher.Arbitrum.TrackingBatchesStatuses do @moduledoc """ - Manages the tracking and updating of the statuses of rollup batches, confirmations, and cross-chain message executions for an Arbitrum rollup. + Manages the tracking and updating of the statuses of rollup batches, + confirmations, and cross-chain message executions for an Arbitrum rollup. This module orchestrates the workflow for discovering new and historical - batches of rollup transactions, confirmations of rollup blocks, and - executions of L2-to-L1 messages. It ensures the accurate tracking and - updating of the rollup process stages. + batches of rollup transactions, confirmations of rollup blocks, and executions + of L2-to-L1 messages. It ensures the accurate tracking and updating of the + rollup process stages. The fetcher's operation cycle begins with the `:init_worker` message, which establishes the initial state with the necessary configuration. @@ -14,12 +15,13 @@ defmodule Indexer.Fetcher.Arbitrum.TrackingBatchesStatuses do specific messages: - `:check_new_batches`: Discovers new batches of rollup transactions and updates their statuses. - - `:check_new_confirmations`: Identifies new confirmations of rollup blocks - to update their statuses. - - `:check_new_executions`: Finds new executions of L2-to-L1 messages to + - `:check_new_confirmations`: Identifies new confirmations of rollup blocks to update their statuses. + - `:check_new_executions`: Finds new executions of L2-to-L1 messages to update + their statuses. - `:check_historical_batches`: Processes historical batches of rollup transactions. + - `:check_missing_batches`: Inspects for missing batches of rollup transactions. - `:check_historical_confirmations`: Handles historical confirmations of rollup blocks. - `:check_historical_executions`: Manages historical executions of L2-to-L1 @@ -28,12 +30,12 @@ defmodule Indexer.Fetcher.Arbitrum.TrackingBatchesStatuses do transactions, confirming the blocks and messages involved. Discovery of rollup transaction batches is executed by requesting logs on L1 - that correspond to the `SequencerBatchDelivered` event emitted by the - Arbitrum `SequencerInbox` contract. + that correspond to the `SequencerBatchDelivered` event emitted by the Arbitrum + `SequencerInbox` contract. Discovery of rollup block confirmations is executed by requesting logs on L1 - that correspond to the `SendRootUpdated` event emitted by the Arbitrum - `Outbox` contract. + that correspond to the `SendRootUpdated` event emitted by the Arbitrum `Outbox` + contract. Discovery of the L2-to-L1 message executions occurs by requesting logs on L1 that correspond to the `OutBoxTransactionExecuted` event emitted by the @@ -90,6 +92,7 @@ defmodule Indexer.Fetcher.Arbitrum.TrackingBatchesStatuses do finalized_confirmations = config_tracker[:finalized_confirmations] confirmation_batches_depth = config_tracker[:confirmation_batches_depth] new_batches_limit = config_tracker[:new_batches_limit] + missing_batches_range = config_tracker[:missing_batches_range] node_interface_address = config_tracker[:node_interface_contract] Process.send(self(), :init_worker, []) @@ -113,6 +116,7 @@ defmodule Indexer.Fetcher.Arbitrum.TrackingBatchesStatuses do l1_start_block: l1_start_block, l1_rollup_init_block: l1_rollup_init_block, new_batches_limit: new_batches_limit, + missing_batches_range: missing_batches_range, messages_to_blocks_shift: messages_to_blocks_shift, confirmation_batches_depth: confirmation_batches_depth, node_interface_address: node_interface_address @@ -179,6 +183,8 @@ defmodule Indexer.Fetcher.Arbitrum.TrackingBatchesStatuses do new_executions_start_block = Db.l1_block_to_discover_latest_execution(l1_start_block) historical_executions_end_block = Db.l1_block_to_discover_earliest_execution(l1_start_block - 1) + {lowest_batch, missing_batches_end_batch} = Db.get_min_max_batch_numbers() + Process.send(self(), :check_new_batches, []) new_state = @@ -188,7 +194,8 @@ defmodule Indexer.Fetcher.Arbitrum.TrackingBatchesStatuses do Map.merge(state.config, %{ l1_start_block: l1_start_block, l1_outbox_address: outbox_address, - l1_sequencer_inbox_address: sequencer_inbox_address + l1_sequencer_inbox_address: sequencer_inbox_address, + lowest_batch: lowest_batch }) ) |> Map.put( @@ -200,7 +207,8 @@ defmodule Indexer.Fetcher.Arbitrum.TrackingBatchesStatuses do historical_confirmations_end_block: nil, historical_confirmations_start_block: nil, new_executions_start_block: new_executions_start_block, - historical_executions_end_block: historical_executions_end_block + historical_executions_end_block: historical_executions_end_block, + missing_batches_end_batch: missing_batches_end_batch }) ) @@ -320,8 +328,8 @@ defmodule Indexer.Fetcher.Arbitrum.TrackingBatchesStatuses do # status of the L2-to-L1 messages included in the corresponding rollup blocks is # also updated. # - # After processing, it immediately transitions to checking historical - # confirmations of rollup blocks by sending the `:check_historical_confirmations` + # After processing, it immediately transitions to inspecting for missing batches + # of rollup blocks by sending the `:check_missing_batches` # message. # # ## Parameters @@ -336,7 +344,7 @@ defmodule Indexer.Fetcher.Arbitrum.TrackingBatchesStatuses do def handle_info(:check_historical_batches, state) do {handle_duration, {:ok, start_block}} = :timer.tc(&NewBatches.discover_historical_batches/1, [state]) - Process.send(self(), :check_historical_confirmations, []) + Process.send(self(), :check_missing_batches, []) new_data = Map.merge(state.data, %{ @@ -347,6 +355,48 @@ defmodule Indexer.Fetcher.Arbitrum.TrackingBatchesStatuses do {:noreply, %{state | data: new_data}} end + # Initiates the process of inspecting for missing batches of rollup transactions. + # + # This function inspects the database for missing batches within the calculated + # batch range. If a missing batch is identified, the L1 block range to look up + # for the transaction that committed the batch is built based on the neighboring + # batches. Then logs within the block range are fetched to get the batch data. + # After discovery, the linkage between batches and the corresponding rollup + # blocks and transactions is built. The status of the L2-to-L1 messages included + # in the corresponding rollup blocks is also updated. + # + # After processing, it immediately transitions to checking historical + # confirmations of rollup blocks by sending the `:check_historical_confirmations` + # message. + # + # ## Parameters + # - `:check_missing_batches`: The message that triggers the function. + # - `state`: The current state of the fetcher, containing configuration and data + # needed for inspection of the missed batches. + # + # ## Returns + # - `{:noreply, new_state}`: Where `new_state` is updated with the new end batch + # for the next iteration of missing batches inspection. + @impl GenServer + def handle_info(:check_missing_batches, state) do + # At the moment of the very first fetcher running, no batches were found yet + new_data = + if is_nil(state.config.lowest_batch) do + state.data + else + {handle_duration, {:ok, start_batch}} = :timer.tc(&NewBatches.inspect_for_missing_batches/1, [state]) + + Map.merge(state.data, %{ + duration: increase_duration(state.data, handle_duration), + missing_batches_end_batch: start_batch - 1 + }) + end + + Process.send(self(), :check_historical_confirmations, []) + + {:noreply, %{state | data: new_data}} + end + # Initiates the process of discovering and handling historical confirmations of rollup blocks. # # This function fetches logs within the calculated range to identify the diff --git a/apps/indexer/lib/indexer/fetcher/arbitrum/utils/db.ex b/apps/indexer/lib/indexer/fetcher/arbitrum/utils/db.ex index 703b0693fe..1c304e623b 100644 --- a/apps/indexer/lib/indexer/fetcher/arbitrum/utils/db.ex +++ b/apps/indexer/lib/indexer/fetcher/arbitrum/utils/db.ex @@ -688,6 +688,134 @@ defmodule Indexer.Fetcher.Arbitrum.Utils.Db do |> Enum.map(&logs_to_map/1) end + @doc """ + Retrieves L1 block ranges that could be used to re-discover missing batches + within a specified range of batch numbers. + + This function identifies the L1 block ranges corresponding to missing L1 batches + within the given range of batch numbers. It first finds the missing batches, + then determines their neighboring ranges, and finally maps these ranges to the + corresponding L1 block numbers. + + ## Parameters + - `start_batch_number`: The starting batch number of the search range. + - `end_batch_number`: The ending batch number of the search range. + - `block_for_batch_0`: The L1 block number corresponding to the batch number 0. + + ## Returns + - A list of tuples, each containing a start and end L1 block number for the + ranges corresponding to the missing batches. + + ## Examples + + Example #1 + - Within the range from 1 to 10, the missing batch is 2. The L1 block for the + batch #1 is 10, and the L1 block for the batch #3 is 31. + - The output will be `[{11, 30}]`. + + Example #2 + - Within the range from 1 to 10, the missing batches are 2 and 6, and + - The L1 block for the batch #1 is 10. + - The L1 block for the batch #3 is 31. + - The L1 block for the batch #5 is 64. + - The L1 block for the batch #7 is 90. + - The output will be `[{11, 30}, {65, 89}]`. + + Example #3 + - Within the range from 1 to 10, the missing batches are 2 and 4, and + - The L1 block for the batch #1 is 10. + - The L1 block for the batch #3 is 31. + - The L1 block for the batch #5 is 64. + - The output will be `[{11, 30}, {32, 63}]`. + """ + @spec get_l1_block_ranges_for_missing_batches(non_neg_integer(), non_neg_integer(), FullBlock.block_number()) :: [ + {FullBlock.block_number(), FullBlock.block_number()} + ] + def get_l1_block_ranges_for_missing_batches(start_batch_number, end_batch_number, block_for_batch_0) + when is_integer(start_batch_number) and is_integer(end_batch_number) and end_batch_number >= start_batch_number do + # credo:disable-for-lines:4 Credo.Check.Refactor.PipeChainStart + neighbors_of_missing_batches = + Reader.find_missing_batches(start_batch_number, end_batch_number) + |> list_to_chunks() + |> chunks_to_neighbor_ranges() + + if neighbors_of_missing_batches == [] do + [] + else + l1_blocks = + neighbors_of_missing_batches + |> Enum.reduce(MapSet.new(), fn {start_batch, end_batch}, acc -> + acc + |> MapSet.put(start_batch) + |> MapSet.put(end_batch) + end) + # To avoid error in getting L1 block for the batch 0 + |> MapSet.delete(0) + |> MapSet.to_list() + |> Reader.get_l1_blocks_of_batches_by_numbers() + # It is safe to add the block for the batch 0 even if the batch 1 is missing + |> Map.put(0, block_for_batch_0) + + neighbors_of_missing_batches + |> Enum.map(fn {start_batch, end_batch} -> + {l1_blocks[start_batch] + 1, l1_blocks[end_batch] - 1} + end) + end + end + + # Splits a list into chunks of consecutive numbers, e.g., [1, 2, 3, 5, 6, 8] becomes [[1, 2, 3], [5, 6], [8]]. + @spec list_to_chunks([non_neg_integer()]) :: [[non_neg_integer()]] + defp list_to_chunks(list) do + chunk_fun = fn current, acc -> + case acc do + [] -> + {:cont, [current]} + + [last | _] = acc when current == last + 1 -> + {:cont, [current | acc]} + + acc -> + {:cont, Enum.reverse(acc), [current]} + end + end + + after_fun = fn acc -> + case acc do + # Special case to handle the situation when the initial list is empty + [] -> {:cont, []} + _ -> {:cont, Enum.reverse(acc), []} + end + end + + list + |> Enum.chunk_while([], chunk_fun, after_fun) + end + + # Converts chunks of elements into neighboring ranges, e.g., [[1, 2], [4]] becomes [{0, 3}, {3, 5}]. + @spec chunks_to_neighbor_ranges([[non_neg_integer()]]) :: [{non_neg_integer(), non_neg_integer()}] + defp chunks_to_neighbor_ranges([]), do: [] + + defp chunks_to_neighbor_ranges(list_of_chunks) do + list_of_chunks + |> Enum.map(fn current -> + case current do + [one_element] -> {one_element - 1, one_element + 1} + chunk -> {List.first(chunk) - 1, List.last(chunk) + 1} + end + end) + end + + @doc """ + Retrieves the minimum and maximum batch numbers of L1 batches. + + ## Returns + - A tuple containing the minimum and maximum batch numbers or `{nil, nil}` if no batches are found. + """ + @spec get_min_max_batch_numbers() :: {non_neg_integer(), non_neg_integer()} | {nil | nil} + def get_min_max_batch_numbers do + Reader.get_min_max_batch_numbers() + end + @doc """ Returns 32-byte signature of the event `L2ToL1Tx` """ diff --git a/apps/indexer/lib/indexer/fetcher/arbitrum/workers/new_batches.ex b/apps/indexer/lib/indexer/fetcher/arbitrum/workers/new_batches.ex index 8a222a8d6a..cf768c0416 100644 --- a/apps/indexer/lib/indexer/fetcher/arbitrum/workers/new_batches.ex +++ b/apps/indexer/lib/indexer/fetcher/arbitrum/workers/new_batches.ex @@ -44,6 +44,8 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.NewBatches do # keccak256("SequencerBatchDelivered(uint256,bytes32,bytes32,bytes32,uint256,(uint64,uint64,uint64,uint64),uint8)") @event_sequencer_batch_delivered "0x7394f4a19a13c7b92b5bb71033245305946ef78452f7b4986ac1390b5df4ebd7" + @max_depth_for_safe_block 1000 + @doc """ Discovers and imports new batches of rollup transactions within the current L1 block range. @@ -91,7 +93,11 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.NewBatches do :node_interface_address => binary(), optional(any()) => any() }, - :data => %{:new_batches_start_block => non_neg_integer(), optional(any()) => any()}, + :data => %{ + :new_batches_start_block => non_neg_integer(), + :historical_batches_end_block => non_neg_integer(), + optional(any()) => any() + }, optional(any()) => any() }) :: {:ok, non_neg_integer()} def discover_new_batches( @@ -104,7 +110,7 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.NewBatches do new_batches_limit: new_batches_limit, node_interface_address: node_interface_address }, - data: %{new_batches_start_block: start_block} + data: %{new_batches_start_block: start_block, historical_batches_end_block: historical_batches_end_block} } = _state ) do # Requesting the "latest" block instead of "safe" allows to catch new batches @@ -116,21 +122,62 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.NewBatches do Rpc.get_resend_attempts() ) + {safe_chain_block, _} = IndexerHelper.get_safe_block(l1_rpc_config.json_rpc_named_arguments) + + # max() cannot be used here since l1_rpc_config.logs_block_range must not + # be taken into account to identify if it is L3 or not + safe_block = + if safe_chain_block < latest_block + 1 - @max_depth_for_safe_block do + # The case of L3, the safe block is too far behind the latest block, + # therefore it is assumed that there is no so deep re-orgs there. + latest_block + 1 - min(@max_depth_for_safe_block, l1_rpc_config.logs_block_range) + else + safe_chain_block + end + + # It is necessary to re-visit some amount of the previous blocks to ensure that + # no batches are missed due to reorgs. The amount of blocks to re-visit depends + # either on the current safe block but must not exceed @max_depth_for_safe_block + # (or L1 RPC max block range for getting logs) since on L3 chains the safe block + # could be too far behind the latest block. + # At the same time it does not make sense to re-visit blocks that will be + # re-visited by the historical batches discovery process. + # If the new batches discovery process does not reach the chain head previously + # no need to re-visit the blocks. + safe_start_block = max(min(start_block, safe_block), historical_batches_end_block + 1) + end_block = min(start_block + l1_rpc_config.logs_block_range - 1, latest_block) - if start_block <= end_block do - log_info("Block range for new batches discovery: #{start_block}..#{end_block}") + if safe_start_block <= end_block do + log_info("Block range for new batches discovery: #{safe_start_block}..#{end_block}") - discover( - sequencer_inbox_address, - start_block, - end_block, - new_batches_limit, - messages_to_blocks_shift, - l1_rpc_config, - node_interface_address, - rollup_rpc_config - ) + # Since with taking the safe block into account, the range safe_start_block..end_block + # could be larger than L1 RPC max block range for getting logs, it is necessary to + # divide the range into the chunks + safe_start_block + |> Stream.unfold(fn + current when current > end_block -> + nil + + current -> + next = min(current + l1_rpc_config.logs_block_range - 1, end_block) + {current, next + 1} + end) + |> Stream.each(fn chunk_start -> + chunk_end = min(chunk_start + l1_rpc_config.logs_block_range - 1, end_block) + + discover( + sequencer_inbox_address, + chunk_start, + chunk_end, + new_batches_limit, + messages_to_blocks_shift, + l1_rpc_config, + node_interface_address, + rollup_rpc_config + ) + end) + |> Stream.run() {:ok, end_block} else @@ -187,7 +234,7 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.NewBatches do :node_interface_address => binary(), optional(any()) => any() }, - :data => %{:historical_batches_end_block => any(), optional(any()) => any()}, + :data => %{:historical_batches_end_block => non_neg_integer(), optional(any()) => any()}, optional(any()) => any() }) :: {:ok, non_neg_integer()} def discover_historical_batches( @@ -226,6 +273,107 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.NewBatches do end end + @doc """ + Inspects and imports missing batches within a specified range of batch numbers. + + This function first finds the missing batches, then determines their + neighboring ranges, maps these ranges to the corresponding L1 block numbers, + and for every such range it retrieves logs representing the + SequencerBatchDelivered events emitted by the SequencerInbox contract. + These logs are processed to identify the batches and their details. The + function then constructs comprehensive data structures for batches, + lifecycle transactions, rollup blocks, and rollup transactions. Additionally, + it identifies L2-to-L1 messages that have been committed within these batches + and updates their status. All discovered and processed data are then imported + into the database. + + ## Parameters + - A map containing: + - `config`: Configuration settings including the L1 rollup initialization block, + RPC configurations, SequencerInbox address, a shift for the message + to block number mapping, a limit for new batches discovery, and the + max size of the range for missing batches inspection. + - `data`: Contains the ending batch number for the missing batches inspection. + + ## Returns + - `{:ok, start_batch}`: On successful inspection of the given batch range, where + `start_batch` is the calculated starting batch for the inspected range, + indicating the need to consider another batch range in the next iteration of + missing batch inspection. + - `{:ok, lowest_batch}`: If the discovery process has been finished, indicating + that all batches up to the rollup origins have been checked and no further + action is needed. + """ + @spec inspect_for_missing_batches(%{ + :config => %{ + :l1_rollup_init_block => non_neg_integer(), + :l1_rpc => %{ + :json_rpc_named_arguments => EthereumJSONRPC.json_rpc_named_arguments(), + :logs_block_range => non_neg_integer(), + optional(any()) => any() + }, + :l1_sequencer_inbox_address => binary(), + :lowest_batch => non_neg_integer(), + :messages_to_blocks_shift => non_neg_integer(), + :missing_batches_range => non_neg_integer(), + :new_batches_limit => non_neg_integer(), + :node_interface_address => binary(), + :rollup_rpc => %{ + :json_rpc_named_arguments => EthereumJSONRPC.json_rpc_named_arguments(), + :chunk_size => non_neg_integer(), + optional(any()) => any() + }, + optional(any()) => any() + }, + :data => %{:missing_batches_end_batch => non_neg_integer(), optional(any()) => any()}, + optional(any()) => any() + }) :: {:ok, non_neg_integer()} + def inspect_for_missing_batches( + %{ + config: %{ + l1_rpc: l1_rpc_config, + rollup_rpc: rollup_rpc_config, + l1_sequencer_inbox_address: sequencer_inbox_address, + messages_to_blocks_shift: messages_to_blocks_shift, + l1_rollup_init_block: l1_rollup_init_block, + new_batches_limit: new_batches_limit, + missing_batches_range: missing_batches_range, + lowest_batch: lowest_batch, + node_interface_address: node_interface_address + }, + data: %{missing_batches_end_batch: end_batch} + } = _state + ) + when not is_nil(lowest_batch) and not is_nil(end_batch) do + # No need to inspect for missing batches below the lowest batch + # since it is assumed that they are picked up by historical batches + # discovery process + if end_batch > lowest_batch do + start_batch = max(lowest_batch, end_batch - missing_batches_range + 1) + + log_info("Batch range for missing batches inspection: #{start_batch}..#{end_batch}") + + l1_block_ranges_for_missing_batches = + Db.get_l1_block_ranges_for_missing_batches(start_batch, end_batch, l1_rollup_init_block - 1) + + unless l1_block_ranges_for_missing_batches == [] do + discover_missing_batches( + sequencer_inbox_address, + l1_block_ranges_for_missing_batches, + new_batches_limit, + messages_to_blocks_shift, + l1_rpc_config, + node_interface_address, + rollup_rpc_config + ) + end + + {:ok, start_batch} + else + {:ok, lowest_batch} + end + end + # Initiates the discovery process for batches within a specified block range. # # Invokes the actual discovery process for new batches by calling `do_discover` @@ -243,6 +391,24 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.NewBatches do # # ## Returns # - N/A + @spec discover( + binary(), + non_neg_integer(), + non_neg_integer(), + non_neg_integer(), + non_neg_integer(), + %{ + :json_rpc_named_arguments => EthereumJSONRPC.json_rpc_named_arguments(), + :chunk_size => non_neg_integer(), + optional(any()) => any() + }, + binary(), + %{ + :json_rpc_named_arguments => EthereumJSONRPC.json_rpc_named_arguments(), + :chunk_size => non_neg_integer(), + optional(any()) => any() + } + ) :: any() defp discover( sequencer_inbox_address, start_block, @@ -282,6 +448,24 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.NewBatches do # # ## Returns # - N/A + @spec discover_historical( + binary(), + non_neg_integer(), + non_neg_integer(), + non_neg_integer(), + non_neg_integer(), + %{ + :json_rpc_named_arguments => EthereumJSONRPC.json_rpc_named_arguments(), + :chunk_size => non_neg_integer(), + optional(any()) => any() + }, + binary(), + %{ + :json_rpc_named_arguments => EthereumJSONRPC.json_rpc_named_arguments(), + :chunk_size => non_neg_integer(), + optional(any()) => any() + } + ) :: any() defp discover_historical( sequencer_inbox_address, start_block, @@ -304,6 +488,72 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.NewBatches do ) end + # Initiates the discovery process for missing batches within specified block ranges. + # + # This function divides each L1 block range into chunks to call `discover_historical` + # for every chunk to discover missing batches. + # + # ## Parameters + # - `sequencer_inbox_address`: The SequencerInbox contract address. + # - `l1_block_ranges`: The L1 block ranges to look for missing batches. + # - `new_batches_limit`: Limit of new batches to process in one iteration. + # - `messages_to_blocks_shift`: Shift value for message to block number mapping. + # - `l1_rpc_config`: Configuration for L1 RPC calls. + # - `node_interface_address`: The address of the NodeInterface contract on the rollup. + # - `rollup_rpc_config`: Configuration for rollup RPC calls. + # + # ## Returns + # - N/A + @spec discover_missing_batches( + binary(), + [{non_neg_integer(), non_neg_integer()}], + non_neg_integer(), + non_neg_integer(), + %{ + :json_rpc_named_arguments => EthereumJSONRPC.json_rpc_named_arguments(), + :logs_block_range => non_neg_integer(), + :chunk_size => non_neg_integer(), + optional(any()) => any() + }, + binary(), + %{ + :json_rpc_named_arguments => EthereumJSONRPC.json_rpc_named_arguments(), + :chunk_size => non_neg_integer(), + optional(any()) => any() + } + ) :: :ok + defp discover_missing_batches( + sequencer_inbox_address, + l1_block_ranges, + new_batches_limit, + messages_to_blocks_shift, + l1_rpc_config, + node_interface_address, + rollup_rpc_config + ) do + Enum.each(l1_block_ranges, fn {start_block, end_block} -> + Enum.each(0..div(end_block - start_block, l1_rpc_config.logs_block_range), fn i -> + start_block = start_block + i * l1_rpc_config.logs_block_range + end_block = min(start_block + l1_rpc_config.logs_block_range - 1, end_block) + + log_info("Block range for missing batches discovery: #{start_block}..#{end_block}") + + # `do_discover` is not used here to demonstrate the need to fetch batches + # which are already historical + discover_historical( + sequencer_inbox_address, + start_block, + end_block, + new_batches_limit, + messages_to_blocks_shift, + l1_rpc_config, + node_interface_address, + rollup_rpc_config + ) + end) + end) + end + # Performs the discovery of new or historical batches within a specified block range, # processing and importing the relevant data into the database. # @@ -346,7 +596,7 @@ defmodule Indexer.Fetcher.Arbitrum.Workers.NewBatches do :chunk_size => non_neg_integer(), optional(any()) => any() } - ) :: :ok + ) :: any() defp do_discover( sequencer_inbox_address, start_block, diff --git a/config/runtime.exs b/config/runtime.exs index 3d972e2009..3c91ccbc4a 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -899,7 +899,8 @@ config :indexer, Indexer.Fetcher.Arbitrum.TrackingBatchesStatuses, finalized_confirmations: ConfigHelper.parse_bool_env_var("INDEXER_ARBITRUM_CONFIRMATIONS_TRACKING_FINALIZED", "true"), new_batches_limit: ConfigHelper.parse_integer_env_var("INDEXER_ARBITRUM_NEW_BATCHES_LIMIT", 10), node_interface_contract: - ConfigHelper.safe_get_env("INDEXER_ARBITRUM_NODE_INTERFACE_CONTRACT", "0x00000000000000000000000000000000000000C8") + ConfigHelper.safe_get_env("INDEXER_ARBITRUM_NODE_INTERFACE_CONTRACT", "0x00000000000000000000000000000000000000C8"), + missing_batches_range: ConfigHelper.parse_integer_env_var("INDEXER_ARBITRUM_MISSING_BATCHES_RANGE", 10000) config :indexer, Indexer.Fetcher.Arbitrum.TrackingBatchesStatuses.Supervisor, enabled: ConfigHelper.parse_bool_env_var("INDEXER_ARBITRUM_BATCHES_TRACKING_ENABLED") diff --git a/docker-compose/envs/common-blockscout.env b/docker-compose/envs/common-blockscout.env index e37f225202..f2dcbd3c46 100644 --- a/docker-compose/envs/common-blockscout.env +++ b/docker-compose/envs/common-blockscout.env @@ -239,6 +239,7 @@ INDEXER_DISABLE_INTERNAL_TRANSACTIONS_FETCHER=false # INDEXER_ARBITRUM_BATCHES_TRACKING_ENABLED= # INDEXER_ARBITRUM_BATCHES_TRACKING_RECHECK_INTERVAL= # INDEXER_ARBITRUM_NEW_BATCHES_LIMIT= +# INDEXER_ARBITRUM_MISSING_BATCHES_RANGE= # INDEXER_ARBITRUM_BATCHES_TRACKING_MESSAGES_TO_BLOCKS_SHIFT= # INDEXER_ARBITRUM_CONFIRMATIONS_TRACKING_FINALIZED= # INDEXER_ARBITRUM_BATCHES_TRACKING_L1_FINALIZATION_CHECK_ENABLED=