Fix issue where prune blobs can OOM

Makefile

@@ -14,7 +14,7 @@ BUILD_PATH_AARCH64 = "target/$(AARCH64_TAG)/release"
 PINNED_NIGHTLY ?= nightly
 
 # List of features to use when cross-compiling. Can be overridden via the environment.
-CROSS_FEATURES ?= gnosis,slasher-lmdb,slasher-mdbx,slasher-redb,jemalloc
+CROSS_FEATURES ?= gnosis,slasher-lmdb,slasher-mdbx,slasher-redb,jemalloc,beacon-node-leveldb,beacon-node-redb
 
 # Cargo profile for Cross builds. Default is for local builds, CI uses an override.
 CROSS_PROFILE ?= release

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -1399,6 +1399,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             sync_committee_period,
             count,
             &self.spec,
+            self.log.clone(),
         )
     }
 

beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs

@@ -325,7 +325,6 @@ impl<E: EthSpec> PendingComponents<E> {
                 (None, Some(verified_data_columns))
             }
         };
-
         let executed_block = recover(diet_executed_block)?;
 
         let AvailabilityPendingExecutedBlock {
@@ -716,7 +715,7 @@ mod test {
     use slog::{info, Logger};
     use state_processing::ConsensusContext;
     use std::collections::VecDeque;
-    use store::{HotColdDB, ItemStore, LevelDB, StoreConfig};
+    use store::{database::interface::BeaconNodeBackend, HotColdDB, ItemStore, StoreConfig};
     use tempfile::{tempdir, TempDir};
     use types::non_zero_usize::new_non_zero_usize;
     use types::{ExecPayload, MinimalEthSpec};
@@ -728,7 +727,7 @@ mod test {
         db_path: &TempDir,
         spec: Arc<ChainSpec>,
         log: Logger,
-    ) -> Arc<HotColdDB<E, LevelDB<E>, LevelDB<E>>> {
+    ) -> Arc<HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>> {
         let hot_path = db_path.path().join("hot_db");
         let cold_path = db_path.path().join("cold_db");
         let blobs_path = db_path.path().join("blobs_db");
@@ -902,7 +901,11 @@ mod test {
     )
     where
         E: EthSpec,
-        T: BeaconChainTypes<HotStore = LevelDB<E>, ColdStore = LevelDB<E>, EthSpec = E>,
+        T: BeaconChainTypes<
+            HotStore = BeaconNodeBackend<E>,
+            ColdStore = BeaconNodeBackend<E>,
+            EthSpec = E,
+        >,
     {
         let log = test_logger();
         let chain_db_path = tempdir().expect("should get temp dir");

beacon_node/beacon_chain/src/light_client_server_cache.rs

@@ -2,7 +2,7 @@ use crate::errors::BeaconChainError;
 use crate::{metrics, BeaconChainTypes, BeaconStore};
 use parking_lot::{Mutex, RwLock};
 use safe_arith::SafeArith;
-use slog::{debug, Logger};
+use slog::{debug, error, Logger};
 use ssz::Decode;
 use std::num::NonZeroUsize;
 use std::sync::Arc;
@@ -270,13 +270,33 @@ impl<T: BeaconChainTypes> LightClientServerCache<T> {
         start_period: u64,
         count: u64,
         chain_spec: &ChainSpec,
+        log: Logger,
     ) -> Result<Vec<LightClientUpdate<T::EthSpec>>, BeaconChainError> {
         let column = DBColumn::LightClientUpdate;
         let mut light_client_updates = vec![];
-        for res in store
-            .hot_db
-            .iter_column_from::<Vec<u8>>(column, &start_period.to_le_bytes())
-        {
+
+        let results = store.hot_db.iter_column_from::<Vec<u8>>(
+            column,
+            &start_period.to_le_bytes(),
+            move |sync_committee_bytes, _| match u64::from_ssz_bytes(sync_committee_bytes) {
+                Ok(sync_committee_period) => {
+                    if sync_committee_period >= start_period + count {
+                        return false;
+                    }
+                    true
+                }
+                Err(e) => {
+                    error!(
+                        log,
+                        "Error decoding sync committee bytes from the db";
+                        "error" => ?e
+                    );
+                    false
+                }
+            },
+        );
+
+        for res in results? {
             let (sync_committee_bytes, light_client_update_bytes) = res?;
             let sync_committee_period = u64::from_ssz_bytes(&sync_committee_bytes)
                 .map_err(store::errors::Error::SszDecodeError)?;
@@ -290,11 +310,8 @@ impl<T: BeaconChainTypes> LightClientServerCache<T> {
                     .map_err(store::errors::Error::SszDecodeError)?;
 
             light_client_updates.push(light_client_update);
-
-            if sync_committee_period >= start_period + count {
-                break;
-            }
         }
+
         Ok(light_client_updates)
     }
 

beacon_node/beacon_chain/src/otb_verification_service.rs

@@ -120,7 +120,7 @@ pub fn load_optimistic_transition_blocks<T: BeaconChainTypes>(
     chain: &BeaconChain<T>,
 ) -> Result<Vec<OptimisticTransitionBlock>, StoreError> {
     process_results(
-        chain.store.hot_db.iter_column::<Hash256>(OTBColumn),
+        chain.store.hot_db.iter_column::<Hash256>(OTBColumn)?,
         |iter| {
             iter.map(|(_, bytes)| OptimisticTransitionBlock::from_store_bytes(&bytes))
                 .collect()

beacon_node/beacon_chain/src/schema_change/migration_schema_v21.rs

@@ -3,9 +3,7 @@ use crate::validator_pubkey_cache::DatabasePubkey;
 use slog::{info, Logger};
 use ssz::{Decode, Encode};
 use std::sync::Arc;
-use store::{
-    get_key_for_col, DBColumn, Error, HotColdDB, KeyValueStore, KeyValueStoreOp, StoreItem,
-};
+use store::{DBColumn, Error, HotColdDB, KeyValueStore, KeyValueStoreOp, StoreItem};
 use types::{Hash256, PublicKey};
 
 const LOG_EVERY: usize = 200_000;
@@ -21,7 +19,7 @@ pub fn upgrade_to_v21<T: BeaconChainTypes>(
     // Iterate through all pubkeys and decompress them.
     for (i, res) in db
         .hot_db
-        .iter_column::<Hash256>(DBColumn::PubkeyCache)
+        .iter_column::<Hash256>(DBColumn::PubkeyCache)?
         .enumerate()
     {
         let (key, value) = res?;
@@ -53,7 +51,7 @@ pub fn downgrade_from_v21<T: BeaconChainTypes>(
     // Iterate through all pubkeys and recompress them.
     for (i, res) in db
         .hot_db
-        .iter_column::<Hash256>(DBColumn::PubkeyCache)
+        .iter_column::<Hash256>(DBColumn::PubkeyCache)?
         .enumerate()
     {
         let (key, value) = res?;
@@ -62,9 +60,10 @@ pub fn downgrade_from_v21<T: BeaconChainTypes>(
             message: format!("{e:?}"),
         })?;
 
-        let db_key = get_key_for_col(DBColumn::PubkeyCache.into(), key.as_slice());
+        let column: &str = DBColumn::PubkeyCache.into();
         ops.push(KeyValueStoreOp::PutKeyValue(
-            db_key,
+            column.to_owned(),
+            key.as_slice().to_vec(),
             pubkey_bytes.as_ssz_bytes(),
         ));
 

beacon_node/beacon_chain/src/test_utils.rs

@@ -56,7 +56,8 @@ use std::str::FromStr;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, LazyLock};
 use std::time::Duration;
-use store::{config::StoreConfig, HotColdDB, ItemStore, LevelDB, MemoryStore};
+use store::database::interface::BeaconNodeBackend;
+use store::{config::StoreConfig, HotColdDB, ItemStore, MemoryStore};
 use task_executor::TaskExecutor;
 use task_executor::{test_utils::TestRuntime, ShutdownReason};
 use tree_hash::TreeHash;
@@ -116,7 +117,7 @@ pub fn get_kzg(spec: &ChainSpec) -> Arc<Kzg> {
 pub type BaseHarnessType<E, THotStore, TColdStore> =
     Witness<TestingSlotClock, CachingEth1Backend<E>, E, THotStore, TColdStore>;
 
-pub type DiskHarnessType<E> = BaseHarnessType<E, LevelDB<E>, LevelDB<E>>;
+pub type DiskHarnessType<E> = BaseHarnessType<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>;
 pub type EphemeralHarnessType<E> = BaseHarnessType<E, MemoryStore<E>, MemoryStore<E>>;
 
 pub type BoxedMutator<E, Hot, Cold> = Box<
@@ -299,7 +300,10 @@ impl<E: EthSpec> Builder<EphemeralHarnessType<E>> {
 
 impl<E: EthSpec> Builder<DiskHarnessType<E>> {
     /// Disk store, start from genesis.
-    pub fn fresh_disk_store(mut self, store: Arc<HotColdDB<E, LevelDB<E>, LevelDB<E>>>) -> Self {
+    pub fn fresh_disk_store(
+        mut self,
+        store: Arc<HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>>,
+    ) -> Self {
         let validator_keypairs = self
             .validator_keypairs
             .clone()
@@ -324,7 +328,10 @@ impl<E: EthSpec> Builder<DiskHarnessType<E>> {
     }
 
     /// Disk store, resume.
-    pub fn resumed_disk_store(mut self, store: Arc<HotColdDB<E, LevelDB<E>, LevelDB<E>>>) -> Self {
+    pub fn resumed_disk_store(
+        mut self,
+        store: Arc<HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>>,
+    ) -> Self {
         let mutator = move |builder: BeaconChainBuilder<_>| {
             builder
                 .resume_from_db()

beacon_node/beacon_chain/tests/op_verification.rs

@@ -14,7 +14,8 @@ use state_processing::per_block_processing::errors::{
     AttesterSlashingInvalid, BlockOperationError, ExitInvalid, ProposerSlashingInvalid,
 };
 use std::sync::{Arc, LazyLock};
-use store::{LevelDB, StoreConfig};
+use store::database::interface::BeaconNodeBackend;
+use store::StoreConfig;
 use tempfile::{tempdir, TempDir};
 use types::*;
 
@@ -26,7 +27,7 @@ static KEYPAIRS: LazyLock<Vec<Keypair>> =
 
 type E = MinimalEthSpec;
 type TestHarness = BeaconChainHarness<DiskHarnessType<E>>;
-type HotColdDB = store::HotColdDB<E, LevelDB<E>, LevelDB<E>>;
+type HotColdDB = store::HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>;
 
 fn get_store(db_path: &TempDir) -> Arc<HotColdDB> {
     let spec = Arc::new(test_spec::<E>());

beacon_node/beacon_chain/tests/store_tests.rs

@@ -26,12 +26,13 @@ use std::convert::TryInto;
 use std::sync::{Arc, LazyLock};
 use std::time::Duration;
 use store::chunked_vector::Chunk;
+use store::database::interface::BeaconNodeBackend;
 use store::metadata::{SchemaVersion, CURRENT_SCHEMA_VERSION, STATE_UPPER_LIMIT_NO_RETAIN};
+use store::KeyValueStore;
 use store::{
     chunked_vector::{chunk_key, Field},
-    get_key_for_col,
     iter::{BlockRootsIterator, StateRootsIterator},
-    BlobInfo, DBColumn, HotColdDB, KeyValueStore, KeyValueStoreOp, LevelDB, StoreConfig,
+    BlobInfo, DBColumn, HotColdDB, KeyValueStoreOp, StoreConfig,
 };
 use tempfile::{tempdir, TempDir};
 use tokio::time::sleep;
@@ -49,15 +50,15 @@ static KEYPAIRS: LazyLock<Vec<Keypair>> =
 type E = MinimalEthSpec;
 type TestHarness = BeaconChainHarness<DiskHarnessType<E>>;
 
-fn get_store(db_path: &TempDir) -> Arc<HotColdDB<E, LevelDB<E>, LevelDB<E>>> {
+fn get_store(db_path: &TempDir) -> Arc<HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>> {
     get_store_generic(db_path, StoreConfig::default(), test_spec::<E>())
 }
 
 fn get_store_generic(
     db_path: &TempDir,
     config: StoreConfig,
     spec: ChainSpec,
-) -> Arc<HotColdDB<E, LevelDB<E>, LevelDB<E>>> {
+) -> Arc<HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>> {
     let hot_path = db_path.path().join("hot_db");
     let cold_path = db_path.path().join("cold_db");
     let blobs_path = db_path.path().join("blobs_db");
@@ -76,7 +77,7 @@ fn get_store_generic(
 }
 
 fn get_harness(
-    store: Arc<HotColdDB<E, LevelDB<E>, LevelDB<E>>>,
+    store: Arc<HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>>,
     validator_count: usize,
 ) -> TestHarness {
     // Most tests expect to retain historic states, so we use this as the default.
@@ -88,7 +89,7 @@ fn get_harness(
 }
 
 fn get_harness_generic(
-    store: Arc<HotColdDB<E, LevelDB<E>, LevelDB<E>>>,
+    store: Arc<HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>>,
     validator_count: usize,
     chain_config: ChainConfig,
 ) -> TestHarness {
@@ -269,10 +270,12 @@ async fn heal_freezer_block_roots_at_split() {
     let chunk_index = <store::chunked_vector::BlockRoots as Field<E>>::chunk_index(
         last_restore_point_slot.as_usize(),
     );
-    let key_chunk = get_key_for_col(DBColumn::BeaconBlockRoots.as_str(), &chunk_key(chunk_index));
     store
         .cold_db
-        .do_atomically(vec![KeyValueStoreOp::DeleteKey(key_chunk)])
+        .do_atomically(vec![KeyValueStoreOp::DeleteKey(
+            DBColumn::BeaconBlockRoots.as_str().to_owned(),
+            chunk_key(chunk_index).to_vec(),
+        )])
         .unwrap();
 
     let block_root_err = store
@@ -348,10 +351,12 @@ async fn heal_freezer_block_roots_with_skip_slots() {
     let chunk_index = <store::chunked_vector::BlockRoots as Field<E>>::chunk_index(
         last_restore_point_slot.as_usize(),
     );
-    let key_chunk = get_key_for_col(DBColumn::BeaconBlockRoots.as_str(), &chunk_key(chunk_index));
     store
         .cold_db
-        .do_atomically(vec![KeyValueStoreOp::DeleteKey(key_chunk)])
+        .do_atomically(vec![KeyValueStoreOp::DeleteKey(
+            DBColumn::BeaconBlockRoots.as_str().to_owned(),
+            chunk_key(chunk_index).to_vec(),
+        )])
         .unwrap();
 
     let block_root_err = store
@@ -493,7 +498,6 @@ async fn full_participation_no_skips() {
             AttestationStrategy::AllValidators,
         )
         .await;
-
     check_finalization(&harness, num_blocks_produced);
     check_split_slot(&harness, store);
     check_chain_dump(&harness, num_blocks_produced + 1);
@@ -2388,7 +2392,7 @@ async fn garbage_collect_temp_states_from_failed_block_on_startup() {
             .unwrap_err();
 
         assert_eq!(
-            store.iter_temporary_state_roots().count(),
+            store.iter_temporary_state_roots().unwrap().count(),
             block_slot.as_usize() - 1
         );
         store
@@ -2407,7 +2411,7 @@ async fn garbage_collect_temp_states_from_failed_block_on_startup() {
 
     // On startup, the store should garbage collect all the temporary states.
     let store = get_store(&db_path);
-    assert_eq!(store.iter_temporary_state_roots().count(), 0);
+    assert_eq!(store.iter_temporary_state_roots().unwrap().count(), 0);
 }
 
 #[tokio::test]
@@ -2443,7 +2447,7 @@ async fn garbage_collect_temp_states_from_failed_block_on_finalization() {
         .unwrap_err();
 
     assert_eq!(
-        store.iter_temporary_state_roots().count(),
+        store.iter_temporary_state_roots().unwrap().count(),
         block_slot.as_usize() - 1
     );
 
@@ -2462,7 +2466,7 @@ async fn garbage_collect_temp_states_from_failed_block_on_finalization() {
     assert_ne!(store.get_split_slot(), 0);
 
     // Check that temporary states have been pruned.
-    assert_eq!(store.iter_temporary_state_roots().count(), 0);
+    assert_eq!(store.iter_temporary_state_roots().unwrap().count(), 0);
 }
 
 #[tokio::test]
@@ -3745,7 +3749,10 @@ fn check_finalization(harness: &TestHarness, expected_slot: u64) {
 }
 
 /// Check that the HotColdDB's split_slot is equal to the start slot of the last finalized epoch.
-fn check_split_slot(harness: &TestHarness, store: Arc<HotColdDB<E, LevelDB<E>, LevelDB<E>>>) {
+fn check_split_slot(
+    harness: &TestHarness,
+    store: Arc<HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>>,
+) {
     let split_slot = store.get_split_slot();
     assert_eq!(
         harness