feat: runs dry-run only once per contract and block (#778)

lib/ae_mdw/application.ex

@@ -244,6 +244,8 @@ defmodule AeMdw.Application do
   end
 
   def start_phase(:start_sync, _start_type, []) do
+    AeMdw.Db.Aex9BalancesCache.init()
+
     if Application.fetch_env!(:ae_mdw, :sync) do
       Watcher.start_sync()
     end

lib/ae_mdw/db/aex9_balances_cache.ex

@@ -0,0 +1,44 @@
+defmodule AeMdw.Db.Aex9BalancesCache do
+  @moduledoc """
+  Cache of aex9 balances in order to call the contract a single time for the same block.
+  """
+  alias AeMdw.Node.Db, as: NodeDb
+
+  alias AeMdw.Blocks
+  alias AeMdw.EtsCache
+
+  @typep balances :: %{{:address, NodeDb.pubkey()} => integer()}
+  @table :aex9_balances
+  @expire_minutes 24 * 60
+
+  @spec init() :: :ok
+  def init do
+    EtsCache.new(@table, @expire_minutes, :ordered_set)
+    :ok
+  end
+
+  @spec get(NodeDb.pubkey(), Blocks.block_index(), Blocks.block_hash()) ::
+          {:ok, balances()} | :not_found
+  def get(contract_pk, block_index, block_hash) do
+    case EtsCache.get(@table, {contract_pk, block_index, block_hash}) do
+      {balances, _time} -> {:ok, balances}
+      nil -> :not_found
+    end
+  end
+
+  @spec put(NodeDb.pubkey(), Blocks.block_index(), Blocks.block_hash(), balances()) :: :ok
+  def put(contract_pk, height, block_hash, balances) do
+    EtsCache.put(@table, {contract_pk, height, block_hash}, balances)
+    :ok
+  end
+
+  @spec purge(NodeDb.pubkey(), Blocks.block_index()) :: :ok
+  def purge(contract_pk, block_index) do
+    with {^contract_pk, ^block_index, hash} <-
+           EtsCache.next(@table, {contract_pk, block_index, <<>>}) do
+      EtsCache.del(@table, {contract_pk, block_index, hash})
+    end
+
+    :ok
+  end
+end

lib/ae_mdw/ets_cache.ex

@@ -1,4 +1,7 @@
 defmodule AeMdw.EtsCache do
+  # credo:disable-for-this-file
+  @moduledoc false
+
   require Ex2ms
 
   @cache_types [:set, :ordered_set, :bag, :duplicate_bag]
@@ -36,6 +39,20 @@ defmodule AeMdw.EtsCache do
   def del(table, key),
     do: :ets.delete(table, key)
 
+  def next(table, key) do
+    case :ets.next(table, key) do
+      :"$end_of_table" -> nil
+      next_key -> next_key
+    end
+  end
+
+  def prev(table, key) do
+    case :ets.prev(table, key) do
+      :"$end_of_table" -> nil
+      prev_key -> prev_key
+    end
+  end
+
   def purge(table, max_age_msecs) do
     boundary = time() - max_age_msecs
 

lib/ae_mdw/sync/async_tasks/update_aex9_state.ex

@@ -6,6 +6,7 @@ defmodule AeMdw.Sync.AsyncTasks.UpdateAex9State do
 
   alias AeMdw.Node.Db, as: DBN
 
+  alias AeMdw.Db.Aex9BalancesCache
   alias AeMdw.Db.Model
   alias AeMdw.Db.Mutation
   alias AeMdw.Db.State
@@ -34,12 +35,8 @@ defmodule AeMdw.Sync.AsyncTasks.UpdateAex9State do
   end
 
   @spec mutations(args :: list()) :: [Mutation.t()]
-  def mutations([contract_pk, {kbi, mbi} = block_index, call_txi]) do
-    next_kb_hash = DBN.get_key_block_hash(kbi + 1)
-    next_hash = DBN.get_next_hash(next_kb_hash, mbi)
-    type = if next_hash == next_kb_hash, do: :key, else: :micro
-
-    {balances, _height_hash} = DBN.aex9_balances(contract_pk, {type, kbi, next_hash})
+  def mutations([contract_pk, block_index, call_txi]) do
+    balances = aex9_balances(contract_pk, block_index)
 
     if map_size(balances) == 0 do
       m_empty_balance = Model.aex9_balance(index: {contract_pk, <<>>})
@@ -48,15 +45,33 @@ defmodule AeMdw.Sync.AsyncTasks.UpdateAex9State do
         WriteMutation.new(Model.Aex9Balance, m_empty_balance)
       ]
     else
-      balances =
+      balances_list =
         Enum.map(balances, fn {{:address, account_pk}, amount} -> {account_pk, amount} end)
 
       [
-        UpdateAex9StateMutation.new(contract_pk, block_index, call_txi, balances)
+        UpdateAex9StateMutation.new(contract_pk, block_index, call_txi, balances_list)
       ]
     end
   end
 
+  defp aex9_balances(contract_pk, {kbi, mbi} = block_index) do
+    next_kb_hash = DBN.get_key_block_hash(kbi + 1)
+    next_hash = DBN.get_next_hash(next_kb_hash, mbi)
+    type = if next_hash == next_kb_hash, do: :key, else: :micro
+
+    case Aex9BalancesCache.get(contract_pk, block_index, next_hash) do
+      {:ok, balances} ->
+        balances
+
+      :not_found ->
+        Aex9BalancesCache.purge(contract_pk, block_index)
+        {balances, _height_hash} = DBN.aex9_balances(contract_pk, {type, kbi, next_hash})
+        Aex9BalancesCache.put(contract_pk, block_index, next_hash, balances)
+
+        balances
+    end
+  end
+
   defp enc_ct(<<pk::binary-32>>), do: :aeser_api_encoder.encode(:contract_pubkey, pk)
   defp enc_ct(invalid_pk), do: invalid_pk
 end

test/ae_mdw/sync/async_tasks/update_aex9_state_test.exs

@@ -2,6 +2,7 @@ defmodule AeMdw.Sync.AsyncTasks.UpdateAex9StateTest do
   use ExUnit.Case
 
   alias AeMdw.Database
+  alias AeMdw.Db.Aex9BalancesCache
   alias AeMdw.Db.Model
   alias AeMdw.Sync.AsyncTasks.UpdateAex9State
   alias AeMdw.Validate
@@ -85,5 +86,47 @@ defmodule AeMdw.Sync.AsyncTasks.UpdateAex9StateTest do
         assert Database.exists?(Model.Aex9Balance, {contract_pk, <<>>})
       end
     end
+
+    test "uses cached aex9 balances when already dry-runned" do
+      kbi = 123
+      mbi = 1
+      block_index = {kbi, mbi}
+      call_txi = 123_456
+      kb_hash = :crypto.strong_rand_bytes(32)
+      next_mb_hash = :crypto.strong_rand_bytes(32)
+      contract_pk = :crypto.strong_rand_bytes(32)
+      account_pk1 = :crypto.strong_rand_bytes(32)
+      account_pk2 = :crypto.strong_rand_bytes(32)
+      amount1 = Enum.random(1_000_000..9_999_999)
+      amount2 = Enum.random(1_000_000..9_999_999)
+
+      Aex9BalancesCache.put(contract_pk, block_index, next_mb_hash, %{
+        {:address, account_pk1} => amount1,
+        {:address, account_pk2} => amount2
+      })
+
+      with_mocks [
+        {AeMdw.Node.Db, [],
+         [
+           get_key_block_hash: fn height ->
+             assert ^height = kbi + 1
+             kb_hash
+           end,
+           get_next_hash: fn ^kb_hash, ^mbi -> next_mb_hash end,
+           aex9_balances: fn ^contract_pk, {:micro, ^kbi, ^next_mb_hash} = block_tuple ->
+             {%{}, block_tuple}
+           end
+         ]}
+      ] do
+        UpdateAex9State.process([contract_pk, block_index, call_txi])
+        refute Database.exists?(Model.Aex9Balance, {contract_pk, <<>>})
+
+        assert Model.aex9_balance(block_index: ^block_index, txi: ^call_txi, amount: ^amount1) =
+                 Database.fetch!(Model.Aex9Balance, {contract_pk, account_pk1})
+
+        assert Model.aex9_balance(block_index: ^block_index, txi: ^call_txi, amount: ^amount2) =
+                 Database.fetch!(Model.Aex9Balance, {contract_pk, account_pk2})
+      end
+    end
   end
 end