feat: add blobs pruning (#1102)

Author: avilagaston9

lib/lambda_ethereum_consensus/beacon/beacon_node.ex

@@ -56,7 +56,8 @@ defmodule LambdaEthereumConsensus.Beacon.BeaconNode do
         LambdaEthereumConsensus.P2P.Gossip.BlobSideCar,
         LambdaEthereumConsensus.P2P.Gossip.OperationsCollector,
         {Task.Supervisor, name: PruneStatesSupervisor},
-        {Task.Supervisor, name: PruneBlocksSupervisor}
+        {Task.Supervisor, name: PruneBlocksSupervisor},
+        {Task.Supervisor, name: PruneBlobsSupervisor}
       ] ++ validator_children
 
     Supervisor.init(children, strategy: :one_for_all)

lib/lambda_ethereum_consensus/fork_choice/fork_choice.ex

@@ -10,6 +10,7 @@ defmodule LambdaEthereumConsensus.ForkChoice do
   alias LambdaEthereumConsensus.ForkChoice.Handlers
   alias LambdaEthereumConsensus.ForkChoice.Head
   alias LambdaEthereumConsensus.P2P.Gossip.OperationsCollector
+  alias LambdaEthereumConsensus.Store.BlobDb
   alias LambdaEthereumConsensus.Store.BlockDb
   alias LambdaEthereumConsensus.Store.BlockDb.BlockInfo
   alias LambdaEthereumConsensus.Store.Blocks
@@ -129,6 +130,11 @@ defmodule LambdaEthereumConsensus.ForkChoice do
         PruneBlocksSupervisor,
         fn -> BlockDb.prune_blocks_older_than(new_finalized_slot) end
       )
+
+      Task.Supervisor.start_child(
+        PruneBlobsSupervisor,
+        fn -> BlobDb.prune_old_blobs(new_finalized_slot) end
+      )
     end
   end
 

lib/lambda_ethereum_consensus/store/blob_db.ex

@@ -2,12 +2,16 @@ defmodule LambdaEthereumConsensus.Store.BlobDb do
   @moduledoc """
   Storage and retrieval of blobs.
   """
+  require Logger
+
   alias LambdaEthereumConsensus.Store.Db
+  alias LambdaEthereumConsensus.Store.Utils
   alias Types.Blobdata
   alias Types.BlobSidecar
 
   @blob_sidecar_prefix "blob_sidecar"
   @blobdata_prefix "blobdata"
+  @block_root_prefix "block_root"
 
   @spec store_blob(BlobSidecar.t()) :: :ok
   def store_blob(%BlobSidecar{signed_block_header: %{message: block_header}} = blob) do
@@ -22,6 +26,9 @@ defmodule LambdaEthereumConsensus.Store.BlobDb do
 
     key = blobdata_key(block_root, blob.index)
     Db.put(key, encoded_blobdata)
+
+    block_root_key = block_root_key(block_header.slot, blob.index)
+    Db.put(block_root_key, block_root)
   end
 
   # TODO: this is only used for tests
@@ -43,6 +50,16 @@ defmodule LambdaEthereumConsensus.Store.BlobDb do
     end
   end
 
+  @spec get_blob_by_slot_index(non_neg_integer(), non_neg_integer()) ::
+          {:ok, BlobSidecar.t()} | {:error, String.t()} | :not_found
+  def get_blob_by_slot_index(slot, index) do
+    block_root_key = block_root_key(slot, index)
+
+    with {:ok, block_root} <- Db.get(block_root_key) do
+      get_blob_sidecar(block_root, index)
+    end
+  end
+
   @spec get_blob_with_proof(Types.root(), Types.blob_index()) ::
           {:ok, {Types.blob(), Types.kzg_proof()}} | {:error, String.t()} | :not_found
   def get_blob_with_proof(block_root, blob_index) do
@@ -55,8 +72,71 @@ defmodule LambdaEthereumConsensus.Store.BlobDb do
     end
   end
 
+  @spec prune_old_blobs(non_neg_integer()) :: :ok | {:error, String.t()} | :not_found
+  def prune_old_blobs(current_finalized_slot) do
+    slot =
+      current_finalized_slot -
+        ChainSpec.get("MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS") * ChainSpec.get("SLOTS_PER_EPOCH")
+
+    Logger.info("[BlobDb] Pruning started.", slot: slot)
+    last_finalized_key = slot |> block_root_key(0)
+
+    with {:ok, it} <- Db.iterate(),
+         {:ok, @block_root_prefix <> _, _value} <-
+           Exleveldb.iterator_move(it, last_finalized_key),
+         {:ok, keys_to_remove} <- get_block_root_keys_to_remove(it),
+         :ok <- Exleveldb.iterator_close(it) do
+      total_removed =
+        keys_to_remove
+        |> Enum.reduce_while(0, fn
+          key, acc_removed_count ->
+            case remove_blob_by_block_root_key(key) do
+              :ok ->
+                {:cont, acc_removed_count + 1}
+
+              _ ->
+                Logger.error("[BlobDb] Error while removing key.", key: key)
+                {:halt, acc_removed_count}
+            end
+        end)
+
+      Logger.info("[BlobDb] Pruning finished. #{total_removed} blobs removed.")
+    end
+  end
+
+  @spec get_block_root_keys_to_remove(list(binary()), :eleveldb.itr_ref()) ::
+          {:ok, list(binary())}
+  defp get_block_root_keys_to_remove(keys_to_remove \\ [], iterator) do
+    case Exleveldb.iterator_move(iterator, :prev) do
+      {:ok, <<@block_root_prefix, _rest::binary>> = block_root_key, _root} ->
+        [block_root_key | keys_to_remove] |> get_block_root_keys_to_remove(iterator)
+
+      _ ->
+        {:ok, keys_to_remove}
+    end
+  end
+
+  @spec remove_blob_by_block_root_key(binary()) :: :ok | :not_found
+  defp remove_blob_by_block_root_key(block_root_key) do
+    <<@block_root_prefix, _slot::unsigned-size(64), index>> =
+      block_root_key
+
+    with {:ok, block_root} <- Db.get(block_root_key) do
+      key_blob = blob_sidecar_key(block_root, index)
+      key_data = blobdata_key(block_root, index)
+
+      Db.delete(block_root_key)
+      Db.delete(key_blob)
+      Db.delete(key_data)
+    end
+  end
+
   defp blob_sidecar_key(block_root, blob_index),
     do: @blob_sidecar_prefix <> block_root <> <<blob_index>>
 
   defp blobdata_key(block_root, blob_index), do: @blobdata_prefix <> block_root <> <<blob_index>>
+
+  defp block_root_key(blob_slot, blob_index) do
+    Utils.get_key(@block_root_prefix, blob_slot) <> <<blob_index>>
+  end
 end