Remove block consensus on import fail

CHANGELOG.md

@@ -14,6 +14,7 @@
 - [#8530](https://github.com/blockscout/blockscout/pull/8530) - Add `block_type` to search results
 - [#8180](https://github.com/blockscout/blockscout/pull/8180) - Deposits and Withdrawals for Polygon Edge
 - [#7996](https://github.com/blockscout/blockscout/pull/7996) - Add CoinBalance fetcher init query limit
+- [#8658](https://github.com/blockscout/blockscout/pull/8658) - Remove block consensus on import fail
 
 ### Fixes
 

apps/explorer/lib/explorer/chain/import.ex

@@ -321,6 +321,18 @@ defmodule Explorer.Chain.Import do
     runner_to_changes_list
     |> runner_to_changes_list_to_multis(options)
     |> logged_import(options)
+    |> case do
+      {:ok, result} ->
+        {:ok, result}
+
+      error ->
+        remove_consensus_from_partially_imported_blocks(options)
+        error
+    end
+  rescue
+    exception ->
+      remove_consensus_from_partially_imported_blocks(options)
+      reraise exception, __STACKTRACE__
   end
 
   defp logged_import(multis, options) when is_list(multis) and is_map(options) do
@@ -348,6 +360,15 @@ defmodule Explorer.Chain.Import do
     Repo.logged_transaction(multi, timeout: Map.get(options, :timeout, @transaction_timeout))
   end
 
+  defp remove_consensus_from_partially_imported_blocks(%{blocks: %{params: blocks_params}}) do
+    block_numbers = Enum.map(blocks_params, & &1.number)
+    Import.Runner.Blocks.invalidate_consensus_blocks(block_numbers)
+
+    Logger.warning("Consensus removed from partially imported block because of error: #{inspect(block_numbers)}")
+  end
+
+  defp remove_consensus_from_partially_imported_blocks(_options), do: :ok
+
   @spec timestamps() :: timestamps
   def timestamps do
     now = DateTime.utc_now()

apps/explorer/test/explorer/chain/import_test.exs

@@ -350,6 +350,19 @@ defmodule Explorer.Chain.ImportTest do
               }} = Import.all(@import_data)
     end
 
+    test "block consensus removed if there was an exception in further steps" do
+      not_existing_block_hash = "0xf6b4b8c88df3ebd252ec476328334dc026cf66606a84fb769b3d3cbccc8471db"
+
+      incorrect_data =
+        update_in(@import_data, [:transactions, :params], fn params ->
+          [params |> Enum.at(0) |> Map.put(:block_hash, not_existing_block_hash)]
+        end)
+
+      assert_raise(Postgrex.Error, fn -> Import.all(incorrect_data) end)
+      assert [] = Repo.all(Transaction)
+      assert %{consensus: false} = Repo.one(Block)
+    end
+
     test "inserts a token_balance" do
       params = %{
         addresses: %{

apps/indexer/lib/indexer/fetcher/empty_blocks_sanitizer.ex

@@ -41,9 +41,11 @@ defmodule Indexer.Fetcher.EmptyBlocksSanitizer do
 
   @impl GenServer
   def init(opts) when is_list(opts) do
+    interval = Application.get_env(:indexer, __MODULE__)[:interval]
+
     state = %__MODULE__{
       json_rpc_named_arguments: Keyword.fetch!(opts, :json_rpc_named_arguments),
-      interval: opts[:interval] || @interval
+      interval: interval || @interval
     }
 
     Process.send_after(self(), :sanitize_empty_blocks, state.interval)

config/runtime.exs

@@ -524,7 +524,8 @@ config :indexer, Indexer.Fetcher.TokenInstance.LegacySanitize.Supervisor,
   disabled?: ConfigHelper.parse_bool_env_var("INDEXER_DISABLE_TOKEN_INSTANCE_LEGACY_SANITIZE_FETCHER", "true")
 
 config :indexer, Indexer.Fetcher.EmptyBlocksSanitizer,
-  batch_size: ConfigHelper.parse_integer_env_var("INDEXER_EMPTY_BLOCKS_SANITIZER_BATCH_SIZE", 100)
+  batch_size: ConfigHelper.parse_integer_env_var("INDEXER_EMPTY_BLOCKS_SANITIZER_BATCH_SIZE", 100),
+  interval: ConfigHelper.parse_time_env_var("INDEXER_EMPTY_BLOCKS_SANITIZER_INTERVAL", "5m")
 
 config :indexer, Indexer.Block.Realtime.Fetcher,
   max_gap: ConfigHelper.parse_integer_env_var("INDEXER_REALTIME_FETCHER_MAX_GAP", 1000)

docker-compose/envs/common-blockscout.env

@@ -107,6 +107,7 @@ INDEXER_DISABLE_INTERNAL_TRANSACTIONS_FETCHER=false
 # INDEXER_CATCHUP_BLOCKS_BATCH_SIZE=
 # INDEXER_CATCHUP_BLOCKS_CONCURRENCY=
 # INDEXER_CATCHUP_BLOCK_INTERVAL=
+# INDEXER_EMPTY_BLOCKS_SANITIZER_INTERVAL=
 # INDEXER_INTERNAL_TRANSACTIONS_BATCH_SIZE=
 # INDEXER_INTERNAL_TRANSACTIONS_CONCURRENCY=
 # INDEXER_BLOCK_REWARD_BATCH_SIZE=