Improve local connection

lib/archethic.ex

@@ -87,7 +87,7 @@ defmodule Archethic do
       |> Enum.reject(&(&1.first_public_key == welcome_node_key))
       |> Enum.sort_by(& &1.first_public_key)
       |> P2P.nearest_nodes(welcome_node_patch)
-      |> Enum.filter(&Node.locally_available?/1)
+      |> Enum.filter(&P2P.node_connected?/1)
 
     this_node = Crypto.first_node_public_key()
 

lib/archethic/beacon_chain.ex

@@ -354,7 +354,7 @@ defmodule Archethic.BeaconChain do
     |> Enum.map(fn subset ->
       subset
       |> Election.beacon_storage_nodes(next_summary_date, authorized_nodes)
-      |> Enum.filter(&Node.locally_available?/1)
+      |> Enum.filter(&P2P.node_connected?/1)
       |> P2P.nearest_nodes()
       |> Enum.take(3)
       |> Enum.map(&{&1, subset})

lib/archethic/bootstrap.ex

@@ -115,7 +115,7 @@ defmodule Archethic.Bootstrap do
           bootstrapping_seeds
 
         nodes ->
-          Enum.each(nodes, &P2P.connect_node/1)
+          P2P.connect_nodes(nodes)
           {:ok, closest_nodes} = Sync.get_closest_nodes_and_renew_seeds(nodes, network_patch)
           closest_nodes
       end

lib/archethic/bootstrap/sync.ex

@@ -18,8 +18,6 @@ defmodule Archethic.Bootstrap.Sync do
 
   alias Archethic.SharedSecrets
 
-  alias Archethic.TaskSupervisor
-
   alias Archethic.TransactionChain
   alias Archethic.TransactionChain.Transaction
 
@@ -138,20 +136,7 @@ defmodule Archethic.Bootstrap.Sync do
   def connect_current_node(closest_nodes) do
     case P2P.fetch_nodes_list(true, closest_nodes) do
       {:ok, nodes} ->
-        nodes =
-          Task.Supervisor.async_stream(
-            TaskSupervisor,
-            nodes,
-            fn node ->
-              P2P.connect_node(node)
-              # Wait connection time
-              Process.sleep(500)
-              %Node{node | availability_history: <<1::1>>}
-            end
-          )
-          |> Stream.filter(&match?({:ok, _}, &1))
-          |> Enum.map(fn {:ok, node} -> node end)
-
+        P2P.connect_nodes(nodes)
         Logger.info("Node list refreshed")
         {:ok, nodes}
 
@@ -207,15 +192,8 @@ defmodule Archethic.Bootstrap.Sync do
           (new_seeds ++ closest_nodes)
           |> P2P.distinct_nodes()
           |> Enum.reject(&(&1.first_public_key == Crypto.first_node_public_key()))
-          |> Task.async_stream(fn node ->
-            P2P.connect_node(node)
-            # Wait for connection time
-            Process.sleep(500)
-            # Set node locally available after connection
-            %Node{node | availability_history: <<1::1>>}
-          end)
-          |> Enum.filter(&match?({:ok, _}, &1))
-          |> Enum.map(fn {:ok, node} -> node end)
+
+        P2P.connect_nodes(closest_nodes)
 
         Logger.info("Closest nodes and seeds loaded in the P2P view")
 

lib/archethic/mining/pending_transaction_validation.ex

@@ -849,7 +849,7 @@ defmodule Archethic.Mining.PendingTransactionValidation do
     previous_address
     |> Election.chain_storage_nodes(P2P.authorized_and_available_nodes())
     |> P2P.nearest_nodes()
-    |> Enum.filter(&Node.locally_available?/1)
+    |> Enum.filter(&P2P.node_connected?/1)
     |> get_first_public_key(previous_address)
   end
 

lib/archethic/mining/validation_context.ex

@@ -112,20 +112,20 @@ defmodule Archethic.Mining.ValidationContext do
 
     iex> ValidationContext.new(
     ...>   transaction: %Transaction{},
-    ...>   welcome_node: %Node{last_public_key: "key1", availability_history: <<1::1>>},
-    ...>   coordinator_node: %Node{last_public_key: "key2", availability_history: <<1::1>>},
-    ...>   cross_validation_nodes: [%Node{last_public_key: "key3", availability_history: <<1::1>>}],
-    ...>   chain_storage_nodes: [%Node{last_public_key: "key4", availability_history: <<1::1>>}, %Node{last_public_key: "key5", availability_history: <<1::1>>}],
-    ...>   beacon_storage_nodes: [%Node{last_public_key: "key6", availability_history: <<1::1>>}, %Node{last_public_key: "key7", availability_history: <<1::1>>}]
+    ...>   welcome_node: %Node{last_public_key: "key1"},
+    ...>   coordinator_node: %Node{last_public_key: "key2"},
+    ...>   cross_validation_nodes: [%Node{last_public_key: "key3"}],
+    ...>   chain_storage_nodes: [%Node{last_public_key: "key4"}, %Node{last_public_key: "key5"}],
+    ...>   beacon_storage_nodes: [%Node{last_public_key: "key6"}, %Node{last_public_key: "key7"}]
     ...> )
     %ValidationContext{
       transaction: %Transaction{},
-      welcome_node: %Node{last_public_key: "key1", availability_history: <<1::1>>},
-      coordinator_node: %Node{last_public_key: "key2", availability_history: <<1::1>>},
-      cross_validation_nodes: [%Node{last_public_key: "key3", availability_history: <<1::1>>}],
+      welcome_node: %Node{last_public_key: "key1"},
+      coordinator_node: %Node{last_public_key: "key2"},
+      cross_validation_nodes: [%Node{last_public_key: "key3"}],
       cross_validation_nodes_confirmation: <<0::1>>,
-      chain_storage_nodes: [%Node{last_public_key: "key4", availability_history: <<1::1>>}, %Node{last_public_key: "key5", availability_history: <<1::1>>}],
-      beacon_storage_nodes: [%Node{last_public_key: "key6", availability_history: <<1::1>>}, %Node{last_public_key: "key7", availability_history: <<1::1>>}]
+      chain_storage_nodes: [%Node{last_public_key: "key4"}, %Node{last_public_key: "key5"}],
+      beacon_storage_nodes: [%Node{last_public_key: "key6"}, %Node{last_public_key: "key7"}]
     }
   """
   @spec new(opts :: Keyword.t()) :: t()

lib/archethic/networking/scheduler.ex

@@ -150,7 +150,7 @@ defmodule Archethic.Networking.Scheduler do
   defp handle_new_ip(%Transaction{address: tx_address, data: transaction_data}) do
     nodes =
       P2P.authorized_and_available_nodes()
-      |> Enum.filter(&Node.locally_available?/1)
+      |> Enum.filter(&P2P.node_connected?/1)
       |> P2P.nearest_nodes()
 
     case Utils.await_confirmation(tx_address, nodes) do

lib/archethic/p2p.ex

@@ -40,19 +40,43 @@ defmodule Archethic.P2P do
   def add_and_connect_node(node = %Node{first_public_key: first_public_key}) do
     :ok = MemTable.add_node(node)
     node = get_node_info!(first_public_key)
-    connect_node(node)
+    do_connect_node(node)
   end
 
   @doc """
-  Establish a connection with a node
+  Establish a connection on a list of node in parallel
   """
-  @spec connect_node(Node.t()) :: :ok
-  def connect_node(%Node{
-        ip: ip,
-        port: port,
-        transport: transport,
-        first_public_key: first_public_key
-      }) do
+  @spec connect_nodes(list(Node.t())) :: :ok
+  def connect_nodes(nodes) do
+    Task.Supervisor.async_stream(
+      TaskSupervisor,
+      nodes,
+      fn node = %Node{first_public_key: first_public_key} ->
+        unless node_connected?(node) do
+          do_connect_node(node)
+
+          receive do
+            :connected ->
+              :ok
+
+            {:error, reason} ->
+              Logger.warning("First connection attempt failed for #{inspect(reason)}",
+                node: Base.encode16(first_public_key)
+              )
+          end
+        end
+      end,
+      on_timeout: :kill_task
+    )
+    |> Stream.run()
+  end
+
+  defp do_connect_node(%Node{
+         ip: ip,
+         port: port,
+         transport: transport,
+         first_public_key: first_public_key
+       }) do
     if first_public_key == Crypto.first_node_public_key() do
       :ok
     else
@@ -61,6 +85,18 @@ defmodule Archethic.P2P do
     end
   end
 
+  @doc """
+  Return true if the node connection is established
+  """
+  @spec node_connected?(node :: Node.t()) :: boolean()
+  def node_connected?(%Node{first_public_key: first_public_key}) do
+    if first_public_key == Crypto.first_node_public_key() do
+      true
+    else
+      Client.connected?(first_public_key)
+    end
+  end
+
   @doc """
   List the nodes registered.
   """
@@ -496,29 +532,6 @@ defmodule Archethic.P2P do
   @spec new_bootstrapping_seeds(list(Node.t())) :: :ok
   defdelegate new_bootstrapping_seeds(nodes), to: BootstrappingSeeds, as: :update
 
-  @doc """
-  Create a binary sequence from a list node and set bit regarding their availability
-
-  ## Examples
-
-      iex> P2P.nodes_availability_as_bits([
-      ...>   %Node{availability_history: <<1::1, 0::1>>},
-      ...>   %Node{availability_history: <<0::1, 1::1>>},
-      ...>   %Node{availability_history: <<1::1, 0::1>>}
-      ...> ])
-      <<1::1, 0::1, 1::1>>
-  """
-  @spec nodes_availability_as_bits(list(Node.t())) :: bitstring()
-  def nodes_availability_as_bits(node_list) when is_list(node_list) do
-    Enum.reduce(node_list, <<>>, fn node, acc ->
-      if Node.locally_available?(node) do
-        <<acc::bitstring, 1::1>>
-      else
-        <<acc::bitstring, 0::1>>
-      end
-    end)
-  end
-
   @doc """
   Create a sequence of bits from a list of node and a subset
   by setting the bit where the subset node if found
@@ -573,7 +586,7 @@ defmodule Archethic.P2P do
     previous_public_key
     |> TransactionChain.get_first_public_key()
     |> get_node_info!()
-    |> connect_node()
+    |> do_connect_node()
   end
 
   def load_transaction(tx), do: MemTableLoader.load_transaction(tx)
@@ -695,7 +708,7 @@ defmodule Archethic.P2P do
         consistency_level
       ) do
     nodes
-    |> Enum.filter(&Node.locally_available?/1)
+    |> Enum.filter(&node_connected?/1)
     |> nearest_nodes()
     |> unprioritize_node(Crypto.first_node_public_key())
     |> do_quorum_read(

lib/archethic/p2p/client.ex

@@ -28,4 +28,6 @@ defmodule Archethic.P2P.Client do
               | {:error, :closed}
 
   @callback get_availability_timer(Crypto.key(), boolean()) :: non_neg_integer()
+
+  @callback connected?(Crypto.key()) :: boolean()
 end