Issue #811 offload transaction indexes memory tables

config/config.exs

@@ -111,6 +111,11 @@ config :archethic, Archethic.P2P.Message, floor_upload_speed: 125_000
 
 config :archethic, Archethic.SelfRepair.Sync, last_sync_file: "p2p/last_sync"
 
+# Default cachae size for the chain index is 300MB
+config :archethic,
+       Archethic.DB.ChainIndex.MaxCacheSize,
+       System.get_env("CHAIN_INDEX_MAX_CACHE_SIZE", "300000000")
+
 # Configure the endpoint
 config :archethic, ArchethicWeb.Endpoint,
   secret_key_base: "5mFu4p5cPMY5Ii0HvjkLfhYZYtC0JAJofu70bzmi5x3xzFIJNlXFgIY5g8YdDPMf",

lib/archethic/db/embedded_impl/chain_index.ex

@@ -10,11 +10,14 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndex do
   alias Archethic.DB
   alias Archethic.DB.EmbeddedImpl.ChainWriter
   alias Archethic.TransactionChain.Transaction
+  alias ArchethicCache.LRU
 
-  @archethic_db_tx_index :archethic_db_tx_index
   @archethic_db_chain_stats :archethic_db_chain_stats
   @archethic_db_last_index :archethic_db_last_index
   @archethic_db_type_stats :archethic_db_type_stats
+  @archetic_db_tx_index_cache Archethic.Db.ChainIndex.LRU
+
+  require Logger
 
   def start_link(arg \\ []) do
     GenServer.start_link(__MODULE__, arg, name: __MODULE__)
@@ -23,7 +26,12 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndex do
   def init(opts) do
     db_path = Keyword.fetch!(opts, :path)
 
-    :ets.new(@archethic_db_tx_index, [:set, :named_table, :public, read_concurrency: true])
+    ## Start the LRU cache for chain indexes. Default max size is 300 Mb
+    cache_max_size =
+      String.to_integer(Application.get_env(:archethic, Archethic.DB.ChainIndex.MaxCacheSize))
+
+    LRU.start_link(@archetic_db_tx_index_cache, cache_max_size)
+
     :ets.new(@archethic_db_chain_stats, [:set, :named_table, :public, read_concurrency: true])
     :ets.new(@archethic_db_last_index, [:set, :named_table, :public, read_concurrency: true])
     :ets.new(@archethic_db_type_stats, [:set, :named_table, :public, read_concurrency: true])
@@ -33,6 +41,33 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndex do
     {:ok, %{db_path: db_path}}
   end
 
+  def code_change("1.0.7", state, _extra) do
+    index = :ets.first(:archethic_db_tx_index)
+    :ok = migrate_lru_cache(index)
+    :ets.delete(:archethic_db_tx_index)
+    {:ok, state}
+  end
+
+  defp migrate_lru_cache(:end_of_table) do
+    :ok
+  end
+
+  defp migrate_lru_cache(index) do
+    case :ets.lookup(:archethic_db_tx_index, index) do
+      [{key, %{size: size, offset: offset, genesis_address: genesis_address}}] ->
+        LRU.put(@archetic_db_tx_index_cache, key, %{
+          size: size,
+          offset: offset,
+          genesis_address: genesis_address
+        })
+
+      _ ->
+        :ok
+    end
+
+    migrate_lru_cache(:ets.next(:archethic_db_tx_index, index))
+  end
+
   defp fill_tables(db_path) do
     Enum.each(0..255, fn subset ->
       subset_summary_filename = index_summary_path(db_path, subset)
@@ -64,12 +99,6 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndex do
       current_address = <<current_curve_id::8, current_hash_type::8, current_digest::binary>>
       genesis_address = <<genesis_curve_id::8, genesis_hash_type::8, genesis_digest::binary>>
 
-      true =
-        :ets.insert(
-          @archethic_db_tx_index,
-          {current_address, %{size: size, offset: offset, genesis_address: genesis_address}}
-        )
-
       :ets.update_counter(
         @archethic_db_chain_stats,
         genesis_address,
@@ -80,6 +109,13 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndex do
         {genesis_address, 0, 0}
       )
 
+      ## Store each transaction in the LRU cache. If cache is full, the oldest entries will be evicted
+      LRU.put(@archetic_db_tx_index_cache, current_address, %{
+        size: size,
+        offset: offset,
+        genesis_address: genesis_address
+      })
+
       do_scan_summary_table(fd)
     else
       :eof ->
@@ -145,12 +181,13 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndex do
       [:binary, :append]
     )
 
-    # Write fast lookup entry for this transaction on memory
-    true =
-      :ets.insert(
-        @archethic_db_tx_index,
-        {tx_address, %{size: size, offset: last_offset, genesis_address: genesis_address}}
-      )
+    # Write fast lookup entry for this transaction on LRU cache
+    :ok =
+      LRU.put(@archetic_db_tx_index_cache, tx_address, %{
+        size: size,
+        offset: last_offset,
+        genesis_address: genesis_address
+      })
 
     :ets.update_counter(
       @archethic_db_chain_stats,
@@ -210,13 +247,14 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndex do
   """
   @spec get_tx_entry(binary(), String.t()) :: {:ok, map()} | {:error, :not_exists}
   def get_tx_entry(address, db_path) do
-    case :ets.lookup(@archethic_db_tx_index, address) do
-      [] ->
+    case LRU.get(@archetic_db_tx_index_cache, address) do
+      nil ->
         # If the transaction is not found in the in memory lookup
         # we scan the index file for the subset of the transaction to find the relative information
+        # This will update the LRU cache if the transaction is found
         search_tx_entry(address, db_path)
 
-      [{_address, entry}] ->
+      entry = %{} ->
         {:ok, entry}
     end
   end
@@ -233,6 +271,13 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndex do
 
           {genesis_address, size, offset} ->
             :file.close(fd)
+
+            LRU.put(@archetic_db_tx_index_cache, search_address, %{
+              size: size,
+              offset: offset,
+              genesis_address: genesis_address
+            })
+
             {:ok, %{genesis_address: genesis_address, size: size, offset: offset}}
         end
 

test/archethic/db/embedded_impl/chain_index_test.exs

@@ -1,8 +1,10 @@
 defmodule Archethic.DB.EmbeddedImpl.ChainIndexTest do
   use ArchethicCase
 
+  alias Mint.Core.Util
   alias Archethic.DB.EmbeddedImpl.ChainIndex
   alias Archethic.DB.EmbeddedImpl.ChainWriter
+  alias ArchethicCache.LRU
 
   setup do
     db_path = Application.app_dir(:archethic, "data_test")
@@ -45,9 +47,9 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndexTest do
       ChainIndex.add_tx(tx_address, genesis_address, 100, db_path)
       ChainIndex.set_last_chain_address(genesis_address, tx_address, DateTime.utc_now(), db_path)
 
-      GenServer.stop(pid)
+      # GenServer.stop(pid)
 
-      ChainIndex.start_link(path: db_path)
+      # ChainIndex.start_link(path: db_path)
 
       assert {:ok, %{genesis_address: ^genesis_address, size: 100}} =
                ChainIndex.get_tx_entry(tx_address, db_path)
@@ -57,7 +59,7 @@ defmodule Archethic.DB.EmbeddedImpl.ChainIndexTest do
       assert {^tx_address, _} = ChainIndex.get_last_chain_address(genesis_address, db_path)
 
       # Remove the transaction from the cache and try to fetch from the file instead
-      :ets.delete(:archethic_db_tx_index, tx_address)
+      LRU.purge(Archethic.Db.ChainIndex.LRU)
       assert true == ChainIndex.transaction_exists?(tx_address, db_path)
       assert false == ChainIndex.transaction_exists?(:crypto.strong_rand_bytes(32), db_path)
     end