feat: add blocks pruning (#1092)

Author: avilagaston9

lib/lambda_ethereum_consensus/fork_choice/fork_choice.ex

@@ -11,6 +11,7 @@ defmodule LambdaEthereumConsensus.ForkChoice do
   alias LambdaEthereumConsensus.ForkChoice.Handlers
   alias LambdaEthereumConsensus.ForkChoice.Head
   alias LambdaEthereumConsensus.P2P.Gossip.OperationsCollector
+  alias LambdaEthereumConsensus.Store.BlockDb
   alias LambdaEthereumConsensus.Store.Blocks
   alias LambdaEthereumConsensus.Store.StateDb
   alias LambdaEthereumConsensus.Store.StoreDb
@@ -157,6 +158,7 @@ defmodule LambdaEthereumConsensus.ForkChoice do
         new_finalized_epoch * ChainSpec.get("SLOTS_PER_EPOCH")
 
       Task.async(StateDb, :prune_states_older_than, [new_finalized_slot])
+      Task.async(BlockDb, :prune_blocks_older_than, [new_finalized_slot])
     end
   end
 

lib/lambda_ethereum_consensus/store/block_db.ex

@@ -2,6 +2,7 @@ defmodule LambdaEthereumConsensus.Store.BlockDb do
   @moduledoc """
   Storage and retrieval of blocks.
   """
+  require Logger
   alias LambdaEthereumConsensus.Store.Db
   alias LambdaEthereumConsensus.Store.Utils
   alias Types.SignedBeaconBlock
@@ -58,36 +59,37 @@ defmodule LambdaEthereumConsensus.Store.BlockDb do
     end
   end
 
-  def stream_missing_blocks_desc() do
-    Stream.resource(
-      fn -> 0xFFFFFFFFFFFFFFFF |> block_root_by_slot_key() |> init_keycursor() end,
-      &next_slot(&1, :prev),
-      &close_cursor/1
-    )
-    # TODO: we should remove this when we have a genesis block
-    |> Stream.concat([-1])
-    |> Stream.transform(nil, &get_missing_desc/2)
-  end
+  @spec prune_blocks_older_than(non_neg_integer()) :: :ok | {:error, String.t()} | :not_found
+  def prune_blocks_older_than(slot) do
+    Logger.info("[BlockDb] Pruning started.", slot: slot)
+    initial_key = slot |> block_root_by_slot_key()
 
-  def stream_missing_blocks_asc(starting_slot) do
-    initial_key = block_root_by_slot_key(starting_slot)
-
-    [starting_slot - 1]
-    |> Stream.concat(
+    slots_to_remove =
       Stream.resource(
         fn -> init_keycursor(initial_key) end,
-        &next_slot(&1, :next),
+        &next_slot(&1, :prev),
         &close_cursor/1
       )
-    )
-    |> Stream.transform(nil, &get_missing_asc/2)
+      |> Enum.to_list()
+
+    slots_to_remove |> Enum.each(&remove_block_by_slot/1)
+    Logger.info("[BlockDb] Pruning finished. #{Enum.count(slots_to_remove)} blocks removed.")
+  end
+
+  @spec remove_block_by_slot(non_neg_integer()) :: :ok | :not_found
+  defp remove_block_by_slot(slot) do
+    slothash_key = block_root_by_slot_key(slot)
+
+    with {:ok, block_root} <- Db.get(slothash_key) do
+      key_block = block_key(block_root)
+      Db.delete(slothash_key)
+      Db.delete(key_block)
+    end
   end
 
   defp init_keycursor(initial_key) do
     with {:ok, it} <- Db.iterate_keys(),
-         {:ok, key} <- Exleveldb.iterator_move(it, initial_key),
-         {:ok, _} <-
-           if(key == initial_key, do: Exleveldb.iterator_move(it, :prev), else: {:ok, nil}) do
+         {:ok, _key} <- Exleveldb.iterator_move(it, initial_key) do
       it
     else
       # DB is empty
@@ -110,12 +112,6 @@ defmodule LambdaEthereumConsensus.Store.BlockDb do
   defp close_cursor(nil), do: :ok
   defp close_cursor(it), do: :ok = Exleveldb.iterator_close(it)
 
-  def get_missing_desc(slot, nil), do: {[], slot}
-  def get_missing_desc(slot, prev), do: {(prev - 1)..(slot + 1)//-1, slot}
-
-  def get_missing_asc(slot, nil), do: {[], slot}
-  def get_missing_asc(slot, prev), do: {(prev + 1)..(slot - 1)//1, slot}
-
   defp block_key(root), do: Utils.get_key(@block_prefix, root)
   defp block_root_by_slot_key(slot), do: Utils.get_key(@blockslot_prefix, slot)
 

lib/lambda_ethereum_consensus/store/state_db.ex

@@ -42,7 +42,7 @@ defmodule LambdaEthereumConsensus.Store.StateDb do
          {:ok, slots_to_remove} <- get_slots_to_remove(it),
          :ok <- Exleveldb.iterator_close(it) do
       slots_to_remove |> Enum.each(&remove_state_by_slot/1)
-      Logger.info("[StateDb] Pruning finished. #{length(slots_to_remove)} slots removed.")
+      Logger.info("[StateDb] Pruning finished. #{length(slots_to_remove)} states removed.")
     end
   end