Derive transaction forks for transacion in blocks that go non-consensus

apps/explorer/.sobelow-conf

@@ -1,7 +1,7 @@
 [
   verbose: false,
   private: true,
-  skip: false,
+  skip: true,
   exit: "low",
   format: "compact",
   ignore: ["Config.HTTPS"],

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

@@ -6,6 +6,7 @@ defmodule Explorer.Chain.Import do
   import Ecto.Query, only: [from: 2, update: 2]
 
   alias Ecto.{Changeset, Multi}
+  alias Ecto.Adapters.SQL
 
   alias Explorer.Chain.{
     Address,
@@ -378,12 +379,22 @@ defmodule Explorer.Chain.Import do
       %{Block => blocks_changes} ->
         timestamps = Map.fetch!(options, :timestamps)
         blocks_timeout = options[:blocks][:timeout] || @insert_blocks_timeout
+        where_forked = where_forked(blocks_changes)
 
         multi
+        |> Multi.run(:derive_transaction_forks, fn _ ->
+          derive_transaction_forks(%{
+            timeout: options[:transaction_forks][:timeout] || @insert_transaction_forks_timeout,
+            timestamps: timestamps,
+            where_forked: where_forked
+          })
+        end)
+        # MUST be after `:derive_transaction_forks`, which depends on values in `transactions` table
         |> Multi.run(:fork_transactions, fn _ ->
-          fork_transactions(blocks_changes, %{
+          fork_transactions(%{
             timeout: options[:transactions][:timeout] || @insert_transactions_timeout,
-            timestamps: timestamps
+            timestamps: timestamps,
+            where_forked: where_forked
           })
         end)
         |> Multi.run(:lose_consenus, fn _ ->
@@ -980,14 +991,9 @@ defmodule Explorer.Chain.Import do
     {:ok, inserted}
   end
 
-  defp fork_transactions(blocks_changes, %{
-         timeout: timeout,
-         timestamps: %{updated_at: updated_at}
-       })
-       when is_list(blocks_changes) do
+  defp fork_transactions(%{timeout: timeout, timestamps: %{updated_at: updated_at}, where_forked: where_forked}) do
     query =
-      Transaction
-      |> where_forked(blocks_changes)
+      where_forked
       |> update(
         set: [
           block_hash: nil,
@@ -1007,12 +1013,12 @@ defmodule Explorer.Chain.Import do
       {:ok, result}
     rescue
       postgrex_error in Postgrex.Error ->
-        {:error, %{exception: postgrex_error, blocks_changes: blocks_changes}}
+        {:error, %{exception: postgrex_error}}
     end
   end
 
-  defp where_forked(query, blocks_changes) when is_list(blocks_changes) do
-    Enum.reduce(blocks_changes, query, fn %{consensus: consensus, hash: hash, number: number}, acc ->
+  defp where_forked(blocks_changes) when is_list(blocks_changes) do
+    Enum.reduce(blocks_changes, Transaction, fn %{consensus: consensus, hash: hash, number: number}, acc ->
       case consensus do
         false ->
           from(transaction in acc, or_where: transaction.block_hash == ^hash and transaction.block_number == ^number)
@@ -1023,6 +1029,42 @@ defmodule Explorer.Chain.Import do
     end)
   end
 
+  # sobelow_skip ["SQL.Query"]
+  defp derive_transaction_forks(%{
+         timeout: timeout,
+         timestamps: %{inserted_at: inserted_at, updated_at: updated_at},
+         where_forked: where_forked
+       }) do
+    query =
+      from(transaction in where_forked,
+        select: [
+          transaction.block_hash,
+          transaction.index,
+          transaction.hash,
+          type(^inserted_at, transaction.inserted_at),
+          type(^updated_at, transaction.updated_at)
+        ]
+      )
+
+    {sql, parameters} = SQL.to_sql(:all, Repo, query)
+
+    {:ok, %Postgrex.Result{columns: ["uncle_hash", "hash"], command: :insert, rows: rows}} =
+      SQL.query(
+        Repo,
+        """
+        INSERT INTO transaction_forks (uncle_hash, index, hash, inserted_at, updated_at)
+        #{sql}
+        RETURNING uncle_hash, hash
+        """,
+        parameters,
+        timeout: timeout
+      )
+
+    derived_transaction_forks = Enum.map(rows, fn [uncle_hash, hash] -> %{uncle_hash: uncle_hash, hash: hash} end)
+
+    {:ok, derived_transaction_forks}
+  end
+
   defp lose_consensus(blocks_changes, %{timeout: timeout, timestamps: %{updated_at: updated_at}})
        when is_list(blocks_changes) do
     ordered_block_number =

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

@@ -1351,5 +1351,113 @@ defmodule Explorer.Chain.ImportTest do
       assert transaction_after.index == nil
       assert transaction_after.status == nil
     end
+
+    test "reorganizations fork transactions that end up in non-consensus blocks" do
+      block_number = 0
+
+      miner_hash_before = address_hash()
+      from_address_hash_before = address_hash()
+      block_hash_before = block_hash()
+      index_before = 0
+
+      transaction_hash = transaction_hash()
+
+      assert {:ok, _} =
+               Import.all(%{
+                 addresses: %{
+                   params: [
+                     %{hash: miner_hash_before},
+                     %{hash: from_address_hash_before}
+                   ]
+                 },
+                 blocks: %{
+                   params: [
+                     %{
+                       consensus: true,
+                       difficulty: 0,
+                       gas_limit: 0,
+                       gas_used: 0,
+                       hash: block_hash_before,
+                       miner_hash: miner_hash_before,
+                       nonce: 0,
+                       number: block_number,
+                       parent_hash: block_hash(),
+                       size: 0,
+                       timestamp: Timex.parse!("2019-01-01T01:00:00Z", "{ISO:Extended:Z}"),
+                       total_difficulty: 0
+                     }
+                   ]
+                 },
+                 transactions: %{
+                   params: [
+                     %{
+                       block_hash: block_hash_before,
+                       block_number: block_number,
+                       from_address_hash: from_address_hash_before,
+                       gas: 21_000,
+                       gas_price: 1,
+                       gas_used: 21_000,
+                       cumulative_gas_used: 21_000,
+                       hash: transaction_hash,
+                       index: index_before,
+                       input: "0x",
+                       nonce: 0,
+                       r: 0,
+                       s: 0,
+                       v: 0,
+                       value: 0,
+                       status: :ok
+                     }
+                   ],
+                   on_conflict: :replace_all
+                 }
+               })
+
+      %Block{consensus: true, number: ^block_number} = Repo.get(Block, block_hash_before)
+
+      assert Repo.one!(from(transaction_fork in Transaction.Fork, select: fragment("COUNT(*)"))) == 0
+
+      miner_hash_after = address_hash()
+      from_address_hash_after = address_hash()
+      block_hash_after = block_hash()
+
+      assert {:ok, _} =
+               Import.all(%{
+                 addresses: %{
+                   params: [
+                     %{hash: miner_hash_after},
+                     %{hash: from_address_hash_after}
+                   ]
+                 },
+                 blocks: %{
+                   params: [
+                     %{
+                       consensus: true,
+                       difficulty: 1,
+                       gas_limit: 1,
+                       gas_used: 1,
+                       hash: block_hash_after,
+                       miner_hash: miner_hash_after,
+                       nonce: 1,
+                       number: block_number,
+                       parent_hash: block_hash(),
+                       size: 1,
+                       timestamp: Timex.parse!("2019-01-01T02:00:00Z", "{ISO:Extended:Z}"),
+                       total_difficulty: 1
+                     }
+                   ]
+                 }
+               })
+
+      assert Repo.one!(from(transaction_fork in Transaction.Fork, select: fragment("COUNT(*)"))) == 1
+
+      assert %Transaction.Fork{index: ^index_before} =
+               Repo.one(
+                 from(transaction_fork in Transaction.Fork,
+                   where:
+                     transaction_fork.uncle_hash == ^block_hash_before and transaction_fork.hash == ^transaction_hash
+                 )
+               )
+    end
   end
 end