Fix replication chain ingestion

lib/archethic/replication.ex

@@ -39,6 +39,7 @@ defmodule Archethic.Replication do
   alias Archethic.TransactionChain
   alias Archethic.TransactionChain.Transaction
   alias Archethic.TransactionChain.Transaction.ValidationStamp
+  alias Archethic.TransactionChain.Transaction.ValidationStamp.LedgerOperations
 
   alias Archethic.Utils
 
@@ -107,18 +108,39 @@ defmodule Archethic.Replication do
           |> stream_previous_chain(download_nodes)
           |> Stream.reject(&Enum.empty?/1)
           |> Stream.flat_map(& &1)
-          |> Stream.each(fn tx = %Transaction{address: address} ->
-            TransactionChain.write_transaction(tx)
-
-            # There is some case where a transaction is not replicated while it should
-            # because of some latency or network issue. So when we replicate a past chain
-            # we also ingest the transaction if we are storage node of it
-
-            if address
-               |> Election.chain_storage_nodes(download_nodes)
-               |> Utils.key_in_node_list?(Crypto.first_node_public_key()),
-               do: ingest_transaction(tx, false)
-          end)
+          |> Stream.each(
+            fn tx = %Transaction{
+                 address: address,
+                 validation_stamp: %ValidationStamp{
+                   ledger_operations: %LedgerOperations{
+                     transaction_movements: transaction_movements
+                   }
+                 }
+               } ->
+              TransactionChain.write_transaction(tx)
+
+              # There is some case where a transaction is not replicated while it should
+              # because of some latency or network issue. So when we replicate a past chain
+              # we also ingest the transaction if we are storage node of it
+
+              io_addresses = Enum.map(transaction_movements, & &1.to)
+
+              cond do
+                address
+                |> Election.chain_storage_nodes(download_nodes)
+                |> Utils.key_in_node_list?(Crypto.first_node_public_key()) ->
+                  ingest_transaction(tx, false)
+
+                io_addresses
+                |> Election.io_storage_nodes(download_nodes)
+                |> Utils.key_in_node_list?(Crypto.first_node_public_key()) ->
+                  ingest_transaction(tx, true)
+
+                true ->
+                  :ok
+              end
+            end
+          )
           |> Stream.run()
 
           TransactionChain.write_transaction(tx)

test/archethic/mining/pending_transaction_validation_test.exs

@@ -785,7 +785,7 @@ defmodule Archethic.Mining.PendingTransactionValidationTest do
             })
         })
 
-      assert {:error, error} = PendingTransactionValidation.validate(tx, DateTime.utc_now())
+      assert {:error, _error} = PendingTransactionValidation.validate(tx, DateTime.utc_now())
     end
 
     test "should return :error when we deploy a wrong aeweb ref transaction" do
@@ -810,7 +810,7 @@ defmodule Archethic.Mining.PendingTransactionValidationTest do
             })
         })
 
-      assert {:error, reason} = PendingTransactionValidation.validate(tx, DateTime.utc_now())
+      assert {:error, _reason} = PendingTransactionValidation.validate(tx, DateTime.utc_now())
     end
   end
 end