Discard child block with parent_hash not matching hash of imported block

CHANGELOG.md

@@ -12,6 +12,7 @@
  - [#1691](https://github.com/poanetwork/blockscout/pull/1691) - decrease token metadata update interval
  - [#1688](https://github.com/poanetwork/blockscout/pull/1688) - do not fail if failure reason is atom
  - [#1692](https://github.com/poanetwork/blockscout/pull/1692) - exclude decompiled smart contract from encoding
+ - [#1684](https://github.com/poanetwork/blockscout/pull/1684) - Discard child block with parent_hash not matching hash of imported block
 
 ### Chore
 

apps/explorer/lib/explorer/chain.ex

@@ -2648,24 +2648,6 @@ defmodule Explorer.Chain do
   @spec data() :: Dataloader.Ecto.t()
   def data, do: DataloaderEcto.new(Repo)
 
-  @doc """
-  Returns a list of block numbers with invalid consensus.
-  """
-  @spec list_block_numbers_with_invalid_consensus :: [integer()]
-  def list_block_numbers_with_invalid_consensus do
-    query =
-      from(
-        block in Block,
-        join: parent in Block,
-        on: parent.hash == block.parent_hash,
-        where: block.consensus == true,
-        where: parent.consensus == false,
-        select: parent.number
-      )
-
-    Repo.all(query, timeout: :infinity)
-  end
-
   def list_decompiled_contracts(limit, offset, not_decompiled_with_version \\ nil) do
     query =
       from(

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

@@ -46,7 +46,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
       |> Map.put(:timestamps, timestamps)
 
     ordered_consensus_block_numbers = ordered_consensus_block_numbers(changes_list)
-    where_invalid_parent = where_invalid_parent(changes_list)
+    where_invalid_neighbour = where_invalid_neighbour(changes_list)
     where_forked = where_forked(changes_list)
 
     multi
@@ -70,8 +70,8 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     |> Multi.run(:lose_consensus, fn repo, _ ->
       lose_consensus(repo, ordered_consensus_block_numbers, insert_options)
     end)
-    |> Multi.run(:lose_invalid_parent_consensus, fn repo, _ ->
-      lose_invalid_parent_consensus(repo, where_invalid_parent, insert_options)
+    |> Multi.run(:lose_invalid_neighbour_consensus, fn repo, _ ->
+      lose_invalid_neighbour_consensus(repo, where_invalid_neighbour, insert_options)
     end)
     |> Multi.run(:delete_address_token_balances, fn repo, _ ->
       delete_address_token_balances(repo, ordered_consensus_block_numbers, insert_options)
@@ -316,13 +316,13 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     end
   end
 
-  defp lose_invalid_parent_consensus(repo, where_invalid_parent, %{
+  defp lose_invalid_neighbour_consensus(repo, where_invalid_neighbour, %{
          timeout: timeout,
          timestamps: %{updated_at: updated_at}
        }) do
     query =
       from(
-        block in where_invalid_parent,
+        block in where_invalid_neighbour,
         update: [
           set: [
             consensus: false,
@@ -338,7 +338,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
       {:ok, result}
     rescue
       postgrex_error in Postgrex.Error ->
-        {:error, %{exception: postgrex_error, where_invalid_parent: where_invalid_parent}}
+        {:error, %{exception: postgrex_error, where_invalid_neighbour: where_invalid_neighbour}}
     end
   end
 
@@ -581,12 +581,22 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     end)
   end
 
-  defp where_invalid_parent(blocks_changes) when is_list(blocks_changes) do
+  defp where_invalid_neighbour(blocks_changes) when is_list(blocks_changes) do
     initial = from(b in Block, where: false)
 
-    Enum.reduce(blocks_changes, initial, fn %{consensus: consensus, parent_hash: parent_hash, number: number}, acc ->
+    Enum.reduce(blocks_changes, initial, fn %{
+                                              consensus: consensus,
+                                              hash: hash,
+                                              parent_hash: parent_hash,
+                                              number: number
+                                            },
+                                            acc ->
       if consensus do
-        from(block in acc, or_where: block.number == ^(number - 1) and block.hash != ^parent_hash)
+        from(
+          block in acc,
+          or_where: block.number == ^(number - 1) and block.hash != ^parent_hash,
+          or_where: block.number == ^(number + 1) and block.parent_hash != ^hash
+        )
       else
         acc
       end

apps/explorer/test/explorer/chain/import/runner/blocks_test.exs

@@ -261,34 +261,36 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
     end
 
     # Regression test for https://github.com/poanetwork/blockscout/issues/1644
-    test "discards parent block if it isn't related to the current one because of reorg",
+    test "discards neighbouring blocks if they aren't related to the current one because of reorg and/or import timeout",
          %{consensus_block: %Block{number: block_number, hash: block_hash, miner_hash: miner_hash}, options: options} do
-      old_block = insert(:block, parent_hash: block_hash, number: block_number + 1)
-      insert(:block, parent_hash: old_block.hash, number: old_block.number + 1)
+      old_block1 = params_for(:block, miner_hash: miner_hash, parent_hash: block_hash, number: block_number + 1)
 
-      new_block1 = params_for(:block, parent_hash: block_hash, number: block_number + 1, miner_hash: miner_hash)
+      new_block1 = params_for(:block, miner_hash: miner_hash, parent_hash: block_hash, number: block_number + 1)
+      new_block2 = params_for(:block, miner_hash: miner_hash, parent_hash: new_block1.hash, number: block_number + 2)
 
-      new_block2 =
-        params_for(:block, parent_hash: new_block1.hash, number: new_block1.number + 1, miner_hash: miner_hash)
+      range = block_number..(block_number + 2)
 
-      %Ecto.Changeset{valid?: true, changes: block_changes} = Block.changeset(%Block{}, new_block2)
-      changes_list = [block_changes]
+      insert_block(new_block1, options)
+      insert_block(new_block2, options)
+      assert Chain.missing_block_number_ranges(range) == []
 
-      Multi.new()
-      |> Blocks.run(changes_list, options)
-      |> Repo.transaction()
+      insert_block(old_block1, options)
+      assert Chain.missing_block_number_ranges(range) == [(block_number + 2)..(block_number + 2)]
 
-      assert Chain.missing_block_number_ranges(block_number..new_block2.number) == [old_block.number..old_block.number]
+      insert_block(new_block2, options)
+      assert Chain.missing_block_number_ranges(range) == [(block_number + 1)..(block_number + 1)]
 
-      %Ecto.Changeset{valid?: true, changes: block_changes} = Block.changeset(%Block{}, new_block1)
-      changes_list = [block_changes]
+      insert_block(new_block1, options)
+      assert Chain.missing_block_number_ranges(range) == []
+    end
+  end
 
-      Multi.new()
-      |> Blocks.run(changes_list, options)
-      |> Repo.transaction()
+  defp insert_block(block_params, options) do
+    %Ecto.Changeset{valid?: true, changes: block_changes} = Block.changeset(%Block{}, block_params)
 
-      assert Chain.missing_block_number_ranges(block_number..new_block2.number) == []
-    end
+    Multi.new()
+    |> Blocks.run([block_changes], options)
+    |> Repo.transaction()
   end
 
   defp count(schema) do

apps/explorer/test/explorer/chain_test.exs

@@ -3667,27 +3667,6 @@ defmodule Explorer.ChainTest do
     end
   end
 
-  describe "list_block_numbers_with_invalid_consensus/0" do
-    test "returns a list of block numbers with invalid consensus" do
-      block1 = insert(:block)
-      block2_with_invalid_consensus = insert(:block, parent_hash: block1.hash, consensus: false)
-      _block2 = insert(:block, parent_hash: block1.hash, number: block2_with_invalid_consensus.number)
-      block3 = insert(:block, parent_hash: block2_with_invalid_consensus.hash)
-      block4 = insert(:block, parent_hash: block3.hash)
-      block5 = insert(:block, parent_hash: block4.hash)
-      block6_without_consensus = insert(:block, parent_hash: block5.hash, consensus: false)
-      block6 = insert(:block, parent_hash: block5.hash, number: block6_without_consensus.number)
-      block7 = insert(:block, parent_hash: block6.hash)
-      block8_with_invalid_consensus = insert(:block, parent_hash: block7.hash, consensus: false)
-      _block8 = insert(:block, parent_hash: block7.hash, number: block8_with_invalid_consensus.number)
-      block9 = insert(:block, parent_hash: block8_with_invalid_consensus.hash)
-      _block10 = insert(:block, parent_hash: block9.hash)
-
-      assert Chain.list_block_numbers_with_invalid_consensus() ==
-               [block2_with_invalid_consensus.number, block8_with_invalid_consensus.number]
-    end
-  end
-
   describe "block_combined_rewards/1" do
     test "sums the block_rewards values" do
       block = insert(:block)

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

@@ -27,7 +27,7 @@ defmodule Indexer.Block.Realtime.Fetcher do
   alias Explorer.Chain.TokenTransfer
   alias Explorer.Counters.AverageBlockTime
   alias Indexer.{AddressExtraction, Block, TokenBalances, Tracer}
-  alias Indexer.Block.Realtime.{ConsensusEnsurer, TaskSupervisor}
+  alias Indexer.Block.Realtime.TaskSupervisor
   alias Timex.Duration
 
   @behaviour Block.Fetcher
@@ -269,12 +269,7 @@ defmodule Indexer.Block.Realtime.Fetcher do
   @decorate span(tracer: Tracer)
   defp do_fetch_and_import_block(block_number_to_fetch, block_fetcher, retry) do
     case fetch_and_import_range(block_fetcher, block_number_to_fetch..block_number_to_fetch) do
-      {:ok, %{inserted: inserted, errors: []}} ->
-        for block <- Map.get(inserted, :blocks, []) do
-          args = [block.parent_hash, block.number - 1, block_fetcher]
-          Task.Supervisor.start_child(TaskSupervisor, ConsensusEnsurer, :perform, args)
-        end
-
+      {:ok, %{inserted: _, errors: []}} ->
         Logger.debug("Fetched and imported.")
 
       {:ok, %{inserted: _, errors: [_ | _] = errors}} ->