feat: persist deposits snapshot in DB (#1019)

Co-authored-by: Tomás Arjovsky <[email protected]>

Author: MegaRedHand

lib/lambda_ethereum_consensus/beacon/store_setup.ex

@@ -88,7 +88,13 @@ defmodule LambdaEthereumConsensus.Beacon.StoreSetup do
 
   @spec get_deposit_snapshot!(store_setup_strategy()) :: DepositTreeSnapshot.t() | nil
   def get_deposit_snapshot!({:checkpoint_sync_url, url}), do: fetch_deposit_snapshot(url)
-  def get_deposit_snapshot!(:db), do: nil
+
+  def get_deposit_snapshot!(:db) do
+    case StoreDb.fetch_deposits_snapshot() do
+      {:ok, snapshot} -> snapshot
+      _ -> nil
+    end
+  end
 
   def get_deposit_snapshot!({:file, %{eth1_data: %Eth1Data{} = eth1_data}}) do
     if eth1_data.deposit_count == 0 do

lib/lambda_ethereum_consensus/execution/execution_chain.ex

@@ -7,6 +7,7 @@ defmodule LambdaEthereumConsensus.Execution.ExecutionChain do
   use GenServer
 
   alias LambdaEthereumConsensus.Execution.ExecutionClient
+  alias LambdaEthereumConsensus.Store.StoreDb
   alias Types.Deposit
   alias Types.DepositTree
   alias Types.DepositTreeSnapshot
@@ -23,6 +24,9 @@ defmodule LambdaEthereumConsensus.Execution.ExecutionChain do
     GenServer.call(__MODULE__, {:get_eth1_vote, slot})
   end
 
+  @spec get_eth1_vote(Types.slot()) :: DepositTreeSnapshot.t()
+  def get_deposit_snapshot(), do: GenServer.call(__MODULE__, :get_deposit_snapshot)
+
   @spec get_deposits(Eth1Data.t(), Eth1Data.t(), Range.t()) ::
           {:ok, [Deposit.t()] | nil} | {:error, any}
   def get_deposits(current_eth1_data, eth1_vote, deposit_range) do
@@ -53,6 +57,8 @@ defmodule LambdaEthereumConsensus.Execution.ExecutionChain do
 
     updated_state = Enum.reduce(eth1_votes, state, &update_state_with_vote(&2, &1))
 
+    StoreDb.persist_deposits_snapshot(snapshot)
+
     {:ok, updated_state}
   end
 
@@ -61,6 +67,11 @@ defmodule LambdaEthereumConsensus.Execution.ExecutionChain do
     {:reply, compute_eth1_vote(state, slot), state}
   end
 
+  @impl true
+  def handle_call(:get_deposit_snapshot, _from, state) do
+    {:reply, DepositTree.get_snapshot(state.deposit_tree), state}
+  end
+
   def handle_call({:get_deposits, current_eth1_data, eth1_vote, deposit_range}, _from, state) do
     votes = state.eth1_data_votes
 
@@ -142,6 +153,8 @@ defmodule LambdaEthereumConsensus.Execution.ExecutionChain do
          {:ok, deposits} <- ExecutionClient.get_deposit_logs(start_block..end_block) do
       # TODO: check if the result should be sorted by index
       deposit_tree = DepositTree.finalize(state.deposit_tree, old_eth1_data, start_block)
+      # TODO: delay persisting until it's finalized
+      deposit_tree |> DepositTree.get_snapshot() |> StoreDb.persist_deposits_snapshot()
       {:ok, update_tree_with_deposits(deposit_tree, deposits)}
     end
   end

lib/lambda_ethereum_consensus/store/store_db.ex

@@ -5,17 +5,32 @@ defmodule LambdaEthereumConsensus.Store.StoreDb do
   alias LambdaEthereumConsensus.Store.Db
 
   @store_prefix "store"
+  @snapshot_prefix "snapshot"
 
   @spec fetch_store() :: {:ok, Types.Store.t()} | :not_found
-  def fetch_store() do
-    with {:ok, encoded_store} <- Db.get(@store_prefix) do
-      {:ok, :erlang.binary_to_term(encoded_store)}
-    end
-  end
+  def fetch_store(), do: get(@store_prefix)
 
   @spec persist_store(Types.Store.t()) :: :ok
   def persist_store(%Types.Store{} = store) do
-    # Compress the store before storing it. This doubles the time it takes to dump, but reduces size by 5 times.
-    Db.put(@store_prefix, :erlang.term_to_binary(store, [{:compressed, 1}]))
+    put(@store_prefix, store)
+  end
+
+  @spec fetch_deposits_snapshot() :: {:ok, Types.DepositTreeSnapshot.t()} | :not_found
+  def fetch_deposits_snapshot(), do: get(@snapshot_prefix)
+
+  @spec persist_deposits_snapshot(Types.DepositTreeSnapshot.t()) :: :ok
+  def persist_deposits_snapshot(%Types.DepositTreeSnapshot{} = snapshot) do
+    put(@snapshot_prefix, snapshot)
+  end
+
+  defp get(key) do
+    with {:ok, value} <- Db.get(key) do
+      {:ok, :erlang.binary_to_term(value)}
+    end
+  end
+
+  defp put(key, value) do
+    # Compress before storing. This doubles the time it takes to dump, but reduces size by 5 times.
+    Db.put(key, :erlang.term_to_binary(value, [{:compressed, 1}]))
   end
 end

lib/types/deposit_tree.ex

@@ -45,6 +45,21 @@ defmodule Types.DepositTree do
     }
   end
 
+  @spec get_snapshot(t()) :: DepositTreeSnapshot.t()
+  def get_snapshot(%__MODULE__{} = tree) do
+    finalized = get_finalized(tree.inner)
+    deposit_root = get_root(tree)
+    {el_hash, el_height} = tree.finalized_execution_block
+
+    %DepositTreeSnapshot{
+      finalized: finalized,
+      deposit_root: deposit_root,
+      deposit_count: tree.deposit_count,
+      execution_block_hash: el_hash,
+      execution_block_height: el_height
+    }
+  end
+
   @spec finalize(t(), Eth1Data.t(), non_neg_integer()) :: t()
   def finalize(%__MODULE__{} = tree, %Eth1Data{} = eth1_data, execution_block_height) do
     finalized_block = {eth1_data.block_hash, execution_block_height}
@@ -55,7 +70,7 @@ defmodule Types.DepositTree do
   @spec get_deposit(t(), non_neg_integer()) :: {:ok, Deposit.t()} | {:error, String.t()}
   def get_deposit(%__MODULE__{} = tree, index) do
     cond do
-      index < get_finalized(tree.inner) ->
+      index < count_finalized(tree.inner) ->
         {:error, "deposit already finalized"}
 
       index >= tree.deposit_count ->
@@ -171,10 +186,15 @@ defmodule Types.DepositTree do
   defp full?({:zero, _}), do: false
   defp full?(_), do: true
 
-  defp get_finalized({:finalized, {_, count}}), do: count
-  defp get_finalized({:node, {left, right}}), do: get_finalized(left) + get_finalized(right)
-  defp get_finalized({:leaf, _}), do: 0
-  defp get_finalized({:zero, _}), do: 0
+  defp count_finalized({:finalized, {_, count}}), do: count
+  defp count_finalized({:node, {left, right}}), do: count_finalized(left) + count_finalized(right)
+  defp count_finalized({:leaf, _}), do: 0
+  defp count_finalized({:zero, _}), do: 0
+
+  defp get_finalized({:finalized, {hash, _}}), do: [hash]
+  defp get_finalized({:node, {left, right}}), do: get_finalized(right) ++ get_finalized(left)
+  defp get_finalized({:leaf, _}), do: []
+  defp get_finalized({:zero, _}), do: []
 
   defp mix_in_length(%__MODULE__{deposit_count: count}),
     do: SszEx.hash_tree_root!(count, TypeAliases.uint64())

test/unit/deposit_tree_test.exs

@@ -105,5 +105,7 @@ defmodule Unit.DepositTreeTest do
       |> DepositTree.finalize(eth1_data, @snapshot_2.execution_block_height)
 
     assert tree == DepositTree.from_snapshot(@snapshot_2)
+
+    assert DepositTree.get_snapshot(tree) == @snapshot_2
   end
 end

test/unit/ssz_ex_test.exs

@@ -13,6 +13,12 @@ defmodule Unit.SSZExTest do
 
   use ExUnit.Case
 
+  setup_all do
+    Application.fetch_env!(:lambda_ethereum_consensus, ChainSpec)
+    |> Keyword.put(:config, MainnetConfig)
+    |> then(&Application.put_env(:lambda_ethereum_consensus, ChainSpec, &1))
+  end
+
   def assert_roundtrip(serialized, deserialized, schema) do
     assert {:ok, ^serialized} = SszEx.encode(deserialized, schema)
     assert {:ok, deserialized} === SszEx.decode(serialized, schema)