cerc-io/lighthouse
16
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

Use the database to persist the pubkey cache (#2234)

Browse Source 
## Issue Addressed

Closes #1787

## Proposed Changes

* Abstract the `ValidatorPubkeyCache` over a "backing" which is either a file (legacy), or the database.
* Implement a migration from schema v2 to schema v3, whereby the contents of the cache file are copied to the DB, and then the file is deleted. The next release to include this change must be a minor version bump, and we will need to warn users of the inability to downgrade (this is our first DB schema change since mainnet genesis).
* Move the schema migration code from the `store` crate into the `beacon_chain` crate so that it can access the datadir and the `ValidatorPubkeyCache`, etc. It gets injected back into the `store` via a closure (similar to what we do in fork choice).
This commit is contained in:
Michael Sproul 2021-03-04 01:25:12 +00:00
parent c370100719
commit 363f15f362
17 changed files with 262 additions and 146 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
beacon_node/beacon_chain/src/beacon_chain.rs
View File

@ -175,6 +175,14 @@ pub type BeaconForkChoice<T> = ForkChoice<
<T as BeaconChainTypes>::EthSpec,
>;
pub type BeaconStore<T> = Arc<
HotColdDB<
<T as BeaconChainTypes>::EthSpec,
<T as BeaconChainTypes>::HotStore,
<T as BeaconChainTypes>::ColdStore,
>,
>;
/// Represents the "Beacon Chain" component of Ethereum 2.0. Allows import of blocks and block
/// operations and chooses a canonical head.
pub struct BeaconChain<T: BeaconChainTypes> {
@ -182,7 +190,7 @@ pub struct BeaconChain<T: BeaconChainTypes> {
/// Configuration for `BeaconChain` runtime behaviour.
pub config: ChainConfig,
/// Persistent storage for blocks, states, etc. Typically an on-disk store, such as LevelDB.
pub store: Arc<HotColdDB<T::EthSpec, T::HotStore, T::ColdStore>>,
pub store: BeaconStore<T>,
/// Database migrator for running background maintenance on the store.
pub store_migrator: BackgroundMigrator<T::EthSpec, T::HotStore, T::ColdStore>,
/// Reports the current slot, typically based upon the system clock.
@ -237,7 +245,7 @@ pub struct BeaconChain<T: BeaconChainTypes> {
/// Caches the beacon block proposer shuffling for a given epoch and shuffling key root.
pub(crate) beacon_proposer_cache: Mutex<BeaconProposerCache>,
/// Caches a map of `validator_index -> validator_pubkey`.
pub(crate) validator_pubkey_cache: TimeoutRwLock<ValidatorPubkeyCache>,
pub(crate) validator_pubkey_cache: TimeoutRwLock<ValidatorPubkeyCache<T>>,
/// A list of any hard-coded forks that have been disabled.
pub disabled_forks: Vec<String>,
/// Sender given to tasks, so that if they encounter a state in which execution cannot
@ -300,9 +308,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
}
/// Load fork choice from disk, returning `None` if it isn't found.
pub fn load_fork_choice(
store: Arc<HotColdDB<T::EthSpec, T::HotStore, T::ColdStore>>,
) -> Result<Option<BeaconForkChoice<T>>, Error> {
pub fn load_fork_choice(store: BeaconStore<T>) -> Result<Option<BeaconForkChoice<T>>, Error> {
let persisted_fork_choice =
match store.get_item::<PersistedForkChoice>(&FORK_CHOICE_DB_KEY)? {
Some(fc) => fc,

10
beacon_node/beacon_chain/src/block_verification.rs
View File

@ -1336,7 +1336,7 @@ fn cheap_state_advance_to_obtain_committees<'a, E: EthSpec>(
/// Obtains a read-locked `ValidatorPubkeyCache` from the `chain`.
fn get_validator_pubkey_cache<T: BeaconChainTypes>(
chain: &BeaconChain<T>,
) -> Result<RwLockReadGuard<ValidatorPubkeyCache>, BlockError<T::EthSpec>> {
) -> Result<RwLockReadGuard<ValidatorPubkeyCache<T>>, BlockError<T::EthSpec>> {
chain
.validator_pubkey_cache
.try_read_for(VALIDATOR_PUBKEY_CACHE_LOCK_TIMEOUT)
@ -1348,11 +1348,11 @@ fn get_validator_pubkey_cache<T: BeaconChainTypes>(
///
/// The signature verifier is empty because it does not yet have any of this block's signatures
/// added to it. Use `Self::apply_to_signature_verifier` to apply the signatures.
fn get_signature_verifier<'a, E: EthSpec>(
state: &'a BeaconState<E>,
validator_pubkey_cache: &'a ValidatorPubkeyCache,
fn get_signature_verifier<'a, T: BeaconChainTypes>(
state: &'a BeaconState<T::EthSpec>,
validator_pubkey_cache: &'a ValidatorPubkeyCache<T>,
spec: &'a ChainSpec,
) -> BlockSignatureVerifier<'a, E, impl Fn(usize) -> Option<Cow<'a, PublicKey>> + Clone> {
) -> BlockSignatureVerifier<'a, T::EthSpec, impl Fn(usize) -> Option<Cow<'a, PublicKey>> + Clone> {
BlockSignatureVerifier::new(
state,
move |validator_index| {

20
beacon_node/beacon_chain/src/builder.rs
View File

@ -31,8 +31,6 @@ use types::{
SignedBeaconBlock, Slot,
};
pub const PUBKEY_CACHE_FILENAME: &str = "pubkey_cache.ssz";
/// An empty struct used to "witness" all the `BeaconChainTypes` traits. It has no user-facing
/// functionality and only exists to satisfy the type system.
pub struct Witness<TSlotClock, TEth1Backend, TEthSpec, THotStore, TColdStore>(
@ -81,8 +79,7 @@ pub struct BeaconChainBuilder<T: BeaconChainTypes> {
shutdown_sender: Option<Sender<&'static str>>,
head_tracker: Option<HeadTracker>,
data_dir: Option<PathBuf>,
pubkey_cache_path: Option<PathBuf>,
validator_pubkey_cache: Option<ValidatorPubkeyCache>,
validator_pubkey_cache: Option<ValidatorPubkeyCache<T>>,
spec: ChainSpec,
chain_config: ChainConfig,
disabled_forks: Vec<String>,
@ -119,7 +116,6 @@ where
slot_clock: None,
shutdown_sender: None,
head_tracker: None,
pubkey_cache_path: None,
data_dir: None,
disabled_forks: Vec::new(),
validator_pubkey_cache: None,
@ -182,7 +178,6 @@ where
///
/// Should generally be called early in the build chain.
pub fn data_dir(mut self, path: PathBuf) -> Self {
self.pubkey_cache_path = Some(path.join(PUBKEY_CACHE_FILENAME));
self.data_dir = Some(path);
self
}
@ -224,11 +219,6 @@ where
pub fn resume_from_db(mut self) -> Result<Self, String> {
let log = self.log.as_ref().ok_or("resume_from_db requires a log")?;
let pubkey_cache_path = self
.pubkey_cache_path
.as_ref()
.ok_or("resume_from_db requires a data_dir")?;
info!(
log,
"Starting beacon chain";
@ -274,7 +264,7 @@ where
.unwrap_or_else(OperationPool::new),
);
let pubkey_cache = ValidatorPubkeyCache::load_from_file(pubkey_cache_path)
let pubkey_cache = ValidatorPubkeyCache::load_from_store(store)
.map_err(|e| format!("Unable to open persisted pubkey cache: {:?}", e))?;
self.genesis_block_root = Some(chain.genesis_block_root);
@ -496,12 +486,8 @@ where
}
}
let pubkey_cache_path = self
.pubkey_cache_path
.ok_or("Cannot build without a pubkey cache path")?;
let validator_pubkey_cache = self.validator_pubkey_cache.map(Ok).unwrap_or_else(|| {
ValidatorPubkeyCache::new(&canonical_head.beacon_state, pubkey_cache_path)
ValidatorPubkeyCache::new(&canonical_head.beacon_state, store.clone())
.map_err(|e| format!("Unable to init validator pubkey cache: {:?}", e))
})?;

3
beacon_node/beacon_chain/src/lib.rs
View File

@ -20,6 +20,7 @@ mod observed_block_producers;
pub mod observed_operations;
mod persisted_beacon_chain;
mod persisted_fork_choice;
pub mod schema_change;
mod shuffling_cache;
mod snapshot_cache;
pub mod state_advance_timer;
@ -29,7 +30,7 @@ pub mod validator_monitor;
mod validator_pubkey_cache;
pub use self::beacon_chain::{
AttestationProcessingOutcome, BeaconChain, BeaconChainTypes, ChainSegmentResult,
AttestationProcessingOutcome, BeaconChain, BeaconChainTypes, BeaconStore, ChainSegmentResult,
ForkChoiceError, StateSkipConfig, MAXIMUM_GOSSIP_CLOCK_DISPARITY,
};
pub use self::beacon_snapshot::BeaconSnapshot;

64
beacon_node/beacon_chain/src/schema_change.rs Normal file
View File

@ -0,0 +1,64 @@
//! Utilities for managing database schema changes.
use crate::beacon_chain::BeaconChainTypes;
use crate::validator_pubkey_cache::ValidatorPubkeyCache;
use std::fs;
use std::path::Path;
use std::sync::Arc;
use store::hot_cold_store::{HotColdDB, HotColdDBError};
use store::metadata::{SchemaVersion, CURRENT_SCHEMA_VERSION};
use store::Error as StoreError;
const PUBKEY_CACHE_FILENAME: &str = "pubkey_cache.ssz";
/// Migrate the database from one schema version to another, applying all requisite mutations.
pub fn migrate_schema<T: BeaconChainTypes>(
db: Arc<HotColdDB<T::EthSpec, T::HotStore, T::ColdStore>>,
datadir: &Path,
from: SchemaVersion,
to: SchemaVersion,
) -> Result<(), StoreError> {
match (from, to) {
// Migrating from the current schema version to iself is always OK, a no-op.
(_, _) if from == to && to == CURRENT_SCHEMA_VERSION => Ok(()),
// Migrate across multiple versions by recursively migrating one step at a time.
(_, _) if from.as_u64() + 1 < to.as_u64() => {
let next = SchemaVersion(from.as_u64() + 1);
migrate_schema::<T>(db.clone(), datadir, from, next)?;
migrate_schema::<T>(db, datadir, next, to)
}
// Migration from v0.3.0 to v0.3.x, adding the temporary states column.
// Nothing actually needs to be done, but once a DB uses v2 it shouldn't go back.
(SchemaVersion(1), SchemaVersion(2)) => {
db.store_schema_version(to)?;
Ok(())
}
// Migration for removing the pubkey cache.
(SchemaVersion(2), SchemaVersion(3)) => {
let pk_cache_path = datadir.join(PUBKEY_CACHE_FILENAME);
// Load from file, store to DB.
ValidatorPubkeyCache::<T>::load_from_file(&pk_cache_path)
.and_then(|cache| ValidatorPubkeyCache::convert(cache, db.clone()))
.map_err(|e| StoreError::SchemaMigrationError(format!("{:?}", e)))?;
db.store_schema_version(to)?;
// Delete cache file now that keys are stored in the DB.
fs::remove_file(&pk_cache_path).map_err(|e| {
StoreError::SchemaMigrationError(format!(
"unable to delete {}: {:?}",
pk_cache_path.display(),
e
))
})?;
Ok(())
}
// Anything else is an error.
(_, _) => Err(HotColdDBError::UnsupportedSchemaVersion {
target_version: to,
current_version: from,
}
.into()),
}
}

190
beacon_node/beacon_chain/src/validator_pubkey_cache.rs
View File

@ -1,11 +1,13 @@
use crate::errors::BeaconChainError;
use crate::{BeaconChainTypes, BeaconStore};
use ssz::{Decode, DecodeError, Encode};
use std::collections::HashMap;
use std::convert::TryInto;
use std::fs::{File, OpenOptions};
use std::io::{self, Read, Write};
use std::path::Path;
use types::{BeaconState, EthSpec, PublicKey, PublicKeyBytes, Validator};
use store::{DBColumn, Error as StoreError, StoreItem};
use types::{BeaconState, Hash256, PublicKey, PublicKeyBytes};
/// Provides a mapping of `validator_index -> validator_publickey`.
///
@ -16,33 +18,35 @@ use types::{BeaconState, EthSpec, PublicKey, PublicKeyBytes, Validator};
/// keys in compressed form and they are needed in decompressed form for signature verification.
/// Decompression is expensive when many keys are involved.
///
/// The cache has a `persistence_file` that it uses to maintain a persistent, on-disk
/// The cache has a `backing` that it uses to maintain a persistent, on-disk
/// copy of itself. This allows it to be restored between process invocations.
pub struct ValidatorPubkeyCache {
pub struct ValidatorPubkeyCache<T: BeaconChainTypes> {
pubkeys: Vec<PublicKey>,
indices: HashMap<PublicKeyBytes, usize>,
persitence_file: ValidatorPubkeyCacheFile,
backing: PubkeyCacheBacking<T>,
}
impl ValidatorPubkeyCache {
pub fn load_from_file<P: AsRef<Path>>(path: P) -> Result<Self, BeaconChainError> {
ValidatorPubkeyCacheFile::open(&path)
.and_then(ValidatorPubkeyCacheFile::into_cache)
.map_err(Into::into)
}
/// Abstraction over on-disk backing.
///
/// `File` backing is legacy, `Database` is current.
enum PubkeyCacheBacking<T: BeaconChainTypes> {
File(ValidatorPubkeyCacheFile),
Database(BeaconStore<T>),
}
impl<T: BeaconChainTypes> ValidatorPubkeyCache<T> {
/// Create a new public key cache using the keys in `state.validators`.
///
/// Also creates a new persistence file, returning an error if there is already a file at
/// `persistence_path`.
pub fn new<T: EthSpec, P: AsRef<Path>>(
state: &BeaconState<T>,
persistence_path: P,
pub fn new(
state: &BeaconState<T::EthSpec>,
store: BeaconStore<T>,
) -> Result<Self, BeaconChainError> {
let mut cache = Self {
persitence_file: ValidatorPubkeyCacheFile::create(persistence_path)?,
pubkeys: vec![],
indices: HashMap::new(),
backing: PubkeyCacheBacking::Database(store),
};
cache.import_new_pubkeys(state)?;
@ -50,33 +54,83 @@ impl ValidatorPubkeyCache {
Ok(cache)
}
/// Load the pubkey cache from the given on-disk database.
pub fn load_from_store(store: BeaconStore<T>) -> Result<Self, BeaconChainError> {
let mut pubkeys = vec![];
let mut indices = HashMap::new();
for validator_index in 0.. {
if let Some(DatabasePubkey(pubkey)) =
store.get_item(&DatabasePubkey::key_for_index(validator_index))?
{
pubkeys.push((&pubkey).try_into().map_err(Error::PubkeyDecode)?);
indices.insert(pubkey, validator_index);
} else {
break;
}
}
Ok(ValidatorPubkeyCache {
pubkeys,
indices,
backing: PubkeyCacheBacking::Database(store),
})
}
/// DEPRECATED: used only for migration
pub fn load_from_file<P: AsRef<Path>>(path: P) -> Result<Self, BeaconChainError> {
ValidatorPubkeyCacheFile::open(&path)
.and_then(ValidatorPubkeyCacheFile::into_cache)
.map_err(Into::into)
}
/// Convert a cache using `File` backing to one using `Database` backing.
///
/// This will write all of the keys from `existing_cache` to `store`.
pub fn convert(existing_cache: Self, store: BeaconStore<T>) -> Result<Self, BeaconChainError> {
let mut result = ValidatorPubkeyCache {
pubkeys: Vec::with_capacity(existing_cache.pubkeys.len()),
indices: HashMap::with_capacity(existing_cache.indices.len()),
backing: PubkeyCacheBacking::Database(store),
};
result.import(existing_cache.pubkeys.iter().map(PublicKeyBytes::from))?;
Ok(result)
}
/// Scan the given `state` and add any new validator public keys.
///
/// Does not delete any keys from `self` if they don't appear in `state`.
pub fn import_new_pubkeys<T: EthSpec>(
pub fn import_new_pubkeys(
&mut self,
state: &BeaconState<T>,
state: &BeaconState<T::EthSpec>,
) -> Result<(), BeaconChainError> {
if state.validators.len() > self.pubkeys.len() {
self.import(&state.validators[self.pubkeys.len()..])
self.import(
state.validators[self.pubkeys.len()..]
.iter()
.map(|v| v.pubkey),
)
} else {
Ok(())
}
}
/// Adds zero or more validators to `self`.
fn import(&mut self, validators: &[Validator]) -> Result<(), BeaconChainError> {
self.pubkeys.reserve(validators.len());
self.indices.reserve(validators.len());
fn import<I>(&mut self, validator_keys: I) -> Result<(), BeaconChainError>
where
I: Iterator<Item = PublicKeyBytes> + ExactSizeIterator,
{
self.pubkeys.reserve(validator_keys.len());
self.indices.reserve(validator_keys.len());
for v in validators.iter() {
for pubkey in validator_keys {
let i = self.pubkeys.len();
if self.indices.contains_key(&v.pubkey) {
if self.indices.contains_key(&pubkey) {
return Err(BeaconChainError::DuplicateValidatorPublicKey);
}
// The item is written to disk (the persistence file) _before_ it is written into
// The item is written to disk _before_ it is written into
// the local struct.
//
// This means that a pubkey cache read from disk will always be equivalent to or
@ -85,15 +139,22 @@ impl ValidatorPubkeyCache {
// The motivation behind this ordering is that we do not want to have states that
// reference a pubkey that is not in our cache. However, it's fine to have pubkeys
// that are never referenced in a state.
self.persitence_file.append(i, &v.pubkey)?;
match &mut self.backing {
PubkeyCacheBacking::File(persistence_file) => {
persistence_file.append(i, &pubkey)?;
}
PubkeyCacheBacking::Database(store) => {
store.put_item(&DatabasePubkey::key_for_index(i), &DatabasePubkey(pubkey))?;
}
}
self.pubkeys.push(
(&v.pubkey)
(&pubkey)
.try_into()
.map_err(BeaconChainError::InvalidValidatorPubkeyBytes)?,
);
self.indices.insert(v.pubkey, i);
self.indices.insert(pubkey, i);
}
Ok(())
@ -115,6 +176,31 @@ impl ValidatorPubkeyCache {
}
}
/// Wrapper for a public key stored in the database.
///
/// Keyed by the validator index as `Hash256::from_low_u64_be(index)`.
struct DatabasePubkey(PublicKeyBytes);
impl StoreItem for DatabasePubkey {
fn db_column() -> DBColumn {
DBColumn::PubkeyCache
}
fn as_store_bytes(&self) -> Vec<u8> {
self.0.as_ssz_bytes()
}
fn from_store_bytes(bytes: &[u8]) -> Result<Self, StoreError> {
Ok(Self(PublicKeyBytes::from_ssz_bytes(bytes)?))
}
}
impl DatabasePubkey {
fn key_for_index(index: usize) -> Hash256 {
Hash256::from_low_u64_be(index as u64)
}
}
/// Allows for maintaining an on-disk copy of the `ValidatorPubkeyCache`. The file is raw SSZ bytes
/// (not ASCII encoded).
///
@ -149,17 +235,6 @@ impl From<Error> for BeaconChainError {
}
impl ValidatorPubkeyCacheFile {
/// Creates a file for reading and writing.
pub fn create<P: AsRef<Path>>(path: P) -> Result<Self, Error> {
OpenOptions::new()
.create(true)
.write(true)
.truncate(true)
.open(path)
.map(Self)
.map_err(Error::Io)
}
/// Opens an existing file for reading and writing.
pub fn open<P: AsRef<Path>>(path: P) -> Result<Self, Error> {
OpenOptions::new()
@ -181,7 +256,7 @@ impl ValidatorPubkeyCacheFile {
}
/// Creates a `ValidatorPubkeyCache` by reading and parsing the underlying file.
pub fn into_cache(mut self) -> Result<ValidatorPubkeyCache, Error> {
pub fn into_cache<T: BeaconChainTypes>(mut self) -> Result<ValidatorPubkeyCache<T>, Error> {
let mut bytes = vec![];
self.0.read_to_end(&mut bytes).map_err(Error::Io)?;
@ -208,7 +283,7 @@ impl ValidatorPubkeyCacheFile {
Ok(ValidatorPubkeyCache {
pubkeys,
indices,
persitence_file: self,
backing: PubkeyCacheBacking::File(self),
})
}
}
@ -225,21 +300,33 @@ fn append_to_file(file: &mut File, index: usize, pubkey: &PublicKeyBytes) -> Res
#[cfg(test)]
mod test {
use super::*;
use crate::test_utils::{test_logger, EphemeralHarnessType};
use std::sync::Arc;
use store::HotColdDB;
use tempfile::tempdir;
use types::{
test_utils::{generate_deterministic_keypair, TestingBeaconStateBuilder},
BeaconState, EthSpec, Keypair, MainnetEthSpec,
};
fn get_state(validator_count: usize) -> (BeaconState<MainnetEthSpec>, Vec<Keypair>) {
let spec = MainnetEthSpec::default_spec();
type E = MainnetEthSpec;
type T = EphemeralHarnessType<E>;
fn get_state(validator_count: usize) -> (BeaconState<E>, Vec<Keypair>) {
let spec = E::default_spec();
let builder =
TestingBeaconStateBuilder::from_deterministic_keypairs(validator_count, &spec);
builder.build()
}
fn get_store() -> BeaconStore<T> {
Arc::new(
HotColdDB::open_ephemeral(<_>::default(), E::default_spec(), test_logger()).unwrap(),
)
}
#[allow(clippy::needless_range_loop)]
fn check_cache_get(cache: &ValidatorPubkeyCache, keypairs: &[Keypair]) {
fn check_cache_get(cache: &ValidatorPubkeyCache<T>, keypairs: &[Keypair]) {
let validator_count = keypairs.len();
for i in 0..validator_count + 1 {
@ -270,10 +357,9 @@ mod test {
fn basic_operation() {
let (state, keypairs) = get_state(8);
let dir = tempdir().expect("should create tempdir");
let path = dir.path().join("cache.ssz");
let store = get_store();
let mut cache = ValidatorPubkeyCache::new(&state, path).expect("should create cache");
let mut cache = ValidatorPubkeyCache::new(&state, store).expect("should create cache");
check_cache_get(&cache, &keypairs[..]);
@ -303,16 +389,16 @@ mod test {
fn persistence() {
let (state, keypairs) = get_state(8);
let dir = tempdir().expect("should create tempdir");
let path = dir.path().join("cache.ssz");
let store = get_store();
// Create a new cache.
let cache = ValidatorPubkeyCache::new(&state, &path).expect("should create cache");
let cache = ValidatorPubkeyCache::new(&state, store.clone()).expect("should create cache");
check_cache_get(&cache, &keypairs[..]);
drop(cache);
// Re-init the cache from the file.
let mut cache = ValidatorPubkeyCache::load_from_file(&path).expect("should open cache");
let mut cache =
ValidatorPubkeyCache::load_from_store(store.clone()).expect("should open cache");
check_cache_get(&cache, &keypairs[..]);
// Add some more keypairs.
@ -324,7 +410,7 @@ mod test {
drop(cache);
// Re-init the cache from the file.
let cache = ValidatorPubkeyCache::load_from_file(&path).expect("should open cache");
let cache = ValidatorPubkeyCache::load_from_store(store).expect("should open cache");
check_cache_get(&cache, &keypairs[..]);
}
@ -338,7 +424,7 @@ mod test {
append_to_file(&mut file, 0, &pubkey).expect("should write to file");
drop(file);
let cache = ValidatorPubkeyCache::load_from_file(&path).expect("should open cache");
let cache = ValidatorPubkeyCache::<T>::load_from_file(&path).expect("should open cache");
drop(cache);
let mut file = OpenOptions::new()
@ -351,7 +437,7 @@ mod test {
drop(file);
assert!(
ValidatorPubkeyCache::load_from_file(&path).is_err(),
ValidatorPubkeyCache::<T>::load_from_file(&path).is_err(),
"should not parse invalid file"
);
}

6
beacon_node/beacon_chain/tests/op_verification.rs
View File

@ -37,10 +37,8 @@ fn get_store(db_path: &TempDir) -> Arc<HotColdDB> {
let cold_path = db_path.path().join("cold_db");
let config = StoreConfig::default();
let log = NullLoggerBuilder.build().expect("logger should build");
Arc::new(
HotColdDB::open(&hot_path, &cold_path, config, spec, log)
.expect("disk store should initialize"),
)
HotColdDB::open(&hot_path, &cold_path, |_, _, _| Ok(()), config, spec, log)
.expect("disk store should initialize")
}
fn get_harness(store: Arc<HotColdDB>, validator_count: usize) -> TestHarness {

6
beacon_node/beacon_chain/tests/persistence_tests.rs
View File

@ -29,10 +29,8 @@ fn get_store(db_path: &TempDir) -> Arc<HotColdDB<E, LevelDB<E>, LevelDB<E>>> {
let cold_path = db_path.path().join("cold_db");
let config = StoreConfig::default();
let log = NullLoggerBuilder.build().expect("logger should build");
Arc::new(
HotColdDB::open(&hot_path, &cold_path, config, spec, log)
.expect("disk store should initialize"),
)
HotColdDB::open(&hot_path, &cold_path, |_, _, _| Ok(()), config, spec, log)
.expect("disk store should initialize")
}
#[test]

6
beacon_node/beacon_chain/tests/store_tests.rs
View File

@ -40,10 +40,8 @@ fn get_store(db_path: &TempDir) -> Arc<HotColdDB<E, LevelDB<E>, LevelDB<E>>> {
let config = StoreConfig::default();
let log = test_logger();
Arc::new(
HotColdDB::open(&hot_path, &cold_path, config, spec, log)
.expect("disk store should initialize"),
)
HotColdDB::open(&hot_path, &cold_path, |_, _, _| Ok(()), config, spec, log)
.expect("disk store should initialize")
}
fn get_harness(

19
beacon_node/client/src/builder.rs
View File

@ -1,6 +1,7 @@
use crate::config::{ClientGenesis, Config as ClientConfig};
use crate::notifier::spawn_notifier;
use crate::Client;
use beacon_chain::schema_change::migrate_schema;
use beacon_chain::{
builder::{BeaconChainBuilder, Witness},
eth1_chain::{CachingEth1Backend, Eth1Chain},
@ -544,6 +545,7 @@ where
/// Specifies that the `Client` should use a `HotColdDB` database.
pub fn disk_store(
mut self,
datadir: &Path,
hot_path: &Path,
cold_path: &Path,
config: StoreConfig,
@ -561,9 +563,20 @@ where
self.db_path = Some(hot_path.into());
self.freezer_db_path = Some(cold_path.into());
let store = HotColdDB::open(hot_path, cold_path, config, spec, context.log().clone())
.map_err(|e| format!("Unable to open database: {:?}", e))?;
self.store = Some(Arc::new(store));
let schema_upgrade = |db, from, to| {
migrate_schema::<Witness<TSlotClock, TEth1Backend, _, _, _>>(db, datadir, from, to)
};
let store = HotColdDB::open(
hot_path,
cold_path,
schema_upgrade,
config,
spec,
context.log().clone(),
)
.map_err(|e| format!("Unable to open database: {:?}", e))?;
self.store = Some(store);
Ok(self)
}
}

8
beacon_node/src/config.rs
View File

@ -1,4 +1,3 @@
use beacon_chain::builder::PUBKEY_CACHE_FILENAME;
use clap::ArgMatches;
use clap_utils::BAD_TESTNET_DIR_MESSAGE;
use client::{ClientConfig, ClientGenesis};
@ -45,13 +44,6 @@ pub fn get_config<E: EthSpec>(
.ok_or("Failed to get freezer db path")?,
)
.map_err(|err| format!("Failed to remove chain_db: {}", err))?;
// Remove the pubkey cache file if it exists
let pubkey_cache_file = client_config.data_dir.join(PUBKEY_CACHE_FILENAME);
if pubkey_cache_file.exists() {
fs::remove_file(&pubkey_cache_file)
.map_err(|e| format!("Failed to remove {:?}: {:?}", pubkey_cache_file, e))?;
}
}
// Create `datadir` and any non-existing parent directories.

6
beacon_node/src/lib.rs
View File

@ -58,10 +58,10 @@ impl<E: EthSpec> ProductionBeaconNode<E> {
mut client_config: ClientConfig,
) -> Result<Self, String> {
let spec = context.eth2_config().spec.clone();
let client_config_1 = client_config.clone();
let client_genesis = client_config.genesis.clone();
let store_config = client_config.store.clone();
let log = context.log().clone();
let datadir = client_config.create_data_dir()?;
let db_path = client_config.create_db_path()?;
let freezer_db_path_res = client_config.create_freezer_db_path();
let executor = context.executor.clone();
@ -70,7 +70,7 @@ impl<E: EthSpec> ProductionBeaconNode<E> {
.runtime_context(context)
.chain_spec(spec)
.http_api_config(client_config.http_api.clone())
.disk_store(&db_path, &freezer_db_path_res?, store_config)?;
.disk_store(&datadir, &db_path, &freezer_db_path_res?, store_config)?;
let builder = if let Some(slasher_config) = client_config.slasher.clone() {
let slasher = Arc::new(
@ -83,7 +83,7 @@ impl<E: EthSpec> ProductionBeaconNode<E> {
};
let builder = builder
.beacon_chain_builder(client_genesis, client_config_1)
.beacon_chain_builder(client_genesis, client_config.clone())
.await?;
let builder = if client_config.sync_eth1_chain && !client_config.dummy_eth1_backend {
info!(

1
beacon_node/store/src/errors.rs
View File

@ -19,6 +19,7 @@ pub enum Error {
NoContinuationData,
SplitPointModified(Slot, Slot),
ConfigError(StoreConfigError),
SchemaMigrationError(String),
}
impl From<DecodeError> for Error {

14
beacon_node/store/src/hot_cold_store.rs
View File

@ -135,16 +135,20 @@ impl<E: EthSpec> HotColdDB<E, LevelDB<E>, LevelDB<E>> {
/// Open a new or existing database, with the given paths to the hot and cold DBs.
///
/// The `slots_per_restore_point` parameter must be a divisor of `SLOTS_PER_HISTORICAL_ROOT`.
///
/// The `migrate_schema` function is passed in so that the parent `BeaconChain` can provide
/// context and access `BeaconChain`-level code without creating a circular dependency.
pub fn open(
hot_path: &Path,
cold_path: &Path,
migrate_schema: impl FnOnce(Arc<Self>, SchemaVersion, SchemaVersion) -> Result<(), Error>,
config: StoreConfig,
spec: ChainSpec,
log: Logger,
) -> Result<HotColdDB<E, LevelDB<E>, LevelDB<E>>, Error> {
) -> Result<Arc<Self>, Error> {
Self::verify_slots_per_restore_point(config.slots_per_restore_point)?;
let db = HotColdDB {
let db = Arc::new(HotColdDB {
split: RwLock::new(Split::default()),
cold_db: LevelDB::open(cold_path)?,
hot_db: LevelDB::open(hot_path)?,
@ -153,7 +157,7 @@ impl<E: EthSpec> HotColdDB<E, LevelDB<E>, LevelDB<E>> {
spec,
log,
_phantom: PhantomData,
};
});
// Ensure that the schema version of the on-disk database matches the software.
// If the version is mismatched, an automatic migration will be attempted.
@ -164,7 +168,7 @@ impl<E: EthSpec> HotColdDB<E, LevelDB<E>, LevelDB<E>> {
"from_version" => schema_version.as_u64(),
"to_version" => CURRENT_SCHEMA_VERSION.as_u64(),
);
db.migrate_schema(schema_version, CURRENT_SCHEMA_VERSION)?;
migrate_schema(db.clone(), schema_version, CURRENT_SCHEMA_VERSION)?;
} else {
db.store_schema_version(CURRENT_SCHEMA_VERSION)?;
}
@ -830,7 +834,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
}
/// Store the database schema version.
pub(crate) fn store_schema_version(&self, schema_version: SchemaVersion) -> Result<(), Error> {
pub fn store_schema_version(&self, schema_version: SchemaVersion) -> Result<(), Error> {
self.hot_db.put(&SCHEMA_VERSION_KEY, &schema_version)
}

5
beacon_node/store/src/lib.rs
View File

@ -20,10 +20,9 @@ pub mod hot_cold_store;
mod impls;
mod leveldb_store;
mod memory_store;
mod metadata;
pub mod metadata;
mod metrics;
mod partial_beacon_state;
mod schema_change;
pub mod iter;
@ -153,6 +152,7 @@ pub enum DBColumn {
OpPool,
Eth1Cache,
ForkChoice,
PubkeyCache,
/// For the table mapping restore point numbers to state roots.
BeaconRestorePoint,
/// For the mapping from state roots to their slots or summaries.
@ -178,6 +178,7 @@ impl Into<&'static str> for DBColumn {
DBColumn::OpPool => "opo",
DBColumn::Eth1Cache => "etc",
DBColumn::ForkChoice => "frk",
DBColumn::PubkeyCache => "pkc",
DBColumn::BeaconRestorePoint => "brp",
DBColumn::BeaconStateSummary => "bss",
DBColumn::BeaconStateTemporary => "bst",

2
beacon_node/store/src/metadata.rs
View File

@ -2,7 +2,7 @@ use crate::{DBColumn, Error, StoreItem};
use ssz::{Decode, Encode};
use types::{Checkpoint, Hash256};
pub const CURRENT_SCHEMA_VERSION: SchemaVersion = SchemaVersion(2);
pub const CURRENT_SCHEMA_VERSION: SchemaVersion = SchemaVersion(3);
// All the keys that get stored under the `BeaconMeta` column.
//

32
beacon_node/store/src/schema_change.rs
View File

@ -1,32 +0,0 @@
//! Utilities for managing database schema changes.
use crate::hot_cold_store::{HotColdDB, HotColdDBError};
use crate::metadata::{SchemaVersion, CURRENT_SCHEMA_VERSION};
use crate::{Error, ItemStore};
use types::EthSpec;
impl<E, Hot, Cold> HotColdDB<E, Hot, Cold>
where
E: EthSpec,
Hot: ItemStore<E>,
Cold: ItemStore<E>,
{
/// Migrate the database from one schema version to another, applying all requisite mutations.
pub fn migrate_schema(&self, from: SchemaVersion, to: SchemaVersion) -> Result<(), Error> {
match (from, to) {
// Migration from v0.3.0 to v0.3.x, adding the temporary states column.
// Nothing actually needs to be done, but once a DB uses v2 it shouldn't go back.
(SchemaVersion(1), SchemaVersion(2)) => {
self.store_schema_version(to)?;
Ok(())
}
// Migrating from the current schema version to iself is always OK, a no-op.
(_, _) if from == to && to == CURRENT_SCHEMA_VERSION => Ok(()),
// Anything else is an error.
(_, _) => Err(HotColdDBError::UnsupportedSchemaVersion {
target_version: to,
current_version: from,
}
.into()),
}
}
}