Merge pull request #4823 from blockscout/vb-extended-error-handlers

Various error handlers with unresponsive JSON RPC endpoint

CHANGELOG.md

@@ -58,6 +58,7 @@
 - [#4582](https://github.com/blockscout/blockscout/pull/4582) - Fix NaN input on write contract page
 
 ### Chore
+- [#4823](https://github.com/blockscout/blockscout/pull/4823) - Various error handlers with unresponsive JSON RPC endpoint
 - [#4821](https://github.com/blockscout/blockscout/pull/4821) - Block Details page: Remove crossing at the Burnt Fee line
 - [#4819](https://github.com/blockscout/blockscout/pull/4819) - Add config for GasUsage Cache
 - [#4735](https://github.com/blockscout/blockscout/pull/4735) - Code clean up: Remove clauses for outdated ganache bugs

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

@@ -262,6 +262,19 @@ defmodule Indexer.Block.Catchup.BoundIntervalSupervisor do
     {:noreply, %__MODULE__{state | task: nil}}
   end
 
+  def handle_info(
+        {ref, {:error, :etimedout}},
+        %__MODULE__{
+          task: %Task{ref: ref}
+        } = state
+      ) do
+    Logger.info("Index had to catch up, but the but request is timing out, so retrying immediately.")
+
+    send(self(), :catchup_index)
+
+    {:noreply, %__MODULE__{state | task: nil}}
+  end
+
   def handle_info(
         {:DOWN, ref, :process, pid, reason},
         %__MODULE__{task: %Task{pid: pid, ref: ref}} = state
@@ -272,4 +285,15 @@ defmodule Indexer.Block.Catchup.BoundIntervalSupervisor do
 
     {:noreply, %__MODULE__{state | task: nil}}
   end
+
+  def handle_info(
+        {:DOWN, _ref, :process, _pid, reason},
+        %__MODULE__{task: nil} = state
+      ) do
+    Logger.error(fn -> "Catchup index stream exited with reason (#{inspect(reason)}). Restarting" end)
+
+    send(self(), :catchup_index)
+
+    {:noreply, %__MODULE__{state | task: nil}}
+  end
 end

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

@@ -71,59 +71,67 @@ defmodule Indexer.Block.Catchup.Fetcher do
       ) do
     Logger.metadata(fetcher: :block_catchup)
 
-    {:ok, latest_block_number} =
-      case latest_block() do
-        nil ->
-          EthereumJSONRPC.fetch_block_number_by_tag("latest", json_rpc_named_arguments)
-
-        number ->
-          {:ok, number}
+    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 ->
+          %{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()
+
+          Logger.metadata(first_block_number: first, last_block_number: last)
+
+          missing_ranges = Chain.missing_block_number_ranges(first..last)
+
+          range_count = Enum.count(missing_ranges)
+
+          missing_block_count =
+            missing_ranges
+            |> Stream.map(&Enum.count/1)
+            |> Enum.sum()
+
+          Logger.debug(fn -> "Missed blocks in ranges." end,
+            missing_block_range_count: range_count,
+            missing_block_count: missing_block_count
+          )
+
+          shrunk =
+            case missing_block_count do
+              0 ->
+                false
+
+              _ ->
+                step = step(first, last, blocks_batch_size)
+                sequence_opts = put_memory_monitor([ranges: missing_ranges, step: step], state)
+                gen_server_opts = [name: @sequence_name]
+                {:ok, sequence} = Sequence.start_link(sequence_opts, gen_server_opts)
+                Sequence.cap(sequence)
+
+                stream_fetch_and_import(state, sequence)
+
+                Shrinkable.shrunk?(sequence)
+            end
+
+          %{
+            first_block_number: first,
+            last_block_number: last,
+            missing_block_count: missing_block_count,
+            shrunk: shrunk
+          }
       end
+    end
+  end
 
-    case latest_block_number do
-      # let realtime indexer get the genesis block
-      0 ->
-        %{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()
-
-        Logger.metadata(first_block_number: first, last_block_number: last)
-
-        missing_ranges = Chain.missing_block_number_ranges(first..last)
-
-        range_count = Enum.count(missing_ranges)
-
-        missing_block_count =
-          missing_ranges
-          |> Stream.map(&Enum.count/1)
-          |> Enum.sum()
-
-        Logger.debug(fn -> "Missed blocks in ranges." end,
-          missing_block_range_count: range_count,
-          missing_block_count: missing_block_count
-        )
-
-        shrunk =
-          case missing_block_count do
-            0 ->
-              false
-
-            _ ->
-              step = step(first, last, blocks_batch_size)
-              sequence_opts = put_memory_monitor([ranges: missing_ranges, step: step], state)
-              gen_server_opts = [name: @sequence_name]
-              {:ok, sequence} = Sequence.start_link(sequence_opts, gen_server_opts)
-              Sequence.cap(sequence)
-
-              stream_fetch_and_import(state, sequence)
-
-              Shrinkable.shrunk?(sequence)
-          end
+  defp fetch_last_block(json_rpc_named_arguments) do
+    case latest_block() do
+      nil ->
+        EthereumJSONRPC.fetch_block_number_by_tag("latest", json_rpc_named_arguments)
 
-        %{first_block_number: first, last_block_number: last, missing_block_count: missing_block_count, shrunk: shrunk}
+      number ->
+        {:ok, number}
     end
   end
 

apps/indexer/lib/indexer/fetcher/pending_transaction.ex

@@ -136,6 +136,11 @@ defmodule Indexer.Fetcher.PendingTransaction do
 
         :ok
 
+      {:error, :etimedout} ->
+        Logger.error("timeout")
+
+        :ok
+
       {:error, {:bad_gateway, _}} ->
         Logger.error("bad_gateway")