"Rolling" SelfRepair

config/test.exs

@@ -143,6 +143,7 @@ config :archethic, Archethic.SelfRepair.Scheduler,
   interval: "0 0 * * * * *",
   availability_application: 1
 
+config :archethic, Archethic.SelfRepair.NetworkView, enabled: false
 config :archethic, Archethic.SelfRepair.Notifier, enabled: false
 
 config :archethic, Archethic.SelfRepair.Sync,

lib/archethic.ex

@@ -7,9 +7,12 @@ defmodule Archethic do
   alias __MODULE__.{SelfRepair, TransactionChain, SharedSecrets, TaskSupervisor}
   alias __MODULE__.Contracts.{Interpreter, Contract}
 
+  alias SelfRepair.NetworkView
+  alias SelfRepair.NetworkChain
+
   alias Message.{NewTransaction, NotFound, StartMining}
   alias Message.{Balance, GetBalance, GetTransactionSummary}
-  alias Message.{StartMining, Ok, TransactionSummaryMessage}
+  alias Message.{StartMining, Ok, Error, TransactionSummaryMessage}
 
   alias TransactionChain.{Transaction, TransactionInput, TransactionSummary}
 
@@ -57,15 +60,16 @@ defmodule Archethic do
         welcome_node_key \\ Crypto.first_node_public_key()
       ) do
     if P2P.authorized_and_available_node?() do
-      case SharedSecrets.verify_synchronization() do
+      case NetworkChain.verify_synchronization(:node_shared_secrets) do
         :ok ->
           do_send_transaction(tx, welcome_node_key)
 
         :error ->
           forward_transaction(tx, welcome_node_key)
 
-        {:error, last_address_to_sync} ->
-          SelfRepair.resync(last_address_to_sync)
+        {:error, addresses} ->
+          SharedSecrets.genesis_address(:node_shared_secrets) |> SelfRepair.resync(addresses, [])
+
           forward_transaction(tx, welcome_node_key)
       end
     else
@@ -142,12 +146,73 @@ defmodule Archethic do
     message = %StartMining{
       transaction: tx,
       welcome_node_public_key: get_welcome_node_public_key(tx_type, welcome_node_key),
-      validation_node_public_keys: Enum.map(validation_nodes, & &1.last_public_key)
+      validation_node_public_keys: Enum.map(validation_nodes, & &1.last_public_key),
+      network_chains_view_hash: NetworkView.get_chains_hash(),
+      p2p_view_hash: NetworkView.get_p2p_hash()
     }
 
-    P2P.broadcast_message(validation_nodes, message)
+    Task.Supervisor.async_stream_nolink(
+      Archethic.TaskSupervisor,
+      validation_nodes,
+      &P2P.send_message(&1, message),
+      ordered: false,
+      on_timeout: :kill_task,
+      timeout: Message.get_timeout(message) + 2000
+    )
+    |> Stream.filter(&match?({:ok, _}, &1))
+    |> Stream.map(fn {:ok, res} -> res end)
+    |> Enum.reduce(
+      %{
+        ok: 0,
+        network_chains_resync_needed: false,
+        p2p_resync_needed: false
+      },
+      &reduce_start_mining_responses/2
+    )
+    |> then(fn aggregated_responses ->
+      maybe_start_resync(aggregated_responses)
+
+      if should_forward_transaction?(aggregated_responses, length(validation_nodes)) do
+        forward_transaction(tx, welcome_node_key)
+      else
+        :ok
+      end
+    end)
+  end
+
+  defp reduce_start_mining_responses({:ok, %Ok{}}, acc) do
+    %{acc | ok: acc.ok + 1}
+  end
+
+  defp reduce_start_mining_responses({:ok, %Error{reason: :network_chains_sync}}, acc) do
+    %{acc | network_chains_resync_needed: true}
+  end
+
+  defp reduce_start_mining_responses({:ok, %Error{reason: :p2p_sync}}, acc) do
+    %{acc | p2p_resync_needed: true}
+  end
+
+  defp reduce_start_mining_responses({:ok, %Error{reason: :both_sync}}, acc) do
+    %{acc | network_chains_resync_needed: true, p2p_resync_needed: true}
   end
 
+  defp reduce_start_mining_responses(_, acc) do
+    acc
+  end
+
+  defp maybe_start_resync(aggregated_responses) do
+    if aggregated_responses.network_chains_resync_needed do
+      NetworkChain.asynchronous_resync_many([:origin, :oracle, :node_shared_secrets])
+    end
+
+    if aggregated_responses.p2p_resync_needed do
+      NetworkChain.asynchronous_resync(:node)
+    end
+  end
+
+  defp should_forward_transaction?(_, 1), do: false
+  defp should_forward_transaction?(%{ok: ok_count}, _), do: ok_count < 2
+
   # Since welcome node is not anymore constant, as we want unauthorised
   # nodes to do some labor. Following bootstrapping, the txn of a new node
   # is sent, with the welcome node being the same new node whose information

lib/archethic/application.ex

@@ -97,7 +97,10 @@ defmodule Archethic.Application do
          http_port: http_port,
          transport: transport
        )},
-      MetricSupervisor
+      MetricSupervisor,
+
+      # a registry used in Utils to ensure a function is executed at most once concurrently
+      {Registry, keys: :unique, name: Archethic.RunExclusiveRegistry}
     ]
 
     opts = [strategy: :rest_for_one, name: Archethic.Supervisor]

lib/archethic/bootstrap.ex

@@ -5,7 +5,7 @@ defmodule Archethic.Bootstrap do
 
   alias Archethic
   alias Archethic.{Bootstrap, Crypto, Networking, P2P, P2P.Node}
-  alias Archethic.{SelfRepair, SelfRepair, TransactionChain}
+  alias Archethic.{SelfRepair, SelfRepair.NetworkChain, TransactionChain}
 
   alias Bootstrap.{NetworkInit, Sync, TransactionHandler}
   alias TransactionChain.{Transaction, TransactionData}
@@ -210,7 +210,9 @@ defmodule Archethic.Bootstrap do
     end
 
     Archethic.Bootstrap.NetworkConstraints.persist_genesis_address()
-    resync_network_chain()
+
+    Logger.info("Enforced Resync: Started!")
+    NetworkChain.synchronous_resync_many([:node, :oracle, :origin, :node_shared_secrets])
 
     Sync.publish_end_of_sync()
     SelfRepair.start_scheduler()
@@ -219,23 +221,6 @@ defmodule Archethic.Bootstrap do
     Archethic.PubSub.notify_node_status(:node_up)
   end
 
-  def resync_network_chain() do
-    Logger.info("Enforced Resync: Started!")
-
-    if P2P.authorized_and_available_node?() do
-      # evict this node
-      nodes =
-        Enum.reject(
-          P2P.authorized_and_available_nodes(),
-          &(&1.first_public_key == Crypto.first_node_public_key())
-        )
-
-      # blocking code
-      [:oracle, :node_shared_secrets, :reward]
-      |> Enum.each(&SelfRepair.resync_network_chain(&1, nodes))
-    end
-  end
-
   defp first_initialization(
          ip,
          port,

lib/archethic/bootstrap/network_constraints.ex

@@ -68,7 +68,7 @@ defmodule Archethic.Bootstrap.NetworkConstraints do
       OracleChain.update_summ_gen_addr()
       Logger.info("Oracle Gen Addr Table: Loaded")
 
-      if gen_addr = OracleChain.genesis_address() do
+      if gen_addr = OracleChain.genesis_addresses() do
         Logger.debug("New Oracle Gen Addr")
         Logger.debug(gen_addr)
       end

lib/archethic/bootstrap/sync.ex

@@ -13,8 +13,6 @@ defmodule Archethic.Bootstrap.Sync do
   alias Archethic.P2P.Message.EncryptedStorageNonce
   alias Archethic.P2P.Message.GetBootstrappingNodes
   alias Archethic.P2P.Message.GetStorageNonce
-  alias Archethic.P2P.Message.ListNodes
-  alias Archethic.P2P.Message.NodeList
   alias Archethic.P2P.Message.NotifyEndOfNodeSync
   alias Archethic.P2P.Node
 
@@ -137,44 +135,16 @@ defmodule Archethic.Bootstrap.Sync do
   """
   @spec load_node_list() :: :ok | {:error, :network_issue}
   def load_node_list() do
-    current_nodes = P2P.authorized_and_available_nodes()
-
-    last_updated_nodes =
-      fn new_node = %Node{
-           first_public_key: public_key,
-           last_update_date: update_date
-         },
-         acc ->
-        previous_node =
-          %Node{last_update_date: previous_update_date} = Map.get(acc, public_key, new_node)
-
-        node =
-          if DateTime.compare(update_date, previous_update_date) == :gt,
-            do: new_node,
-            else: previous_node
-
-        Map.put(acc, public_key, node)
-      end
-
-    conflict_resolver = fn results ->
-      nodes =
-        Enum.flat_map(results, fn %NodeList{nodes: nodes} -> nodes end)
-        |> Enum.reduce(%{}, fn node, acc -> last_updated_nodes.(node, acc) end)
-        |> Map.values()
-
-      %NodeList{nodes: nodes}
-    end
-
-    case P2P.quorum_read(current_nodes, %ListNodes{}, conflict_resolver) do
-      {:ok, %NodeList{nodes: nodes}} ->
+    case P2P.fetch_nodes_list() do
+      {:ok, nodes} ->
         Enum.each(nodes, &P2P.add_and_connect_node/1)
         # After loading all current nodes, we update the p2p view with the last one stored in DB
         # to have a proper view for the next beacon summary to self repair
         SelfRepair.last_sync_date() |> P2P.reload_last_view()
         Logger.info("Node list refreshed")
 
-      error ->
-        error
+      {:error, :network_issue} ->
+        {:error, :network_issue}
     end
   end
 

lib/archethic/mining/pending_transaction_validation.ex

@@ -97,7 +97,7 @@ defmodule Archethic.Mining.PendingTransactionValidation do
           %NotFound{}
 
         res ->
-          Enum.sort_by(res, & &1.timestamp, {:desc, DateTime})
+          Enum.sort_by(res, & &1.transaction_summary.timestamp, {:desc, DateTime})
           |> List.first()
       end
     end
@@ -704,7 +704,7 @@ defmodule Archethic.Mining.PendingTransactionValidation do
        when type in [:oracle, :oracle_summary] do
     # multipe txn chain based on summary date
 
-    case OracleChain.genesis_address() do
+    case OracleChain.genesis_addresses() do
       nil ->
         false
 

lib/archethic/networking/scheduler.ex

@@ -5,7 +5,8 @@ defmodule Archethic.Networking.Scheduler do
   @vsn Mix.Project.config()[:version]
 
   alias Archethic.{Crypto, P2P, P2P.Node, Networking, TaskSupervisor, Utils, PubSub}
-  alias Archethic.{SelfRepair, TransactionChain}
+  alias Archethic.TransactionChain
+  alias Archethic.SelfRepair.NetworkChain
 
   alias Networking.{IPLookup, PortForwarding}
   alias TransactionChain.{Transaction, TransactionData}
@@ -139,11 +140,6 @@ defmodule Archethic.Networking.Scheduler do
 
     Utils.await_confirmation(tx_address, nodes)
 
-    types = [:node, :oracle, :node_shared_secrets, :reward]
-
-    Task.Supervisor.async_stream_nolink(Archethic.TaskSupervisor, types, fn type ->
-      SelfRepair.resync_network_chain(type, nodes)
-    end)
-    |> Stream.run()
+    NetworkChain.asynchronous_resync_many([:node, :oracle, :node_shared_secrets, :origin])
   end
 end

lib/archethic/oracle_chain.ex

@@ -192,8 +192,22 @@ defmodule Archethic.OracleChain do
   @doc """
   Returns current and previous summary_time genesis address of oracle chain
   """
-  @spec genesis_address() :: map() | nil
-  defdelegate genesis_address(),
+  @spec genesis_addresses() :: map() | nil
+  defdelegate genesis_addresses(),
     to: MemTable,
     as: :get_addr
+
+  @doc """
+  Returns current genesis address of oracle chain
+  """
+  @spec genesis_address() :: binary() | nil
+  def genesis_address() do
+    case genesis_addresses() do
+      %{current: {address, _time}} ->
+        address
+
+      _ ->
+        nil
+    end
+  end
 end

lib/archethic/p2p.ex

@@ -8,6 +8,9 @@ defmodule Archethic.P2P do
 
   alias __MODULE__.{BootstrappingSeeds, Client, GeoPatch, MemTable, MemTableLoader, Message, Node}
 
+  alias __MODULE__.Message.NodeList
+  alias __MODULE__.Message.ListNodes
+
   require Logger
 
   @type supported_transport :: :tcp
@@ -66,6 +69,46 @@ defmodule Archethic.P2P do
   @spec list_nodes() :: list(Node.t())
   defdelegate list_nodes, to: MemTable
 
+  @doc """
+  Fetch the list of nodes from close nodes
+  """
+  @spec fetch_nodes_list() :: {:ok, list(Node.t())} | {:error, :network_issue}
+  def fetch_nodes_list() do
+    last_updated_nodes =
+      fn new_node = %Node{
+           first_public_key: public_key,
+           last_update_date: update_date
+         },
+         acc ->
+        previous_node =
+          %Node{last_update_date: previous_update_date} = Map.get(acc, public_key, new_node)
+
+        node =
+          if DateTime.compare(update_date, previous_update_date) == :gt,
+            do: new_node,
+            else: previous_node
+
+        Map.put(acc, public_key, node)
+      end
+
+    conflict_resolver = fn results ->
+      nodes =
+        Enum.flat_map(results, fn %NodeList{nodes: nodes} -> nodes end)
+        |> Enum.reduce(%{}, fn node, acc -> last_updated_nodes.(node, acc) end)
+        |> Map.values()
+
+      %NodeList{nodes: nodes}
+    end
+
+    case quorum_read(authorized_and_available_nodes(), %ListNodes{}, conflict_resolver) do
+      {:ok, %NodeList{nodes: nodes}} ->
+        {:ok, nodes}
+
+      {:error, :network_issue} ->
+        {:error, :network_issue}
+    end
+  end
+
   @doc """
   Return the list of available nodes
   """

lib/archethic/p2p/message/error.ex

@@ -10,6 +10,9 @@ defmodule Archethic.P2P.Message.Error do
           | :invalid_transaction
           | :invalid_attestation
           | :transaction_already_exists
+          | :network_chains_sync
+          | :p2p_sync
+          | :both_sync
 
   @type t :: %__MODULE__{
           reason: reason()
@@ -33,6 +36,9 @@ defmodule Archethic.P2P.Message.Error do
   def serialize_reason(:invalid_transaction), do: 1
   def serialize_reason(:invalid_attestation), do: 2
   def serialize_reason(:transaction_already_exists), do: 3
+  def serialize_reason(:network_chains_sync), do: 4
+  def serialize_reason(:p2p_sync), do: 5
+  def serialize_reason(:both_sync), do: 6
 
   @doc """
   Deserialize an error reason
@@ -42,4 +48,7 @@ defmodule Archethic.P2P.Message.Error do
   def deserialize_reason(1), do: :invalid_transaction
   def deserialize_reason(2), do: :invalid_attestation
   def deserialize_reason(3), do: :transaction_already_exists
+  def deserialize_reason(4), do: :network_chains_sync
+  def deserialize_reason(5), do: :p2p_sync
+  def deserialize_reason(6), do: :both_sync
 end