From 5aa8d5c456241bba041f00cbb23a6dcb37ae0149 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Tue, 10 Sep 2019 12:07:16 +0300
Subject: [PATCH 01/10] remove nonconsensus token transfers

---
 .../explorer/chain/import/runner/blocks.ex    | 55 ++++++++++++++-
 .../chain/import/runner/blocks_test.exs       | 68 +++++++++++++++----
 2 files changed, 110 insertions(+), 13 deletions(-)

diff --git a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
index 209d5d2fa6..3c949270cf 100644
--- a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
+++ b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
@@ -9,7 +9,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
 
   alias Ecto.Adapters.SQL
   alias Ecto.{Changeset, Multi, Repo}
-  alias Explorer.Chain.{Address, Block, Hash, Import, InternalTransaction, Transaction}
+  alias Explorer.Chain.{Address, Block, Hash, Import, InternalTransaction, Transaction, TokenTransfer}
   alias Explorer.Chain.Block.Reward
   alias Explorer.Chain.Import.Runner
   alias Explorer.Chain.Import.Runner.Address.CurrentTokenBalances
@@ -73,6 +73,21 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     |> Multi.run(:lose_invalid_neighbour_consensus, fn repo, _ ->
       lose_invalid_neighbour_consensus(repo, where_invalid_neighbour, insert_options)
     end)
+    |> Multi.run(:remove_nonconsensus_data, fn repo,
+                                               %{
+                                                 lose_consensus: lost_consensus_blocks,
+                                                 lose_invalid_neighbour_consensus: lost_consensus_neighbours
+                                               } ->
+      nonconsensus_block_numbers =
+        lost_consensus_blocks
+        |> Kernel.++(lost_consensus_neighbours)
+        |> Enum.map(fn %{number: number} ->
+          number
+        end)
+        |> Enum.sort()
+
+      remove_nonconsensus_data(repo, nonconsensus_block_numbers, insert_options)
+    end)
     |> Multi.run(:delete_address_token_balances, fn repo, _ ->
       delete_address_token_balances(repo, ordered_consensus_block_numbers, insert_options)
     end)
@@ -342,6 +357,44 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     end
   end
 
+  defp remove_nonconsensus_data(repo, nonconsensus_block_numbers, insert_options) do
+    with {:ok, deleted_token_transfers} <-
+           remove_nonconsensus_token_transfers(repo, nonconsensus_block_numbers, insert_options) do
+      {:ok, %{token_transfers: deleted_token_transfers}}
+    end
+  end
+
+  defp remove_nonconsensus_token_transfers(repo, nonconsensus_block_numbers, %{timeout: timeout}) do
+    ordered_token_transfers =
+      from(token_transfer in TokenTransfer,
+        where: token_transfer.block_number in ^nonconsensus_block_numbers,
+        select: map(token_transfer, [:transaction_hash, :log_index]),
+        order_by: [
+          token_transfer.transaction_hash,
+          token_transfer.log_index
+        ],
+        lock: "FOR UPDATE"
+      )
+
+    query =
+      from(token_transfer in TokenTransfer,
+        select: map(token_transfer, [:transaction_hash, :log_index]),
+        inner_join: ordered_token_transfer in subquery(ordered_token_transfers),
+        on:
+            ordered_token_transfer.transaction_hash ==
+              token_transfer.transaction_hash and
+            ordered_token_transfer.log_index == token_transfer.log_index
+      )
+
+    try do
+      {_count, deleted_token_transfers} = repo.delete_all(query, timeout: timeout)
+
+      {:ok, deleted_token_transfers}
+    rescue
+      postgrex_error in Postgrex.Error ->
+        {:error, %{exception: postgrex_error, block_numbers: nonconsensus_block_numbers}}
+    end
+  end
   defp delete_address_token_balances(_, [], _), do: {:ok, []}
 
   defp delete_address_token_balances(repo, ordered_consensus_block_numbers, %{timeout: timeout}) do
diff --git a/apps/explorer/test/explorer/chain/import/runner/blocks_test.exs b/apps/explorer/test/explorer/chain/import/runner/blocks_test.exs
index 63110cba40..f1ebaa4c77 100644
--- a/apps/explorer/test/explorer/chain/import/runner/blocks_test.exs
+++ b/apps/explorer/test/explorer/chain/import/runner/blocks_test.exs
@@ -7,13 +7,14 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
 
   alias Ecto.Multi
   alias Explorer.Chain.Import.Runner.{Blocks, Transactions}
-  alias Explorer.Chain.{Address, Block, Transaction}
+  alias Explorer.Chain.{Address, Block, Transaction, TokenTransfer}
   alias Explorer.Chain
   alias Explorer.Repo
 
   describe "run/1" do
     setup do
-      block = insert(:block, consensus: true)
+      miner = insert(:address)
+      block = params_for(:block, consensus: true, miner_hash: miner.hash)
 
       timestamp = DateTime.utc_now()
       options = %{timestamps: %{inserted_at: timestamp, updated_at: timestamp}}
@@ -22,9 +23,11 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
     end
 
     test "derive_transaction_forks replaces hash on conflicting (uncle_hash, index)", %{
-      consensus_block: %Block{hash: block_hash, miner_hash: miner_hash, number: block_number} = consensus_block,
+      consensus_block: %{hash: block_hash, miner_hash: miner_hash, number: block_number},
       options: options
     } do
+      consensus_block = insert(:block, %{hash: block_hash, number: block_number})
+
       transaction =
         :transaction
         |> insert()
@@ -81,7 +84,7 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
     end
 
     test "delete_address_current_token_balances deletes rows with matching block number when consensus is true",
-         %{consensus_block: %Block{number: block_number} = block, options: options} do
+         %{consensus_block: %{number: block_number} = block, options: options} do
       %Address.CurrentTokenBalance{address_hash: address_hash, token_contract_address_hash: token_contract_address_hash} =
         insert(:address_current_token_balance, block_number: block_number)
 
@@ -98,7 +101,7 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
     end
 
     test "delete_address_current_token_balances does not delete rows with matching block number when consensus is false",
-         %{consensus_block: %Block{number: block_number} = block, options: options} do
+         %{consensus_block: %{number: block_number} = block, options: options} do
       %Address.CurrentTokenBalance{} = insert(:address_current_token_balance, block_number: block_number)
 
       count = 1
@@ -113,8 +116,47 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
       assert count(Address.CurrentTokenBalance) == count
     end
 
+    test "remove_nonconsensus_data deletes rows with matching block number when new consensus block is inserted",
+         %{consensus_block: %{number: block_number} = block, options: options} do
+      insert(:block, number: block_number, consensus: true)
+
+      %TokenTransfer{transaction_hash: transaction_hash, log_index: log_index} =
+        insert(:token_transfer, block_number: block_number, transaction: insert(:transaction))
+
+      assert count(TokenTransfer) == 1
+
+      assert {:ok,
+              %{
+                remove_nonconsensus_data: %{
+                  token_transfers: [
+                    %{transaction_hash: ^transaction_hash, log_index: ^log_index}
+                  ]
+                }
+              }} = run_block_consensus_change(block, true, options)
+
+      assert count(TokenTransfer) == 0
+    end
+
+    test "delete_token_transfers does not delete rows with matching block number when consensus is false",
+         %{consensus_block: %{number: block_number} = block, options: options} do
+      insert(:token_transfer, block_number: block_number, transaction: insert(:transaction))
+
+      count = 1
+
+      assert count(TokenTransfer) == count
+
+      assert {:ok,
+              %{
+                remove_nonconsensus_data: %{
+                  token_transfers: []
+                }
+              }} = run_block_consensus_change(block, false, options)
+
+      assert count(TokenTransfer) == count
+    end
+
     test "derive_address_current_token_balances inserts rows if there is an address_token_balance left for the rows deleted by delete_address_current_token_balances",
-         %{consensus_block: %Block{number: block_number} = block, options: options} do
+         %{consensus_block: %{number: block_number} = block, options: options} do
       token = insert(:token)
       token_contract_address_hash = token.contract_address_hash
 
@@ -172,7 +214,7 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
     end
 
     test "a non-holder reverting to a holder increases the holder_count",
-         %{consensus_block: %Block{hash: block_hash, miner_hash: miner_hash, number: block_number}, options: options} do
+         %{consensus_block: %{hash: block_hash, miner_hash: miner_hash, number: block_number}, options: options} do
       token = insert(:token)
       token_contract_address_hash = token.contract_address_hash
 
@@ -204,7 +246,7 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
     end
 
     test "a holder reverting to a non-holder decreases the holder_count",
-         %{consensus_block: %Block{hash: block_hash, miner_hash: miner_hash, number: block_number}, options: options} do
+         %{consensus_block: %{hash: block_hash, miner_hash: miner_hash, number: block_number}, options: options} do
       token = insert(:token)
       token_contract_address_hash = token.contract_address_hash
 
@@ -236,7 +278,7 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
     end
 
     test "a non-holder becoming and a holder becoming while a holder becomes a non-holder cancels out and holder_count does not change",
-         %{consensus_block: %Block{number: block_number} = block, options: options} do
+         %{consensus_block: %{number: block_number} = block, options: options} do
       token = insert(:token)
       token_contract_address_hash = token.contract_address_hash
 
@@ -262,7 +304,8 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
 
     # Regression test for https://github.com/poanetwork/blockscout/issues/1644
     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
+         %{consensus_block: %{number: block_number, hash: block_hash, miner_hash: miner_hash}, options: options} do
+      insert(:block, %{number: block_number, hash: block_hash})
       old_block1 = params_for(:block, miner_hash: miner_hash, parent_hash: block_hash, number: block_number + 1)
 
       new_block1 = params_for(:block, miner_hash: miner_hash, parent_hash: block_hash, number: block_number + 1)
@@ -286,7 +329,8 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
 
     # Regression test for https://github.com/poanetwork/blockscout/issues/1911
     test "forces block refetch if transaction is re-collated in a different block",
-         %{consensus_block: %Block{number: block_number, hash: block_hash, miner_hash: miner_hash}, options: options} do
+         %{consensus_block: %{number: block_number, hash: block_hash, miner_hash: miner_hash}, options: options} do
+      insert(:block, %{number: block_number, hash: block_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)
 
@@ -365,7 +409,7 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
   end
 
   defp run_block_consensus_change(
-         %Block{hash: block_hash, miner_hash: miner_hash, number: block_number},
+         %{hash: block_hash, miner_hash: miner_hash, number: block_number},
          consensus,
          options
        ) do

From c195428634d9d2f917ee4f21f1dc0fde36023b62 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Tue, 10 Sep 2019 13:04:15 +0300
Subject: [PATCH 02/10] remove nonconsensus logs

---
 .../explorer/chain/import/runner/blocks.ex    | 79 +++++++++++++++----
 .../chain/import/runner/blocks_test.exs       | 28 ++++++-
 2 files changed, 88 insertions(+), 19 deletions(-)

diff --git a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
index 3c949270cf..5c4c42d898 100644
--- a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
+++ b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
@@ -9,7 +9,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
 
   alias Ecto.Adapters.SQL
   alias Ecto.{Changeset, Multi, Repo}
-  alias Explorer.Chain.{Address, Block, Hash, Import, InternalTransaction, Transaction, TokenTransfer}
+  alias Explorer.Chain.{Address, Block, Hash, Import, InternalTransaction, Log, TokenTransfer, Transaction}
   alias Explorer.Chain.Block.Reward
   alias Explorer.Chain.Import.Runner
   alias Explorer.Chain.Import.Runner.Address.CurrentTokenBalances
@@ -58,15 +58,6 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
         where_forked: where_forked
       })
     end)
-    # MUST be after `:derive_transaction_forks`, which depends on values in `transactions` table
-    |> Multi.run(:fork_transactions, fn repo, _ ->
-      fork_transactions(%{
-        repo: repo,
-        timeout: options[Runner.Transactions.option_key()][:timeout] || Runner.Transactions.timeout(),
-        timestamps: timestamps,
-        where_forked: where_forked
-      })
-    end)
     |> Multi.run(:lose_consensus, fn repo, _ ->
       lose_consensus(repo, ordered_consensus_block_numbers, insert_options)
     end)
@@ -86,7 +77,20 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
         end)
         |> Enum.sort()
 
-      remove_nonconsensus_data(repo, nonconsensus_block_numbers, insert_options)
+      remove_nonconsensus_data(
+        repo,
+        nonconsensus_block_numbers,
+        insert_options
+      )
+    end)
+    # MUST be after `:derive_transaction_forks`, which depends on values in `transactions` table
+    |> Multi.run(:fork_transactions, fn repo, _ ->
+      fork_transactions(%{
+        repo: repo,
+        timeout: options[Runner.Transactions.option_key()][:timeout] || Runner.Transactions.timeout(),
+        timestamps: timestamps,
+        where_forked: where_forked
+      })
     end)
     |> Multi.run(:delete_address_token_balances, fn repo, _ ->
       delete_address_token_balances(repo, ordered_consensus_block_numbers, insert_options)
@@ -357,10 +361,15 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     end
   end
 
-  defp remove_nonconsensus_data(repo, nonconsensus_block_numbers, insert_options) do
+  defp remove_nonconsensus_data(
+         repo,
+         nonconsensus_block_numbers,
+         insert_options
+       ) do
     with {:ok, deleted_token_transfers} <-
-           remove_nonconsensus_token_transfers(repo, nonconsensus_block_numbers, insert_options) do
-      {:ok, %{token_transfers: deleted_token_transfers}}
+           remove_nonconsensus_token_transfers(repo, nonconsensus_block_numbers, insert_options),
+         {:ok, deleted_logs} <- remove_nonconsensus_logs(repo, nonconsensus_block_numbers, insert_options) do
+      {:ok, %{token_transfers: deleted_token_transfers, logs: deleted_logs}}
     end
   end
 
@@ -381,8 +390,8 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
         select: map(token_transfer, [:transaction_hash, :log_index]),
         inner_join: ordered_token_transfer in subquery(ordered_token_transfers),
         on:
-            ordered_token_transfer.transaction_hash ==
-              token_transfer.transaction_hash and
+          ordered_token_transfer.transaction_hash ==
+            token_transfer.transaction_hash and
             ordered_token_transfer.log_index == token_transfer.log_index
       )
 
@@ -395,6 +404,44 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
         {:error, %{exception: postgrex_error, block_numbers: nonconsensus_block_numbers}}
     end
   end
+
+  defp remove_nonconsensus_logs(repo, nonconsensus_block_numbers, %{timeout: timeout}) do
+    transaction_query =
+      from(transaction in Transaction,
+        where: transaction.block_number in ^nonconsensus_block_numbers,
+        select: map(transaction, [:hash]),
+        order_by: transaction.hash
+      )
+
+    ordered_logs =
+      from(log in Log,
+        inner_join: transaction in subquery(transaction_query),
+        on: log.transaction_hash == transaction.hash,
+        select: map(log, [:transaction_hash, :index]),
+        order_by: [
+          log.transaction_hash,
+          log.index
+        ],
+        lock: "FOR UPDATE"
+      )
+
+    query =
+      from(log in Log,
+        select: map(log, [:transaction_hash, :index]),
+        inner_join: ordered_log in subquery(ordered_logs),
+        on: ordered_log.transaction_hash == log.transaction_hash and ordered_log.index == log.index
+      )
+
+    try do
+      {_count, deleted_logs} = repo.delete_all(query, timeout: timeout)
+
+      {:ok, deleted_logs}
+    rescue
+      postgrex_error in Postgrex.Error ->
+        {:error, %{exception: postgrex_error, block_numbers: nonconsensus_block_numbers}}
+    end
+  end
+
   defp delete_address_token_balances(_, [], _), do: {:ok, []}
 
   defp delete_address_token_balances(repo, ordered_consensus_block_numbers, %{timeout: timeout}) do
diff --git a/apps/explorer/test/explorer/chain/import/runner/blocks_test.exs b/apps/explorer/test/explorer/chain/import/runner/blocks_test.exs
index f1ebaa4c77..459363839b 100644
--- a/apps/explorer/test/explorer/chain/import/runner/blocks_test.exs
+++ b/apps/explorer/test/explorer/chain/import/runner/blocks_test.exs
@@ -7,7 +7,7 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
 
   alias Ecto.Multi
   alias Explorer.Chain.Import.Runner.{Blocks, Transactions}
-  alias Explorer.Chain.{Address, Block, Transaction, TokenTransfer}
+  alias Explorer.Chain.{Address, Block, Log, Transaction, TokenTransfer}
   alias Explorer.Chain
   alias Explorer.Repo
 
@@ -116,7 +116,7 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
       assert count(Address.CurrentTokenBalance) == count
     end
 
-    test "remove_nonconsensus_data deletes rows with matching block number when new consensus block is inserted",
+    test "remove_nonconsensus_data deletes token transfer rows with matching block number when new consensus block is inserted",
          %{consensus_block: %{number: block_number} = block, options: options} do
       insert(:block, number: block_number, consensus: true)
 
@@ -137,7 +137,7 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
       assert count(TokenTransfer) == 0
     end
 
-    test "delete_token_transfers does not delete rows with matching block number when consensus is false",
+    test "remove_nonconsensus_data does not delete token transfer rows with matching block number when new consensus block wasn't inserted",
          %{consensus_block: %{number: block_number} = block, options: options} do
       insert(:token_transfer, block_number: block_number, transaction: insert(:transaction))
 
@@ -155,6 +155,28 @@ defmodule Explorer.Chain.Import.Runner.BlocksTest do
       assert count(TokenTransfer) == count
     end
 
+    test "remove_nonconsensus_data deletes nonconsensus logs", %{
+      consensus_block: %{number: block_number} = block,
+      options: options
+    } do
+      old_block = insert(:block, number: block_number, consensus: true)
+      forked_transaction = :transaction |> insert() |> with_block(old_block)
+      %Log{transaction_hash: hash, index: index} = insert(:log, transaction: forked_transaction)
+
+      assert count(Log) == 1
+
+      assert {:ok,
+              %{
+                remove_nonconsensus_data: %{
+                  logs: [
+                    %{transaction_hash: ^hash, index: ^index}
+                  ]
+                }
+              }} = run_block_consensus_change(block, true, options)
+
+      assert count(Log) == 0
+    end
+
     test "derive_address_current_token_balances inserts rows if there is an address_token_balance left for the rows deleted by delete_address_current_token_balances",
          %{consensus_block: %{number: block_number} = block, options: options} do
       token = insert(:token)

From ab655103f60ccce8427224e5649d576b9671ba29 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Tue, 10 Sep 2019 13:11:13 +0300
Subject: [PATCH 03/10] use existing forked transactions query

---
 .../lib/explorer/chain/import/runner/blocks.ex  | 17 +++++++++--------
 1 file changed, 9 insertions(+), 8 deletions(-)

diff --git a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
index 5c4c42d898..cbf7532ca2 100644
--- a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
+++ b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
@@ -79,7 +79,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
 
       remove_nonconsensus_data(
         repo,
-        nonconsensus_block_numbers,
+        %{nonconsensus_block_numbers: nonconsensus_block_numbers, forked_transactions_query: where_forked},
         insert_options
       )
     end)
@@ -363,12 +363,15 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
 
   defp remove_nonconsensus_data(
          repo,
-         nonconsensus_block_numbers,
+         %{
+           nonconsensus_block_numbers: nonconsensus_block_numbers,
+           forked_transactions_query: forked_transactions_query
+         },
          insert_options
        ) do
     with {:ok, deleted_token_transfers} <-
            remove_nonconsensus_token_transfers(repo, nonconsensus_block_numbers, insert_options),
-         {:ok, deleted_logs} <- remove_nonconsensus_logs(repo, nonconsensus_block_numbers, insert_options) do
+         {:ok, deleted_logs} <- remove_nonconsensus_logs(repo, forked_transactions_query, insert_options) do
       {:ok, %{token_transfers: deleted_token_transfers, logs: deleted_logs}}
     end
   end
@@ -405,11 +408,9 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     end
   end
 
-  defp remove_nonconsensus_logs(repo, nonconsensus_block_numbers, %{timeout: timeout}) do
+  defp remove_nonconsensus_logs(repo, forked_transactions_query, %{timeout: timeout}) do
     transaction_query =
-      from(transaction in Transaction,
-        where: transaction.block_number in ^nonconsensus_block_numbers,
-        select: map(transaction, [:hash]),
+      from(transaction in subquery(forked_transactions_query),
         order_by: transaction.hash
       )
 
@@ -438,7 +439,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
       {:ok, deleted_logs}
     rescue
       postgrex_error in Postgrex.Error ->
-        {:error, %{exception: postgrex_error, block_numbers: nonconsensus_block_numbers}}
+        {:error, %{exception: postgrex_error, forked_transactions_query: forked_transactions_query}}
     end
   end
 

From c038511cf2c99f244f4502d2c3491e748aaae89c Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Tue, 10 Sep 2019 13:58:00 +0300
Subject: [PATCH 04/10] use query across all inserted blocks

---
 .../lib/explorer/chain/import/runner/blocks.ex  | 17 ++++++++---------
 1 file changed, 8 insertions(+), 9 deletions(-)

diff --git a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
index cbf7532ca2..5c4c42d898 100644
--- a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
+++ b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
@@ -79,7 +79,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
 
       remove_nonconsensus_data(
         repo,
-        %{nonconsensus_block_numbers: nonconsensus_block_numbers, forked_transactions_query: where_forked},
+        nonconsensus_block_numbers,
         insert_options
       )
     end)
@@ -363,15 +363,12 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
 
   defp remove_nonconsensus_data(
          repo,
-         %{
-           nonconsensus_block_numbers: nonconsensus_block_numbers,
-           forked_transactions_query: forked_transactions_query
-         },
+         nonconsensus_block_numbers,
          insert_options
        ) do
     with {:ok, deleted_token_transfers} <-
            remove_nonconsensus_token_transfers(repo, nonconsensus_block_numbers, insert_options),
-         {:ok, deleted_logs} <- remove_nonconsensus_logs(repo, forked_transactions_query, insert_options) do
+         {:ok, deleted_logs} <- remove_nonconsensus_logs(repo, nonconsensus_block_numbers, insert_options) do
       {:ok, %{token_transfers: deleted_token_transfers, logs: deleted_logs}}
     end
   end
@@ -408,9 +405,11 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     end
   end
 
-  defp remove_nonconsensus_logs(repo, forked_transactions_query, %{timeout: timeout}) do
+  defp remove_nonconsensus_logs(repo, nonconsensus_block_numbers, %{timeout: timeout}) do
     transaction_query =
-      from(transaction in subquery(forked_transactions_query),
+      from(transaction in Transaction,
+        where: transaction.block_number in ^nonconsensus_block_numbers,
+        select: map(transaction, [:hash]),
         order_by: transaction.hash
       )
 
@@ -439,7 +438,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
       {:ok, deleted_logs}
     rescue
       postgrex_error in Postgrex.Error ->
-        {:error, %{exception: postgrex_error, forked_transactions_query: forked_transactions_query}}
+        {:error, %{exception: postgrex_error, block_numbers: nonconsensus_block_numbers}}
     end
   end
 

From 9f04a12d31fa52035b1908fc3a64c5aab7252a19 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Tue, 10 Sep 2019 14:38:59 +0300
Subject: [PATCH 05/10] fix CR issues

---
 apps/explorer/lib/explorer/chain/import/runner/blocks.ex | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
index 5c4c42d898..08ad720a0f 100644
--- a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
+++ b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
@@ -70,8 +70,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
                                                  lose_invalid_neighbour_consensus: lost_consensus_neighbours
                                                } ->
       nonconsensus_block_numbers =
-        lost_consensus_blocks
-        |> Kernel.++(lost_consensus_neighbours)
+        (lost_consensus_blocks ++ lost_consensus_neighbours)
         |> Enum.map(fn %{number: number} ->
           number
         end)
@@ -378,6 +377,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
       from(token_transfer in TokenTransfer,
         where: token_transfer.block_number in ^nonconsensus_block_numbers,
         select: map(token_transfer, [:transaction_hash, :log_index]),
+        # Enforce TokenTransfer ShareLocks order (see docs: sharelocks.md)
         order_by: [
           token_transfer.transaction_hash,
           token_transfer.log_index
@@ -418,11 +418,12 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
         inner_join: transaction in subquery(transaction_query),
         on: log.transaction_hash == transaction.hash,
         select: map(log, [:transaction_hash, :index]),
+        # Enforce Log ShareLocks order (see docs: sharelocks.md)
         order_by: [
           log.transaction_hash,
           log.index
         ],
-        lock: "FOR UPDATE"
+        lock: "FOR UPDATE OF l0"
       )
 
     query =

From ddfec17f604c3e8e752be8ddabdfe1dc391411f8 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Tue, 10 Sep 2019 14:40:09 +0300
Subject: [PATCH 06/10] add CHANGELOG entry

---
 CHANGELOG.md | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 343301eb65..f942ada294 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,13 +1,14 @@
 ## Current
 
 ### Features
-- [#2679](https://github.com/poanetwork/blockscout/pull/2679) - added fixed height for card chain blocks and card chain transactions 
+- [#2679](https://github.com/poanetwork/blockscout/pull/2679) - added fixed height for card chain blocks and card chain transactions
 - [#2678](https://github.com/poanetwork/blockscout/pull/2678) - fixed dashboard banner height bug
-- [#2672](https://github.com/poanetwork/blockscout/pull/2672) - added new theme for xUSDT 
+- [#2672](https://github.com/poanetwork/blockscout/pull/2672) - added new theme for xUSDT
 - [#2663](https://github.com/poanetwork/blockscout/pull/2663) - Fetch address counters in parallel
 
 ### Fixes
 - [#2682](https://github.com/poanetwork/blockscout/pull/2682) - Use Task.start instead of Task.async in caches
+- [#2687](https://github.com/poanetwork/blockscout/pull/2687) - remove non-consensus token transfers, logs when inserting new consensus blocks
 
 ### Chore
 

From e569c5a7987de95b910a6393d17d12e32a6e0015 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Tue, 10 Sep 2019 20:10:06 +0300
Subject: [PATCH 07/10] add block_number index

---
 ...r_block_number_in_token_transfers_and_transactions.exs | 8 ++++++++
 1 file changed, 8 insertions(+)
 create mode 100644 apps/explorer/priv/repo/migrations/20190910170703_create_indexes_for_block_number_in_token_transfers_and_transactions.exs

diff --git a/apps/explorer/priv/repo/migrations/20190910170703_create_indexes_for_block_number_in_token_transfers_and_transactions.exs b/apps/explorer/priv/repo/migrations/20190910170703_create_indexes_for_block_number_in_token_transfers_and_transactions.exs
new file mode 100644
index 0000000000..ac15f632d5
--- /dev/null
+++ b/apps/explorer/priv/repo/migrations/20190910170703_create_indexes_for_block_number_in_token_transfers_and_transactions.exs
@@ -0,0 +1,8 @@
+defmodule Explorer.Repo.Migrations.CreateIndexesForBlockNumberInTokenTransfersAndTransactions do
+  use Ecto.Migration
+
+  def change do
+    create_if_not_exists(index(:token_transfers, [:block_number]))
+    create_if_not_exists(index(:transactions, [:block_number]))
+  end
+end

From 71ee060f3f1d77222eea2ea21f3deb323c4c56b5 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Tue, 10 Sep 2019 20:19:55 +0300
Subject: [PATCH 08/10] remove transaction index creation

---
 ...exes_for_block_number_in_token_transfers_and_transactions.exs | 1 -
 1 file changed, 1 deletion(-)

diff --git a/apps/explorer/priv/repo/migrations/20190910170703_create_indexes_for_block_number_in_token_transfers_and_transactions.exs b/apps/explorer/priv/repo/migrations/20190910170703_create_indexes_for_block_number_in_token_transfers_and_transactions.exs
index ac15f632d5..b718224622 100644
--- a/apps/explorer/priv/repo/migrations/20190910170703_create_indexes_for_block_number_in_token_transfers_and_transactions.exs
+++ b/apps/explorer/priv/repo/migrations/20190910170703_create_indexes_for_block_number_in_token_transfers_and_transactions.exs
@@ -3,6 +3,5 @@ defmodule Explorer.Repo.Migrations.CreateIndexesForBlockNumberInTokenTransfersAn
 
   def change do
     create_if_not_exists(index(:token_transfers, [:block_number]))
-    create_if_not_exists(index(:transactions, [:block_number]))
   end
 end

From a883e4ac68f351a5f7c98d4300b89326dc6dacd2 Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Wed, 11 Sep 2019 15:20:16 +0300
Subject: [PATCH 09/10] deduplicate numbers

---
 apps/explorer/lib/explorer/chain/import/runner/blocks.ex | 1 +
 1 file changed, 1 insertion(+)

diff --git a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
index 08ad720a0f..aa80e6c251 100644
--- a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
+++ b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
@@ -75,6 +75,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
           number
         end)
         |> Enum.sort()
+        |> Enum.uniq()
 
       remove_nonconsensus_data(
         repo,

From b674fa675e943ed888aa212b10a6ddd859922cdd Mon Sep 17 00:00:00 2001
From: Ayrat Badykov <ayratin555@gmail.com>
Date: Wed, 11 Sep 2019 16:08:23 +0300
Subject: [PATCH 10/10] Update blocks.ex

---
 apps/explorer/lib/explorer/chain/import/runner/blocks.ex | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
index aa80e6c251..825a533743 100644
--- a/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
+++ b/apps/explorer/lib/explorer/chain/import/runner/blocks.ex
@@ -75,7 +75,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
           number
         end)
         |> Enum.sort()
-        |> Enum.uniq()
+        |> Enum.dedup()
 
       remove_nonconsensus_data(
         repo,