Allow to setup multiple ranges of blocks to index (#5783)

* Add BLOCK_RANGES env var

* Allow to setup multiple ranges of blocks to index

* Add tests for BLOCK_RANGES

* Add CHANGELOG entry

Co-authored-by: Viktor Baranov <baranov.viktor.27@gmail.com>

CHANGELOG.md

@@ -1,6 +1,7 @@
 ## Current
 
 ### Features
+- [#5783](https://github.com/blockscout/blockscout/pull/5783) - Allow to setup multiple ranges of blocks to index
 
 ### Fixes
 

apps/indexer/lib/indexer/block/catchup/fetcher.ex

@@ -71,20 +71,24 @@ defmodule Indexer.Block.Catchup.Fetcher do
       ) do
     Logger.metadata(fetcher: :block_catchup)
 
-    with {:ok, latest_block_number} <- fetch_last_block(json_rpc_named_arguments) do
-      case latest_block_number do
-        # let realtime indexer get the genesis block
-        0 ->
+    with {:ok, ranges} <- block_ranges(json_rpc_named_arguments) do
+      case ranges do
+        # -1 means that latest block is 0, so let realtime indexer get the genesis block
+        [_..-1] ->
           %{first_block_number: 0, missing_block_count: 0, last_block_number: 0, shrunk: false}
 
         _ ->
           # realtime indexer gets the current latest block
-          first = latest_block_number - 1
-          last = last_block()
+          _..first = List.last(ranges)
+          last.._ = List.first(ranges)
 
           Logger.metadata(first_block_number: first, last_block_number: last)
 
-          missing_ranges = Chain.missing_block_number_ranges(first..last)
+          missing_ranges =
+            ranges
+            # let it fetch from newest to oldest block
+            |> Enum.reverse()
+            |> Enum.flat_map(fn f..l -> Chain.missing_block_number_ranges(l..f) end)
 
           range_count = Enum.count(missing_ranges)
 
@@ -347,6 +351,83 @@ defmodule Indexer.Block.Catchup.Fetcher do
     end
   end
 
+  @doc false
+  def block_ranges(json_rpc_named_arguments) do
+    block_ranges_string = Application.get_env(:indexer, :block_ranges)
+
+    ranges =
+      block_ranges_string
+      |> String.split(",")
+      |> Enum.map(fn string_range ->
+        case String.split(string_range, "..") do
+          [from_string, "latest"] ->
+            parse_integer(from_string)
+
+          [from_string, to_string] ->
+            with {from, ""} <- Integer.parse(from_string),
+                 {to, ""} <- Integer.parse(to_string) do
+              if from <= to, do: from..to, else: nil
+            else
+              _ -> nil
+            end
+
+          _ ->
+            nil
+        end
+      end)
+      |> sanitize_ranges()
+
+    case List.last(ranges) do
+      _from.._to ->
+        {:ok, ranges}
+
+      nil ->
+        with {:ok, latest_block_number} <- fetch_last_block(json_rpc_named_arguments) do
+          {:ok, [last_block()..(latest_block_number - 1)]}
+        end
+
+      num ->
+        with {:ok, latest_block_number} <-
+               EthereumJSONRPC.fetch_block_number_by_tag("latest", json_rpc_named_arguments) do
+          {:ok, List.update_at(ranges, -1, fn _ -> num..(latest_block_number - 1) end)}
+        end
+    end
+  end
+
+  defp sanitize_ranges(ranges) do
+    ranges
+    |> Enum.filter(&(not is_nil(&1)))
+    |> Enum.sort_by(
+      fn
+        from.._to -> from
+        el -> el
+      end,
+      :asc
+    )
+    |> Enum.chunk_while(
+      nil,
+      fn
+        _from.._to = chunk, nil ->
+          {:cont, chunk}
+
+        _ch_from..ch_to = chunk, acc_from..acc_to = acc ->
+          if Range.disjoint?(chunk, acc),
+            do: {:cont, acc, chunk},
+            else: {:cont, acc_from..max(ch_to, acc_to)}
+
+        num, nil ->
+          {:halt, num}
+
+        num, acc_from.._ = acc ->
+          if Range.disjoint?(num..num, acc), do: {:cont, acc, num}, else: {:halt, acc_from}
+
+        _, num ->
+          {:halt, num}
+      end,
+      fn reminder -> {:cont, reminder, nil} end
+    )
+  end
+
   defp last_block do
     string_value = Application.get_env(:indexer, :first_block)
 
@@ -366,8 +447,11 @@ defmodule Indexer.Block.Catchup.Fetcher do
 
   defp latest_block do
     string_value = Application.get_env(:indexer, :last_block)
+    parse_integer(string_value)
+  end
 
-    case Integer.parse(string_value) do
+  defp parse_integer(integer_string) do
+    case Integer.parse(integer_string) do
       {integer, ""} -> integer
       _ -> nil
     end

apps/indexer/test/indexer/block/catchup/fetcher_test.exs

@@ -572,6 +572,57 @@ defmodule Indexer.Block.Catchup.FetcherTest do
     end
   end
 
+  describe "block_ranges/0" do
+    setup do
+      initial_env = Application.get_all_env(:indexer)
+      on_exit(fn -> Application.put_all_env([{:indexer, initial_env}]) end)
+    end
+
+    test "ignores bad ranges", %{
+      json_rpc_named_arguments: json_rpc_named_arguments
+    } do
+      EthereumJSONRPC.Mox
+      |> expect(:json_rpc, fn %{method: "eth_getBlockByNumber", params: ["latest", false]}, _options ->
+        {:ok, %{"number" => "0x100"}}
+      end)
+
+      # doing such workaround is safe since this module is not async
+      Application.put_env(:indexer, :block_ranges, "1..5,3..5,2qw1..12,10..11a,,asd..qwe,10..latest")
+      # latest block is left for realtime_index
+      assert Fetcher.block_ranges(json_rpc_named_arguments) == {:ok, [1..5, 10..255]}
+    end
+
+    test "ignores FIRST_BLOCK/LAST_BLOCK when BLOCK_RANGES defined", %{
+      json_rpc_named_arguments: json_rpc_named_arguments
+    } do
+      Application.put_env(:indexer, :first_block, "1")
+      Application.put_env(:indexer, :last_block, "10")
+      Application.put_env(:indexer, :block_ranges, "2..5,10..100")
+      assert Fetcher.block_ranges(json_rpc_named_arguments) == {:ok, [2..5, 10..100]}
+    end
+
+    test "uses FIRST_BLOCK/LAST_BLOCK when BLOCK_RANGES is undefined or invalid", %{
+      json_rpc_named_arguments: json_rpc_named_arguments
+    } do
+      Application.put_env(:indexer, :first_block, "1")
+      Application.put_env(:indexer, :last_block, "10")
+      assert Fetcher.block_ranges(json_rpc_named_arguments) == {:ok, [1..9]}
+
+      Application.put_env(:indexer, :block_ranges, "latest..123,,fvdskvjglav!@#$%^&,2..1")
+      assert Fetcher.block_ranges(json_rpc_named_arguments) == {:ok, [1..9]}
+    end
+
+    test "all ranges are disjoint", %{json_rpc_named_arguments: json_rpc_named_arguments} do
+      EthereumJSONRPC.Mox
+      |> expect(:json_rpc, fn %{method: "eth_getBlockByNumber", params: ["latest", false]}, _options ->
+        {:ok, %{"number" => "0x100"}}
+      end)
+
+      Application.put_env(:indexer, :block_ranges, "10..20,5..15,18..25,35..40,30..50,100..latest,150..200")
+      assert Fetcher.block_ranges(json_rpc_named_arguments) == {:ok, [5..25, 30..50, 100..255]}
+    end
+  end
+
   defp count(schema) do
     Repo.one!(select(schema, fragment("COUNT(*)")))
   end

config/runtime.exs

@@ -327,6 +327,7 @@ config :indexer,
   block_transformer: block_transformer,
   metadata_updater_seconds_interval:
     String.to_integer(System.get_env("TOKEN_METADATA_UPDATE_INTERVAL") || "#{2 * 24 * 60 * 60}"),
+  block_ranges: System.get_env("BLOCK_RANGES") || "",
   first_block: System.get_env("FIRST_BLOCK") || "",
   last_block: System.get_env("LAST_BLOCK") || "",
   trace_first_block: System.get_env("TRACE_FIRST_BLOCK") || "",

docker-compose/envs/common-blockscout.env

@@ -45,6 +45,7 @@ BLOCKSCOUT_VERSION=
 RELEASE_LINK=
 BLOCK_TRANSFORMER=base
 # GRAPHIQL_TRANSACTION=
+# BLOCK_RANGES=
 # FIRST_BLOCK=
 # LAST_BLOCK=
 # TRACE_FIRST_BLOCK=

docker/Makefile

@@ -115,6 +115,9 @@ endif
 ifdef GRAPHIQL_TRANSACTION
 	BLOCKSCOUT_CONTAINER_PARAMS += -e 'GRAPHIQL_TRANSACTION=$(GRAPHIQL_TRANSACTION)'
 endif
+ifdef BLOCK_RANGES
+	BLOCKSCOUT_CONTAINER_PARAMS += -e 'BLOCK_RANGES=$(BLOCK_RANGES)'
+endif
 ifdef FIRST_BLOCK
 	BLOCKSCOUT_CONTAINER_PARAMS += -e 'FIRST_BLOCK=$(FIRST_BLOCK)'
 endif