improve_synchronization_when_node_ip_changes_#887

lib/archethic/networking/scheduler.ex

@@ -4,24 +4,12 @@ defmodule Archethic.Networking.Scheduler do
   use GenServer
   @vsn Mix.Project.config()[:version]
 
-  alias Archethic.Crypto
+  alias Archethic.{Crypto, P2P, P2P.Node, Networking, TaskSupervisor, Utils, PubSub}
+  alias Archethic.{SelfRepair, TransactionChain}
 
-  alias Archethic.P2P
-  alias Archethic.P2P.Listener, as: P2PListener
-  alias Archethic.P2P.Node
-
-  alias Archethic.Networking.IPLookup
-  alias Archethic.Networking.PortForwarding
-
-  alias Archethic.TaskSupervisor
-
-  alias Archethic.TransactionChain
-  alias Archethic.TransactionChain.Transaction
-  alias Archethic.TransactionChain.TransactionData
-
-  alias Archethic.Utils
-
-  alias Archethic.PubSub
+  alias P2P.Listener, as: P2PListener
+  alias Networking.{IPLookup, PortForwarding}
+  alias TransactionChain.{Transaction, TransactionData}
 
   alias ArchethicWeb.Endpoint, as: WebEndpoint
 
@@ -84,6 +72,22 @@ defmodule Archethic.Networking.Scheduler do
     {:noreply, Map.put(state, :timer, timer)}
   end
 
+  def handle_info({:new_transaction, address}, state) do
+    PubSub.unregister_to_new_transaction_by_address(address)
+    types = [:node, :oracle, :node_shared_secrets, :reward]
+
+    nodes =
+      P2P.authorized_and_available_nodes()
+      |> Enum.filter(&Node.locally_available?/1)
+
+    Task.Supervisor.async_stream_nolink(Archethic.TaskSupervisor, types, fn type ->
+      SelfRepair.resync_network_chain(type, nodes)
+    end)
+    |> Stream.run()
+
+    {:noreply, state}
+  end
+
   defp schedule_update(interval) do
     Process.send_after(self(), :update, Utils.time_offset(interval) * 1000)
   end
@@ -101,20 +105,23 @@ defmodule Archethic.Networking.Scheduler do
       origin_public_key = Crypto.origin_node_public_key()
       key_certificate = Crypto.get_key_certificate(origin_public_key)
 
-      Transaction.new(:node, %TransactionData{
-        code: code,
-        content:
-          Node.encode_transaction_content(
-            ip,
-            p2p_port,
-            web_port,
-            transport,
-            reward_address,
-            origin_public_key,
-            key_certificate
-          )
-      })
-      |> Archethic.send_new_transaction()
+      tx =
+        Transaction.new(:node, %TransactionData{
+          code: code,
+          content:
+            Node.encode_transaction_content(
+              ip,
+              p2p_port,
+              web_port,
+              transport,
+              reward_address,
+              origin_public_key,
+              key_certificate
+            )
+        })
+
+      Archethic.send_new_transaction(tx)
+      PubSub.register_to_new_transaction_by_address(tx.address)
     else
       :error ->
         Logger.warning("Cannot open port")

lib/archethic/self_repair.ex

@@ -224,7 +224,7 @@ defmodule Archethic.SelfRepair do
     :ok
   end
 
-  @spec resync_network_chain(atom(), list(Node.t()) | []) :: :ok | :error
+  @spec resync_network_chain(type :: atom(), nodes :: list(Node.t()) | []) :: :ok | :error
   def resync_network_chain(_, []),
     do: Logger.notice("Enforce Resync of Network Txs: No-Nodes")