Modularize beacon node backend

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

@@ -1398,6 +1398,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

@@ -304,7 +304,6 @@ impl<E: EthSpec> PendingComponents<E> {
             };
             (Some(verified_blobs), None)
         };
-
         let executed_block = recover(diet_executed_block)?;
 
         let AvailabilityPendingExecutedBlock {
@@ -676,7 +675,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};
@@ -688,7 +687,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");
@@ -862,7 +861,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/historical_blocks.rs

@@ -10,10 +10,7 @@ use std::borrow::Cow;
 use std::iter;
 use std::time::Duration;
 use store::metadata::DataColumnInfo;
-use store::{
-    get_key_for_col, AnchorInfo, BlobInfo, DBColumn, Error as StoreError, KeyValueStore,
-    KeyValueStoreOp,
-};
+use store::{AnchorInfo, BlobInfo, DBColumn, Error as StoreError, KeyValueStore, KeyValueStoreOp};
 use strum::IntoStaticStr;
 use types::{FixedBytesExtended, Hash256, Slot};
 
@@ -153,7 +150,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             // Store block roots, including at all skip slots in the freezer DB.
             for slot in (block.slot().as_u64()..prev_block_slot.as_u64()).rev() {
                 cold_batch.push(KeyValueStoreOp::PutKeyValue(
-                    get_key_for_col(DBColumn::BeaconBlockRoots.into(), &slot.to_be_bytes()),
+                    DBColumn::BeaconBlockRoots,
+                    slot.to_be_bytes().to_vec(),
                     block_root.as_slice().to_vec(),
                 ));
             }
@@ -169,7 +167,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 let genesis_slot = self.spec.genesis_slot;
                 for slot in genesis_slot.as_u64()..prev_block_slot.as_u64() {
                     cold_batch.push(KeyValueStoreOp::PutKeyValue(
-                        get_key_for_col(DBColumn::BeaconBlockRoots.into(), &slot.to_be_bytes()),
+                        DBColumn::BeaconBlockRoots,
+                        slot.to_be_bytes().to_vec(),
                         self.genesis_block_root.as_slice().to_vec(),
                     ));
                 }

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;
@@ -272,13 +272,32 @@ 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>>(
+            DBColumn::LightClientUpdate,
+            &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)?;
@@ -298,6 +317,7 @@ impl<T: BeaconChainTypes> LightClientServerCache<T> {
 
             light_client_updates.push(light_client_update);
         }
+
         Ok(light_client_updates)
     }
 

beacon_node/beacon_chain/src/otb_verification_service.rs

@@ -64,7 +64,7 @@ impl OptimisticTransitionBlock {
         store
             .as_ref()
             .hot_db
-            .key_delete(OTBColumn.into(), self.root.as_slice())
+            .key_delete(OTBColumn, self.root.as_slice())
     }
 
     fn is_canonical<T: BeaconChainTypes>(
@@ -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,9 @@ pub fn downgrade_from_v21<T: BeaconChainTypes>(
             message: format!("{e:?}"),
         })?;
 
-        let db_key = get_key_for_col(DBColumn::PubkeyCache.into(), key.as_slice());
         ops.push(KeyValueStoreOp::PutKeyValue(
-            db_key,
+            DBColumn::PubkeyCache,
+            key.as_slice().to_vec(),
             pubkey_bytes.as_ssz_bytes(),
         ));
 

beacon_node/beacon_chain/src/schema_change/migration_schema_v22.rs

@@ -4,7 +4,6 @@ use std::sync::Arc;
 use store::chunked_iter::ChunkedVectorIter;
 use store::{
     chunked_vector::BlockRootsChunked,
-    get_key_for_col,
     metadata::{
         SchemaVersion, ANCHOR_FOR_ARCHIVE_NODE, ANCHOR_UNINITIALIZED, STATE_UPPER_LIMIT_NO_RETAIN,
     },
@@ -21,7 +20,7 @@ fn load_old_schema_frozen_state<T: BeaconChainTypes>(
 ) -> Result<Option<BeaconState<T::EthSpec>>, Error> {
     let Some(partial_state_bytes) = db
         .cold_db
-        .get_bytes(DBColumn::BeaconState.into(), state_root.as_slice())?
+        .get_bytes(DBColumn::BeaconState, state_root.as_slice())?
     else {
         return Ok(None);
     };
@@ -134,12 +133,9 @@ pub fn delete_old_schema_freezer_data<T: BeaconChainTypes>(
     ];
 
     for column in columns {
-        for res in db.cold_db.iter_column_keys::<Vec<u8>>(column) {
+        for res in db.cold_db.iter_column_keys::<Vec<u8>>(column)? {
             let key = res?;
-            cold_ops.push(KeyValueStoreOp::DeleteKey(get_key_for_col(
-                column.as_str(),
-                &key,
-            )));
+            cold_ops.push(KeyValueStoreOp::DeleteKey(column, key));
         }
     }
     let delete_ops = cold_ops.len();
@@ -175,7 +171,8 @@ pub fn write_new_schema_block_roots<T: BeaconChainTypes>(
     // Store the genesis block root if it would otherwise not be stored.
     if oldest_block_slot != 0 {
         cold_ops.push(KeyValueStoreOp::PutKeyValue(
-            get_key_for_col(DBColumn::BeaconBlockRoots.into(), &0u64.to_be_bytes()),
+            DBColumn::BeaconBlockRoots,
+            0u64.to_be_bytes().to_vec(),
             genesis_block_root.as_slice().to_vec(),
         ));
     }
@@ -192,10 +189,8 @@ pub fn write_new_schema_block_roots<T: BeaconChainTypes>(
     // OK to hold these in memory (10M slots * 43 bytes per KV ~= 430 MB).
     for (i, (slot, block_root)) in block_root_iter.enumerate() {
         cold_ops.push(KeyValueStoreOp::PutKeyValue(
-            get_key_for_col(
-                DBColumn::BeaconBlockRoots.into(),
-                &(slot as u64).to_be_bytes(),
-            ),
+            DBColumn::BeaconBlockRoots,
+            slot.to_be_bytes().to_vec(),
             block_root.as_slice().to_vec(),
         ));
 

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>());