Merge pull request #69 from DOS/sync-upstream-v11.0.3

Sync upstream v11.0.3 (merge conflicts)

Author: JOY

CHANGELOG.md

@@ -1,15 +1,29 @@
 # Changelog
 
+## 11.0.3
+
+### 🐛 Bug Fixes
+
+- Insert AddressIdToAddressHash via safe_insert_all ([#14333](https://github.com/blockscout/blockscout/pull/14333))
+- Force search for contract creator if internal transactions module is disabled ([#14324](https://github.com/blockscout/blockscout/issues/14324))
+- Add transactions uniqueness before insert ([#14329](https://github.com/blockscout/blockscout/issues/14329))
+
+### ⚙️ Miscellaneous Tasks
+
+- Don't lock tables if foreign keys are already dropped ([#14321](https://github.com/blockscout/blockscout/issues/14321))
+- Dev branch + CI, remove obsolete GA workflows ([#14317](https://github.com/blockscout/blockscout/issues/14317))
+
+
 ## 11.0.2
 
 ### 🐛 Bug Fixes
 
-- Process empty list of changes on fetching contract codes ((#14312)[https://github.com/blockscout/blockscout/pull/14312])
+- Process empty list of changes on fetching contract codes ([#14312](https://github.com/blockscout/blockscout/pull/14312))
 - Add fallback for empty "to" in Geth selfdestruct ([#14256](https://github.com/blockscout/blockscout/issues/14256))
 - Trim contractaddresses in getcontractcreation ([#14306](https://github.com/blockscout/blockscout/issues/14306))
 - Adapt maybe_reject_zero_value for empty blocks ([#14309](https://github.com/blockscout/blockscout/issues/14309))
 - Add missing internal transactions address preload ([#14308](https://github.com/blockscout/blockscout/issues/14308))
-- Fix some web tests ([#14310][https://github.com/blockscout/blockscout/pull/14310])
+- Fix some web tests ([#14310](https://github.com/blockscout/blockscout/pull/14310))
 
 ### ⚙️ Miscellaneous Tasks
 

apps/block_scout_web/mix.exs

@@ -19,7 +19,7 @@ defmodule BlockScoutWeb.Mixfile do
       lockfile: "../../mix.lock",
       package: package(),
       start_permanent: Mix.env() == :prod,
-      version: "11.0.2",
+      version: "11.0.3",
       xref: [
         exclude: [
           Explorer.Chain.Beacon.Reader,

apps/ethereum_jsonrpc/mix.exs

@@ -19,7 +19,7 @@ defmodule EthereumJSONRPC.MixProject do
       elixirc_paths: elixirc_paths(Mix.env()),
       lockfile: "../../mix.lock",
       start_permanent: Mix.env() == :prod,
-      version: "11.0.2"
+      version: "11.0.3"
     ]
   end
 

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

@@ -19,7 +19,6 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     BlockNumberHelper,
     DenormalizationHelper,
     Import,
-    PendingBlockOperation,
     PendingOperationsHelper,
     SmartContract,
     Token,
@@ -106,16 +105,6 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
         :blocks
       )
     end)
-    |> Multi.run(:new_pending_block_operations, fn repo, %{blocks: blocks} ->
-      Instrumenter.block_import_stage_runner(
-        fn ->
-          new_pending_block_operations(repo, blocks, insert_options)
-        end,
-        :address_referencing,
-        :blocks,
-        :new_pending_block_operations
-      )
-    end)
     |> Multi.run(:uncle_fetched_block_second_degree_relations, fn repo, _ ->
       Instrumenter.block_import_stage_runner(
         fn ->
@@ -695,32 +684,6 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     lose_consensus(repo, blocks_changes, opts)
   end
 
-  defp new_pending_block_operations(repo, inserted_blocks, %{timeout: timeout, timestamps: timestamps}) do
-    case PendingOperationsHelper.pending_operations_type() do
-      "blocks" ->
-        sorted_pending_ops =
-          inserted_blocks
-          |> RangesHelper.filter_by_height_range(&RangesHelper.traceable_block_number?(&1.number))
-          |> Enum.filter(& &1.consensus)
-          |> Enum.map(&%{block_hash: &1.hash, block_number: &1.number})
-          |> Enum.sort()
-
-        Import.insert_changes_list(
-          repo,
-          sorted_pending_ops,
-          conflict_target: :block_hash,
-          on_conflict: :nothing,
-          for: PendingBlockOperation,
-          returning: true,
-          timeout: timeout,
-          timestamps: timestamps
-        )
-
-      _other_type ->
-        {:ok, []}
-    end
-  end
-
   defp delete_address_coin_balances(_repo, [], _options), do: {:ok, []}
 
   defp delete_address_coin_balances(repo, non_consensus_blocks, %{timeout: timeout}) do

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

@@ -10,7 +10,17 @@ defmodule Explorer.Chain.Import.Runner.Transactions do
 
   alias Ecto.{Multi, Repo}
   alias EthereumJSONRPC.Utility.RangesHelper
-  alias Explorer.Chain.{Block, Hash, Import, PendingOperationsHelper, PendingTransactionOperation, Transaction}
+
+  alias Explorer.Chain.{
+    Block,
+    Hash,
+    Import,
+    PendingBlockOperation,
+    PendingOperationsHelper,
+    PendingTransactionOperation,
+    Transaction
+  }
+
   alias Explorer.Chain.Import.Runner.TokenTransfers
   alias Explorer.Prometheus.Instrumenter
   alias Explorer.Utility.MissingBlockRange
@@ -66,14 +76,14 @@ defmodule Explorer.Chain.Import.Runner.Transactions do
         :transactions
       )
     end)
-    |> Multi.run(:new_pending_transaction_operations, fn repo, %{transactions: transactions} ->
+    |> Multi.run(:new_pending_operations, fn repo, %{transactions: transactions} ->
       Instrumenter.block_import_stage_runner(
         fn ->
-          new_pending_transaction_operations(repo, transactions, insert_options)
+          new_pending_operations(repo, transactions, insert_options)
         end,
         :block_referencing,
         :transactions,
-        :new_pending_transaction_operations
+        :new_pending_operations
       )
     end)
   end
@@ -105,7 +115,10 @@ defmodule Explorer.Chain.Import.Runner.Transactions do
     on_conflict = Map.get_lazy(options, :on_conflict, &default_on_conflict/0)
 
     # Enforce Transaction ShareLocks order (see docs: sharelocks.md)
-    ordered_changes_list = Enum.sort_by(changes_list, & &1.hash)
+    ordered_changes_list =
+      changes_list
+      |> Enum.uniq_by(& &1.hash)
+      |> Enum.sort_by(& &1.hash)
 
     Import.insert_changes_list(
       repo,
@@ -119,15 +132,21 @@ defmodule Explorer.Chain.Import.Runner.Transactions do
     )
   end
 
-  defp new_pending_transaction_operations(repo, inserted_transactions, %{timeout: timeout, timestamps: timestamps}) do
+  defp new_pending_operations(repo, inserted_transactions, %{timeout: timeout, timestamps: timestamps}) do
+    traceable_consensus_transactions =
+      inserted_transactions
+      |> RangesHelper.filter_by_height_range(&RangesHelper.traceable_block_number?(&1.block_number))
+      |> Transaction.filter_non_traceable_transactions()
+
+    traceable_consensus_block_numbers = Enum.map(traceable_consensus_transactions, & &1.block_number) |> Enum.uniq()
+    block_numbers_with_priorities = MissingBlockRange.find_priority_by_numbers(traceable_consensus_block_numbers)
+
     case PendingOperationsHelper.pending_operations_type() do
       "transactions" ->
         sorted_pending_ops =
-          inserted_transactions
-          |> RangesHelper.filter_by_height_range(&RangesHelper.traceable_block_number?(&1.block_number))
-          |> Transaction.filter_non_traceable_transactions()
+          traceable_consensus_transactions
           |> Enum.reject(&is_nil(&1.block_number))
-          |> Enum.map(&%{transaction_hash: &1.hash})
+          |> Enum.map(&%{transaction_hash: &1.hash, priority: Map.get(block_numbers_with_priorities, &1.block_number)})
           |> Enum.sort()
 
         Import.insert_changes_list(
@@ -141,8 +160,29 @@ defmodule Explorer.Chain.Import.Runner.Transactions do
           timestamps: timestamps
         )
 
-      _other_type ->
-        {:ok, []}
+      "blocks" ->
+        sorted_pending_ops =
+          traceable_consensus_transactions
+          |> Enum.reject(&is_nil(&1.block_number))
+          |> Enum.map(
+            &%{
+              block_hash: &1.block_hash,
+              block_number: &1.block_number,
+              priority: Map.get(block_numbers_with_priorities, &1.block_number)
+            }
+          )
+          |> Enum.sort()
+
+        Import.insert_changes_list(
+          repo,
+          sorted_pending_ops,
+          conflict_target: :block_hash,
+          on_conflict: :nothing,
+          for: PendingBlockOperation,
+          returning: true,
+          timeout: timeout,
+          timestamps: timestamps
+        )
     end
   end
 

apps/explorer/lib/explorer/chain/pending_block_operation.ex

@@ -28,6 +28,8 @@ defmodule Explorer.Chain.PendingBlockOperation do
       type: Hash.Full,
       null: false
     )
+
+    field(:priority, :integer)
   end
 
   def changeset(%__MODULE__{} = pending_ops, attrs) do
@@ -80,7 +82,7 @@ defmodule Explorer.Chain.PendingBlockOperation do
           limited? :: boolean()
         ) :: {:ok, accumulator}
         when accumulator: term()
-  def stream_blocks_with_unfetched_internal_transactions(initial, reducer, limited? \\ false)
+  def stream_blocks_with_unfetched_internal_transactions(initial, reducer, limited? \\ false, with_priority? \\ false)
       when is_function(reducer, 2) do
     direction = Application.get_env(:indexer, :internal_transactions_fetch_order)
 
@@ -93,7 +95,16 @@ defmodule Explorer.Chain.PendingBlockOperation do
       )
 
     query
+    |> maybe_add_priority_filter(with_priority?)
     |> add_fetcher_limit(limited?)
     |> Repo.stream_reduce(initial, reducer)
   end
+
+  defp maybe_add_priority_filter(query, false), do: query
+
+  defp maybe_add_priority_filter(query, true) do
+    from(pbo in query,
+      where: not is_nil(pbo.priority)
+    )
+  end
 end

apps/explorer/lib/explorer/chain/pending_operations_helper.ex

@@ -3,7 +3,7 @@ defmodule Explorer.Chain.PendingOperationsHelper do
 
   import Ecto.Query
 
-  alias Explorer.Chain.{Hash, PendingBlockOperation, PendingTransactionOperation, Transaction}
+  alias Explorer.Chain.{Block, Hash, PendingBlockOperation, PendingTransactionOperation, Transaction}
   alias Explorer.{Helper, Repo}
 
   defp transactions_batch_size,
@@ -81,7 +81,7 @@ defmodule Explorer.Chain.PendingOperationsHelper do
       from(
         pto in PendingTransactionOperation,
         join: t in assoc(pto, :transaction),
-        select: %{block_hash: t.block_hash, block_number: t.block_number},
+        select: %{block_hash: t.block_hash, block_number: t.block_number, priority: pto.priority},
         limit: ^batch_size
       )
 
@@ -246,4 +246,82 @@ defmodule Explorer.Chain.PendingOperationsHelper do
     |> block_range_in_query(max_block_number)
     |> Repo.exists?()
   end
+
+  @doc """
+    Inserts pending operations for the given block numbers.
+  """
+  @spec insert_pending_operations([integer()], integer() | nil) :: {[integer()], [Explorer.Chain.Transaction.t()]}
+  def insert_pending_operations(block_numbers, priority \\ nil) do
+    case pending_operations_type() do
+      "transactions" ->
+        default_on_conflict = default_pto_on_conflict()
+        transactions = Transaction.get_transactions_of_block_numbers(block_numbers)
+
+        pto_params =
+          transactions
+          |> Transaction.filter_non_traceable_transactions()
+          |> Enum.map(&%{transaction_hash: &1.hash, priority: priority})
+          |> Helper.add_timestamps()
+
+        Repo.insert_all(PendingTransactionOperation, pto_params,
+          on_conflict: default_on_conflict,
+          conflict_target: [:transaction_hash]
+        )
+
+        {[], transactions}
+
+      "blocks" ->
+        default_on_conflict = default_pbo_on_conflict()
+
+        pbo_params =
+          Block
+          |> where([b], b.number in ^block_numbers)
+          |> where([b], b.consensus == true)
+          |> select([b], %{block_hash: b.hash, block_number: b.number})
+          |> Repo.all()
+          |> add_priority(priority)
+          |> Helper.add_timestamps()
+
+        {_total, inserted} =
+          Repo.insert_all(PendingBlockOperation, pbo_params,
+            on_conflict: default_on_conflict,
+            conflict_target: [:block_hash],
+            returning: [:block_number]
+          )
+
+        {Enum.map(inserted, & &1.block_number), []}
+    end
+  end
+
+  defp default_pbo_on_conflict do
+    from(
+      pending_block_operation in PendingBlockOperation,
+      update: [
+        set: [
+          priority: fragment("EXCLUDED.priority"),
+          inserted_at: fragment("LEAST(?, EXCLUDED.inserted_at)", pending_block_operation.inserted_at),
+          updated_at: fragment("GREATEST(?, EXCLUDED.updated_at)", pending_block_operation.updated_at)
+        ]
+      ],
+      where: is_nil(pending_block_operation.priority) and fragment("EXCLUDED.priority IS NOT NULL")
+    )
+  end
+
+  defp default_pto_on_conflict do
+    from(
+      pending_transaction_operation in PendingTransactionOperation,
+      update: [
+        set: [
+          priority: fragment("EXCLUDED.priority"),
+          inserted_at: fragment("LEAST(?, EXCLUDED.inserted_at)", pending_transaction_operation.inserted_at),
+          updated_at: fragment("GREATEST(?, EXCLUDED.updated_at)", pending_transaction_operation.updated_at)
+        ]
+      ],
+      where: is_nil(pending_transaction_operation.priority) and fragment("EXCLUDED.priority IS NOT NULL")
+    )
+  end
+
+  defp add_priority(params, priority) do
+    Enum.map(params, &Map.merge(&1, %{priority: priority}))
+  end
 end

apps/explorer/lib/explorer/chain/pending_transaction_operation.ex

@@ -26,6 +26,8 @@ defmodule Explorer.Chain.PendingTransactionOperation do
       type: Hash.Full,
       null: false
     )
+
+    field(:priority, :integer)
   end
 
   def changeset(%__MODULE__{} = pending_ops, attrs) do
@@ -69,10 +71,17 @@ defmodule Explorer.Chain.PendingTransactionOperation do
   """
   @spec stream_transactions_with_unfetched_internal_transactions(
           initial :: accumulator,
-          reducer :: (entry :: term(), accumulator -> accumulator)
+          reducer :: (entry :: term(), accumulator -> accumulator),
+          limited? :: boolean(),
+          with_priority? :: boolean()
         ) :: {:ok, accumulator}
         when accumulator: term()
-  def stream_transactions_with_unfetched_internal_transactions(initial, reducer, limited? \\ false)
+  def stream_transactions_with_unfetched_internal_transactions(
+        initial,
+        reducer,
+        limited? \\ false,
+        with_priority? \\ false
+      )
       when is_function(reducer, 2) do
     direction = Application.get_env(:indexer, :internal_transactions_fetch_order)
 
@@ -85,7 +94,16 @@ defmodule Explorer.Chain.PendingTransactionOperation do
       )
 
     query
+    |> maybe_add_priority_filter(with_priority?)
     |> add_fetcher_limit(limited?)
     |> Repo.stream_reduce(initial, reducer)
   end
+
+  defp maybe_add_priority_filter(query, false), do: query
+
+  defp maybe_add_priority_filter(query, true) do
+    from(pto in query,
+      where: not is_nil(pto.priority)
+    )
+  end
 end