feat: missing Arbitrum batches re-discovery (#10446)

* functionality to re-discover missed batches

* Proper request for safe block

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

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

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`
   """

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}")
+
+      # 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,
-        start_block,
-        end_block,
+          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,

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")

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=