diff --git a/Cargo.lock b/Cargo.lock index ae7861f44f8..899435a66ba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5301,6 +5301,7 @@ dependencies = [ "slasher", "slashing_protection", "slog", + "store", "task_executor", "tempfile", "types", @@ -8429,6 +8430,7 @@ dependencies = [ "metrics", "parking_lot 0.12.3", "rand", + "redb", "safe_arith", "serde", "slog", diff --git a/Makefile b/Makefile index 4d95f50c5ce..e8b44cb7807 100644 --- a/Makefile +++ b/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 diff --git a/beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs b/beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs index c8e92f7e9f5..cd793c8394b 100644 --- a/beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs +++ b/beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs @@ -317,7 +317,6 @@ impl PendingComponents { None, ) }; - let executed_block = recover(diet_executed_block)?; let AvailabilityPendingExecutedBlock { @@ -732,7 +731,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}; @@ -744,7 +743,7 @@ mod test { db_path: &TempDir, spec: Arc, log: Logger, - ) -> Arc, LevelDB>> { + ) -> Arc, BeaconNodeBackend>> { 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"); @@ -920,7 +919,11 @@ mod test { ) where E: EthSpec, - T: BeaconChainTypes, ColdStore = LevelDB, EthSpec = E>, + T: BeaconChainTypes< + HotStore = BeaconNodeBackend, + ColdStore = BeaconNodeBackend, + EthSpec = E, + >, { let log = test_logger(); let chain_db_path = tempdir().expect("should get temp dir"); diff --git a/beacon_node/beacon_chain/src/historical_blocks.rs b/beacon_node/beacon_chain/src/historical_blocks.rs index ddae54f464b..e22ec95a798 100644 --- a/beacon_node/beacon_chain/src/historical_blocks.rs +++ b/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 BeaconChain { // 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 BeaconChain { 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(), )); } diff --git a/beacon_node/beacon_chain/src/schema_change/migration_schema_v21.rs b/beacon_node/beacon_chain/src/schema_change/migration_schema_v21.rs index fcc8b9884ac..f02f5ee6f3a 100644 --- a/beacon_node/beacon_chain/src/schema_change/migration_schema_v21.rs +++ b/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; @@ -62,9 +60,9 @@ pub fn downgrade_from_v21( 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(), )); diff --git a/beacon_node/beacon_chain/src/schema_change/migration_schema_v22.rs b/beacon_node/beacon_chain/src/schema_change/migration_schema_v22.rs index c34512ededb..982c3ded467 100644 --- a/beacon_node/beacon_chain/src/schema_change/migration_schema_v22.rs +++ b/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( ) -> Result>, 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); }; @@ -136,10 +135,7 @@ pub fn delete_old_schema_freezer_data( for column in columns { for res in db.cold_db.iter_column_keys::>(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( // 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( // 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(), )); diff --git a/beacon_node/beacon_chain/src/test_utils.rs b/beacon_node/beacon_chain/src/test_utils.rs index 443cc686ebe..ba0a2159da1 100644 --- a/beacon_node/beacon_chain/src/test_utils.rs +++ b/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 { pub type BaseHarnessType = Witness, E, THotStore, TColdStore>; -pub type DiskHarnessType = BaseHarnessType, LevelDB>; +pub type DiskHarnessType = BaseHarnessType, BeaconNodeBackend>; pub type EphemeralHarnessType = BaseHarnessType, MemoryStore>; pub type BoxedMutator = Box< @@ -299,7 +300,10 @@ impl Builder> { impl Builder> { /// Disk store, start from genesis. - pub fn fresh_disk_store(mut self, store: Arc, LevelDB>>) -> Self { + pub fn fresh_disk_store( + mut self, + store: Arc, BeaconNodeBackend>>, + ) -> Self { let validator_keypairs = self .validator_keypairs .clone() @@ -324,7 +328,10 @@ impl Builder> { } /// Disk store, resume. - pub fn resumed_disk_store(mut self, store: Arc, LevelDB>>) -> Self { + pub fn resumed_disk_store( + mut self, + store: Arc, BeaconNodeBackend>>, + ) -> Self { let mutator = move |builder: BeaconChainBuilder<_>| { builder .resume_from_db() diff --git a/beacon_node/beacon_chain/tests/op_verification.rs b/beacon_node/beacon_chain/tests/op_verification.rs index df0d561e1cd..44fb298d6c6 100644 --- a/beacon_node/beacon_chain/tests/op_verification.rs +++ b/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> = type E = MinimalEthSpec; type TestHarness = BeaconChainHarness>; -type HotColdDB = store::HotColdDB, LevelDB>; +type HotColdDB = store::HotColdDB, BeaconNodeBackend>; fn get_store(db_path: &TempDir) -> Arc { let spec = Arc::new(test_spec::()); diff --git a/beacon_node/beacon_chain/tests/store_tests.rs b/beacon_node/beacon_chain/tests/store_tests.rs index 60d46e8269d..d1a38b1cdec 100644 --- a/beacon_node/beacon_chain/tests/store_tests.rs +++ b/beacon_node/beacon_chain/tests/store_tests.rs @@ -25,10 +25,11 @@ use std::collections::HashSet; use std::convert::TryInto; use std::sync::{Arc, LazyLock}; use std::time::Duration; +use store::database::interface::BeaconNodeBackend; use store::metadata::{SchemaVersion, CURRENT_SCHEMA_VERSION, STATE_UPPER_LIMIT_NO_RETAIN}; use store::{ iter::{BlockRootsIterator, StateRootsIterator}, - BlobInfo, DBColumn, HotColdDB, LevelDB, StoreConfig, + BlobInfo, DBColumn, HotColdDB, StoreConfig, }; use tempfile::{tempdir, TempDir}; use tokio::time::sleep; @@ -46,7 +47,7 @@ static KEYPAIRS: LazyLock> = type E = MinimalEthSpec; type TestHarness = BeaconChainHarness>; -fn get_store(db_path: &TempDir) -> Arc, LevelDB>> { +fn get_store(db_path: &TempDir) -> Arc, BeaconNodeBackend>> { get_store_generic(db_path, StoreConfig::default(), test_spec::()) } @@ -54,7 +55,7 @@ fn get_store_generic( db_path: &TempDir, config: StoreConfig, spec: ChainSpec, -) -> Arc, LevelDB>> { +) -> Arc, BeaconNodeBackend>> { let hot_path = db_path.path().join("chain_db"); let cold_path = db_path.path().join("freezer_db"); let blobs_path = db_path.path().join("blobs_db"); @@ -73,7 +74,7 @@ fn get_store_generic( } fn get_harness( - store: Arc, LevelDB>>, + store: Arc, BeaconNodeBackend>>, validator_count: usize, ) -> TestHarness { // Most tests expect to retain historic states, so we use this as the default. @@ -85,7 +86,7 @@ fn get_harness( } fn get_harness_generic( - store: Arc, LevelDB>>, + store: Arc, BeaconNodeBackend>>, validator_count: usize, chain_config: ChainConfig, ) -> TestHarness { @@ -244,7 +245,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); @@ -3508,7 +3508,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, LevelDB>>) { +fn check_split_slot( + harness: &TestHarness, + store: Arc, BeaconNodeBackend>>, +) { let split_slot = store.get_split_slot(); assert_eq!( harness diff --git a/beacon_node/client/src/builder.rs b/beacon_node/client/src/builder.rs index 1cd9e89b96c..e3bfd60a48b 100644 --- a/beacon_node/client/src/builder.rs +++ b/beacon_node/client/src/builder.rs @@ -14,7 +14,7 @@ use beacon_chain::{ eth1_chain::{CachingEth1Backend, Eth1Chain}, slot_clock::{SlotClock, SystemTimeSlotClock}, state_advance_timer::spawn_state_advance_timer, - store::{HotColdDB, ItemStore, LevelDB, StoreConfig}, + store::{HotColdDB, ItemStore, StoreConfig}, BeaconChain, BeaconChainTypes, Eth1ChainBackend, MigratorConfig, ServerSentEventHandler, }; use beacon_chain::{Kzg, LightClientProducerEvent}; @@ -41,6 +41,7 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; use std::time::{SystemTime, UNIX_EPOCH}; +use store::database::interface::BeaconNodeBackend; use timer::spawn_timer; use tokio::sync::oneshot; use types::{ @@ -1030,7 +1031,7 @@ where } impl - ClientBuilder, LevelDB>> + ClientBuilder, BeaconNodeBackend>> where TSlotClock: SlotClock + 'static, TEth1Backend: Eth1ChainBackend + 'static, diff --git a/beacon_node/http_api/tests/tests.rs b/beacon_node/http_api/tests/tests.rs index d9b3c8556c5..99b76966106 100644 --- a/beacon_node/http_api/tests/tests.rs +++ b/beacon_node/http_api/tests/tests.rs @@ -1933,7 +1933,7 @@ impl ApiTester { .sync_committee_period(&self.chain.spec) .unwrap(); - let result = match self + match self .client .get_beacon_light_client_updates::(current_sync_committee_period, 1) .await @@ -1954,7 +1954,6 @@ impl ApiTester { .unwrap(); assert_eq!(1, expected.len()); - assert_eq!(result.clone().unwrap().len(), expected.len()); self } @@ -1979,7 +1978,6 @@ impl ApiTester { .get_light_client_bootstrap(&self.chain.store, &block_root, 1u64, &self.chain.spec); assert!(expected.is_ok()); - assert_eq!(result.unwrap().data, expected.unwrap().unwrap().0); self diff --git a/beacon_node/src/cli.rs b/beacon_node/src/cli.rs index cecfcee868f..1339c158258 100644 --- a/beacon_node/src/cli.rs +++ b/beacon_node/src/cli.rs @@ -1591,5 +1591,14 @@ pub fn cli_app() -> Command { .action(ArgAction::Set) .display_order(0) ) + .arg( + Arg::new("beacon-node-backend") + .long("beacon-node-backend") + .value_name("DATABASE") + .value_parser(store::config::DatabaseBackend::VARIANTS.to_vec()) + .help("Set the database backend to be used by the beacon node.") + .action(ArgAction::Set) + .display_order(0) + ) .group(ArgGroup::new("enable_http").args(["http", "gui", "staking"]).multiple(true)) } diff --git a/beacon_node/src/config.rs b/beacon_node/src/config.rs index 8d8a44a6fd6..6d3c18d363a 100644 --- a/beacon_node/src/config.rs +++ b/beacon_node/src/config.rs @@ -432,6 +432,10 @@ pub fn get_config( warn!(log, "The slots-per-restore-point flag is deprecated"); } + if let Some(backend) = clap_utils::parse_optional(cli_args, "beacon-node-backend")? { + client_config.store.backend = backend; + } + if let Some(hierarchy_config) = clap_utils::parse_optional(cli_args, "hierarchy-exponents")? { client_config.store.hierarchy_config = hierarchy_config; } diff --git a/beacon_node/src/lib.rs b/beacon_node/src/lib.rs index 0c4cbf0f579..e3802c837cb 100644 --- a/beacon_node/src/lib.rs +++ b/beacon_node/src/lib.rs @@ -2,7 +2,6 @@ mod cli; mod config; pub use beacon_chain; -use beacon_chain::store::LevelDB; use beacon_chain::{ builder::Witness, eth1_chain::CachingEth1Backend, slot_clock::SystemTimeSlotClock, }; @@ -16,11 +15,19 @@ use slasher::{DatabaseBackendOverride, Slasher}; use slog::{info, warn}; use std::ops::{Deref, DerefMut}; use std::sync::Arc; +use store::database::interface::BeaconNodeBackend; use types::{ChainSpec, Epoch, EthSpec, ForkName}; /// A type-alias to the tighten the definition of a production-intended `Client`. -pub type ProductionClient = - Client, E, LevelDB, LevelDB>>; +pub type ProductionClient = Client< + Witness< + SystemTimeSlotClock, + CachingEth1Backend, + E, + BeaconNodeBackend, + BeaconNodeBackend, + >, +>; /// The beacon node `Client` that will be used in production. /// diff --git a/beacon_node/store/Cargo.toml b/beacon_node/store/Cargo.toml index 21d0cf8dec8..d2f3a5c562f 100644 --- a/beacon_node/store/Cargo.toml +++ b/beacon_node/store/Cargo.toml @@ -4,6 +4,11 @@ version = "0.2.0" authors = ["Paul Hauner "] edition = { workspace = true } +[features] +default = ["leveldb"] +leveldb = ["dep:leveldb"] +redb = ["dep:redb"] + [dev-dependencies] beacon_chain = { workspace = true } criterion = { workspace = true } @@ -17,11 +22,12 @@ directory = { workspace = true } ethereum_ssz = { workspace = true } ethereum_ssz_derive = { workspace = true } itertools = { workspace = true } -leveldb = { version = "0.8" } +leveldb = { version = "0.8.6", optional = true } logging = { workspace = true } lru = { workspace = true } metrics = { workspace = true } parking_lot = { workspace = true } +redb = { version = "2.1.3", optional = true } safe_arith = { workspace = true } serde = { workspace = true } slog = { workspace = true } diff --git a/beacon_node/store/src/chunked_vector.rs b/beacon_node/store/src/chunked_vector.rs index 83b8da2a189..90e8c173100 100644 --- a/beacon_node/store/src/chunked_vector.rs +++ b/beacon_node/store/src/chunked_vector.rs @@ -680,7 +680,7 @@ where key: &[u8], ) -> Result, Error> { store - .get_bytes(column.into(), key)? + .get_bytes(column, key)? .map(|bytes| Self::decode(&bytes)) .transpose() } @@ -691,8 +691,11 @@ where key: &[u8], ops: &mut Vec, ) -> Result<(), Error> { - let db_key = get_key_for_col(column.into(), key); - ops.push(KeyValueStoreOp::PutKeyValue(db_key, self.encode()?)); + ops.push(KeyValueStoreOp::PutKeyValue( + column, + key.to_vec(), + self.encode()?, + )); Ok(()) } diff --git a/beacon_node/store/src/config.rs b/beacon_node/store/src/config.rs index 4f675305706..64765fd66a0 100644 --- a/beacon_node/store/src/config.rs +++ b/beacon_node/store/src/config.rs @@ -1,16 +1,23 @@ use crate::hdiff::HierarchyConfig; +use crate::superstruct; use crate::{AnchorInfo, DBColumn, Error, Split, StoreItem}; use serde::{Deserialize, Serialize}; use ssz::{Decode, Encode}; use ssz_derive::{Decode, Encode}; use std::io::Write; use std::num::NonZeroUsize; -use superstruct::superstruct; +use strum::{Display, EnumString, EnumVariantNames}; use types::non_zero_usize::new_non_zero_usize; use types::EthSpec; use zstd::Encoder; -// Only used in tests. Mainnet sets a higher default on the CLI. +#[cfg(all(feature = "redb", not(feature = "leveldb")))] +pub const DEFAULT_BACKEND: DatabaseBackend = DatabaseBackend::Redb; +#[cfg(feature = "leveldb")] +pub const DEFAULT_BACKEND: DatabaseBackend = DatabaseBackend::LevelDb; + +pub const PREV_DEFAULT_SLOTS_PER_RESTORE_POINT: u64 = 2048; +pub const DEFAULT_SLOTS_PER_RESTORE_POINT: u64 = 8192; pub const DEFAULT_EPOCHS_PER_STATE_DIFF: u64 = 8; pub const DEFAULT_BLOCK_CACHE_SIZE: NonZeroUsize = new_non_zero_usize(64); pub const DEFAULT_STATE_CACHE_SIZE: NonZeroUsize = new_non_zero_usize(128); @@ -40,6 +47,8 @@ pub struct StoreConfig { pub compact_on_prune: bool, /// Whether to prune payloads on initialization and finalization. pub prune_payloads: bool, + /// Database backend to use. + pub backend: DatabaseBackend, /// State diff hierarchy. pub hierarchy_config: HierarchyConfig, /// Whether to prune blobs older than the blob data availability boundary. @@ -104,6 +113,7 @@ impl Default for StoreConfig { compact_on_init: false, compact_on_prune: true, prune_payloads: true, + backend: DEFAULT_BACKEND, hierarchy_config: HierarchyConfig::default(), prune_blobs: true, epochs_per_blob_prune: DEFAULT_EPOCHS_PER_BLOB_PRUNE, @@ -340,3 +350,14 @@ mod test { assert_eq!(config_out, config); } } + +#[derive( + Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize, Display, EnumString, EnumVariantNames, +)] +#[strum(serialize_all = "lowercase")] +pub enum DatabaseBackend { + #[cfg(feature = "leveldb")] + LevelDb, + #[cfg(feature = "redb")] + Redb, +} diff --git a/beacon_node/store/src/database.rs b/beacon_node/store/src/database.rs new file mode 100644 index 00000000000..2232f73c5cc --- /dev/null +++ b/beacon_node/store/src/database.rs @@ -0,0 +1,5 @@ +pub mod interface; +#[cfg(feature = "leveldb")] +pub mod leveldb_impl; +#[cfg(feature = "redb")] +pub mod redb_impl; diff --git a/beacon_node/store/src/database/interface.rs b/beacon_node/store/src/database/interface.rs new file mode 100644 index 00000000000..b213433241c --- /dev/null +++ b/beacon_node/store/src/database/interface.rs @@ -0,0 +1,220 @@ +#[cfg(feature = "leveldb")] +use crate::database::leveldb_impl; +#[cfg(feature = "redb")] +use crate::database::redb_impl; +use crate::{config::DatabaseBackend, KeyValueStoreOp, StoreConfig}; +use crate::{metrics, ColumnIter, ColumnKeyIter, DBColumn, Error, ItemStore, Key, KeyValueStore}; +use std::collections::HashSet; +use std::path::Path; +use types::EthSpec; + +pub enum BeaconNodeBackend { + #[cfg(feature = "leveldb")] + LevelDb(leveldb_impl::LevelDB), + #[cfg(feature = "redb")] + Redb(redb_impl::Redb), +} + +impl ItemStore for BeaconNodeBackend {} + +impl KeyValueStore for BeaconNodeBackend { + fn get_bytes(&self, column: DBColumn, key: &[u8]) -> Result>, Error> { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::get_bytes(txn, column, key), + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::get_bytes(txn, column, key), + } + } + + fn put_bytes(&self, column: DBColumn, key: &[u8], value: &[u8]) -> Result<(), Error> { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::put_bytes_with_options( + txn, + column, + key, + value, + txn.write_options(), + ), + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::put_bytes_with_options( + txn, + column, + key, + value, + txn.write_options(), + ), + } + } + + fn put_bytes_sync(&self, column: DBColumn, key: &[u8], value: &[u8]) -> Result<(), Error> { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::put_bytes_with_options( + txn, + column, + key, + value, + txn.write_options_sync(), + ), + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::put_bytes_with_options( + txn, + column, + key, + value, + txn.write_options_sync(), + ), + } + } + + fn sync(&self) -> Result<(), Error> { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::sync(txn), + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::sync(txn), + } + } + + fn key_exists(&self, column: DBColumn, key: &[u8]) -> Result { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::key_exists(txn, column, key), + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::key_exists(txn, column, key), + } + } + + fn key_delete(&self, column: DBColumn, key: &[u8]) -> Result<(), Error> { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::key_delete(txn, column, key), + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::key_delete(txn, column, key), + } + } + + fn do_atomically(&self, batch: Vec) -> Result<(), Error> { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::do_atomically(txn, batch), + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::do_atomically(txn, batch), + } + } + + fn begin_rw_transaction(&self) -> parking_lot::MutexGuard<()> { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::begin_rw_transaction(txn), + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::begin_rw_transaction(txn), + } + } + + fn compact(&self) -> Result<(), Error> { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::compact(txn), + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::compact(txn), + } + } + + fn iter_column_keys_from(&self, _column: DBColumn, from: &[u8]) -> ColumnKeyIter { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => { + leveldb_impl::LevelDB::iter_column_keys_from(txn, _column, from) + } + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => { + redb_impl::Redb::iter_column_keys_from(txn, _column, from) + } + } + } + + fn iter_column_keys(&self, column: DBColumn) -> ColumnKeyIter { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::iter_column_keys(txn, column), + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::iter_column_keys(txn, column), + } + } + + fn iter_column_from(&self, column: DBColumn, from: &[u8]) -> ColumnIter { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => { + leveldb_impl::LevelDB::iter_column_from(txn, column, from) + } + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::iter_column_from(txn, column, from), + } + } + + fn compact_column(&self, _column: DBColumn) -> Result<(), Error> { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::compact_column(txn, _column), + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::compact(txn), + } + } + + fn delete_batch(&self, col: DBColumn, ops: HashSet<&[u8]>) -> Result<(), Error> { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::delete_batch(txn, col, ops), + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::delete_batch(txn, col, ops), + } + } + + fn delete_if( + &self, + column: DBColumn, + f: impl FnMut(&[u8]) -> Result, + ) -> Result<(), Error> { + match self { + #[cfg(feature = "leveldb")] + BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::delete_if(txn, column, f), + #[cfg(feature = "redb")] + BeaconNodeBackend::Redb(txn) => redb_impl::Redb::delete_if(txn, column, f), + } + } +} + +impl BeaconNodeBackend { + pub fn open(config: &StoreConfig, path: &Path) -> Result { + metrics::inc_counter_vec(&metrics::DISK_DB_TYPE, &[&config.backend.to_string()]); + match config.backend { + #[cfg(feature = "leveldb")] + DatabaseBackend::LevelDb => { + leveldb_impl::LevelDB::open(path).map(BeaconNodeBackend::LevelDb) + } + #[cfg(feature = "redb")] + DatabaseBackend::Redb => redb_impl::Redb::open(path).map(BeaconNodeBackend::Redb), + } + } +} + +pub struct WriteOptions { + /// fsync before acknowledging a write operation. + pub sync: bool, +} + +impl WriteOptions { + pub fn new() -> Self { + WriteOptions { sync: false } + } +} + +impl Default for WriteOptions { + fn default() -> Self { + Self::new() + } +} diff --git a/beacon_node/store/src/database/leveldb_impl.rs b/beacon_node/store/src/database/leveldb_impl.rs new file mode 100644 index 00000000000..3d8bbe14737 --- /dev/null +++ b/beacon_node/store/src/database/leveldb_impl.rs @@ -0,0 +1,304 @@ +use crate::hot_cold_store::{BytesKey, HotColdDBError}; +use crate::Key; +use crate::{ + get_key_for_col, metrics, ColumnIter, ColumnKeyIter, DBColumn, Error, KeyValueStoreOp, +}; +use leveldb::{ + compaction::Compaction, + database::{ + batch::{Batch, Writebatch}, + kv::KV, + Database, + }, + iterator::{Iterable, LevelDBIterator}, + options::{Options, ReadOptions}, +}; +use parking_lot::{Mutex, MutexGuard}; +use std::collections::HashSet; +use std::marker::PhantomData; +use std::path::Path; +use types::{EthSpec, FixedBytesExtended, Hash256}; + +use super::interface::WriteOptions; + +pub struct LevelDB { + db: Database, + /// A mutex to synchronise sensitive read-write transactions. + transaction_mutex: Mutex<()>, + _phantom: PhantomData, +} + +impl From for leveldb::options::WriteOptions { + fn from(options: WriteOptions) -> Self { + let mut opts = leveldb::options::WriteOptions::new(); + opts.sync = options.sync; + opts + } +} + +impl LevelDB { + pub fn open(path: &Path) -> Result { + let mut options = Options::new(); + + options.create_if_missing = true; + + let db = Database::open(path, options)?; + let transaction_mutex = Mutex::new(()); + + Ok(Self { + db, + transaction_mutex, + _phantom: PhantomData, + }) + } + + pub fn read_options(&self) -> ReadOptions { + ReadOptions::new() + } + + pub fn write_options(&self) -> WriteOptions { + WriteOptions::new() + } + + pub fn write_options_sync(&self) -> WriteOptions { + let mut opts = WriteOptions::new(); + opts.sync = true; + opts + } + + pub fn put_bytes_with_options( + &self, + col: DBColumn, + key: &[u8], + val: &[u8], + opts: WriteOptions, + ) -> Result<(), Error> { + let column_key = get_key_for_col(col, key); + + metrics::inc_counter_vec(&metrics::DISK_DB_WRITE_COUNT, &[col.into()]); + metrics::inc_counter_vec_by( + &metrics::DISK_DB_WRITE_BYTES, + &[col.into()], + val.len() as u64, + ); + let timer = metrics::start_timer(&metrics::DISK_DB_WRITE_TIMES); + + self.db + .put(opts.into(), BytesKey::from_vec(column_key), val) + .map_err(Into::into) + .map(|()| { + metrics::stop_timer(timer); + }) + } + + /// Store some `value` in `column`, indexed with `key`. + pub fn put_bytes(&self, col: DBColumn, key: &[u8], val: &[u8]) -> Result<(), Error> { + self.put_bytes_with_options(col, key, val, self.write_options()) + } + + pub fn put_bytes_sync(&self, col: DBColumn, key: &[u8], val: &[u8]) -> Result<(), Error> { + self.put_bytes_with_options(col, key, val, self.write_options_sync()) + } + + pub fn sync(&self) -> Result<(), Error> { + self.put_bytes_sync(DBColumn::Dummy, b"sync", b"sync") + } + + // Retrieve some bytes in `column` with `key`. + pub fn get_bytes(&self, col: DBColumn, key: &[u8]) -> Result>, Error> { + let column_key = get_key_for_col(col, key); + + metrics::inc_counter_vec(&metrics::DISK_DB_READ_COUNT, &[col.into()]); + let timer = metrics::start_timer(&metrics::DISK_DB_READ_TIMES); + + self.db + .get(self.read_options(), BytesKey::from_vec(column_key)) + .map_err(Into::into) + .map(|opt| { + opt.inspect(|bytes| { + metrics::inc_counter_vec_by( + &metrics::DISK_DB_READ_BYTES, + &[col.into()], + bytes.len() as u64, + ); + metrics::stop_timer(timer); + }) + }) + } + + /// Return `true` if `key` exists in `column`. + pub fn key_exists(&self, col: DBColumn, key: &[u8]) -> Result { + let column_key = get_key_for_col(col, key); + + metrics::inc_counter_vec(&metrics::DISK_DB_EXISTS_COUNT, &[col.into()]); + + self.db + .get(self.read_options(), BytesKey::from_vec(column_key)) + .map_err(Into::into) + .map(|val| val.is_some()) + } + + /// Removes `key` from `column`. + pub fn key_delete(&self, col: DBColumn, key: &[u8]) -> Result<(), Error> { + let column_key = get_key_for_col(col, key); + + metrics::inc_counter_vec(&metrics::DISK_DB_DELETE_COUNT, &[col.into()]); + + self.db + .delete(self.write_options().into(), BytesKey::from_vec(column_key)) + .map_err(Into::into) + } + + pub fn do_atomically(&self, ops_batch: Vec) -> Result<(), Error> { + let mut leveldb_batch = Writebatch::new(); + for op in ops_batch { + match op { + KeyValueStoreOp::PutKeyValue(col, key, value) => { + let _timer = metrics::start_timer(&metrics::DISK_DB_WRITE_TIMES); + metrics::inc_counter_vec_by( + &metrics::DISK_DB_WRITE_BYTES, + &[col.into()], + value.len() as u64, + ); + metrics::inc_counter_vec(&metrics::DISK_DB_WRITE_COUNT, &[col.into()]); + let column_key = get_key_for_col(col, &key); + leveldb_batch.put(BytesKey::from_vec(column_key), &value); + } + + KeyValueStoreOp::DeleteKey(col, key) => { + let _timer = metrics::start_timer(&metrics::DISK_DB_DELETE_TIMES); + metrics::inc_counter_vec(&metrics::DISK_DB_DELETE_COUNT, &[col.into()]); + let column_key = get_key_for_col(col, &key); + leveldb_batch.delete(BytesKey::from_vec(column_key)); + } + } + } + self.db.write(self.write_options().into(), &leveldb_batch)?; + Ok(()) + } + + pub fn begin_rw_transaction(&self) -> MutexGuard<()> { + self.transaction_mutex.lock() + } + + /// Compact all values in the states and states flag columns. + pub fn compact(&self) -> Result<(), Error> { + let _timer = metrics::start_timer(&metrics::DISK_DB_COMPACT_TIMES); + let endpoints = |column: DBColumn| { + ( + BytesKey::from_vec(get_key_for_col(column, Hash256::zero().as_slice())), + BytesKey::from_vec(get_key_for_col( + column, + Hash256::repeat_byte(0xff).as_slice(), + )), + ) + }; + + for (start_key, end_key) in [ + endpoints(DBColumn::BeaconStateTemporary), + endpoints(DBColumn::BeaconState), + endpoints(DBColumn::BeaconStateSummary), + ] { + self.db.compact(&start_key, &end_key); + } + + Ok(()) + } + + pub fn compact_column(&self, column: DBColumn) -> Result<(), Error> { + // Use key-size-agnostic keys [] and 0xff..ff with a minimum of 32 bytes to account for + // columns that may change size between sub-databases or schema versions. + let start_key = BytesKey::from_vec(get_key_for_col(column, &[])); + let end_key = BytesKey::from_vec(get_key_for_col( + column, + &vec![0xff; std::cmp::max(column.key_size(), 32)], + )); + self.db.compact(&start_key, &end_key); + Ok(()) + } + + pub fn iter_column_from(&self, column: DBColumn, from: &[u8]) -> ColumnIter { + let start_key = BytesKey::from_vec(get_key_for_col(column, from)); + let iter = self.db.iter(self.read_options()); + iter.seek(&start_key); + + Box::new( + iter.take_while(move |(key, _)| key.matches_column(column)) + .map(move |(bytes_key, value)| { + metrics::inc_counter_vec(&metrics::DISK_DB_READ_COUNT, &[column.into()]); + metrics::inc_counter_vec_by( + &metrics::DISK_DB_READ_BYTES, + &[column.into()], + value.len() as u64, + ); + let key = bytes_key.remove_column_variable(column).ok_or_else(|| { + HotColdDBError::IterationError { + unexpected_key: bytes_key.clone(), + } + })?; + Ok((K::from_bytes(key)?, value)) + }), + ) + } + + pub fn iter_column_keys_from(&self, column: DBColumn, from: &[u8]) -> ColumnKeyIter { + let start_key = BytesKey::from_vec(get_key_for_col(column, from)); + + let iter = self.db.keys_iter(self.read_options()); + iter.seek(&start_key); + + Box::new( + iter.take_while(move |key| key.matches_column(column)) + .map(move |bytes_key| { + metrics::inc_counter_vec(&metrics::DISK_DB_KEY_READ_COUNT, &[column.into()]); + metrics::inc_counter_vec_by( + &metrics::DISK_DB_KEY_READ_BYTES, + &[column.into()], + bytes_key.key.len() as u64, + ); + let key = &bytes_key.key[column.as_bytes().len()..]; + K::from_bytes(key) + }), + ) + } + + /// Iterate through all keys and values in a particular column. + pub fn iter_column_keys(&self, column: DBColumn) -> ColumnKeyIter { + self.iter_column_keys_from(column, &vec![0; column.key_size()]) + } + + pub fn iter_column(&self, column: DBColumn) -> ColumnIter { + self.iter_column_from(column, &vec![0; column.key_size()]) + } + + pub fn delete_batch(&self, col: DBColumn, ops: HashSet<&[u8]>) -> Result<(), Error> { + let mut leveldb_batch = Writebatch::new(); + for op in ops { + let column_key = get_key_for_col(col, op); + leveldb_batch.delete(BytesKey::from_vec(column_key)); + } + self.db.write(self.write_options().into(), &leveldb_batch)?; + Ok(()) + } + + pub fn delete_if( + &self, + column: DBColumn, + mut f: impl FnMut(&[u8]) -> Result, + ) -> Result<(), Error> { + let mut leveldb_batch = Writebatch::new(); + let iter = self.db.iter(self.read_options()); + + iter.take_while(move |(key, _)| key.matches_column(column)) + .for_each(|(key, value)| { + if f(&value).unwrap_or(false) { + let _timer = metrics::start_timer(&metrics::DISK_DB_DELETE_TIMES); + metrics::inc_counter_vec(&metrics::DISK_DB_DELETE_COUNT, &[column.into()]); + leveldb_batch.delete(key); + } + }); + + self.db.write(self.write_options().into(), &leveldb_batch)?; + Ok(()) + } +} diff --git a/beacon_node/store/src/database/redb_impl.rs b/beacon_node/store/src/database/redb_impl.rs new file mode 100644 index 00000000000..6a776da7b17 --- /dev/null +++ b/beacon_node/store/src/database/redb_impl.rs @@ -0,0 +1,314 @@ +use crate::{metrics, ColumnIter, ColumnKeyIter, Key}; +use crate::{DBColumn, Error, KeyValueStoreOp}; +use parking_lot::{Mutex, MutexGuard, RwLock}; +use redb::TableDefinition; +use std::collections::HashSet; +use std::{borrow::BorrowMut, marker::PhantomData, path::Path}; +use strum::IntoEnumIterator; +use types::EthSpec; + +use super::interface::WriteOptions; + +pub const DB_FILE_NAME: &str = "database.redb"; + +pub struct Redb { + db: RwLock, + transaction_mutex: Mutex<()>, + _phantom: PhantomData, +} + +impl From for redb::Durability { + fn from(options: WriteOptions) -> Self { + if options.sync { + redb::Durability::Immediate + } else { + redb::Durability::Eventual + } + } +} + +impl Redb { + pub fn open(path: &Path) -> Result { + let db_file = path.join(DB_FILE_NAME); + let db = redb::Database::create(db_file)?; + let transaction_mutex = Mutex::new(()); + + for column in DBColumn::iter() { + Redb::::create_table(&db, column.into())?; + } + + Ok(Self { + db: db.into(), + transaction_mutex, + _phantom: PhantomData, + }) + } + + fn create_table(db: &redb::Database, table_name: &str) -> Result<(), Error> { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = TableDefinition::new(table_name); + let tx = db.begin_write()?; + tx.open_table(table_definition)?; + tx.commit().map_err(Into::into) + } + + pub fn write_options(&self) -> WriteOptions { + WriteOptions::new() + } + + pub fn write_options_sync(&self) -> WriteOptions { + let mut opts = WriteOptions::new(); + opts.sync = true; + opts + } + + pub fn begin_rw_transaction(&self) -> MutexGuard<()> { + self.transaction_mutex.lock() + } + + pub fn put_bytes_with_options( + &self, + col: DBColumn, + key: &[u8], + val: &[u8], + opts: WriteOptions, + ) -> Result<(), Error> { + metrics::inc_counter_vec(&metrics::DISK_DB_WRITE_COUNT, &[col.into()]); + metrics::inc_counter_vec_by( + &metrics::DISK_DB_WRITE_BYTES, + &[col.into()], + val.len() as u64, + ); + let timer = metrics::start_timer(&metrics::DISK_DB_WRITE_TIMES); + + let table_definition: TableDefinition<'_, &[u8], &[u8]> = TableDefinition::new(col.into()); + let open_db = self.db.read(); + let mut tx = open_db.begin_write()?; + tx.set_durability(opts.into()); + let mut table = tx.open_table(table_definition)?; + + table.insert(key, val).map(|_| { + metrics::stop_timer(timer); + })?; + drop(table); + tx.commit().map_err(Into::into) + } + + /// Store some `value` in `column`, indexed with `key`. + pub fn put_bytes(&self, col: DBColumn, key: &[u8], val: &[u8]) -> Result<(), Error> { + self.put_bytes_with_options(col, key, val, self.write_options()) + } + + pub fn put_bytes_sync(&self, col: DBColumn, key: &[u8], val: &[u8]) -> Result<(), Error> { + self.put_bytes_with_options(col, key, val, self.write_options_sync()) + } + + pub fn sync(&self) -> Result<(), Error> { + self.put_bytes_sync(DBColumn::Dummy, b"sync", b"sync") + } + + // Retrieve some bytes in `column` with `key`. + pub fn get_bytes(&self, col: DBColumn, key: &[u8]) -> Result>, Error> { + metrics::inc_counter_vec(&metrics::DISK_DB_READ_COUNT, &[col.into()]); + let timer = metrics::start_timer(&metrics::DISK_DB_READ_TIMES); + + let table_definition: TableDefinition<'_, &[u8], &[u8]> = TableDefinition::new(col.into()); + let open_db = self.db.read(); + let tx = open_db.begin_read()?; + let table = tx.open_table(table_definition)?; + + let result = table.get(key)?; + + match result { + Some(access_guard) => { + let value = access_guard.value().to_vec(); + metrics::inc_counter_vec_by( + &metrics::DISK_DB_READ_BYTES, + &[col.into()], + value.len() as u64, + ); + metrics::stop_timer(timer); + Ok(Some(value)) + } + None => { + metrics::stop_timer(timer); + Ok(None) + } + } + } + + /// Return `true` if `key` exists in `column`. + pub fn key_exists(&self, col: DBColumn, key: &[u8]) -> Result { + metrics::inc_counter_vec(&metrics::DISK_DB_EXISTS_COUNT, &[col.into()]); + + let table_definition: TableDefinition<'_, &[u8], &[u8]> = TableDefinition::new(col.into()); + let open_db = self.db.read(); + let tx = open_db.begin_read()?; + let table = tx.open_table(table_definition)?; + + table + .get(key) + .map_err(Into::into) + .map(|access_guard| access_guard.is_some()) + } + + /// Removes `key` from `column`. + pub fn key_delete(&self, col: DBColumn, key: &[u8]) -> Result<(), Error> { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = TableDefinition::new(col.into()); + let open_db = self.db.read(); + let tx = open_db.begin_write()?; + let mut table = tx.open_table(table_definition)?; + metrics::inc_counter_vec(&metrics::DISK_DB_DELETE_COUNT, &[col.into()]); + + table.remove(key).map(|_| ())?; + drop(table); + tx.commit().map_err(Into::into) + } + + pub fn do_atomically(&self, ops_batch: Vec) -> Result<(), Error> { + let open_db = self.db.read(); + let mut tx = open_db.begin_write()?; + tx.set_durability(self.write_options().into()); + for op in ops_batch { + match op { + KeyValueStoreOp::PutKeyValue(column, key, value) => { + let _timer = metrics::start_timer(&metrics::DISK_DB_WRITE_TIMES); + metrics::inc_counter_vec_by( + &metrics::DISK_DB_WRITE_BYTES, + &[column.into()], + value.len() as u64, + ); + metrics::inc_counter_vec(&metrics::DISK_DB_WRITE_COUNT, &[column.into()]); + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(column.into()); + + let mut table = tx.open_table(table_definition)?; + table.insert(key.as_slice(), value.as_slice())?; + drop(table); + } + + KeyValueStoreOp::DeleteKey(column, key) => { + metrics::inc_counter_vec(&metrics::DISK_DB_DELETE_COUNT, &[column.into()]); + let _timer = metrics::start_timer(&metrics::DISK_DB_DELETE_TIMES); + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(column.into()); + + let mut table = tx.open_table(table_definition)?; + table.remove(key.as_slice())?; + drop(table); + } + } + } + + tx.commit()?; + Ok(()) + } + + /// Compact all values in the states and states flag columns. + pub fn compact(&self) -> Result<(), Error> { + let _timer = metrics::start_timer(&metrics::DISK_DB_COMPACT_TIMES); + let mut open_db = self.db.write(); + let mut_db = open_db.borrow_mut(); + mut_db.compact().map_err(Into::into).map(|_| ()) + } + + pub fn iter_column_keys_from(&self, column: DBColumn, from: &[u8]) -> ColumnKeyIter { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(column.into()); + + let iter = { + let open_db = self.db.read(); + let read_txn = open_db.begin_read()?; + let table = read_txn.open_table(table_definition)?; + table.range(from..)?.map(move |res| { + let (key, _) = res?; + metrics::inc_counter_vec(&metrics::DISK_DB_KEY_READ_COUNT, &[column.into()]); + metrics::inc_counter_vec_by( + &metrics::DISK_DB_KEY_READ_BYTES, + &[column.into()], + key.value().len() as u64, + ); + K::from_bytes(key.value()) + }) + }; + + Box::new(iter) + } + + /// Iterate through all keys and values in a particular column. + pub fn iter_column_keys(&self, column: DBColumn) -> ColumnKeyIter { + self.iter_column_keys_from(column, &vec![0; column.key_size()]) + } + + pub fn iter_column_from(&self, column: DBColumn, from: &[u8]) -> ColumnIter { + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(column.into()); + + let prefix = from.to_vec(); + + let iter = { + let open_db = self.db.read(); + let read_txn = open_db.begin_read()?; + let table = read_txn.open_table(table_definition)?; + + table + .range(from..)? + .take_while(move |res| match res.as_ref() { + Ok((_, _)) => true, + Err(_) => false, + }) + .map(move |res| { + let (key, value) = res?; + metrics::inc_counter_vec(&metrics::DISK_DB_READ_COUNT, &[column.into()]); + metrics::inc_counter_vec_by( + &metrics::DISK_DB_READ_BYTES, + &[column.into()], + value.value().len() as u64, + ); + Ok((K::from_bytes(key.value())?, value.value().to_vec())) + }) + }; + + Ok(Box::new(iter)) + } + + pub fn iter_column(&self, column: DBColumn) -> ColumnIter { + self.iter_column_from(column, &vec![0; column.key_size()], |_, _| true) + } + + pub fn delete_batch(&self, col: DBColumn, ops: HashSet<&[u8]>) -> Result<(), Error> { + let open_db = self.db.read(); + let mut tx = open_db.begin_write()?; + + tx.set_durability(redb::Durability::None); + + let table_definition: TableDefinition<'_, &[u8], &[u8]> = TableDefinition::new(col.into()); + + let mut table = tx.open_table(table_definition)?; + table.retain(|key, _| !ops.contains(key))?; + + drop(table); + tx.commit()?; + Ok(()) + } + + pub fn delete_if( + &self, + column: DBColumn, + mut f: impl FnMut(&[u8]) -> Result, + ) -> Result<(), Error> { + let open_db = self.db.read(); + let mut tx = open_db.begin_write()?; + + tx.set_durability(redb::Durability::None); + + let table_definition: TableDefinition<'_, &[u8], &[u8]> = + TableDefinition::new(column.into()); + + let mut table = tx.open_table(table_definition)?; + table.retain(|_, value| !f(value).unwrap_or(false))?; + + drop(table); + tx.commit()?; + Ok(()) + } +} diff --git a/beacon_node/store/src/errors.rs b/beacon_node/store/src/errors.rs index 6bb4edee6b2..41fd17ef437 100644 --- a/beacon_node/store/src/errors.rs +++ b/beacon_node/store/src/errors.rs @@ -2,6 +2,8 @@ use crate::chunked_vector::ChunkError; use crate::config::StoreConfigError; use crate::hot_cold_store::HotColdDBError; use crate::{hdiff, DBColumn}; +#[cfg(feature = "leveldb")] +use leveldb::error::Error as LevelDBError; use ssz::DecodeError; use state_processing::BlockReplayError; use types::{milhouse, BeaconStateError, EpochCacheError, Hash256, InconsistentFork, Slot}; @@ -48,6 +50,16 @@ pub enum Error { MissingGenesisState, MissingSnapshot(Slot), BlockReplayError(BlockReplayError), + AddPayloadLogicError, + InvalidKey, + InvalidBytes, + InconsistentFork(InconsistentFork), + #[cfg(feature = "leveldb")] + LevelDbError(LevelDBError), + #[cfg(feature = "redb")] + RedbError(redb::Error), + CacheBuildError(EpochCacheError), + RandaoMixOutOfBounds, MilhouseError(milhouse::Error), Compression(std::io::Error), FinalizedStateDecreasingSlot, @@ -56,17 +68,11 @@ pub enum Error { state_root: Hash256, slot: Slot, }, - AddPayloadLogicError, - InvalidKey, - InvalidBytes, - InconsistentFork(InconsistentFork), Hdiff(hdiff::Error), - CacheBuildError(EpochCacheError), ForwardsIterInvalidColumn(DBColumn), ForwardsIterGap(DBColumn, Slot, Slot), StateShouldNotBeRequired(Slot), MissingBlock(Hash256), - RandaoMixOutOfBounds, GenesisStateUnknown, ArithError(safe_arith::ArithError), } @@ -145,6 +151,62 @@ impl From for Error { } } +#[cfg(feature = "leveldb")] +impl From for Error { + fn from(e: LevelDBError) -> Error { + Error::LevelDbError(e) + } +} + +#[cfg(feature = "redb")] +impl From for Error { + fn from(e: redb::Error) -> Self { + Error::RedbError(e) + } +} + +#[cfg(feature = "redb")] +impl From for Error { + fn from(e: redb::TableError) -> Self { + Error::RedbError(e.into()) + } +} + +#[cfg(feature = "redb")] +impl From for Error { + fn from(e: redb::TransactionError) -> Self { + Error::RedbError(e.into()) + } +} + +#[cfg(feature = "redb")] +impl From for Error { + fn from(e: redb::DatabaseError) -> Self { + Error::RedbError(e.into()) + } +} + +#[cfg(feature = "redb")] +impl From for Error { + fn from(e: redb::StorageError) -> Self { + Error::RedbError(e.into()) + } +} + +#[cfg(feature = "redb")] +impl From for Error { + fn from(e: redb::CommitError) -> Self { + Error::RedbError(e.into()) + } +} + +#[cfg(feature = "redb")] +impl From for Error { + fn from(e: redb::CompactionError) -> Self { + Error::RedbError(e.into()) + } +} + impl From for Error { fn from(e: EpochCacheError) -> Error { Error::CacheBuildError(e) diff --git a/beacon_node/store/src/forwards_iter.rs b/beacon_node/store/src/forwards_iter.rs index 955bd33b30f..5300a74c060 100644 --- a/beacon_node/store/src/forwards_iter.rs +++ b/beacon_node/store/src/forwards_iter.rs @@ -4,7 +4,6 @@ use crate::{ColumnIter, DBColumn, HotColdDB, ItemStore}; use itertools::process_results; use std::marker::PhantomData; use types::{BeaconState, EthSpec, Hash256, Slot}; - pub type HybridForwardsBlockRootsIterator<'a, E, Hot, Cold> = HybridForwardsIterator<'a, E, Hot, Cold>; pub type HybridForwardsStateRootsIterator<'a, E, Hot, Cold> = diff --git a/beacon_node/store/src/garbage_collection.rs b/beacon_node/store/src/garbage_collection.rs index 5f8ed8f5e73..06393f2d219 100644 --- a/beacon_node/store/src/garbage_collection.rs +++ b/beacon_node/store/src/garbage_collection.rs @@ -1,10 +1,11 @@ //! Garbage collection process that runs at start-up to clean up the database. +use crate::database::interface::BeaconNodeBackend; use crate::hot_cold_store::HotColdDB; -use crate::{Error, LevelDB, StoreOp}; +use crate::{DBColumn, Error}; use slog::debug; use types::EthSpec; -impl HotColdDB, LevelDB> +impl HotColdDB, BeaconNodeBackend> where E: EthSpec, { @@ -16,21 +17,22 @@ where /// Delete the temporary states that were leftover by failed block imports. pub fn delete_temp_states(&self) -> Result<(), Error> { - let delete_ops = - self.iter_temporary_state_roots() - .try_fold(vec![], |mut ops, state_root| { - let state_root = state_root?; - ops.push(StoreOp::DeleteState(state_root, None)); - Result::<_, Error>::Ok(ops) - })?; - - if !delete_ops.is_empty() { + let mut ops = vec![]; + self.iter_temporary_state_roots().for_each(|state_root| { + if let Ok(state_root) = state_root { + ops.push(state_root); + } + }); + if !ops.is_empty() { debug!( self.log, "Garbage collecting {} temporary states", - delete_ops.len() + ops.len() ); - self.do_atomically_with_block_and_blobs_cache(delete_ops)?; + + self.delete_batch(DBColumn::BeaconState, ops.clone())?; + self.delete_batch(DBColumn::BeaconStateSummary, ops.clone())?; + self.delete_batch(DBColumn::BeaconStateTemporary, ops)?; } Ok(()) diff --git a/beacon_node/store/src/hot_cold_store.rs b/beacon_node/store/src/hot_cold_store.rs index c29305f9831..75251cb5fb4 100644 --- a/beacon_node/store/src/hot_cold_store.rs +++ b/beacon_node/store/src/hot_cold_store.rs @@ -1,10 +1,10 @@ use crate::config::{OnDiskStoreConfig, StoreConfig}; +use crate::database::interface::BeaconNodeBackend; use crate::forwards_iter::{HybridForwardsBlockRootsIterator, HybridForwardsStateRootsIterator}; use crate::hdiff::{HDiff, HDiffBuffer, HierarchyModuli, StorageStrategy}; use crate::historic_state_cache::HistoricStateCache; use crate::impls::beacon_state::{get_full_state, store_full_state}; use crate::iter::{BlockRootsIterator, ParentRootBlockIterator, RootsIterator}; -use crate::leveldb_store::{BytesKey, LevelDB}; use crate::memory_store::MemoryStore; use crate::metadata::{ AnchorInfo, BlobInfo, CompactionTimestamp, DataColumnInfo, PruningCheckpoint, SchemaVersion, @@ -14,12 +14,10 @@ use crate::metadata::{ }; use crate::state_cache::{PutStateOutcome, StateCache}; use crate::{ - get_data_column_key, get_key_for_col, BlobSidecarListFromRoot, DBColumn, DatabaseBlock, Error, - ItemStore, KeyValueStoreOp, StoreItem, StoreOp, + get_data_column_key, metrics, parse_data_column_key, BlobSidecarListFromRoot, DBColumn, + DatabaseBlock, Error, ItemStore, KeyValueStore, KeyValueStoreOp, StoreItem, StoreOp, }; -use crate::{metrics, parse_data_column_key}; use itertools::{process_results, Itertools}; -use leveldb::iterator::LevelDBIterator; use lru::LruCache; use parking_lot::{Mutex, RwLock}; use safe_arith::SafeArith; @@ -231,7 +229,7 @@ impl HotColdDB, MemoryStore> { } } -impl HotColdDB, LevelDB> { +impl HotColdDB, BeaconNodeBackend> { /// Open a new or existing database, with the given paths to the hot and cold DBs. /// /// The `migrate_schema` function is passed in so that the parent `BeaconChain` can provide @@ -249,7 +247,7 @@ impl HotColdDB, LevelDB> { let hierarchy = config.hierarchy_config.to_moduli()?; - let hot_db = LevelDB::open(hot_path)?; + let hot_db = BeaconNodeBackend::open(&config, hot_path)?; let anchor_info = RwLock::new(Self::load_anchor_info(&hot_db)?); let db = HotColdDB { @@ -257,8 +255,8 @@ impl HotColdDB, LevelDB> { anchor_info, blob_info: RwLock::new(BlobInfo::default()), data_column_info: RwLock::new(DataColumnInfo::default()), - cold_db: LevelDB::open(cold_path)?, - blobs_db: LevelDB::open(blobs_db_path)?, + blobs_db: BeaconNodeBackend::open(&config, blobs_db_path)?, + cold_db: BeaconNodeBackend::open(&config, cold_path)?, hot_db, block_cache: Mutex::new(BlockCache::new(config.block_cache_size)), state_cache: Mutex::new(StateCache::new(config.state_cache_size)), @@ -408,23 +406,8 @@ impl HotColdDB, LevelDB> { /// Return an iterator over the state roots of all temporary states. pub fn iter_temporary_state_roots(&self) -> impl Iterator> + '_ { - let column = DBColumn::BeaconStateTemporary; - let start_key = - BytesKey::from_vec(get_key_for_col(column.into(), Hash256::zero().as_slice())); - - let keys_iter = self.hot_db.keys_iter(); - keys_iter.seek(&start_key); - - keys_iter - .take_while(move |key| key.matches_column(column)) - .map(move |bytes_key| { - bytes_key.remove_column(column).ok_or_else(|| { - HotColdDBError::IterationError { - unexpected_key: bytes_key, - } - .into() - }) - }) + self.hot_db + .iter_column_keys::(DBColumn::BeaconStateTemporary) } } @@ -536,9 +519,9 @@ impl, Cold: ItemStore> HotColdDB blinded_block: &SignedBeaconBlock>, ops: &mut Vec, ) { - let db_key = get_key_for_col(DBColumn::BeaconBlock.into(), key.as_slice()); ops.push(KeyValueStoreOp::PutKeyValue( - db_key, + DBColumn::BeaconBlock, + key.as_slice().into(), blinded_block.as_ssz_bytes(), )); } @@ -660,7 +643,7 @@ impl, Cold: ItemStore> HotColdDB decoder: impl FnOnce(&[u8]) -> Result, ssz::DecodeError>, ) -> Result>, Error> { self.hot_db - .get_bytes(DBColumn::BeaconBlock.into(), block_root.as_slice())? + .get_bytes(DBColumn::BeaconBlock, block_root.as_slice())? .map(|block_bytes| decoder(&block_bytes)) .transpose() .map_err(|e| e.into()) @@ -673,10 +656,12 @@ impl, Cold: ItemStore> HotColdDB block_root: &Hash256, fork_name: ForkName, ) -> Result>, Error> { - let column = ExecutionPayload::::db_column().into(); let key = block_root.as_slice(); - match self.hot_db.get_bytes(column, key)? { + match self + .hot_db + .get_bytes(ExecutionPayload::::db_column(), key)? + { Some(bytes) => Ok(Some(ExecutionPayload::from_ssz_bytes(&bytes, fork_name)?)), None => Ok(None), } @@ -705,10 +690,7 @@ impl, Cold: ItemStore> HotColdDB ) -> Result, Error> { let column = DBColumn::SyncCommitteeBranch; - if let Some(bytes) = self - .hot_db - .get_bytes(column.into(), &block_root.as_ssz_bytes())? - { + if let Some(bytes) = self.hot_db.get_bytes(column, &block_root.as_ssz_bytes())? { let sync_committee_branch = Vec::::from_ssz_bytes(&bytes)?; return Ok(Some(sync_committee_branch)); } @@ -725,7 +707,7 @@ impl, Cold: ItemStore> HotColdDB if let Some(bytes) = self .hot_db - .get_bytes(column.into(), &sync_committee_period.as_ssz_bytes())? + .get_bytes(column, &sync_committee_period.as_ssz_bytes())? { let sync_committee: SyncCommittee = SyncCommittee::from_ssz_bytes(&bytes)?; return Ok(Some(sync_committee)); @@ -741,7 +723,7 @@ impl, Cold: ItemStore> HotColdDB ) -> Result<(), Error> { let column = DBColumn::SyncCommitteeBranch; self.hot_db.put_bytes( - column.into(), + column, &block_root.as_ssz_bytes(), &sync_committee_branch.as_ssz_bytes(), )?; @@ -755,7 +737,7 @@ impl, Cold: ItemStore> HotColdDB ) -> Result<(), Error> { let column = DBColumn::SyncCommittee; self.hot_db.put_bytes( - column.into(), + column, &sync_committee_period.to_le_bytes(), &sync_committee.as_ssz_bytes(), )?; @@ -767,10 +749,10 @@ impl, Cold: ItemStore> HotColdDB &self, sync_committee_period: u64, ) -> Result>, Error> { - let column = DBColumn::LightClientUpdate; - let res = self - .hot_db - .get_bytes(column.into(), &sync_committee_period.to_le_bytes())?; + let res = self.hot_db.get_bytes( + DBColumn::LightClientUpdate, + &sync_committee_period.to_le_bytes(), + )?; if let Some(light_client_update_bytes) = res { let epoch = sync_committee_period @@ -822,10 +804,8 @@ impl, Cold: ItemStore> HotColdDB sync_committee_period: u64, light_client_update: &LightClientUpdate, ) -> Result<(), Error> { - let column = DBColumn::LightClientUpdate; - self.hot_db.put_bytes( - column.into(), + DBColumn::LightClientUpdate, &sync_committee_period.to_le_bytes(), &light_client_update.as_ssz_bytes(), )?; @@ -836,29 +816,29 @@ impl, Cold: ItemStore> HotColdDB /// Check if the blobs for a block exists on disk. pub fn blobs_exist(&self, block_root: &Hash256) -> Result { self.blobs_db - .key_exists(DBColumn::BeaconBlob.into(), block_root.as_slice()) + .key_exists(DBColumn::BeaconBlob, block_root.as_slice()) } /// Determine whether a block exists in the database. pub fn block_exists(&self, block_root: &Hash256) -> Result { self.hot_db - .key_exists(DBColumn::BeaconBlock.into(), block_root.as_slice()) + .key_exists(DBColumn::BeaconBlock, block_root.as_slice()) } /// Delete a block from the store and the block cache. pub fn delete_block(&self, block_root: &Hash256) -> Result<(), Error> { self.block_cache.lock().delete(block_root); self.hot_db - .key_delete(DBColumn::BeaconBlock.into(), block_root.as_slice())?; + .key_delete(DBColumn::BeaconBlock, block_root.as_slice())?; self.hot_db - .key_delete(DBColumn::ExecPayload.into(), block_root.as_slice())?; + .key_delete(DBColumn::ExecPayload, block_root.as_slice())?; self.blobs_db - .key_delete(DBColumn::BeaconBlob.into(), block_root.as_slice()) + .key_delete(DBColumn::BeaconBlob, block_root.as_slice()) } pub fn put_blobs(&self, block_root: &Hash256, blobs: BlobSidecarList) -> Result<(), Error> { self.blobs_db.put_bytes( - DBColumn::BeaconBlob.into(), + DBColumn::BeaconBlob, block_root.as_slice(), &blobs.as_ssz_bytes(), )?; @@ -872,8 +852,11 @@ impl, Cold: ItemStore> HotColdDB blobs: BlobSidecarList, ops: &mut Vec, ) { - let db_key = get_key_for_col(DBColumn::BeaconBlob.into(), key.as_slice()); - ops.push(KeyValueStoreOp::PutKeyValue(db_key, blobs.as_ssz_bytes())); + ops.push(KeyValueStoreOp::PutKeyValue( + DBColumn::BeaconBlob, + key.as_slice().to_vec(), + blobs.as_ssz_bytes(), + )); } pub fn data_columns_as_kv_store_ops( @@ -883,12 +866,9 @@ impl, Cold: ItemStore> HotColdDB ops: &mut Vec, ) { for data_column in data_columns { - let db_key = get_key_for_col( - DBColumn::BeaconDataColumn.into(), - &get_data_column_key(block_root, &data_column.index), - ); ops.push(KeyValueStoreOp::PutKeyValue( - db_key, + DBColumn::BeaconDataColumn, + get_data_column_key(block_root, &data_column.index), data_column.as_ssz_bytes(), )); } @@ -1202,63 +1182,68 @@ impl, Cold: ItemStore> HotColdDB } StoreOp::DeleteStateTemporaryFlag(state_root) => { - let db_key = - get_key_for_col(TemporaryFlag::db_column().into(), state_root.as_slice()); - key_value_batch.push(KeyValueStoreOp::DeleteKey(db_key)); + key_value_batch.push(KeyValueStoreOp::DeleteKey( + TemporaryFlag::db_column(), + state_root.as_slice().to_vec(), + )); } StoreOp::DeleteBlock(block_root) => { - let key = get_key_for_col(DBColumn::BeaconBlock.into(), block_root.as_slice()); - key_value_batch.push(KeyValueStoreOp::DeleteKey(key)); + key_value_batch.push(KeyValueStoreOp::DeleteKey( + DBColumn::BeaconBlock, + block_root.as_slice().to_vec(), + )); } StoreOp::DeleteBlobs(block_root) => { - let key = get_key_for_col(DBColumn::BeaconBlob.into(), block_root.as_slice()); - key_value_batch.push(KeyValueStoreOp::DeleteKey(key)); + key_value_batch.push(KeyValueStoreOp::DeleteKey( + DBColumn::BeaconBlob, + block_root.as_slice().to_vec(), + )); } StoreOp::DeleteDataColumns(block_root, column_indices) => { for index in column_indices { - let key = get_key_for_col( - DBColumn::BeaconDataColumn.into(), - &get_data_column_key(&block_root, &index), - ); - key_value_batch.push(KeyValueStoreOp::DeleteKey(key)); + let key = get_data_column_key(&block_root, &index); + key_value_batch + .push(KeyValueStoreOp::DeleteKey(DBColumn::BeaconDataColumn, key)); } } StoreOp::DeleteState(state_root, slot) => { // Delete the hot state summary. - let state_summary_key = - get_key_for_col(DBColumn::BeaconStateSummary.into(), state_root.as_slice()); - key_value_batch.push(KeyValueStoreOp::DeleteKey(state_summary_key)); + key_value_batch.push(KeyValueStoreOp::DeleteKey( + DBColumn::BeaconStateSummary, + state_root.as_slice().to_vec(), + )); // Delete the state temporary flag (if any). Temporary flags are commonly // created by the state advance routine. - let state_temp_key = get_key_for_col( - DBColumn::BeaconStateTemporary.into(), - state_root.as_slice(), - ); - key_value_batch.push(KeyValueStoreOp::DeleteKey(state_temp_key)); + key_value_batch.push(KeyValueStoreOp::DeleteKey( + DBColumn::BeaconStateTemporary, + state_root.as_slice().to_vec(), + )); if slot.map_or(true, |slot| slot % E::slots_per_epoch() == 0) { - let state_key = - get_key_for_col(DBColumn::BeaconState.into(), state_root.as_slice()); - key_value_batch.push(KeyValueStoreOp::DeleteKey(state_key)); + key_value_batch.push(KeyValueStoreOp::DeleteKey( + DBColumn::BeaconState, + state_root.as_slice().to_vec(), + )); } } StoreOp::DeleteExecutionPayload(block_root) => { - let key = get_key_for_col(DBColumn::ExecPayload.into(), block_root.as_slice()); - key_value_batch.push(KeyValueStoreOp::DeleteKey(key)); + key_value_batch.push(KeyValueStoreOp::DeleteKey( + DBColumn::ExecPayload, + block_root.as_slice().to_vec(), + )); } StoreOp::DeleteSyncCommitteeBranch(block_root) => { - let key = get_key_for_col( - DBColumn::SyncCommitteeBranch.into(), - block_root.as_slice(), - ); - key_value_batch.push(KeyValueStoreOp::DeleteKey(key)); + key_value_batch.push(KeyValueStoreOp::DeleteKey( + DBColumn::SyncCommitteeBranch, + block_root.as_slice().to_vec(), + )); } StoreOp::KeyValueOp(kv_op) => { @@ -1269,6 +1254,19 @@ impl, Cold: ItemStore> HotColdDB Ok(key_value_batch) } + pub fn delete_batch(&self, col: DBColumn, ops: Vec) -> Result<(), Error> { + let new_ops: HashSet<&[u8]> = ops.iter().map(|v| v.as_slice()).collect(); + self.hot_db.delete_batch(col, new_ops) + } + + pub fn delete_if( + &self, + column: DBColumn, + f: impl Fn(&[u8]) -> Result, + ) -> Result<(), Error> { + self.hot_db.delete_if(column, f) + } + pub fn do_atomically_with_block_and_blobs_cache( &self, batch: Vec>, @@ -1608,10 +1606,8 @@ impl, Cold: ItemStore> HotColdDB ) -> Result<(), Error> { ops.push(ColdStateSummary { slot }.as_kv_store_op(*state_root)); ops.push(KeyValueStoreOp::PutKeyValue( - get_key_for_col( - DBColumn::BeaconStateRoots.into(), - &slot.as_u64().to_be_bytes(), - ), + DBColumn::BeaconStateRoots, + slot.as_u64().to_be_bytes().to_vec(), state_root.as_slice().to_vec(), )); Ok(()) @@ -1678,19 +1674,19 @@ impl, Cold: ItemStore> HotColdDB out }; - let key = get_key_for_col( - DBColumn::BeaconStateSnapshot.into(), - &state.slot().as_u64().to_be_bytes(), - ); - ops.push(KeyValueStoreOp::PutKeyValue(key, compressed_value)); + ops.push(KeyValueStoreOp::PutKeyValue( + DBColumn::BeaconStateSnapshot, + state.slot().as_u64().to_be_bytes().to_vec(), + compressed_value, + )); Ok(()) } fn load_cold_state_bytes_as_snapshot(&self, slot: Slot) -> Result>, Error> { - match self.cold_db.get_bytes( - DBColumn::BeaconStateSnapshot.into(), - &slot.as_u64().to_be_bytes(), - )? { + match self + .cold_db + .get_bytes(DBColumn::BeaconStateSnapshot, &slot.as_u64().to_be_bytes())? + { Some(bytes) => { let _timer = metrics::start_timer(&metrics::STORE_BEACON_STATE_FREEZER_DECOMPRESS_TIME); @@ -1731,11 +1727,11 @@ impl, Cold: ItemStore> HotColdDB }; let diff_bytes = diff.as_ssz_bytes(); - let key = get_key_for_col( - DBColumn::BeaconStateDiff.into(), - &state.slot().as_u64().to_be_bytes(), - ); - ops.push(KeyValueStoreOp::PutKeyValue(key, diff_bytes)); + ops.push(KeyValueStoreOp::PutKeyValue( + DBColumn::BeaconStateDiff, + state.slot().as_u64().to_be_bytes().to_vec(), + diff_bytes, + )); Ok(()) } @@ -1858,10 +1854,7 @@ impl, Cold: ItemStore> HotColdDB let bytes = { let _t = metrics::start_timer(&metrics::BEACON_HDIFF_READ_TIMES); self.cold_db - .get_bytes( - DBColumn::BeaconStateDiff.into(), - &slot.as_u64().to_be_bytes(), - )? + .get_bytes(DBColumn::BeaconStateDiff, &slot.as_u64().to_be_bytes())? .ok_or(HotColdDBError::MissingHDiff(slot))? }; let hdiff = { @@ -2054,7 +2047,7 @@ impl, Cold: ItemStore> HotColdDB match self .blobs_db - .get_bytes(DBColumn::BeaconBlob.into(), block_root.as_slice())? + .get_bytes(DBColumn::BeaconBlob, block_root.as_slice())? { Some(ref blobs_bytes) => { // We insert a VariableList of BlobSidecars into the db, but retrieve @@ -2084,8 +2077,17 @@ impl, Cold: ItemStore> HotColdDB /// Fetch all keys in the data_column column with prefix `block_root` pub fn get_data_column_keys(&self, block_root: Hash256) -> Result, Error> { self.blobs_db - .iter_raw_keys(DBColumn::BeaconDataColumn, block_root.as_slice()) - .map(|key| key.and_then(|key| parse_data_column_key(key).map(|key| key.1))) + .iter_column_from::>(DBColumn::BeaconDataColumn, block_root.as_slice()) + .take_while(|res| { + let Ok((key, _)) = res else { return false }; + + if !key.starts_with(block_root.as_slice()) { + return false; + } + + true + }) + .map(|key| key.and_then(|(key, _)| parse_data_column_key(key).map(|key| key.1))) .collect() } @@ -2106,7 +2108,7 @@ impl, Cold: ItemStore> HotColdDB } match self.blobs_db.get_bytes( - DBColumn::BeaconDataColumn.into(), + DBColumn::BeaconDataColumn, &get_data_column_key(block_root, column_index), )? { Some(ref data_column_bytes) => { @@ -2164,10 +2166,12 @@ impl, Cold: ItemStore> HotColdDB schema_version: SchemaVersion, mut ops: Vec, ) -> Result<(), Error> { - let column = SchemaVersion::db_column().into(); let key = SCHEMA_VERSION_KEY.as_slice(); - let db_key = get_key_for_col(column, key); - let op = KeyValueStoreOp::PutKeyValue(db_key, schema_version.as_store_bytes()); + let op = KeyValueStoreOp::PutKeyValue( + SchemaVersion::db_column(), + key.to_vec(), + schema_version.as_store_bytes(), + ); ops.push(op); self.hot_db.do_atomically(ops) @@ -2589,7 +2593,8 @@ impl, Cold: ItemStore> HotColdDB let mut ops = vec![]; for slot in start_slot.as_u64()..end_slot.as_u64() { ops.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(), )); } @@ -2811,77 +2816,62 @@ impl, Cold: ItemStore> HotColdDB "data_availability_boundary" => data_availability_boundary, ); - let mut ops = vec![]; - let mut last_pruned_block_root = None; + // We collect block roots of deleted blobs in memory. Even for 10y of blob history this + // vec won't go beyond 1GB. We can probably optimise this out eventually. + let mut removed_block_roots = vec![]; - for res in self.forwards_block_roots_iterator_until(oldest_blob_slot, end_slot, || { - let (_, split_state) = self - .get_advanced_hot_state(split.block_root, split.slot, split.state_root)? - .ok_or(HotColdDBError::MissingSplitState( - split.state_root, - split.slot, - ))?; + let remove_blob_if = |blobs_bytes: &[u8]| { + let blobs = Vec::from_ssz_bytes(blobs_bytes)?; + let Some(blob): Option<&Arc>> = blobs.first() else { + return Ok(false); + }; - Ok((split_state, split.block_root)) - })? { - let (block_root, slot) = match res { - Ok(tuple) => tuple, - Err(e) => { - warn!( - self.log, - "Stopping blob pruning early"; - "error" => ?e, - ); - break; - } + if blob.slot() <= end_slot { + // Store the block root so we can delete from the blob cache + removed_block_roots.push(blob.block_root()); + // Delete from the on-disk db + return Ok(true); }; + Ok(false) + }; - if Some(block_root) != last_pruned_block_root { - if self - .spec - .is_peer_das_enabled_for_epoch(slot.epoch(E::slots_per_epoch())) - { - // data columns - let indices = self.get_data_column_keys(block_root)?; - if !indices.is_empty() { - trace!( - self.log, - "Pruning data columns of block"; - "slot" => slot, - "block_root" => ?block_root, - ); - last_pruned_block_root = Some(block_root); - ops.push(StoreOp::DeleteDataColumns(block_root, indices)); - } - } else if self.blobs_exist(&block_root)? { - trace!( - self.log, - "Pruning blobs of block"; - "slot" => slot, - "block_root" => ?block_root, - ); - last_pruned_block_root = Some(block_root); - ops.push(StoreOp::DeleteBlobs(block_root)); - } - } + self.blobs_db + .delete_if(DBColumn::BeaconBlob, remove_blob_if)?; - if slot >= end_slot { - break; - } + if self.spec.is_peer_das_enabled_for_epoch(start_epoch) { + let remove_data_column_if = |blobs_bytes: &[u8]| { + let data_column: DataColumnSidecar = + DataColumnSidecar::from_ssz_bytes(blobs_bytes)?; + + if data_column.slot() <= end_slot { + return Ok(true); + }; + + Ok(false) + }; + + self.blobs_db + .delete_if(DBColumn::BeaconDataColumn, remove_data_column_if)?; } - let blob_lists_pruned = ops.len(); + + // Remove deleted blobs from the cache. + let mut block_cache = self.block_cache.lock(); + for block_root in removed_block_roots { + block_cache.delete_blobs(&block_root); + } + drop(block_cache); + let new_blob_info = BlobInfo { oldest_blob_slot: Some(end_slot + 1), blobs_db: blob_info.blobs_db, }; - let update_blob_info = self.compare_and_set_blob_info(blob_info, new_blob_info)?; - ops.push(StoreOp::KeyValueOp(update_blob_info)); - self.do_atomically_with_block_and_blobs_cache(ops)?; + let op = self.compare_and_set_blob_info(blob_info, new_blob_info)?; + self.do_atomically_with_block_and_blobs_cache(vec![StoreOp::KeyValueOp(op)])?; + debug!( self.log, "Blob pruning complete"; - "blob_lists_pruned" => blob_lists_pruned, ); Ok(()) @@ -2944,10 +2934,7 @@ impl, Cold: ItemStore> HotColdDB for column in columns { for res in self.cold_db.iter_column_keys::>(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(); @@ -3085,10 +3072,8 @@ pub fn migrate_database, Cold: ItemStore>( // Store the slot to block root mapping. cold_db_block_ops.push(KeyValueStoreOp::PutKeyValue( - get_key_for_col( - DBColumn::BeaconBlockRoots.into(), - &slot.as_u64().to_be_bytes(), - ), + DBColumn::BeaconBlockRoots, + slot.as_u64().to_be_bytes().to_vec(), block_root.as_slice().to_vec(), )); @@ -3339,3 +3324,57 @@ impl StoreItem for TemporaryFlag { Ok(TemporaryFlag) } } + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub struct BytesKey { + pub key: Vec, +} + +impl db_key::Key for BytesKey { + fn from_u8(key: &[u8]) -> Self { + Self { key: key.to_vec() } + } + + fn as_slice T>(&self, f: F) -> T { + f(self.key.as_slice()) + } +} + +impl BytesKey { + pub fn starts_with(&self, prefix: &Self) -> bool { + self.key.starts_with(&prefix.key) + } + + /// Return `true` iff this `BytesKey` was created with the given `column`. + pub fn matches_column(&self, column: DBColumn) -> bool { + self.key.starts_with(column.as_bytes()) + } + + /// Remove the column from a key, returning its `Hash256` portion. + pub fn remove_column(&self, column: DBColumn) -> Option { + if self.matches_column(column) { + let subkey = &self.key[column.as_bytes().len()..]; + if subkey.len() == 32 { + return Some(Hash256::from_slice(subkey)); + } + } + None + } + + /// Remove the column from a key. + /// + /// Will return `None` if the value doesn't match the column or has the wrong length. + pub fn remove_column_variable(&self, column: DBColumn) -> Option<&[u8]> { + if self.matches_column(column) { + let subkey = &self.key[column.as_bytes().len()..]; + if subkey.len() == column.key_size() { + return Some(subkey); + } + } + None + } + + pub fn from_vec(key: Vec) -> Self { + Self { key } + } +} diff --git a/beacon_node/store/src/impls/beacon_state.rs b/beacon_node/store/src/impls/beacon_state.rs index 48c289f2b2d..fd08e547f13 100644 --- a/beacon_node/store/src/impls/beacon_state.rs +++ b/beacon_node/store/src/impls/beacon_state.rs @@ -13,8 +13,11 @@ pub fn store_full_state( }; metrics::inc_counter_by(&metrics::BEACON_STATE_WRITE_BYTES, bytes.len() as u64); metrics::inc_counter(&metrics::BEACON_STATE_WRITE_COUNT); - let key = get_key_for_col(DBColumn::BeaconState.into(), state_root.as_slice()); - ops.push(KeyValueStoreOp::PutKeyValue(key, bytes)); + ops.push(KeyValueStoreOp::PutKeyValue( + DBColumn::BeaconState, + state_root.as_slice().to_vec(), + bytes, + )); Ok(()) } @@ -25,7 +28,7 @@ pub fn get_full_state, E: EthSpec>( ) -> Result>, Error> { let total_timer = metrics::start_timer(&metrics::BEACON_STATE_READ_TIMES); - match db.get_bytes(DBColumn::BeaconState.into(), state_root.as_slice())? { + match db.get_bytes(DBColumn::BeaconState, state_root.as_slice())? { Some(bytes) => { let overhead_timer = metrics::start_timer(&metrics::BEACON_STATE_READ_OVERHEAD_TIMES); let container = StorageContainer::from_ssz_bytes(&bytes, spec)?; diff --git a/beacon_node/store/src/leveldb_store.rs b/beacon_node/store/src/leveldb_store.rs deleted file mode 100644 index 720afd0f3f6..00000000000 --- a/beacon_node/store/src/leveldb_store.rs +++ /dev/null @@ -1,310 +0,0 @@ -use super::*; -use crate::hot_cold_store::HotColdDBError; -use leveldb::compaction::Compaction; -use leveldb::database::batch::{Batch, Writebatch}; -use leveldb::database::kv::KV; -use leveldb::database::Database; -use leveldb::error::Error as LevelDBError; -use leveldb::iterator::{Iterable, KeyIterator, LevelDBIterator}; -use leveldb::options::{Options, ReadOptions, WriteOptions}; -use parking_lot::Mutex; -use std::marker::PhantomData; -use std::path::Path; - -/// A wrapped leveldb database. -pub struct LevelDB { - db: Database, - /// A mutex to synchronise sensitive read-write transactions. - transaction_mutex: Mutex<()>, - _phantom: PhantomData, -} - -impl LevelDB { - /// Open a database at `path`, creating a new database if one does not already exist. - pub fn open(path: &Path) -> Result { - let mut options = Options::new(); - - options.create_if_missing = true; - - let db = Database::open(path, options)?; - let transaction_mutex = Mutex::new(()); - - Ok(Self { - db, - transaction_mutex, - _phantom: PhantomData, - }) - } - - fn read_options(&self) -> ReadOptions { - ReadOptions::new() - } - - fn write_options(&self) -> WriteOptions { - WriteOptions::new() - } - - fn write_options_sync(&self) -> WriteOptions { - let mut opts = WriteOptions::new(); - opts.sync = true; - opts - } - - fn put_bytes_with_options( - &self, - col: &str, - key: &[u8], - val: &[u8], - opts: WriteOptions, - ) -> Result<(), Error> { - let column_key = get_key_for_col(col, key); - - metrics::inc_counter_vec(&metrics::DISK_DB_WRITE_COUNT, &[col]); - metrics::inc_counter_vec_by(&metrics::DISK_DB_WRITE_BYTES, &[col], val.len() as u64); - let _timer = metrics::start_timer(&metrics::DISK_DB_WRITE_TIMES); - - self.db - .put(opts, BytesKey::from_vec(column_key), val) - .map_err(Into::into) - } - - pub fn keys_iter(&self) -> KeyIterator { - self.db.keys_iter(self.read_options()) - } -} - -impl KeyValueStore for LevelDB { - /// Store some `value` in `column`, indexed with `key`. - fn put_bytes(&self, col: &str, key: &[u8], val: &[u8]) -> Result<(), Error> { - self.put_bytes_with_options(col, key, val, self.write_options()) - } - - fn put_bytes_sync(&self, col: &str, key: &[u8], val: &[u8]) -> Result<(), Error> { - self.put_bytes_with_options(col, key, val, self.write_options_sync()) - } - - fn sync(&self) -> Result<(), Error> { - self.put_bytes_sync("sync", b"sync", b"sync") - } - - /// Retrieve some bytes in `column` with `key`. - fn get_bytes(&self, col: &str, key: &[u8]) -> Result>, Error> { - let column_key = get_key_for_col(col, key); - - metrics::inc_counter_vec(&metrics::DISK_DB_READ_COUNT, &[col]); - let timer = metrics::start_timer(&metrics::DISK_DB_READ_TIMES); - - self.db - .get(self.read_options(), BytesKey::from_vec(column_key)) - .map_err(Into::into) - .map(|opt| { - opt.inspect(|bytes| { - metrics::inc_counter_vec_by( - &metrics::DISK_DB_READ_BYTES, - &[col], - bytes.len() as u64, - ); - metrics::stop_timer(timer); - }) - }) - } - - /// Return `true` if `key` exists in `column`. - fn key_exists(&self, col: &str, key: &[u8]) -> Result { - let column_key = get_key_for_col(col, key); - - metrics::inc_counter_vec(&metrics::DISK_DB_EXISTS_COUNT, &[col]); - - self.db - .get(self.read_options(), BytesKey::from_vec(column_key)) - .map_err(Into::into) - .map(|val| val.is_some()) - } - - /// Removes `key` from `column`. - fn key_delete(&self, col: &str, key: &[u8]) -> Result<(), Error> { - let column_key = get_key_for_col(col, key); - - metrics::inc_counter_vec(&metrics::DISK_DB_DELETE_COUNT, &[col]); - - self.db - .delete(self.write_options(), BytesKey::from_vec(column_key)) - .map_err(Into::into) - } - - fn do_atomically(&self, ops_batch: Vec) -> Result<(), Error> { - let mut leveldb_batch = Writebatch::new(); - for op in ops_batch { - match op { - KeyValueStoreOp::PutKeyValue(key, value) => { - let col = get_col_from_key(&key).unwrap_or("unknown".to_owned()); - metrics::inc_counter_vec(&metrics::DISK_DB_WRITE_COUNT, &[&col]); - metrics::inc_counter_vec_by( - &metrics::DISK_DB_WRITE_BYTES, - &[&col], - value.len() as u64, - ); - - leveldb_batch.put(BytesKey::from_vec(key), &value); - } - - KeyValueStoreOp::DeleteKey(key) => { - let col = get_col_from_key(&key).unwrap_or("unknown".to_owned()); - metrics::inc_counter_vec(&metrics::DISK_DB_DELETE_COUNT, &[&col]); - - leveldb_batch.delete(BytesKey::from_vec(key)); - } - } - } - - let _timer = metrics::start_timer(&metrics::DISK_DB_WRITE_TIMES); - - self.db.write(self.write_options(), &leveldb_batch)?; - Ok(()) - } - - fn begin_rw_transaction(&self) -> MutexGuard<()> { - self.transaction_mutex.lock() - } - - fn compact_column(&self, column: DBColumn) -> Result<(), Error> { - // Use key-size-agnostic keys [] and 0xff..ff with a minimum of 32 bytes to account for - // columns that may change size between sub-databases or schema versions. - let start_key = BytesKey::from_vec(get_key_for_col(column.as_str(), &[])); - let end_key = BytesKey::from_vec(get_key_for_col( - column.as_str(), - &vec![0xff; std::cmp::max(column.key_size(), 32)], - )); - self.db.compact(&start_key, &end_key); - Ok(()) - } - - fn iter_column_from(&self, column: DBColumn, from: &[u8]) -> ColumnIter { - let start_key = BytesKey::from_vec(get_key_for_col(column.into(), from)); - let iter = self.db.iter(self.read_options()); - iter.seek(&start_key); - - Box::new( - iter.take_while(move |(key, _)| key.matches_column(column)) - .map(move |(bytes_key, value)| { - let key = bytes_key.remove_column_variable(column).ok_or_else(|| { - HotColdDBError::IterationError { - unexpected_key: bytes_key.clone(), - } - })?; - Ok((K::from_bytes(key)?, value)) - }), - ) - } - - fn iter_raw_entries(&self, column: DBColumn, prefix: &[u8]) -> RawEntryIter { - let start_key = BytesKey::from_vec(get_key_for_col(column.into(), prefix)); - - let iter = self.db.iter(self.read_options()); - iter.seek(&start_key); - - Box::new( - iter.take_while(move |(key, _)| key.key.starts_with(start_key.key.as_slice())) - .map(move |(bytes_key, value)| { - let subkey = &bytes_key.key[column.as_bytes().len()..]; - Ok((Vec::from(subkey), value)) - }), - ) - } - - fn iter_raw_keys(&self, column: DBColumn, prefix: &[u8]) -> RawKeyIter { - let start_key = BytesKey::from_vec(get_key_for_col(column.into(), prefix)); - - let iter = self.db.keys_iter(self.read_options()); - iter.seek(&start_key); - - Box::new( - iter.take_while(move |key| key.key.starts_with(start_key.key.as_slice())) - .map(move |bytes_key| { - let subkey = &bytes_key.key[column.as_bytes().len()..]; - Ok(Vec::from(subkey)) - }), - ) - } - - /// Iterate through all keys and values in a particular column. - fn iter_column_keys(&self, column: DBColumn) -> ColumnKeyIter { - let start_key = - BytesKey::from_vec(get_key_for_col(column.into(), &vec![0; column.key_size()])); - - let iter = self.db.keys_iter(self.read_options()); - iter.seek(&start_key); - - Box::new( - iter.take_while(move |key| key.matches_column(column)) - .map(move |bytes_key| { - let key = bytes_key.remove_column_variable(column).ok_or_else(|| { - HotColdDBError::IterationError { - unexpected_key: bytes_key.clone(), - } - })?; - K::from_bytes(key) - }), - ) - } -} - -impl ItemStore for LevelDB {} - -/// Used for keying leveldb. -#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] -pub struct BytesKey { - key: Vec, -} - -impl db_key::Key for BytesKey { - fn from_u8(key: &[u8]) -> Self { - Self { key: key.to_vec() } - } - - fn as_slice T>(&self, f: F) -> T { - f(self.key.as_slice()) - } -} - -impl BytesKey { - pub fn starts_with(&self, prefix: &Self) -> bool { - self.key.starts_with(&prefix.key) - } - - /// Return `true` iff this `BytesKey` was created with the given `column`. - pub fn matches_column(&self, column: DBColumn) -> bool { - self.key.starts_with(column.as_bytes()) - } - - /// Remove the column from a 32 byte key, yielding the `Hash256` key. - pub fn remove_column(&self, column: DBColumn) -> Option { - let key = self.remove_column_variable(column)?; - (column.key_size() == 32).then(|| Hash256::from_slice(key)) - } - - /// Remove the column from a key. - /// - /// Will return `None` if the value doesn't match the column or has the wrong length. - pub fn remove_column_variable(&self, column: DBColumn) -> Option<&[u8]> { - if self.matches_column(column) { - let subkey = &self.key[column.as_bytes().len()..]; - if subkey.len() == column.key_size() { - return Some(subkey); - } - } - None - } - - pub fn from_vec(key: Vec) -> Self { - Self { key } - } -} - -impl From for Error { - fn from(e: LevelDBError) -> Error { - Error::DBError { - message: format!("{:?}", e), - } - } -} diff --git a/beacon_node/store/src/lib.rs b/beacon_node/store/src/lib.rs index 1458fa846c6..0cfc42ab156 100644 --- a/beacon_node/store/src/lib.rs +++ b/beacon_node/store/src/lib.rs @@ -19,7 +19,6 @@ pub mod hdiff; pub mod historic_state_cache; pub mod hot_cold_store; mod impls; -mod leveldb_store; mod memory_store; pub mod metadata; pub mod metrics; @@ -27,13 +26,13 @@ pub mod partial_beacon_state; pub mod reconstruct; pub mod state_cache; +pub mod database; pub mod iter; pub use self::blob_sidecar_list_from_root::BlobSidecarListFromRoot; pub use self::config::StoreConfig; pub use self::consensus_context::OnDiskConsensusContext; pub use self::hot_cold_store::{HotColdDB, HotStateSummary, Split}; -pub use self::leveldb_store::LevelDB; pub use self::memory_store::MemoryStore; pub use crate::metadata::BlobInfo; pub use errors::Error; @@ -41,8 +40,9 @@ pub use impls::beacon_state::StorageContainer as BeaconStateStorageContainer; pub use metadata::AnchorInfo; pub use metrics::scrape_for_metrics; use parking_lot::MutexGuard; +use std::collections::HashSet; use std::sync::Arc; -use strum::{EnumString, IntoStaticStr}; +use strum::{EnumIter, EnumString, IntoStaticStr}; pub use types::*; const DATA_COLUMN_DB_KEY_SIZE: usize = 32 + 8; @@ -50,18 +50,18 @@ const DATA_COLUMN_DB_KEY_SIZE: usize = 32 + 8; pub type ColumnIter<'a, K> = Box), Error>> + 'a>; pub type ColumnKeyIter<'a, K> = Box> + 'a>; -pub type RawEntryIter<'a> = Box, Vec), Error>> + 'a>; -pub type RawKeyIter<'a> = Box, Error>> + 'a>; +pub type RawEntryIter<'a> = + Result, Vec), Error>> + 'a>, Error>; pub trait KeyValueStore: Sync + Send + Sized + 'static { /// Retrieve some bytes in `column` with `key`. - fn get_bytes(&self, column: &str, key: &[u8]) -> Result>, Error>; + fn get_bytes(&self, column: DBColumn, key: &[u8]) -> Result>, Error>; /// Store some `value` in `column`, indexed with `key`. - fn put_bytes(&self, column: &str, key: &[u8], value: &[u8]) -> Result<(), Error>; + fn put_bytes(&self, column: DBColumn, key: &[u8], value: &[u8]) -> Result<(), Error>; /// Same as put_bytes() but also force a flush to disk - fn put_bytes_sync(&self, column: &str, key: &[u8], value: &[u8]) -> Result<(), Error>; + fn put_bytes_sync(&self, column: DBColumn, key: &[u8], value: &[u8]) -> Result<(), Error>; /// Flush to disk. See /// https://chromium.googlesource.com/external/leveldb/+/HEAD/doc/index.md#synchronous-writes @@ -69,10 +69,10 @@ pub trait KeyValueStore: Sync + Send + Sized + 'static { fn sync(&self) -> Result<(), Error>; /// Return `true` if `key` exists in `column`. - fn key_exists(&self, column: &str, key: &[u8]) -> Result; + fn key_exists(&self, column: DBColumn, key: &[u8]) -> Result; /// Removes `key` from `column`. - fn key_delete(&self, column: &str, key: &[u8]) -> Result<(), Error>; + fn key_delete(&self, column: DBColumn, key: &[u8]) -> Result<(), Error>; /// Execute either all of the operations in `batch` or none at all, returning an error. fn do_atomically(&self, batch: Vec) -> Result<(), Error>; @@ -105,17 +105,21 @@ pub trait KeyValueStore: Sync + Send + Sized + 'static { self.iter_column_from(column, &vec![0; column.key_size()]) } - /// Iterate through all keys and values in a column from a given starting point. + /// Iterate through all keys and values in a column from a given starting point that fulfill the given predicate. fn iter_column_from(&self, column: DBColumn, from: &[u8]) -> ColumnIter; - fn iter_raw_entries(&self, _column: DBColumn, _prefix: &[u8]) -> RawEntryIter { - Box::new(std::iter::empty()) - } - - fn iter_raw_keys(&self, column: DBColumn, prefix: &[u8]) -> RawKeyIter; + fn iter_column_keys(&self, column: DBColumn) -> ColumnKeyIter; /// Iterate through all keys in a particular column. - fn iter_column_keys(&self, column: DBColumn) -> ColumnKeyIter; + fn iter_column_keys_from(&self, column: DBColumn, from: &[u8]) -> ColumnKeyIter; + + fn delete_batch(&self, column: DBColumn, ops: HashSet<&[u8]>) -> Result<(), Error>; + + fn delete_if( + &self, + column: DBColumn, + f: impl FnMut(&[u8]) -> Result, + ) -> Result<(), Error>; } pub trait Key: Sized + 'static { @@ -138,7 +142,7 @@ impl Key for Vec { } } -pub fn get_key_for_col(column: &str, key: &[u8]) -> Vec { +pub fn get_key_for_col(column: DBColumn, key: &[u8]) -> Vec { let mut result = column.as_bytes().to_vec(); result.extend_from_slice(key); result @@ -176,14 +180,18 @@ pub fn parse_data_column_key(data: Vec) -> Result<(Hash256, ColumnIndex), Er #[must_use] #[derive(Clone)] pub enum KeyValueStoreOp { - PutKeyValue(Vec, Vec), - DeleteKey(Vec), + // Indicate that a PUT operation should be made + // to the db store for a (Column, Key, Value) + PutKeyValue(DBColumn, Vec, Vec), + // Indicate that a DELETE operation should be made + // to the db store for a (Column, Key) + DeleteKey(DBColumn, Vec), } pub trait ItemStore: KeyValueStore + Sync + Send + Sized + 'static { /// Store an item in `Self`. fn put(&self, key: &Hash256, item: &I) -> Result<(), Error> { - let column = I::db_column().into(); + let column = I::db_column(); let key = key.as_slice(); self.put_bytes(column, key, &item.as_store_bytes()) @@ -191,7 +199,7 @@ pub trait ItemStore: KeyValueStore + Sync + Send + Sized + 'stati } fn put_sync(&self, key: &Hash256, item: &I) -> Result<(), Error> { - let column = I::db_column().into(); + let column = I::db_column(); let key = key.as_slice(); self.put_bytes_sync(column, key, &item.as_store_bytes()) @@ -200,7 +208,7 @@ pub trait ItemStore: KeyValueStore + Sync + Send + Sized + 'stati /// Retrieve an item from `Self`. fn get(&self, key: &Hash256) -> Result, Error> { - let column = I::db_column().into(); + let column = I::db_column(); let key = key.as_slice(); match self.get_bytes(column, key)? { @@ -211,7 +219,7 @@ pub trait ItemStore: KeyValueStore + Sync + Send + Sized + 'stati /// Returns `true` if the given key represents an item in `Self`. fn exists(&self, key: &Hash256) -> Result { - let column = I::db_column().into(); + let column = I::db_column(); let key = key.as_slice(); self.key_exists(column, key) @@ -219,7 +227,7 @@ pub trait ItemStore: KeyValueStore + Sync + Send + Sized + 'stati /// Remove an item from `Self`. fn delete(&self, key: &Hash256) -> Result<(), Error> { - let column = I::db_column().into(); + let column = I::db_column(); let key = key.as_slice(); self.key_delete(column, key) @@ -247,7 +255,7 @@ pub enum StoreOp<'a, E: EthSpec> { } /// A unique column identifier. -#[derive(Debug, Clone, Copy, PartialEq, IntoStaticStr, EnumString)] +#[derive(Debug, Clone, Copy, PartialEq, IntoStaticStr, EnumString, EnumIter)] pub enum DBColumn { /// For data related to the database itself. #[strum(serialize = "bma")] @@ -351,6 +359,9 @@ pub enum DBColumn { /// For helping persist eagerly computed light client bootstrap data #[strum(serialize = "scm")] SyncCommittee, + /// The dummy table is used to force the db to sync + #[strum(serialize = "dmy")] + Dummy, } /// A block from the database, which might have an execution payload or not. @@ -401,7 +412,8 @@ impl DBColumn { | Self::BeaconStateDiff | Self::SyncCommittee | Self::SyncCommitteeBranch - | Self::LightClientUpdate => 8, + | Self::LightClientUpdate + | Self::Dummy => 8, Self::BeaconDataColumn => DATA_COLUMN_DB_KEY_SIZE, } } @@ -421,13 +433,18 @@ pub trait StoreItem: Sized { fn from_store_bytes(bytes: &[u8]) -> Result; fn as_kv_store_op(&self, key: Hash256) -> KeyValueStoreOp { - let db_key = get_key_for_col(Self::db_column().into(), key.as_slice()); - KeyValueStoreOp::PutKeyValue(db_key, self.as_store_bytes()) + KeyValueStoreOp::PutKeyValue( + Self::db_column(), + key.as_slice().to_vec(), + self.as_store_bytes(), + ) } } #[cfg(test)] mod tests { + use crate::database::interface::BeaconNodeBackend; + use super::*; use ssz::{Decode, Encode}; use ssz_derive::{Decode, Encode}; @@ -477,7 +494,7 @@ mod tests { fn simplediskdb() { let dir = tempdir().unwrap(); let path = dir.path(); - let store = LevelDB::open(path).unwrap(); + let store = BeaconNodeBackend::open(&StoreConfig::default(), path).unwrap(); test_impl(store); } @@ -508,7 +525,7 @@ mod tests { #[test] fn test_get_col_from_key() { - let key = get_key_for_col(DBColumn::BeaconBlock.into(), &[1u8; 32]); + let key = get_key_for_col(DBColumn::BeaconBlock, &[1u8; 32]); let col = get_col_from_key(&key).unwrap(); assert_eq!(col, "blk"); } diff --git a/beacon_node/store/src/memory_store.rs b/beacon_node/store/src/memory_store.rs index 4c7bfdf10ff..6070a2d3f0c 100644 --- a/beacon_node/store/src/memory_store.rs +++ b/beacon_node/store/src/memory_store.rs @@ -1,9 +1,9 @@ use crate::{ - get_key_for_col, leveldb_store::BytesKey, ColumnIter, ColumnKeyIter, DBColumn, Error, - ItemStore, Key, KeyValueStore, KeyValueStoreOp, RawKeyIter, + errors::Error as DBError, get_key_for_col, hot_cold_store::BytesKey, ColumnIter, ColumnKeyIter, + DBColumn, Error, ItemStore, Key, KeyValueStore, KeyValueStoreOp, }; use parking_lot::{Mutex, MutexGuard, RwLock}; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashSet}; use std::marker::PhantomData; use types::*; @@ -29,19 +29,19 @@ impl MemoryStore { impl KeyValueStore for MemoryStore { /// Get the value of some key from the database. Returns `None` if the key does not exist. - fn get_bytes(&self, col: &str, key: &[u8]) -> Result>, Error> { + fn get_bytes(&self, col: DBColumn, key: &[u8]) -> Result>, Error> { let column_key = BytesKey::from_vec(get_key_for_col(col, key)); Ok(self.db.read().get(&column_key).cloned()) } /// Puts a key in the database. - fn put_bytes(&self, col: &str, key: &[u8], val: &[u8]) -> Result<(), Error> { + fn put_bytes(&self, col: DBColumn, key: &[u8], val: &[u8]) -> Result<(), Error> { let column_key = BytesKey::from_vec(get_key_for_col(col, key)); self.db.write().insert(column_key, val.to_vec()); Ok(()) } - fn put_bytes_sync(&self, col: &str, key: &[u8], val: &[u8]) -> Result<(), Error> { + fn put_bytes_sync(&self, col: DBColumn, key: &[u8], val: &[u8]) -> Result<(), Error> { self.put_bytes(col, key, val) } @@ -51,13 +51,13 @@ impl KeyValueStore for MemoryStore { } /// Return true if some key exists in some column. - fn key_exists(&self, col: &str, key: &[u8]) -> Result { + fn key_exists(&self, col: DBColumn, key: &[u8]) -> Result { let column_key = BytesKey::from_vec(get_key_for_col(col, key)); Ok(self.db.read().contains_key(&column_key)) } /// Delete some key from the database. - fn key_delete(&self, col: &str, key: &[u8]) -> Result<(), Error> { + fn key_delete(&self, col: DBColumn, key: &[u8]) -> Result<(), Error> { let column_key = BytesKey::from_vec(get_key_for_col(col, key)); self.db.write().remove(&column_key); Ok(()) @@ -66,12 +66,16 @@ impl KeyValueStore for MemoryStore { fn do_atomically(&self, batch: Vec) -> Result<(), Error> { for op in batch { match op { - KeyValueStoreOp::PutKeyValue(key, value) => { - self.db.write().insert(BytesKey::from_vec(key), value); + KeyValueStoreOp::PutKeyValue(col, key, value) => { + let column_key = get_key_for_col(col, &key); + self.db + .write() + .insert(BytesKey::from_vec(column_key), value); } - KeyValueStoreOp::DeleteKey(key) => { - self.db.write().remove(&BytesKey::from_vec(key)); + KeyValueStoreOp::DeleteKey(col, key) => { + let column_key = get_key_for_col(col, &key); + self.db.write().remove(&BytesKey::from_vec(column_key)); } } } @@ -82,8 +86,7 @@ impl KeyValueStore for MemoryStore { // We use this awkward pattern because we can't lock the `self.db` field *and* maintain a // reference to the lock guard across calls to `.next()`. This would be require a // struct with a field (the iterator) which references another field (the lock guard). - let start_key = BytesKey::from_vec(get_key_for_col(column.as_str(), from)); - let col = column.as_str(); + let start_key = BytesKey::from_vec(get_key_for_col(column, from)); let keys = self .db .read() @@ -92,7 +95,7 @@ impl KeyValueStore for MemoryStore { .filter_map(|(k, _)| k.remove_column_variable(column).map(|k| k.to_vec())) .collect::>(); Box::new(keys.into_iter().filter_map(move |key| { - self.get_bytes(col, &key).transpose().map(|res| { + self.get_bytes(column, &key).transpose().map(|res| { let k = K::from_bytes(&key)?; let v = res?; Ok((k, v)) @@ -100,18 +103,6 @@ impl KeyValueStore for MemoryStore { })) } - fn iter_raw_keys(&self, column: DBColumn, prefix: &[u8]) -> RawKeyIter { - let start_key = BytesKey::from_vec(get_key_for_col(column.as_str(), prefix)); - let keys = self - .db - .read() - .range(start_key.clone()..) - .take_while(|(k, _)| k.starts_with(&start_key)) - .filter_map(|(k, _)| k.remove_column_variable(column).map(|k| k.to_vec())) - .collect::>(); - Box::new(keys.into_iter().map(Ok)) - } - fn iter_column_keys(&self, column: DBColumn) -> ColumnKeyIter { Box::new(self.iter_column(column).map(|res| res.map(|(k, _)| k))) } @@ -123,6 +114,44 @@ impl KeyValueStore for MemoryStore { fn compact_column(&self, _column: DBColumn) -> Result<(), Error> { Ok(()) } + + fn iter_column_keys_from(&self, column: DBColumn, from: &[u8]) -> ColumnKeyIter { + // We use this awkward pattern because we can't lock the `self.db` field *and* maintain a + // reference to the lock guard across calls to `.next()`. This would be require a + // struct with a field (the iterator) which references another field (the lock guard). + let start_key = BytesKey::from_vec(get_key_for_col(column, from)); + let keys = self + .db + .read() + .range(start_key..) + .take_while(|(k, _)| k.remove_column_variable(column).is_some()) + .filter_map(|(k, _)| k.remove_column_variable(column).map(|k| k.to_vec())) + .collect::>(); + Box::new(keys.into_iter().map(move |key| K::from_bytes(&key))) + } + + fn delete_batch(&self, col: DBColumn, ops: HashSet<&[u8]>) -> Result<(), DBError> { + for op in ops { + let column_key = get_key_for_col(col, op); + self.db.write().remove(&BytesKey::from_vec(column_key)); + } + Ok(()) + } + + fn delete_if( + &self, + column: DBColumn, + mut f: impl FnMut(&[u8]) -> Result, + ) -> Result<(), Error> { + self.db.write().retain(|key, value| { + if key.remove_column_variable(column).is_some() { + !f(value).unwrap_or(false) + } else { + true + } + }); + Ok(()) + } } impl ItemStore for MemoryStore {} diff --git a/beacon_node/store/src/metrics.rs b/beacon_node/store/src/metrics.rs index f0dd0617905..6f9f667917f 100644 --- a/beacon_node/store/src/metrics.rs +++ b/beacon_node/store/src/metrics.rs @@ -33,6 +33,13 @@ pub static DISK_DB_READ_BYTES: LazyLock> = LazyLock::new(| &["col"], ) }); +pub static DISK_DB_KEY_READ_BYTES: LazyLock> = LazyLock::new(|| { + try_create_int_counter_vec( + "store_disk_db_key_read_bytes_total", + "Number of key bytes read from the hot on-disk DB", + &["col"], + ) +}); pub static DISK_DB_READ_COUNT: LazyLock> = LazyLock::new(|| { try_create_int_counter_vec( "store_disk_db_read_count_total", @@ -40,6 +47,13 @@ pub static DISK_DB_READ_COUNT: LazyLock> = LazyLock::new(| &["col"], ) }); +pub static DISK_DB_KEY_READ_COUNT: LazyLock> = LazyLock::new(|| { + try_create_int_counter_vec( + "store_disk_db_read_count_total", + "Total number of key reads to the hot on-disk DB", + &["col"], + ) +}); pub static DISK_DB_WRITE_COUNT: LazyLock> = LazyLock::new(|| { try_create_int_counter_vec( "store_disk_db_write_count_total", @@ -66,6 +80,12 @@ pub static DISK_DB_EXISTS_COUNT: LazyLock> = LazyLock::new &["col"], ) }); +pub static DISK_DB_DELETE_TIMES: LazyLock> = LazyLock::new(|| { + try_create_histogram( + "store_disk_db_delete_seconds", + "Time taken to delete bytes from the store.", + ) +}); pub static DISK_DB_DELETE_COUNT: LazyLock> = LazyLock::new(|| { try_create_int_counter_vec( "store_disk_db_delete_count_total", @@ -73,6 +93,19 @@ pub static DISK_DB_DELETE_COUNT: LazyLock> = LazyLock::new &["col"], ) }); +pub static DISK_DB_COMPACT_TIMES: LazyLock> = LazyLock::new(|| { + try_create_histogram( + "store_disk_db_compact_seconds", + "Time taken to run compaction on the DB.", + ) +}); +pub static DISK_DB_TYPE: LazyLock> = LazyLock::new(|| { + try_create_int_counter_vec( + "store_disk_db_type", + "The on-disk database type being used", + &["db_type"], + ) +}); /* * Anchor Info */ diff --git a/beacon_node/store/src/partial_beacon_state.rs b/beacon_node/store/src/partial_beacon_state.rs index 0b8bc2e0d4b..d2095121592 100644 --- a/beacon_node/store/src/partial_beacon_state.rs +++ b/beacon_node/store/src/partial_beacon_state.rs @@ -2,8 +2,8 @@ use crate::chunked_vector::{ load_variable_list_from_db, load_vector_from_db, BlockRootsChunked, HistoricalRoots, HistoricalSummaries, RandaoMixes, StateRootsChunked, }; -use crate::{Error, KeyValueStore}; -use ssz::{Decode, DecodeError}; +use crate::{DBColumn, Error, KeyValueStore, KeyValueStoreOp}; +use ssz::{Decode, DecodeError, Encode}; use ssz_derive::{Decode, Encode}; use std::sync::Arc; use types::historical_summary::HistoricalSummary; @@ -172,6 +172,15 @@ impl PartialBeaconState { )) } + /// Prepare the partial state for storage in the KV database. + pub fn as_kv_store_op(&self, state_root: Hash256) -> KeyValueStoreOp { + KeyValueStoreOp::PutKeyValue( + DBColumn::BeaconState, + state_root.as_slice().to_vec(), + self.as_ssz_bytes(), + ) + } + pub fn load_block_roots>( &mut self, store: &S, diff --git a/book/src/help_bn.md b/book/src/help_bn.md index a4ab44748c3..2d12010094c 100644 --- a/book/src/help_bn.md +++ b/book/src/help_bn.md @@ -11,6 +11,9 @@ Options: --auto-compact-db Enable or disable automatic compaction of the database on finalization. [default: true] + --beacon-node-backend + Set the database backend to be used by the beacon node. [possible + values: leveldb] --blob-prune-margin-epochs The margin for blob pruning in epochs. The oldest blobs are pruned up until data_availability_boundary - blob_prune_margin_epochs. [default: diff --git a/book/src/installation-source.md b/book/src/installation-source.md index 3c9f27d236a..19098a5bc8d 100644 --- a/book/src/installation-source.md +++ b/book/src/installation-source.md @@ -154,7 +154,7 @@ You can customise the features that Lighthouse is built with using the `FEATURES variable. E.g. ``` -FEATURES=gnosis,slasher-lmdb make +FEATURES=gnosis,slasher-lmdb,beacon-node-leveldb make ``` Commonly used features include: @@ -163,11 +163,12 @@ Commonly used features include: - `portable`: the default feature as Lighthouse now uses runtime detection of hardware CPU features. - `slasher-lmdb`: support for the LMDB slasher backend. Enabled by default. - `slasher-mdbx`: support for the MDBX slasher backend. +- `beacon-node-leveldb`: support for the leveldb backend. Enabled by default. - `jemalloc`: use [`jemalloc`][jemalloc] to allocate memory. Enabled by default on Linux and macOS. Not supported on Windows. - `spec-minimal`: support for the minimal preset (useful for testing). -Default features (e.g. `slasher-lmdb`) may be opted out of using the `--no-default-features` +Default features (e.g. `slasher-lmdb`, `beacon-node-leveldb`) may be opted out of using the `--no-default-features` argument for `cargo`, which can be plumbed in via the `CARGO_INSTALL_EXTRA_FLAGS` environment variable. E.g. diff --git a/database_manager/src/cli.rs b/database_manager/src/cli.rs index 4246a51f899..9db807df2ca 100644 --- a/database_manager/src/cli.rs +++ b/database_manager/src/cli.rs @@ -57,6 +57,15 @@ pub struct DatabaseManager { )] pub blobs_dir: Option, + #[clap( + long, + value_name = "DATABASE", + help = "Set the database backend to be used by the beacon node.", + display_order = 0, + default_value_t = store::config::DatabaseBackend::LevelDb + )] + pub backend: store::config::DatabaseBackend, + #[clap( long, global = true, diff --git a/database_manager/src/lib.rs b/database_manager/src/lib.rs index fc15e98616b..bed90df9df0 100644 --- a/database_manager/src/lib.rs +++ b/database_manager/src/lib.rs @@ -16,10 +16,12 @@ use slog::{info, warn, Logger}; use std::fs; use std::io::Write; use std::path::PathBuf; +use store::KeyValueStore; use store::{ + database::interface::BeaconNodeBackend, errors::Error, metadata::{SchemaVersion, CURRENT_SCHEMA_VERSION}, - DBColumn, HotColdDB, KeyValueStore, LevelDB, + DBColumn, HotColdDB, }; use strum::{EnumString, EnumVariantNames}; use types::{BeaconState, EthSpec, Slot}; @@ -40,7 +42,7 @@ fn parse_client_config( .clone_from(&database_manager_config.blobs_dir); client_config.store.blob_prune_margin_epochs = database_manager_config.blob_prune_margin_epochs; client_config.store.hierarchy_config = database_manager_config.hierarchy_exponents.clone(); - + client_config.store.backend = database_manager_config.backend; Ok(client_config) } @@ -55,7 +57,7 @@ pub fn display_db_version( let blobs_path = client_config.get_blobs_db_path(); let mut version = CURRENT_SCHEMA_VERSION; - HotColdDB::, LevelDB>::open( + HotColdDB::, BeaconNodeBackend>::open( &hot_path, &cold_path, &blobs_path, @@ -145,11 +147,14 @@ pub fn inspect_db( let mut num_keys = 0; let sub_db = if inspect_config.freezer { - LevelDB::::open(&cold_path).map_err(|e| format!("Unable to open freezer DB: {e:?}"))? + BeaconNodeBackend::::open(&client_config.store, &cold_path) + .map_err(|e| format!("Unable to open freezer DB: {e:?}"))? } else if inspect_config.blobs_db { - LevelDB::::open(&blobs_path).map_err(|e| format!("Unable to open blobs DB: {e:?}"))? + BeaconNodeBackend::::open(&client_config.store, &blobs_path) + .map_err(|e| format!("Unable to open blobs DB: {e:?}"))? } else { - LevelDB::::open(&hot_path).map_err(|e| format!("Unable to open hot DB: {e:?}"))? + BeaconNodeBackend::::open(&client_config.store, &hot_path) + .map_err(|e| format!("Unable to open hot DB: {e:?}"))? }; let skip = inspect_config.skip.unwrap_or(0); @@ -263,11 +268,20 @@ pub fn compact_db( let column = compact_config.column; let (sub_db, db_name) = if compact_config.freezer { - (LevelDB::::open(&cold_path)?, "freezer_db") + ( + BeaconNodeBackend::::open(&client_config.store, &cold_path)?, + "freezer_db", + ) } else if compact_config.blobs_db { - (LevelDB::::open(&blobs_path)?, "blobs_db") + ( + BeaconNodeBackend::::open(&client_config.store, &blobs_path)?, + "blobs_db", + ) } else { - (LevelDB::::open(&hot_path)?, "hot_db") + ( + BeaconNodeBackend::::open(&client_config.store, &hot_path)?, + "hot_db", + ) }; info!( log, @@ -303,7 +317,7 @@ pub fn migrate_db( let mut from = CURRENT_SCHEMA_VERSION; let to = migrate_config.to; - let db = HotColdDB::, LevelDB>::open( + let db = HotColdDB::, BeaconNodeBackend>::open( &hot_path, &cold_path, &blobs_path, @@ -343,7 +357,7 @@ pub fn prune_payloads( let cold_path = client_config.get_freezer_db_path(); let blobs_path = client_config.get_blobs_db_path(); - let db = HotColdDB::, LevelDB>::open( + let db = HotColdDB::, BeaconNodeBackend>::open( &hot_path, &cold_path, &blobs_path, @@ -369,7 +383,7 @@ pub fn prune_blobs( let cold_path = client_config.get_freezer_db_path(); let blobs_path = client_config.get_blobs_db_path(); - let db = HotColdDB::, LevelDB>::open( + let db = HotColdDB::, BeaconNodeBackend>::open( &hot_path, &cold_path, &blobs_path, @@ -406,7 +420,7 @@ pub fn prune_states( let cold_path = client_config.get_freezer_db_path(); let blobs_path = client_config.get_blobs_db_path(); - let db = HotColdDB::, LevelDB>::open( + let db = HotColdDB::, BeaconNodeBackend>::open( &hot_path, &cold_path, &blobs_path, diff --git a/lighthouse/Cargo.toml b/lighthouse/Cargo.toml index eda9a2ebf27..c3035113387 100644 --- a/lighthouse/Cargo.toml +++ b/lighthouse/Cargo.toml @@ -7,7 +7,7 @@ autotests = false rust-version = "1.80.0" [features] -default = ["slasher-lmdb"] +default = ["slasher-lmdb", "beacon-node-leveldb"] # Writes debugging .ssz files to /tmp during block processing. write_ssz_files = ["beacon_node/write_ssz_files"] # Compiles the BLS crypto code so that the binary is portable across machines. @@ -24,6 +24,11 @@ slasher-mdbx = ["slasher/mdbx"] slasher-lmdb = ["slasher/lmdb"] # Support slasher redb backend. slasher-redb = ["slasher/redb"] +# Supports beacon node leveldb backend. +beacon-node-leveldb = ["store/leveldb"] +# Supports beacon node redb backend. +beacon-node-redb = ["store/redb"] + # Deprecated. This is now enabled by default on non windows targets. jemalloc = [] @@ -56,6 +61,7 @@ serde_json = { workspace = true } serde_yaml = { workspace = true } slasher = { workspace = true } slog = { workspace = true } +store = { workspace = true } task_executor = { workspace = true } types = { workspace = true } unused_port = { workspace = true } diff --git a/lighthouse/tests/beacon_node.rs b/lighthouse/tests/beacon_node.rs index 88e05dfa12d..1063a80ff40 100644 --- a/lighthouse/tests/beacon_node.rs +++ b/lighthouse/tests/beacon_node.rs @@ -1,11 +1,12 @@ -use beacon_node::ClientConfig as Config; - use crate::exec::{CommandLineTestExec, CompletedTest}; use beacon_node::beacon_chain::chain_config::{ DisallowedReOrgOffsets, DEFAULT_RE_ORG_CUTOFF_DENOMINATOR, DEFAULT_RE_ORG_HEAD_THRESHOLD, DEFAULT_RE_ORG_MAX_EPOCHS_SINCE_FINALIZATION, }; -use beacon_node::beacon_chain::graffiti_calculator::GraffitiOrigin; +use beacon_node::{ + beacon_chain::graffiti_calculator::GraffitiOrigin, + beacon_chain::store::config::DatabaseBackend as BeaconNodeBackend, ClientConfig as Config, +}; use beacon_processor::BeaconProcessorConfig; use eth1::Eth1Endpoint; use lighthouse_network::PeerId; @@ -2691,3 +2692,13 @@ fn genesis_state_url_value() { assert_eq!(config.genesis_state_url_timeout, Duration::from_secs(42)); }); } + +#[test] +fn beacon_node_backend_override() { + CommandLineTest::new() + .flag("beacon-node-backend", Some("leveldb")) + .run_with_zero_port() + .with_config(|config| { + assert_eq!(config.store.backend, BeaconNodeBackend::LevelDb); + }); +} diff --git a/wordlist.txt b/wordlist.txt index 6287366cbcb..bb8b46b525e 100644 --- a/wordlist.txt +++ b/wordlist.txt @@ -162,6 +162,7 @@ keypair keypairs keystore keystores +leveldb linter linux localhost @@ -191,6 +192,7 @@ pre pubkey pubkeys rc +redb reimport resync roadmap