Release 1.1.3

lib/archethic/application.ex

@@ -56,6 +56,10 @@ defmodule Archethic.Application do
   require Logger
 
   def start(_type, _args) do
+    # First start the migration process synchronously
+    Application.spec(:archethic, :vsn) |> Migrate.run(false)
+
+    # Then start the Archethic Supervisor
     p2p_endpoint_conf = Application.get_env(:archethic, Archethic.P2P.Listener)
     web_endpoint_conf = Application.get_env(:archethic, ArchethicWeb.Endpoint)
 

lib/archethic/db.ex

@@ -57,6 +57,8 @@ defmodule Archethic.DB do
   @callback list_chain_public_keys(binary(), DateTime.t()) ::
               Enumerable.t() | list({binary(), DateTime.t()})
 
+  @callback get_last_chain_address_stored(genesis_address :: Crypto.prepended_hash()) ::
+              Crypto.prepended_hash() | nil
   @callback get_last_chain_address(binary()) :: {binary(), DateTime.t()}
   @callback get_last_chain_address(binary(), DateTime.t()) :: {binary(), DateTime.t()}
   @callback get_last_chain_public_key(binary()) :: Crypto.key()

lib/archethic/db/embedded_impl.ex

@@ -229,6 +229,15 @@ defmodule Archethic.DB.EmbeddedImpl do
     ChainIndex.count_transactions_by_type(type)
   end
 
+  @doc """
+  Return the last address stored for a chain
+  """
+  @spec get_last_chain_address_stored(genesis_address :: Crypto.prepended_hash()) ::
+          Crypto.prepended_hash() | nil
+  def get_last_chain_address_stored(genesis_address) when is_binary(genesis_address) do
+    ChainIndex.get_last_chain_address_stored(genesis_address, db_path())
+  end
+
   @doc """
   Return the last address from the given transaction's address until the given date along with its timestamp
   """

lib/archethic/db/embedded_impl/chain_index.ex

@@ -406,6 +406,33 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndex do
     :ok
   end
 
+  @doc """
+  Set the last stored address
+  """
+  @spec set_last_chain_address_stored(
+          genesis_address :: Crypto.prepended_hash(),
+          tx_address :: Crypto.prepended_hash(),
+          db_path :: binary()
+        ) :: :ok
+  def set_last_chain_address_stored(genesis_address, tx_address, db_path),
+    do: last_chain_address_stored_path(db_path, genesis_address) |> File.write!(tx_address)
+
+  @doc """
+  Return the last address stored for a chain
+  """
+  @spec get_last_chain_address_stored(
+          genesis_address :: Crypto.prepended_hash(),
+          db_path :: binary()
+        ) :: Crypto.prepended_hash() | nil
+  def get_last_chain_address_stored(genesis_address, db_path) do
+    filename = last_chain_address_stored_path(db_path, genesis_address)
+
+    case File.read(filename) do
+      {:ok, address} -> address
+      _ -> nil
+    end
+  end
+
   @doc """
   Reference a new transaction address for the genesis address at the transaction time
   """
@@ -713,6 +740,13 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndex do
     ])
   end
 
+  defp last_chain_address_stored_path(db_path, genesis_address) do
+    Path.join([
+      ChainWriter.base_chain_path(db_path),
+      "#{Base.encode16(genesis_address)}-last-stored-address"
+    ])
+  end
+
   defp type_path(db_path, type) do
     Path.join([ChainWriter.base_chain_path(db_path), Atom.to_string(type)])
   end

lib/archethic/db/embedded_impl/chain_reader.ex

@@ -167,7 +167,7 @@ defmodule Archethic.DB.EmbeddedImpl.ChainReader do
           genesis_address :: binary(),
           fields :: list(),
           db_path :: binary()
-        ) :: Enumerable.t()
+        ) :: Enumerable.t() | list(Transaction.t())
   def stream_chain(genesis_address, fields, db_path) do
     filepath = ChainWriter.chain_path(db_path, genesis_address)
 

lib/archethic/db/embedded_impl/chain_writer.ex

@@ -188,6 +188,7 @@ defmodule Archethic.DB.EmbeddedImpl.ChainWriter do
 
     ChainIndex.add_tx(tx_address, genesis_address, encoded_size, db_path)
     ChainIndex.add_tx_type(tx_type, tx_address, db_path)
+    ChainIndex.set_last_chain_address_stored(genesis_address, tx_address, db_path)
     ChainIndex.set_last_chain_address(genesis_address, tx_address, timestamp, db_path)
     ChainIndex.set_public_key(genesis_address, previous_public_key, timestamp, db_path)
 

lib/archethic/p2p/mem_table.ex

@@ -54,21 +54,6 @@ defmodule Archethic.P2P.MemTable do
     {:ok, []}
   end
 
-  def code_change("1.1.1", state, _extra) do
-    # This match_spec remove the key :"$9" which is the availability_history
-    match_spec = [
-      {{:"$1", :"$2", :"$3", :"$4", :"$5", :"$6", :"$7", :"$8", :"$9", :"$10", :"$11", :"$12",
-        :"$13", :"$14", :"$15", :"$16", :"$17", :"$18"}, [],
-       [
-         {{:"$1", :"$2", :"$3", :"$4", :"$5", :"$6", :"$7", :"$8", :"$10", :"$11", :"$12", :"$13",
-           :"$14", :"$15", :"$16", :"$17", :"$18"}}
-       ]}
-    ]
-
-    :ets.select_replace(@discovery_table, match_spec)
-    {:ok, state}
-  end
-
   @doc """
   Add a node into the P2P view.
 

lib/archethic/transaction_chain.ex

@@ -1027,13 +1027,9 @@ defmodule Archethic.TransactionChain do
   @doc """
   Retrieve the last transaction address for a chain stored locally
   """
-  @spec get_last_stored_address(genesis_address :: binary()) :: binary() | nil
-  def get_last_stored_address(genesis_address) do
-    list_chain_addresses(genesis_address)
-    |> Enum.reduce_while(nil, fn {address, _}, acc ->
-      if transaction_exists?(address), do: {:cont, address}, else: {:halt, acc}
-    end)
-  end
+  @spec get_last_stored_address(genesis_address :: Crypto.prepended_hash()) ::
+          Crypto.prepended_hash() | nil
+  defdelegate get_last_stored_address(genesis_address), to: DB, as: :get_last_chain_address_stored
 
   @doc """
   Retrieve the genesis address for a chain from P2P Quorom

mix.exs

@@ -4,7 +4,7 @@ defmodule Archethic.MixProject do
   def project do
     [
       app: :archethic,
-      version: "1.1.2",
+      version: "1.1.3",
       build_path: "_build",
       config_path: "config/config.exs",
       deps_path: "deps",
@@ -33,8 +33,7 @@ defmodule Archethic.MixProject do
         :xmerl,
         :crypto
       ],
-      mod: {Archethic.Application, []},
-      start_phases: [migrate: []]
+      mod: {Archethic.Application, []}
     ]
   end
 

priv/migration_tasks/prod/1.1.3@create_last_stored_index.exs

@@ -0,0 +1,78 @@
+defmodule Migration_1_1_3 do
+  @moduledoc false
+
+  alias Archethic.Crypto
+
+  alias Archethic.DB.EmbeddedImpl
+  alias Archethic.DB.EmbeddedImpl.ChainIndex
+  alias Archethic.DB.EmbeddedImpl.ChainReader
+  alias Archethic.DB.EmbeddedImpl.ChainWriter
+
+  alias Archethic.TransactionChain.Transaction
+
+  def run() do
+    db_path = EmbeddedImpl.db_path()
+
+    get_genesis_addresses(db_path) |> index_last_addresses(db_path)
+  end
+
+  defp get_genesis_addresses(db_path) do
+    if chain_index_started?() do
+      ChainIndex.list_genesis_addresses()
+    else
+      get_genesis_addresses_from_index(db_path)
+    end
+  end
+
+  defp chain_index_started?(), do: Process.whereis(ChainIndex) != nil
+
+  defp get_genesis_addresses_from_index(db_path) do
+    Task.async_stream(0..255, fn subset ->
+      filename = index_summary_path(db_path, subset)
+
+      case File.open(filename, [:binary, :read]) do
+        {:ok, fd} ->
+          do_get_genesis_addresses_from_index(fd, [])
+
+        {:error, _} ->
+          []
+      end
+    end)
+    |> Stream.flat_map(fn {:ok, genesis_addresses} -> genesis_addresses end)
+    |> Stream.uniq()
+  end
+
+  defp index_summary_path(db_path, subset) do
+    Path.join([ChainWriter.base_chain_path(db_path), "#{Base.encode16(<<subset>>)}-summary"])
+  end
+
+  defp do_get_genesis_addresses_from_index(fd, acc) do
+    with {:ok, <<_current_curve_id::8, current_hash_type::8>>} <- :file.read(fd, 2),
+         hash_size <- Crypto.hash_size(current_hash_type),
+         {:ok, _current_digest} <- :file.read(fd, hash_size),
+         {:ok, <<genesis_curve_id::8, genesis_hash_type::8>>} <- :file.read(fd, 2),
+         hash_size <- Crypto.hash_size(genesis_hash_type),
+         {:ok, genesis_digest} <- :file.read(fd, hash_size),
+         {:ok, <<_size::32, _offset::32>>} <- :file.read(fd, 8) do
+      genesis_address = <<genesis_curve_id::8, genesis_hash_type::8, genesis_digest::binary>>
+
+      do_get_genesis_addresses_from_index(fd, [genesis_address | acc])
+    else
+      :eof ->
+        :file.close(fd)
+        acc
+    end
+  end
+
+  defp index_last_addresses(genesis_addresses, db_path) do
+    Task.async_stream(genesis_addresses, &index_last_address(&1, db_path), timeout: 20000)
+    |> Stream.run()
+  end
+
+  defp index_last_address(genesis_address, db_path) do
+    [%Transaction{address: address}] =
+      ChainReader.stream_chain(genesis_address, [:address], db_path) |> Enum.take(-1)
+
+    ChainIndex.set_last_chain_address_stored(genesis_address, address, db_path)
+  end
+end

test/archethic/db/embedded_impl/chain_index_test.exs

@@ -5,6 +5,8 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndexTest do
   alias Archethic.DB.EmbeddedImpl.ChainWriter
   alias ArchethicCache.LRU
 
+  import ArchethicCase
+
   setup do
     db_path = Application.app_dir(:archethic, "data_test")
     ChainWriter.setup_folders!(db_path)
@@ -99,4 +101,17 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndexTest do
       assert {^tx_address_2, _} = ChainIndex.get_last_chain_address(genesis_address, db_path)
     end
   end
+
+  describe "set_last_chain_address_stored" do
+    test "should write a new index containing the address", %{db_path: db_path} do
+      genesis_address = random_address()
+      last_address = random_address()
+
+      assert nil == ChainIndex.get_last_chain_address_stored(genesis_address, db_path)
+
+      ChainIndex.set_last_chain_address_stored(genesis_address, last_address, db_path)
+
+      assert last_address == ChainIndex.get_last_chain_address_stored(genesis_address, db_path)
+    end
+  end
 end

test/archethic/p2p/mem_table_test.exs

@@ -6,21 +6,4 @@ defmodule Archethic.P2P.MemTableTest do
   alias Archethic.P2P.Node
 
   doctest MemTable
-
-  test "code_change" do
-    values =
-      {"key1", "key2", {127, 0, 0, 1}, 3000, 4000, "AFZ", "AAA", 0.9, <<1::1, 1::1>>,
-       ~U[2020-10-22 23:19:45.797109Z], :tcp, "reward_address", "last_address", "origin_key",
-       true, ~U[2020-10-22 23:19:45.797109Z], true, ~U[2020-10-22 23:19:45.797109Z]}
-
-    :ets.insert(:archethic_node_discovery, values)
-    assert {:ok, %{}} = MemTable.code_change("1.1.1", %{}, nil)
-
-    assert [
-             {"key1", "key2", {127, 0, 0, 1}, 3000, 4000, "AFZ", "AAA", 0.9,
-              ~U[2020-10-22 23:19:45.797109Z], :tcp, "reward_address", "last_address",
-              "origin_key", true, ~U[2020-10-22 23:19:45.797109Z], true,
-              ~U[2020-10-22 23:19:45.797109Z]}
-           ] = :ets.lookup(:archethic_node_discovery, "key1")
-  end
 end

test/archethic/replication/transaction_context_test.exs

@@ -56,13 +56,7 @@ defmodule Archethic.Replication.TransactionContextTest do
     addr2 = Crypto.derive_address(pub2)
 
     MockDB
-    |> stub(:list_chain_addresses, fn _ ->
-      [{addr1, DateTime.utc_now()}, {addr2, DateTime.utc_now()}]
-    end)
-    |> stub(:transaction_exists?, fn
-      ^addr1, _ -> true
-      _, _ -> false
-    end)
+    |> stub(:get_last_chain_address_stored, fn _ -> addr1 end)
 
     MockClient
     |> stub(:send_message, fn

test/support/template.ex

@@ -76,6 +76,7 @@ defmodule ArchethicCase do
     |> stub(:stop_inputs_writer, fn _ -> :ok end)
     |> stub(:append_input, fn _, _ -> :ok end)
     |> stub(:get_inputs, fn _, _ -> [] end)
+    |> stub(:get_last_chain_address_stored, fn addr -> addr end)
 
     {:ok, shared_secrets_counter} = Agent.start_link(fn -> 0 end)
     {:ok, network_pool_counter} = Agent.start_link(fn -> 0 end)