enforce network tx sync

lib/archethic/bootstrap.ex

@@ -17,9 +17,11 @@ defmodule Archethic.Bootstrap do
   alias Archethic.SelfRepair
 
   require Logger
-
   use Task
 
+  alias Archethic.TransactionChain
+  alias Archethic.Replication
+
   @doc """
   Start the bootstrapping as a task
   """
@@ -206,12 +208,92 @@ defmodule Archethic.Bootstrap do
       Logger.info("Synchronization finished")
     end
 
+    nodes = P2P.authorized_and_available_nodes()
+    resync_network_chain([:oracle, :node_shared_secrets], nodes)
+
     Sync.publish_end_of_sync()
     SelfRepair.start_scheduler()
-
     :persistent_term.put(:archethic_up, :up)
   end
 
+  # when no node loaded in memtable
+  @spec resync_network_chain(any, maybe_improper_list) :: :ok
+  def resync_network_chain(_type_list, _nodes = []),
+    do: Logger.info("Enforced Resync of Network Txs: failure, No-Nodes")
+
+  def resync_network_chain(type_list, nodes) when is_list(nodes) do
+    # spawn process for each type of txn to sync
+    Task.Supervisor.async_stream_nolink(Archethic.TaskSupervisor, type_list, fn type ->
+      # acquire a(any) address of type txn, (for querying DB)
+      case TransactionChain.list_addresses_by_type(type) |> Enum.at(0) do
+        nil ->
+          # no txn avilable for that type txn
+          # can fetch that "type" network chain
+          Logger.info("Enforced Resync of Network Transactions: failure , Transaction_addr: NA ",
+            transaction_type: type
+          )
+
+          :ok
+
+        addr ->
+          # get last txn locally available
+          last_addr = TransactionChain.get_last_address(addr)
+
+          try do
+            # fetch txs remotely post last_addr
+            TransactionChain.stream_remotely(last_addr, nodes, last_addr)
+            |> Enum.to_list()
+            |> List.flatten()
+            |> Enum.each(fn tx ->
+              if TransactionChain.transaction_exists?(tx.addr) do
+                # if tx already exists , Dont validate_and_store
+                Logger.info(
+                  "Enforced Resyncing  Network Transactions: , #{tx.address |> Base.encode16()}  status: :aklready_available",
+                  transaction_type: type
+                )
+
+                :ok
+              else
+                # when tx , not available Validate and store tx
+
+                case Replication.validate_and_store_transaction(tx) do
+                  :ok ->
+                    Logger.info(
+                      "Enforced Resyncing  Network Transactions: , #{tx.address |> Base.encode16()}  status: :stored",
+                      transaction_type: type
+                    )
+
+                    :ok
+
+                  {:error, e} ->
+                    Logger.info(
+                      "Enforced Resyncing  Network Transactions: , #{tx.address |> Base.encode16()}  status: :error_during_replication",
+                      transaction_type: type
+                    )
+
+                    Logger.debug(e)
+                end
+              end
+            end)
+          rescue
+            e ->
+              # network error
+              Logger.info(
+                "Enforced Resync of Network Transactions: failure, error in fetching Network Transaction Chain ",
+                transaction_type: type
+              )
+
+              Logger.debug(e)
+
+              :ok
+          end
+      end
+    end)
+    |> Stream.run()
+
+    Logger.info("Enforced Resync of Network Transactions: Exiting")
+  end
+
   defp first_initialization(
          ip,
          port,

test/archethic/bootstrap_test.exs

@@ -52,6 +52,11 @@ defmodule Archethic.BootstrapTest do
 
   import Mox
 
+  use ExUnit.Case
+
+  # import ExUnit.CaptureLog
+  require Logger
+
   setup do
     start_supervised!({BeaconSummaryTimer, interval: "0 0 * * * * *"})
     start_supervised!({BeaconSlotTimer, interval: "0 * * * * * *"})
@@ -431,4 +436,111 @@ defmodule Archethic.BootstrapTest do
       Process.sleep(100)
     end
   end
+
+  describe "resync_network_chain/2" do
+    # test "Should Log error when no nodes available" do
+    #   assert capture_log([level: :info], fn ->
+    #            Bootstrap.resync_network_chain(
+    #              [:oracle, :node_shared_secrets, :mint_rewards],
+    #              _nodes = []
+    #            )
+    #          end) =~ "Enforced Resync of Network Txs: failure, No-Nodes"
+    # end
+
+    # test "Should Log error when no Address for network type chain is found" do
+    #   pbk =
+    #     Crypto.derive_keypair("node_for_bootstrap", 0)
+    #     |> elem(0)
+
+    #   P2P.add_and_connect_node(%Node{
+    #     ip: {80, 10, 20, 102},
+    #     port: 3006,
+    #     http_port: 4000,
+    #     last_public_key: pbk,
+    #     first_public_key: pbk,
+    #     network_patch: "AAA",
+    #     geo_patch: "AAA",
+    #     available?: true,
+    #     enrollment_date: DateTime.utc_now(),
+    #     reward_address: pbk |> Crypto.derive_address(),
+    #     authorized?: true,
+    #     authorization_date: DateTime.utc_now() |> DateTime.add(-86_400)
+    #   })
+
+    #   MockDB
+    #   |> expect(:list_addresses_by_type, fn :oracle ->
+    #     []
+    #   end)
+
+    #   assert capture_log([level: :info], fn ->
+    #            Bootstrap.resync_network_chain(
+    #              [:oracle],
+    #              _nodes = P2P.authorized_and_available_nodes()
+    #            )
+    #          end) =~ "Enforced Resync of Network Txs: failure, No-Nodes"
+    # end
+
+    test "should Retrieve and Store Network tx's" do
+      pbk =
+        Crypto.derive_keypair("node_for_bootstrap1", 0)
+        |> elem(0)
+
+      P2P.add_and_connect_node(%Node{
+        ip: {80, 10, 20, 102},
+        port: 3002,
+        http_port: 4000,
+        last_public_key: pbk,
+        first_public_key: pbk,
+        network_patch: "AAA",
+        geo_patch: "AAA",
+        available?: true,
+        enrollment_date: DateTime.utc_now(),
+        reward_address: pbk |> Crypto.derive_address(),
+        authorized?: true,
+        authorization_date: DateTime.utc_now() |> DateTime.add(-86_400),
+        synced?: true
+      })
+
+      addr = Crypto.derive_keypair("oracle_summary_test", 1) |> elem(0) |> Crypto.derive_address()
+
+      MockDB
+      |> expect(:list_addresses_by_type, 1, fn :oracle -> [addr] end)
+      |> expect(:get_last_chain_address, 1, fn addr -> addr end)
+      |> stub(:transaction_exists?, fn _ -> true end)
+
+      MockClient
+      |> stub(:send_message, fn
+        _, %GetTransactionChainLength{}, _ ->
+          %TransactionChainLength{length: 4}
+
+        _, %GetTransactionChain{}, _ ->
+          {:ok,
+           %TransactionList{
+             transactions: [
+               %Transaction{
+                 address:
+                   Crypto.derive_keypair("oracle_summary_test", 2)
+                   |> elem(0)
+                   |> Crypto.derive_address()
+               },
+               %Transaction{
+                 address:
+                   Crypto.derive_keypair("oracle_summary_test", 3)
+                   |> elem(0)
+                   |> Crypto.derive_address()
+               },
+               %Transaction{
+                 address:
+                   Crypto.derive_keypair("oracle_summary_test", 4)
+                   |> elem(0)
+                   |> Crypto.derive_address()
+               }
+             ]
+           }}
+      end)
+
+      :ok =
+        Bootstrap.resync_network_chain([:oracle], _nodes = P2P.authorized_and_available_nodes())
+    end
+  end
 end