gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-04-30 03:03:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/unstable' into tree-states

Browse Source
This commit is contained in:
Michael Sproul
2022-05-24 10:01:05 +10:00
parent 3b4115fdb3 60449849e2
commit e436035c52
237 changed files with 8506 additions and 3598 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

File diff suppressed because it is too large Load Diff

9
beacon_node/beacon_chain/src/beacon_fork_choice_store.rs
View File

@@ -13,7 +13,8 @@ use std::sync::Arc;
use store::{Error as StoreError, HotColdDB, ItemStore};
use superstruct::superstruct;
use types::{
BeaconBlock, BeaconState, BeaconStateError, Checkpoint, Epoch, EthSpec, Hash256, Slot,
BeaconBlock, BeaconState, BeaconStateError, Checkpoint, Epoch, EthSpec, ExecPayload, Hash256,
Slot,
};
#[derive(Debug)]
@@ -255,9 +256,9 @@ where
self.time = slot
}
fn on_verified_block(
fn on_verified_block<Payload: ExecPayload<E>>(
&mut self,
_block: &BeaconBlock<E>,
_block: &BeaconBlock<E, Payload>,
block_root: Hash256,
state: &BeaconState<E>,
) -> Result<(), Self::Error> {
@@ -301,7 +302,7 @@ where
metrics::inc_counter(&metrics::BALANCES_CACHE_MISSES);
let justified_block = self
.store
.get_block(&self.justified_checkpoint.root)
.get_blinded_block(&self.justified_checkpoint.root)
.map_err(Error::FailedToReadBlock)?
.ok_or(Error::MissingBlock(self.justified_checkpoint.root))?
.deconstruct()

17
beacon_node/beacon_chain/src/beacon_snapshot.rs
View File

@@ -1,11 +1,14 @@
use serde_derive::Serialize;
use types::{BeaconState, EthSpec, Hash256, SignedBeaconBlock};
use types::{
BeaconState, EthSpec, ExecPayload, FullPayload, Hash256, SignedBeaconBlock,
SignedBlindedBeaconBlock,
};
/// Represents some block and its associated state. Generally, this will be used for tracking the
/// head, justified head and finalized head.
#[derive(Clone, Serialize, PartialEq, Debug)]
pub struct BeaconSnapshot<E: EthSpec> {
pub beacon_block: SignedBeaconBlock<E>,
pub struct BeaconSnapshot<E: EthSpec, Payload: ExecPayload<E> = FullPayload<E>> {
pub beacon_block: SignedBeaconBlock<E, Payload>,
pub beacon_block_root: Hash256,
pub beacon_state: BeaconState<E>,
}
@@ -19,14 +22,14 @@ pub struct PreProcessingSnapshot<T: EthSpec> {
pub pre_state: BeaconState<T>,
/// This value is only set to `Some` if the `pre_state` was *not* advanced forward.
pub beacon_state_root: Option<Hash256>,
pub beacon_block: SignedBeaconBlock<T>,
pub beacon_block: SignedBlindedBeaconBlock<T>,
pub beacon_block_root: Hash256,
}
impl<E: EthSpec> BeaconSnapshot<E> {
impl<E: EthSpec, Payload: ExecPayload<E>> BeaconSnapshot<E, Payload> {
/// Create a new checkpoint.
pub fn new(
beacon_block: SignedBeaconBlock<E>,
beacon_block: SignedBeaconBlock<E, Payload>,
beacon_block_root: Hash256,
beacon_state: BeaconState<E>,
) -> Self {
@@ -49,7 +52,7 @@ impl<E: EthSpec> BeaconSnapshot<E> {
/// Update all fields of the checkpoint.
pub fn update(
&mut self,
beacon_block: SignedBeaconBlock<E>,
beacon_block: SignedBeaconBlock<E, Payload>,
beacon_block_root: Hash256,
beacon_state: BeaconState<E>,
) {

6
beacon_node/beacon_chain/src/block_reward.rs
View File

@@ -2,12 +2,12 @@ use crate::{BeaconChain, BeaconChainError, BeaconChainTypes};
use eth2::lighthouse::{AttestationRewards, BlockReward, BlockRewardMeta};
use operation_pool::{AttMaxCover, MaxCover, RewardCache};
use state_processing::per_block_processing::altair::sync_committee::compute_sync_aggregate_rewards;
use types::{BeaconBlockRef, BeaconState, EthSpec, Hash256, RelativeEpoch};
use types::{BeaconBlockRef, BeaconState, EthSpec, ExecPayload, Hash256, RelativeEpoch};
impl<T: BeaconChainTypes> BeaconChain<T> {
pub fn compute_block_reward(
pub fn compute_block_reward<Payload: ExecPayload<T::EthSpec>>(
&self,
block: BeaconBlockRef<'_, T::EthSpec>,
block: BeaconBlockRef<'_, T::EthSpec, Payload>,
block_root: Hash256,
state: &BeaconState<T::EthSpec>,
) -> Result<BlockReward, BeaconChainError> {

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

@@ -50,6 +50,7 @@ use crate::{
beacon_chain::{MAXIMUM_GOSSIP_CLOCK_DISPARITY, VALIDATOR_PUBKEY_CACHE_LOCK_TIMEOUT},
metrics, BeaconChain, BeaconChainError, BeaconChainTypes,
};
use derivative::Derivative;
use eth2::types::EventKind;
use execution_layer::PayloadStatus;
use fork_choice::{ForkChoice, ForkChoiceStore, PayloadVerificationStatus};
@@ -70,12 +71,14 @@ use state_processing::{
use std::borrow::Cow;
use std::fs;
use std::io::Write;
use std::sync::Arc;
use store::{Error as DBError, HotColdDB, KeyValueStore, StoreOp};
use tree_hash::TreeHash;
use types::ExecPayload;
use types::{
BeaconBlockRef, BeaconState, BeaconStateError, ChainSpec, Epoch, EthSpec, ExecutionBlockHash,
Hash256, InconsistentFork, PublicKey, PublicKeyBytes, RelativeEpoch, SignedBeaconBlock,
SignedBeaconBlockHeader, Slot,
BeaconBlockRef, BeaconState, BeaconStateError, BlindedPayload, ChainSpec, Epoch, EthSpec,
ExecutionBlockHash, Hash256, InconsistentFork, PublicKey, PublicKeyBytes, RelativeEpoch,
SignedBeaconBlock, SignedBeaconBlockHeader, Slot,
};
const POS_PANDA_BANNER: &str = r#"
@@ -540,7 +543,8 @@ pub fn signature_verify_chain_segment<T: BeaconChainTypes>(
/// A wrapper around a `SignedBeaconBlock` that indicates it has been approved for re-gossiping on
/// the p2p network.
#[derive(Debug)]
#[derive(Derivative)]
#[derivative(Debug(bound = "T: BeaconChainTypes"))]
pub struct GossipVerifiedBlock<T: BeaconChainTypes> {
pub block: SignedBeaconBlock<T::EthSpec>,
pub block_root: Hash256,
@@ -572,7 +576,7 @@ pub struct FullyVerifiedBlock<'a, T: BeaconChainTypes> {
pub block: SignedBeaconBlock<T::EthSpec>,
pub block_root: Hash256,
pub state: BeaconState<T::EthSpec>,
pub parent_block: SignedBeaconBlock<T::EthSpec>,
pub parent_block: SignedBeaconBlock<T::EthSpec, BlindedPayload<T::EthSpec>>,
pub confirmation_db_batch: Vec<StoreOp<'a, T::EthSpec>>,
pub payload_verification_status: PayloadVerificationStatus,
}
@@ -583,7 +587,7 @@ pub struct FullyVerifiedBlock<'a, T: BeaconChainTypes> {
pub trait IntoFullyVerifiedBlock<T: BeaconChainTypes>: Sized {
fn into_fully_verified_block(
self,
chain: &BeaconChain<T>,
chain: &Arc<BeaconChain<T>>,
) -> Result<FullyVerifiedBlock<T>, BlockError<T::EthSpec>> {
self.into_fully_verified_block_slashable(chain)
.map(|fully_verified| {
@@ -599,7 +603,7 @@ pub trait IntoFullyVerifiedBlock<T: BeaconChainTypes>: Sized {
/// Convert the block to fully-verified form while producing data to aid checking slashability.
fn into_fully_verified_block_slashable(
self,
chain: &BeaconChain<T>,
chain: &Arc<BeaconChain<T>>,
) -> Result<FullyVerifiedBlock<T>, BlockSlashInfo<BlockError<T::EthSpec>>>;
fn block(&self) -> &SignedBeaconBlock<T::EthSpec>;
@@ -840,7 +844,7 @@ impl<T: BeaconChainTypes> IntoFullyVerifiedBlock<T> for GossipVerifiedBlock<T> {
/// Completes verification of the wrapped `block`.
fn into_fully_verified_block_slashable(
self,
chain: &BeaconChain<T>,
chain: &Arc<BeaconChain<T>>,
) -> Result<FullyVerifiedBlock<T>, BlockSlashInfo<BlockError<T::EthSpec>>> {
let fully_verified =
SignatureVerifiedBlock::from_gossip_verified_block_check_slashable(self, chain)?;
@@ -963,7 +967,7 @@ impl<T: BeaconChainTypes> IntoFullyVerifiedBlock<T> for SignatureVerifiedBlock<T
/// Completes verification of the wrapped `block`.
fn into_fully_verified_block_slashable(
self,
chain: &BeaconChain<T>,
chain: &Arc<BeaconChain<T>>,
) -> Result<FullyVerifiedBlock<T>, BlockSlashInfo<BlockError<T::EthSpec>>> {
let header = self.block.signed_block_header();
let (parent, block) = if let Some(parent) = self.parent {
@@ -993,7 +997,7 @@ impl<T: BeaconChainTypes> IntoFullyVerifiedBlock<T> for SignedBeaconBlock<T::Eth
/// and then using that implementation of `IntoFullyVerifiedBlock` to complete verification.
fn into_fully_verified_block_slashable(
self,
chain: &BeaconChain<T>,
chain: &Arc<BeaconChain<T>>,
) -> Result<FullyVerifiedBlock<T>, BlockSlashInfo<BlockError<T::EthSpec>>> {
// Perform an early check to prevent wasting time on irrelevant blocks.
let block_root = check_block_relevancy(&self, None, chain)
@@ -1021,23 +1025,27 @@ impl<'a, T: BeaconChainTypes> FullyVerifiedBlock<'a, T> {
block_root: Hash256,
parent: PreProcessingSnapshot<T::EthSpec>,
mut consensus_context: ConsensusContext<T::EthSpec>,
chain: &BeaconChain<T>,
chain: &Arc<BeaconChain<T>>,
) -> Result<Self, BlockError<T::EthSpec>> {
// Reject any block if its parent is not known to fork choice.
//
// A block that is not in fork choice is either:
//
// - Not yet imported: we should reject this block because we should only import a child
// after its parent has been fully imported.
// - Pre-finalized: if the parent block is _prior_ to finalization, we should ignore it
// because it will revert finalization. Note that the finalized block is stored in fork
// choice, so we will not reject any child of the finalized block (this is relevant during
// genesis).
if !chain
.fork_choice
.read()
.contains_block(&block.parent_root())
{
if let Some(parent) = chain.fork_choice.read().get_block(&block.parent_root()) {
// Reject any block where the parent has an invalid payload. It's impossible for a valid
// block to descend from an invalid parent.
if parent.execution_status.is_invalid() {
return Err(BlockError::ParentExecutionPayloadInvalid {
parent_root: block.parent_root(),
});
}
} else {
// Reject any block if its parent is not known to fork choice.
//
// A block that is not in fork choice is either:
//
// - Not yet imported: we should reject this block because we should only import a child
// after its parent has been fully imported.
// - Pre-finalized: if the parent block is _prior_ to finalization, we should ignore it
// because it will revert finalization. Note that the finalized block is stored in fork
// choice, so we will not reject any child of the finalized block (this is relevant during
// genesis).
return Err(BlockError::ParentUnknown(Box::new(block)));
}
@@ -1158,7 +1166,7 @@ impl<'a, T: BeaconChainTypes> FullyVerifiedBlock<'a, T> {
// If the payload did not validate or invalidate the block, check to see if this block is
// valid for optimistic import.
if payload_verification_status == PayloadVerificationStatus::NotVerified {
if payload_verification_status.is_optimistic() {
let current_slot = chain
.slot_clock
.now()
@@ -1292,9 +1300,9 @@ impl<'a, T: BeaconChainTypes> FullyVerifiedBlock<'a, T> {
if valid_merge_transition_block {
info!(chain.log, "{}", POS_PANDA_BANNER);
info!(chain.log, "Proof of Stake Activated"; "slot" => block.slot());
info!(chain.log, ""; "Terminal POW Block Hash" => ?block.message().execution_payload()?.parent_hash.into_root());
info!(chain.log, ""; "Terminal POW Block Hash" => ?block.message().execution_payload()?.parent_hash().into_root());
info!(chain.log, ""; "Merge Transition Block Root" => ?block.message().tree_hash_root());
info!(chain.log, ""; "Merge Transition Execution Hash" => ?block.message().execution_payload()?.block_hash.into_root());
info!(chain.log, ""; "Merge Transition Execution Hash" => ?block.message().execution_payload()?.block_hash().into_root());
}
Ok(Self {
@@ -1523,7 +1531,7 @@ fn load_parent<T: BeaconChainTypes>(
// indicate that we don't yet know the parent.
let root = block.parent_root();
let parent_block = chain
.get_block(&block.parent_root())
.get_blinded_block(&block.parent_root())
.map_err(BlockError::BeaconChainError)?
.ok_or_else(|| {
// Return a `MissingBeaconBlock` error instead of a `ParentUnknown` error since

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

@@ -1,5 +1,6 @@
use crate::beacon_chain::{BEACON_CHAIN_DB_KEY, ETH1_CACHE_DB_KEY, OP_POOL_DB_KEY};
use crate::eth1_chain::{CachingEth1Backend, SszEth1};
use crate::fork_choice_signal::ForkChoiceSignalTx;
use crate::fork_revert::{reset_fork_choice_to_finalization, revert_to_fork_boundary};
use crate::head_tracker::HeadTracker;
use crate::migrate::{BackgroundMigrator, MigratorConfig};
@@ -26,7 +27,7 @@ use std::marker::PhantomData;
use std::sync::Arc;
use std::time::Duration;
use store::{Error as StoreError, HotColdDB, ItemStore, KeyValueStoreOp};
use task_executor::ShutdownReason;
use task_executor::{ShutdownReason, TaskExecutor};
use types::{
BeaconBlock, BeaconState, ChainSpec, Checkpoint, Epoch, EthSpec, Graffiti, Hash256,
PublicKeyBytes, Signature, SignedBeaconBlock, Slot,
@@ -90,6 +91,7 @@ pub struct BeaconChainBuilder<T: BeaconChainTypes> {
// Pending I/O batch that is constructed during building and should be executed atomically
// alongside `PersistedBeaconChain` storage when `BeaconChainBuilder::build` is called.
pending_io_batch: Vec<KeyValueStoreOp>,
task_executor: Option<TaskExecutor>,
}
impl<TSlotClock, TEth1Backend, TEthSpec, THotStore, TColdStore>
@@ -128,6 +130,7 @@ where
slasher: None,
validator_monitor: None,
pending_io_batch: vec![],
task_executor: None,
}
}
@@ -181,6 +184,13 @@ where
self.log = Some(log);
self
}
/// Sets the task executor.
pub fn task_executor(mut self, task_executor: TaskExecutor) -> Self {
self.task_executor = Some(task_executor);
self
}
/// Attempt to load an existing eth1 cache from the builder's `Store`.
pub fn get_persisted_eth1_backend(&self) -> Result<Option<SszEth1>, String> {
let store = self
@@ -239,7 +249,7 @@ where
.ok_or("Fork choice not found in store")?;
let genesis_block = store
.get_block(&chain.genesis_block_root)
.get_blinded_block(&chain.genesis_block_root)
.map_err(|e| descriptive_db_error("genesis block", &e))?
.ok_or("Genesis block not found in store")?;
let genesis_state = store
@@ -617,7 +627,7 @@ where
// Try to decode the head block according to the current fork, if that fails, try
// to backtrack to before the most recent fork.
let (head_block_root, head_block, head_reverted) =
match store.get_block(&initial_head_block_root) {
match store.get_full_block(&initial_head_block_root) {
Ok(Some(block)) => (initial_head_block_root, block, false),
Ok(None) => return Err("Head block not found in store".into()),
Err(StoreError::SszDecodeError(_)) => {
@@ -714,6 +724,16 @@ where
);
}
// If enabled, set up the fork choice signaller.
let (fork_choice_signal_tx, fork_choice_signal_rx) =
if self.chain_config.fork_choice_before_proposal_timeout_ms != 0 {
let tx = ForkChoiceSignalTx::new();
let rx = tx.get_receiver();
(Some(tx), Some(rx))
} else {
(None, None)
};
// Store the `PersistedBeaconChain` in the database atomically with the metadata so that on
// restart we can correctly detect the presence of an initialized database.
//
@@ -772,6 +792,8 @@ where
genesis_block_root,
genesis_state_root,
fork_choice: RwLock::new(fork_choice),
fork_choice_signal_tx,
fork_choice_signal_rx,
event_handler: self.event_handler,
head_tracker,
shuffling_cache: TimeoutRwLock::new(ShufflingCache::new()),
@@ -944,6 +966,7 @@ mod test {
use std::time::Duration;
use store::config::StoreConfig;
use store::{HotColdDB, MemoryStore};
use task_executor::test_utils::TestRuntime;
use types::{EthSpec, MinimalEthSpec, Slot};
type TestEthSpec = MinimalEthSpec;
@@ -977,10 +1000,12 @@ mod test {
.expect("should create interop genesis state");
let (shutdown_tx, _) = futures::channel::mpsc::channel(1);
let runtime = TestRuntime::default();
let chain = BeaconChainBuilder::new(MinimalEthSpec)
.logger(log.clone())
.store(Arc::new(store))
.task_executor(runtime.task_executor.clone())
.genesis_state(genesis_state)
.expect("should build state using recent genesis")
.dummy_eth1_backend()
@@ -1011,10 +1036,10 @@ mod test {
assert_eq!(
chain
.store
.get_block(&Hash256::zero())
.get_blinded_block(&Hash256::zero())
.expect("should read db")
.expect("should find genesis block"),
block,
block.clone().into(),
"should store genesis block under zero hash alias"
);
assert_eq!(

7
beacon_node/beacon_chain/src/chain_config.rs
View File

@@ -1,6 +1,8 @@
use serde_derive::{Deserialize, Serialize};
use types::Checkpoint;
pub const DEFAULT_FORK_CHOICE_BEFORE_PROPOSAL_TIMEOUT: u64 = 250;
#[derive(Debug, PartialEq, Eq, Clone, Deserialize, Serialize)]
pub struct ChainConfig {
/// Maximum number of slots to skip when importing a consensus message (e.g., block,
@@ -18,6 +20,10 @@ pub struct ChainConfig {
pub enable_lock_timeouts: bool,
/// The max size of a message that can be sent over the network.
pub max_network_size: usize,
/// Number of milliseconds to wait for fork choice before proposing a block.
///
/// If set to 0 then block proposal will not wait for fork choice at all.
pub fork_choice_before_proposal_timeout_ms: u64,
}
impl Default for ChainConfig {
@@ -28,6 +34,7 @@ impl Default for ChainConfig {
reconstruct_historic_states: false,
enable_lock_timeouts: true,
max_network_size: 10 * 1_048_576, // 10M
fork_choice_before_proposal_timeout_ms: DEFAULT_FORK_CHOICE_BEFORE_PROPOSAL_TIMEOUT,
}
}
}

4
beacon_node/beacon_chain/src/early_attester_cache.rs
View File

@@ -104,6 +104,10 @@ impl<E: EthSpec> EarlyAttesterCache<E> {
return Ok(None);
}
if request_slot < item.block.slot() {
return Ok(None);
}
let committee_count = item
.committee_lengths
.get_committee_count_per_slot::<E>(spec)?;

30
beacon_node/beacon_chain/src/errors.rs
View File

@@ -9,6 +9,7 @@ use crate::observed_aggregates::Error as ObservedAttestationsError;
use crate::observed_attesters::Error as ObservedAttestersError;
use crate::observed_block_producers::Error as ObservedBlockProducersError;
use execution_layer::PayloadStatus;
use fork_choice::ExecutionStatus;
use futures::channel::mpsc::TrySendError;
use operation_pool::OpPoolError;
use safe_arith::ArithError;
@@ -25,6 +26,7 @@ use state_processing::{
};
use std::time::Duration;
use task_executor::ShutdownReason;
use tokio::task::JoinError;
use types::*;
macro_rules! easy_from_to {
@@ -89,7 +91,7 @@ pub enum BeaconChainError {
BlockSignatureVerifierError(state_processing::block_signature_verifier::Error),
BlockReplayError(BlockReplayError),
DuplicateValidatorPublicKey,
ValidatorPubkeyCacheFileError(String),
ValidatorPubkeyCacheError(String),
ValidatorIndexUnknown(usize),
ValidatorPubkeyUnknown(PublicKeyBytes),
OpPoolError(OpPoolError),
@@ -137,6 +139,18 @@ pub enum BeaconChainError {
},
AltairForkDisabled,
ExecutionLayerMissing,
BlockVariantLacksExecutionPayload(Hash256),
ExecutionLayerErrorPayloadReconstruction(ExecutionBlockHash, execution_layer::Error),
BlockHashMissingFromExecutionLayer(ExecutionBlockHash),
InconsistentPayloadReconstructed {
slot: Slot,
exec_block_hash: ExecutionBlockHash,
canonical_payload_root: Hash256,
reconstructed_payload_root: Hash256,
canonical_transactions_root: Hash256,
reconstructed_transactions_root: Hash256,
},
AddPayloadLogicError,
ExecutionForkChoiceUpdateFailed(execution_layer::Error),
PrepareProposerBlockingFailed(execution_layer::Error),
ExecutionForkChoiceUpdateInvalid {
@@ -162,6 +176,19 @@ pub enum BeaconChainError {
fork_choice: Hash256,
},
InvalidSlot(Slot),
HeadBlockNotFullyVerified {
beacon_block_root: Hash256,
execution_status: ExecutionStatus,
},
CannotAttestToFinalizedBlock {
beacon_block_root: Hash256,
},
RuntimeShutdown,
ProcessInvalidExecutionPayload(JoinError),
ForkChoiceSignalOutOfOrder {
current: Slot,
latest: Slot,
},
}
easy_from_to!(SlotProcessingError, BeaconChainError);
@@ -212,6 +239,7 @@ pub enum BlockProductionError {
FailedToLoadState(store::Error),
MissingFinalizedBlock(Hash256),
BlockTooLarge(usize),
ForkChoiceError(BeaconChainError),
}
easy_from_to!(BlockProcessingError, BlockProductionError);

51
beacon_node/beacon_chain/src/execution_payload.rs
View File

@@ -20,6 +20,7 @@ use state_processing::per_block_processing::{
compute_timestamp_at_slot, is_execution_enabled, is_merge_transition_complete,
partially_verify_execution_payload,
};
use std::sync::Arc;
use types::*;
/// Verify that `execution_payload` contained by `block` is considered valid by an execution
@@ -32,7 +33,7 @@ use types::*;
///
/// https://github.com/ethereum/consensus-specs/blob/v1.1.9/specs/bellatrix/beacon-chain.md#notify_new_payload
pub fn notify_new_payload<T: BeaconChainTypes>(
chain: &BeaconChain<T>,
chain: &Arc<BeaconChain<T>>,
state: &BeaconState<T::EthSpec>,
block: BeaconBlockRef<T::EthSpec>,
) -> Result<PayloadVerificationStatus, BlockError<T::EthSpec>> {
@@ -53,14 +54,15 @@ pub fn notify_new_payload<T: BeaconChainTypes>(
.execution_layer
.as_ref()
.ok_or(ExecutionPayloadError::NoExecutionConnection)?;
let new_payload_response = execution_layer
.block_on(|execution_layer| execution_layer.notify_new_payload(execution_payload));
let new_payload_response = execution_layer.block_on(|execution_layer| {
execution_layer.notify_new_payload(&execution_payload.execution_payload)
});
match new_payload_response {
Ok(status) => match status {
PayloadStatus::Valid => Ok(PayloadVerificationStatus::Verified),
PayloadStatus::Syncing | PayloadStatus::Accepted => {
Ok(PayloadVerificationStatus::NotVerified)
Ok(PayloadVerificationStatus::Optimistic)
}
PayloadStatus::Invalid {
latest_valid_hash, ..
@@ -118,10 +120,10 @@ pub fn validate_merge_block<T: BeaconChainTypes>(
.into());
}
if execution_payload.parent_hash != spec.terminal_block_hash {
if execution_payload.parent_hash() != spec.terminal_block_hash {
return Err(ExecutionPayloadError::InvalidTerminalBlockHash {
terminal_block_hash: spec.terminal_block_hash,
payload_parent_hash: execution_payload.parent_hash,
payload_parent_hash: execution_payload.parent_hash(),
}
.into());
}
@@ -136,14 +138,14 @@ pub fn validate_merge_block<T: BeaconChainTypes>(
let is_valid_terminal_pow_block = execution_layer
.block_on(|execution_layer| {
execution_layer.is_valid_terminal_pow_block_hash(execution_payload.parent_hash, spec)
execution_layer.is_valid_terminal_pow_block_hash(execution_payload.parent_hash(), spec)
})
.map_err(ExecutionPayloadError::from)?;
match is_valid_terminal_pow_block {
Some(true) => Ok(()),
Some(false) => Err(ExecutionPayloadError::InvalidTerminalPoWBlock {
parent_hash: execution_payload.parent_hash,
parent_hash: execution_payload.parent_hash(),
}
.into()),
None => {
@@ -167,7 +169,7 @@ pub fn validate_merge_block<T: BeaconChainTypes>(
debug!(
chain.log,
"Optimistically accepting terminal block";
"block_hash" => ?execution_payload.parent_hash,
"block_hash" => ?execution_payload.parent_hash(),
"msg" => "the terminal block/parent was unavailable"
);
Ok(())
@@ -192,7 +194,7 @@ pub fn validate_execution_payload_for_gossip<T: BeaconChainTypes>(
let is_merge_transition_complete = match parent_block.execution_status {
// Optimistically declare that an "unknown" status block has completed the merge.
ExecutionStatus::Valid(_) | ExecutionStatus::Unknown(_) => true,
ExecutionStatus::Valid(_) | ExecutionStatus::Optimistic(_) => true,
// It's impossible for an irrelevant block to have completed the merge. It is pre-merge
// by definition.
ExecutionStatus::Irrelevant(_) => false,
@@ -215,11 +217,11 @@ pub fn validate_execution_payload_for_gossip<T: BeaconChainTypes>(
))?;
// The block's execution payload timestamp is correct with respect to the slot
if execution_payload.timestamp != expected_timestamp {
if execution_payload.timestamp() != expected_timestamp {
return Err(BlockError::ExecutionPayloadError(
ExecutionPayloadError::InvalidPayloadTimestamp {
expected: expected_timestamp,
found: execution_payload.timestamp,
found: execution_payload.timestamp(),
},
));
}
@@ -241,20 +243,23 @@ pub fn validate_execution_payload_for_gossip<T: BeaconChainTypes>(
/// Equivalent to the `get_execution_payload` function in the Validator Guide:
///
/// https://github.com/ethereum/consensus-specs/blob/v1.1.5/specs/merge/validator.md#block-proposal
pub fn get_execution_payload<T: BeaconChainTypes>(
pub fn get_execution_payload<T: BeaconChainTypes, Payload: ExecPayload<T::EthSpec>>(
chain: &BeaconChain<T>,
state: &BeaconState<T::EthSpec>,
proposer_index: u64,
) -> Result<ExecutionPayload<T::EthSpec>, BlockProductionError> {
Ok(prepare_execution_payload_blocking(chain, state, proposer_index)?.unwrap_or_default())
) -> Result<Payload, BlockProductionError> {
Ok(
prepare_execution_payload_blocking::<T, Payload>(chain, state, proposer_index)?
.unwrap_or_default(),
)
}
/// Wraps the async `prepare_execution_payload` function as a blocking task.
pub fn prepare_execution_payload_blocking<T: BeaconChainTypes>(
pub fn prepare_execution_payload_blocking<T: BeaconChainTypes, Payload: ExecPayload<T::EthSpec>>(
chain: &BeaconChain<T>,
state: &BeaconState<T::EthSpec>,
proposer_index: u64,
) -> Result<Option<ExecutionPayload<T::EthSpec>>, BlockProductionError> {
) -> Result<Option<Payload>, BlockProductionError> {
let execution_layer = chain
.execution_layer
.as_ref()
@@ -262,7 +267,7 @@ pub fn prepare_execution_payload_blocking<T: BeaconChainTypes>(
execution_layer
.block_on_generic(|_| async {
prepare_execution_payload(chain, state, proposer_index).await
prepare_execution_payload::<T, Payload>(chain, state, proposer_index).await
})
.map_err(BlockProductionError::BlockingFailed)?
}
@@ -281,11 +286,11 @@ pub fn prepare_execution_payload_blocking<T: BeaconChainTypes>(
/// Equivalent to the `prepare_execution_payload` function in the Validator Guide:
///
/// https://github.com/ethereum/consensus-specs/blob/v1.1.5/specs/merge/validator.md#block-proposal
pub async fn prepare_execution_payload<T: BeaconChainTypes>(
pub async fn prepare_execution_payload<T: BeaconChainTypes, Payload: ExecPayload<T::EthSpec>>(
chain: &BeaconChain<T>,
state: &BeaconState<T::EthSpec>,
proposer_index: u64,
) -> Result<Option<ExecutionPayload<T::EthSpec>>, BlockProductionError> {
) -> Result<Option<Payload>, BlockProductionError> {
let spec = &chain.spec;
let execution_layer = chain
.execution_layer
@@ -328,19 +333,19 @@ pub async fn prepare_execution_payload<T: BeaconChainTypes>(
} else {
chain
.store
.get_block(&finalized_root)
.get_blinded_block(&finalized_root)
.map_err(BlockProductionError::FailedToReadFinalizedBlock)?
.ok_or(BlockProductionError::MissingFinalizedBlock(finalized_root))?
.message()
.body()
.execution_payload()
.ok()
.map(|ep| ep.block_hash)
.map(|ep| ep.block_hash())
};
// Note: the suggested_fee_recipient is stored in the `execution_layer`, it will add this parameter.
let execution_payload = execution_layer
.get_payload(
.get_payload::<T::EthSpec, Payload>(
parent_hash,
timestamp,
random,

97
beacon_node/beacon_chain/src/fork_choice_signal.rs Normal file
View File

@@ -0,0 +1,97 @@
//! Concurrency helpers for synchronising block proposal with fork choice.
//!
//! The transmitter provides a way for a thread runnning fork choice on a schedule to signal
//! to the receiver that fork choice has been updated for a given slot.
use crate::BeaconChainError;
use parking_lot::{Condvar, Mutex};
use std::sync::Arc;
use std::time::Duration;
use types::Slot;
/// Sender, for use by the per-slot task timer.
pub struct ForkChoiceSignalTx {
pair: Arc<(Mutex<Slot>, Condvar)>,
}
/// Receiver, for use by the beacon chain waiting on fork choice to complete.
pub struct ForkChoiceSignalRx {
pair: Arc<(Mutex<Slot>, Condvar)>,
}
pub enum ForkChoiceWaitResult {
/// Successfully reached a slot greater than or equal to the awaited slot.
Success(Slot),
/// Fork choice was updated to a lower slot, indicative of lag or processing delays.
Behind(Slot),
/// Timed out waiting for the fork choice update from the sender.
TimeOut,
}
impl ForkChoiceSignalTx {
pub fn new() -> Self {
let pair = Arc::new((Mutex::new(Slot::new(0)), Condvar::new()));
Self { pair }
}
pub fn get_receiver(&self) -> ForkChoiceSignalRx {
ForkChoiceSignalRx {
pair: self.pair.clone(),
}
}
/// Signal to the receiver that fork choice has been updated to `slot`.
///
/// Return an error if the provided `slot` is strictly less than any previously provided slot.
pub fn notify_fork_choice_complete(&self, slot: Slot) -> Result<(), BeaconChainError> {
let &(ref lock, ref condvar) = &*self.pair;
let mut current_slot = lock.lock();
if slot < *current_slot {
return Err(BeaconChainError::ForkChoiceSignalOutOfOrder {
current: *current_slot,
latest: slot,
});
} else {
*current_slot = slot;
}
// We use `notify_all` because there may be multiple block proposals waiting simultaneously.
// Usually there'll be 0-1.
condvar.notify_all();
Ok(())
}
}
impl Default for ForkChoiceSignalTx {
fn default() -> Self {
Self::new()
}
}
impl ForkChoiceSignalRx {
pub fn wait_for_fork_choice(&self, slot: Slot, timeout: Duration) -> ForkChoiceWaitResult {
let &(ref lock, ref condvar) = &*self.pair;
let mut current_slot = lock.lock();
// Wait for `current_slot >= slot`.
//
// Do not loop and wait, if we receive an update for the wrong slot then something is
// quite out of whack and we shouldn't waste more time waiting.
if *current_slot < slot {
let timeout_result = condvar.wait_for(&mut current_slot, timeout);
if timeout_result.timed_out() {
return ForkChoiceWaitResult::TimeOut;
}
}
if *current_slot >= slot {
ForkChoiceWaitResult::Success(*current_slot)
} else {
ForkChoiceWaitResult::Behind(*current_slot)
}
}
}

14
beacon_node/beacon_chain/src/fork_revert.rs
View File

@@ -49,7 +49,7 @@ pub fn revert_to_fork_boundary<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>
);
let block_iter = ParentRootBlockIterator::fork_tolerant(&store, head_block_root);
process_results(block_iter, |mut iter| {
let (block_root, blinded_block) = process_results(block_iter, |mut iter| {
iter.find_map(|(block_root, block)| {
if block.slot() < fork_epoch.start_slot(E::slots_per_epoch()) {
Some((block_root, block))
@@ -70,7 +70,13 @@ pub fn revert_to_fork_boundary<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>
e, CORRUPT_DB_MESSAGE
)
})?
.ok_or_else(|| format!("No pre-fork blocks found. {}", CORRUPT_DB_MESSAGE))
.ok_or_else(|| format!("No pre-fork blocks found. {}", CORRUPT_DB_MESSAGE))?;
let block = store
.make_full_block(&block_root, blinded_block)
.map_err(|e| format!("Unable to add payload to new head block: {:?}", e))?;
Ok((block_root, block))
}
/// Reset fork choice to the finalized checkpoint of the supplied head state.
@@ -98,7 +104,7 @@ pub fn reset_fork_choice_to_finalization<E: EthSpec, Hot: ItemStore<E>, Cold: It
let finalized_checkpoint = head_state.finalized_checkpoint();
let finalized_block_root = finalized_checkpoint.root;
let finalized_block = store
.get_block(&finalized_block_root)
.get_full_block(&finalized_block_root)
.map_err(|e| format!("Error loading finalized block: {:?}", e))?
.ok_or_else(|| {
format!(
@@ -175,7 +181,7 @@ pub fn reset_fork_choice_to_finalization<E: EthSpec, Hot: ItemStore<E>, Cold: It
// retro-actively determine if they were valid or not.
//
// This scenario is so rare that it seems OK to double-verify some blocks.
let payload_verification_status = PayloadVerificationStatus::NotVerified;
let payload_verification_status = PayloadVerificationStatus::Optimistic;
let (block, _) = block.deconstruct();
fork_choice

9
beacon_node/beacon_chain/src/historical_blocks.rs
View File

@@ -9,7 +9,7 @@ use std::borrow::Cow;
use std::iter;
use std::time::Duration;
use store::{chunked_vector::BlockRoots, AnchorInfo, ChunkWriter, KeyValueStore};
use types::{Hash256, SignedBeaconBlock, Slot};
use types::{Hash256, SignedBlindedBeaconBlock, Slot};
/// Use a longer timeout on the pubkey cache.
///
@@ -58,7 +58,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
/// Return the number of blocks successfully imported.
pub fn import_historical_block_batch(
&self,
blocks: &[SignedBeaconBlock<T::EthSpec>],
blocks: Vec<SignedBlindedBeaconBlock<T::EthSpec>>,
) -> Result<usize, Error> {
let anchor_info = self
.store
@@ -106,8 +106,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
.into());
}
// Store block in the hot database.
hot_batch.push(self.store.block_as_kv_store_op(&block_root, block));
// Store block in the hot database without payload.
self.store
.blinded_block_as_kv_store_ops(&block_root, block, &mut hot_batch);
// Store block roots, including at all skip slots in the freezer DB.
for slot in (block.slot().as_usize()..prev_block_slot.as_usize()).rev() {

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

@@ -15,6 +15,7 @@ mod errors;
pub mod eth1_chain;
pub mod events;
mod execution_payload;
pub mod fork_choice_signal;
pub mod fork_revert;
mod head_tracker;
pub mod historical_blocks;

4
beacon_node/beacon_chain/src/metrics.rs
View File

@@ -86,6 +86,10 @@ lazy_static! {
);
pub static ref BLOCK_PRODUCTION_TIMES: Result<Histogram> =
try_create_histogram("beacon_block_production_seconds", "Full runtime of block production");
pub static ref BLOCK_PRODUCTION_FORK_CHOICE_TIMES: Result<Histogram> = try_create_histogram(
"beacon_block_production_fork_choice_seconds",
"Time taken to run fork choice before block production"
);
pub static ref BLOCK_PRODUCTION_STATE_LOAD_TIMES: Result<Histogram> = try_create_histogram(
"beacon_block_production_state_load_seconds",
"Time taken to load the base state for block production"

9
beacon_node/beacon_chain/src/migrate.rs
View File

@@ -396,7 +396,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
// so delete it from the head tracker but leave it and its states in the database
// This is suboptimal as it wastes disk space, but it's difficult to fix. A re-sync
// can be used to reclaim the space.
let head_state_root = match store.get_block(&head_hash) {
let head_state_root = match store.get_blinded_block(&head_hash) {
Ok(Some(block)) => block.state_root(),
Ok(None) => {
return Err(BeaconStateError::MissingBeaconBlock(head_hash.into()).into())
@@ -518,7 +518,12 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
let mut batch: Vec<StoreOp<E>> = abandoned_blocks
.into_iter()
.map(Into::into)
.map(StoreOp::DeleteBlock)
.flat_map(|block_root: Hash256| {
[
StoreOp::DeleteBlock(block_root),
StoreOp::DeleteExecutionPayload(block_root),
]
})
.collect();
// Persist the head in case the process is killed or crashes here. This prevents

1
beacon_node/beacon_chain/src/observed_aggregates.rs
View File

@@ -203,6 +203,7 @@ impl<T: TreeHash + SlotData + Consts, E: EthSpec> ObservedAggregates<T, E> {
/// Check to see if the `root` of `item` is in self.
///
/// `root` must equal `a.tree_hash_root()`.
#[allow(clippy::wrong_self_convention)]
pub fn is_known(&mut self, item: &T, root: Hash256) -> Result<bool, Error> {
let index = self.get_set_index(item.get_slot())?;

2
beacon_node/beacon_chain/src/pre_finalization_cache.rs
View File

@@ -71,7 +71,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
}
// 2. Check on disk.
if self.store.get_block(&block_root)?.is_some() {
if self.store.get_blinded_block(&block_root)?.is_some() {
cache.block_roots.put(block_root, ());
return Ok(true);
}

19
beacon_node/beacon_chain/src/proposer_prep_service.rs
View File

@@ -50,12 +50,19 @@ async fn proposer_prep_service<T: BeaconChainTypes>(
let inner_chain = chain.clone();
executor.spawn(
async move {
if let Err(e) = inner_chain.prepare_beacon_proposer_async().await {
error!(
inner_chain.log,
"Proposer prepare routine failed";
"error" => ?e
);
if let Ok(current_slot) = inner_chain.slot() {
if let Err(e) = inner_chain
.prepare_beacon_proposer_async(current_slot)
.await
{
error!(
inner_chain.log,
"Proposer prepare routine failed";
"error" => ?e
);
}
} else {
debug!(inner_chain.log, "No slot for proposer prepare routine");
}
},
"proposer_prep_update",

117
beacon_node/beacon_chain/src/schema_change.rs
View File

@@ -3,24 +3,17 @@ mod migration_schema_v10;
mod migration_schema_v6;
mod migration_schema_v7;
mod migration_schema_v8;
mod migration_schema_v9;
mod types;
use crate::beacon_chain::{BeaconChainTypes, FORK_CHOICE_DB_KEY, OP_POOL_DB_KEY};
use crate::beacon_chain::{BeaconChainTypes, FORK_CHOICE_DB_KEY};
use crate::persisted_fork_choice::{PersistedForkChoiceV1, PersistedForkChoiceV7};
use crate::validator_pubkey_cache::ValidatorPubkeyCache;
use operation_pool::{PersistedOperationPool, PersistedOperationPoolBase};
use slog::{warn, Logger};
use ssz::{Decode, Encode};
use ssz_derive::{Decode, Encode};
use std::fs;
use std::path::Path;
use std::sync::Arc;
use store::config::OnDiskStoreConfig;
use store::hot_cold_store::{HotColdDB, HotColdDBError};
use store::metadata::{SchemaVersion, CONFIG_KEY, CURRENT_SCHEMA_VERSION};
use store::{DBColumn, Error as StoreError, ItemStore, StoreItem};
const PUBKEY_CACHE_FILENAME: &str = "pubkey_cache.ssz";
use store::metadata::{SchemaVersion, CURRENT_SCHEMA_VERSION};
use store::{Error as StoreError, StoreItem};
/// Migrate the database from one schema version to another, applying all requisite mutations.
pub fn migrate_schema<T: BeaconChainTypes>(
@@ -33,75 +26,17 @@ pub fn migrate_schema<T: BeaconChainTypes>(
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.
// Upgrade 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, log.clone())?;
migrate_schema::<T>(db, datadir, next, to, log)
}
// 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)))?;
//
// Migrations from before SchemaVersion(5) are deprecated.
//
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(())
}
// Migration for adding sync committee contributions to the persisted op pool.
(SchemaVersion(3), SchemaVersion(4)) => {
// Deserialize from what exists in the database using the `PersistedOperationPoolBase`
// variant and convert it to the Altair variant.
let pool_opt = db
.get_item::<PersistedOperationPoolBase<T::EthSpec>>(&OP_POOL_DB_KEY)?
.map(PersistedOperationPool::Base)
.map(PersistedOperationPool::base_to_altair);
if let Some(pool) = pool_opt {
// Store the converted pool under the same key.
db.put_item::<PersistedOperationPool<T::EthSpec>>(&OP_POOL_DB_KEY, &pool)?;
}
db.store_schema_version(to)?;
Ok(())
}
// Migration for weak subjectivity sync support and clean up of `OnDiskStoreConfig` (#1784).
(SchemaVersion(4), SchemaVersion(5)) => {
if let Some(OnDiskStoreConfigV4 {
slots_per_restore_point,
..
}) = db.hot_db.get(&CONFIG_KEY)?
{
let new_config = OnDiskStoreConfig {
slots_per_restore_point,
};
db.hot_db.put(&CONFIG_KEY, &new_config)?;
}
db.store_schema_version(to)?;
Ok(())
}
// Migration for adding `execution_status` field to the fork choice store.
(SchemaVersion(5), SchemaVersion(6)) => {
// Database operations to be done atomically
@@ -182,12 +117,21 @@ pub fn migrate_schema<T: BeaconChainTypes>(
Ok(())
}
// Reserved for merge-related changes.
(SchemaVersion(8), SchemaVersion(9)) => Ok(()),
// Upgrade from v8 to v9 to separate the execution payloads into their own column.
(SchemaVersion(8), SchemaVersion(9)) => {
migration_schema_v9::upgrade_to_v9::<T>(db.clone(), log)?;
db.store_schema_version(to)
}
// Downgrade from v9 to v8 to ignore the separation of execution payloads
// NOTE: only works before the Bellatrix fork epoch.
(SchemaVersion(9), SchemaVersion(8)) => {
migration_schema_v9::downgrade_from_v9::<T>(db.clone(), log)?;
db.store_schema_version(to)
}
// Upgrade for tree-states database changes.
(SchemaVersion(9), SchemaVersion(10)) => migration_schema_v10::upgrade_to_v10::<T>(db, log),
// Downgrade for tree-states database changes.
(SchemaVersion(10), SchemaVersion(8)) => {
(SchemaVersion(10), SchemaVersion(9)) => {
migration_schema_v10::downgrade_from_v10::<T>(db, log)
}
// Anything else is an error.
@@ -198,24 +142,3 @@ pub fn migrate_schema<T: BeaconChainTypes>(
.into()),
}
}
// Store config used in v4 schema and earlier.
#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
pub struct OnDiskStoreConfigV4 {
pub slots_per_restore_point: u64,
pub _block_cache_size: usize,
}
impl StoreItem for OnDiskStoreConfigV4 {
fn db_column() -> DBColumn {
DBColumn::BeaconMeta
}
fn as_store_bytes(&self) -> Result<Vec<u8>, StoreError> {
Ok(self.as_ssz_bytes())
}
fn from_store_bytes(bytes: &[u8]) -> Result<Self, StoreError> {
Ok(Self::from_ssz_bytes(bytes)?)
}
}

2
beacon_node/beacon_chain/src/schema_change/migration_schema_v10.rs
View File

@@ -70,7 +70,7 @@ pub fn upgrade_to_v10<T: BeaconChainTypes>(
.zip(ssz_head_tracker.slots)
{
let block = db
.get_block(&head_block_root)?
.get_blinded_block(&head_block_root)?
.ok_or(Error::BlockNotFound(head_block_root))?;
let head_state_root = block.state_root();

2
beacon_node/beacon_chain/src/schema_change/migration_schema_v7.rs
View File

@@ -31,7 +31,7 @@ pub(crate) fn update_with_reinitialized_fork_choice<T: BeaconChainTypes>(
.finalized_checkpoint
.root;
let anchor_block = db
.get_block(&anchor_block_root)
.get_full_block_prior_to_v9(&anchor_block_root)
.map_err(|e| format!("{:?}", e))?
.ok_or_else(|| "Missing anchor beacon block".to_string())?;
let anchor_state = db

2
beacon_node/beacon_chain/src/schema_change/migration_schema_v8.rs
View File

@@ -34,7 +34,7 @@ pub fn update_fork_choice<T: BeaconChainTypes>(
// before schema v8 the cache would always miss on skipped slots.
for item in balances_cache.items {
// Drop any blocks that aren't found, they're presumably too old and this is only a cache.
if let Some(block) = db.get_block(&item.block_root)? {
if let Some(block) = db.get_full_block_prior_to_v9(&item.block_root)? {
fork_choice_store.balances_cache.items.push(CacheItemV8 {
block_root: item.block_root,
epoch: block.slot().epoch(T::EthSpec::slots_per_epoch()),

176
beacon_node/beacon_chain/src/schema_change/migration_schema_v9.rs Normal file
View File

@@ -0,0 +1,176 @@
use crate::beacon_chain::BeaconChainTypes;
use slog::{debug, error, info, Logger};
use slot_clock::SlotClock;
use std::sync::Arc;
use std::time::Duration;
use store::{DBColumn, Error, HotColdDB, KeyValueStore};
use types::{EthSpec, Hash256, Slot};
const OPS_PER_BLOCK_WRITE: usize = 2;
/// The slot clock isn't usually available before the database is initialized, so we construct a
/// temporary slot clock by reading the genesis state. It should always exist if the database is
/// initialized at a prior schema version, however we still handle the lack of genesis state
/// gracefully.
fn get_slot_clock<T: BeaconChainTypes>(
db: &HotColdDB<T::EthSpec, T::HotStore, T::ColdStore>,
log: &Logger,
) -> Result<Option<T::SlotClock>, Error> {
// At schema v8 the genesis block must be a *full* block (with payload). In all likeliness it
// actually has no payload.
let spec = db.get_chain_spec();
let genesis_block = if let Some(block) = db.get_full_block_prior_to_v9(&Hash256::zero())? {
block
} else {
error!(log, "Missing genesis block");
return Ok(None);
};
let genesis_state =
if let Some(state) = db.get_state(&genesis_block.state_root(), Some(Slot::new(0)))? {
state
} else {
error!(log, "Missing genesis state"; "state_root" => ?genesis_block.state_root());
return Ok(None);
};
Ok(Some(T::SlotClock::new(
spec.genesis_slot,
Duration::from_secs(genesis_state.genesis_time()),
Duration::from_secs(spec.seconds_per_slot),
)))
}
pub fn upgrade_to_v9<T: BeaconChainTypes>(
db: Arc<HotColdDB<T::EthSpec, T::HotStore, T::ColdStore>>,
log: Logger,
) -> Result<(), Error> {
// This upgrade is a no-op if the Bellatrix fork epoch has not already passed. This migration
// was implemented before the activation of Bellatrix on all networks except Kiln, so the only
// users who will need to wait for the slow copying migration are Kiln users.
let slot_clock = if let Some(slot_clock) = get_slot_clock::<T>(&db, &log)? {
slot_clock
} else {
error!(
log,
"Unable to complete migration because genesis state or genesis block is missing"
);
return Err(Error::SlotClockUnavailableForMigration);
};
let current_epoch = if let Some(slot) = slot_clock.now() {
slot.epoch(T::EthSpec::slots_per_epoch())
} else {
return Ok(());
};
let bellatrix_fork_epoch = if let Some(fork_epoch) = db.get_chain_spec().bellatrix_fork_epoch {
fork_epoch
} else {
info!(
log,
"Upgrading database schema to v9 (no-op)";
"info" => "To downgrade before the merge run `lighthouse db migrate`"
);
return Ok(());
};
if current_epoch >= bellatrix_fork_epoch {
info!(
log,
"Upgrading database schema to v9";
"info" => "This will take several minutes. Each block will be read from and \
re-written to the database. You may safely exit now (Ctrl-C) and resume \
the migration later. Downgrading is no longer possible."
);
for res in db.hot_db.iter_column_keys(DBColumn::BeaconBlock) {
let block_root = res?;
let block = match db.get_full_block_prior_to_v9(&block_root) {
// A pre-v9 block is present.
Ok(Some(block)) => block,
// A block is missing.
Ok(None) => return Err(Error::BlockNotFound(block_root)),
// There was an error reading a pre-v9 block. Try reading it as a post-v9 block.
Err(_) => {
if db.try_get_full_block(&block_root)?.is_some() {
// The block is present as a post-v9 block, assume that it was already
// correctly migrated.
continue;
} else {
// This scenario should not be encountered since a prior check has ensured
// that this block exists.
return Err(Error::V9MigrationFailure(block_root));
}
}
};
if block.message().execution_payload().is_ok() {
// Overwrite block with blinded block and store execution payload separately.
debug!(
log,
"Rewriting Bellatrix block";
"block_root" => ?block_root,
);
let mut kv_batch = Vec::with_capacity(OPS_PER_BLOCK_WRITE);
db.block_as_kv_store_ops(&block_root, block, &mut kv_batch)?;
db.hot_db.do_atomically(kv_batch)?;
}
}
} else {
info!(
log,
"Upgrading database schema to v9 (no-op)";
"info" => "To downgrade before the merge run `lighthouse db migrate`"
);
}
Ok(())
}
// This downgrade is conditional and will only succeed if the Bellatrix fork epoch hasn't been
// reached.
pub fn downgrade_from_v9<T: BeaconChainTypes>(
db: Arc<HotColdDB<T::EthSpec, T::HotStore, T::ColdStore>>,
log: Logger,
) -> Result<(), Error> {
let slot_clock = if let Some(slot_clock) = get_slot_clock::<T>(&db, &log)? {
slot_clock
} else {
error!(
log,
"Unable to complete migration because genesis state or genesis block is missing"
);
return Err(Error::SlotClockUnavailableForMigration);
};
let current_epoch = if let Some(slot) = slot_clock.now() {
slot.epoch(T::EthSpec::slots_per_epoch())
} else {
return Ok(());
};
let bellatrix_fork_epoch = if let Some(fork_epoch) = db.get_chain_spec().bellatrix_fork_epoch {
fork_epoch
} else {
info!(
log,
"Downgrading database schema from v9";
"info" => "You need to upgrade to v9 again before the merge"
);
return Ok(());
};
if current_epoch >= bellatrix_fork_epoch {
error!(
log,
"Downgrading from schema v9 after the Bellatrix fork epoch is not supported";
"current_epoch" => current_epoch,
"bellatrix_fork_epoch" => bellatrix_fork_epoch,
"reason" => "You need a v9 schema database to run on a merged version of Prater or \
mainnet. On Kiln, you have to re-sync",
);
Err(Error::ResyncRequiredForExecutionPayloadSeparation)
} else {
Ok(())
}
}

89
beacon_node/beacon_chain/src/state_advance_timer.rs
View File

@@ -25,7 +25,7 @@ use std::sync::{
Arc,
};
use task_executor::TaskExecutor;
use tokio::time::sleep;
use tokio::time::{sleep, sleep_until, Instant};
use types::{AttestationShufflingId, BeaconStateError, EthSpec, Hash256, RelativeEpoch, Slot};
/// If the head slot is more than `MAX_ADVANCE_DISTANCE` from the current slot, then don't perform
@@ -117,8 +117,8 @@ async fn state_advance_timer<T: BeaconChainTypes>(
let slot_duration = slot_clock.slot_duration();
loop {
match beacon_chain.slot_clock.duration_to_next_slot() {
Some(duration) => sleep(duration + (slot_duration / 4) * 3).await,
let duration_to_next_slot = match beacon_chain.slot_clock.duration_to_next_slot() {
Some(duration) => duration,
None => {
error!(log, "Failed to read slot clock");
// If we can't read the slot clock, just wait another slot.
@@ -127,7 +127,45 @@ async fn state_advance_timer<T: BeaconChainTypes>(
}
};
// Only start spawn the state advance task if the lock was previously free.
// Run the state advance 3/4 of the way through the slot (9s on mainnet).
let state_advance_offset = slot_duration / 4;
let state_advance_instant = if duration_to_next_slot > state_advance_offset {
Instant::now() + duration_to_next_slot - state_advance_offset
} else {
// Skip the state advance for the current slot and wait until the next one.
Instant::now() + duration_to_next_slot + slot_duration - state_advance_offset
};
// Run fork choice 23/24s of the way through the slot (11.5s on mainnet).
// We need to run after the state advance, so use the same condition as above.
let fork_choice_offset = slot_duration / 24;
let fork_choice_instant = if duration_to_next_slot > state_advance_offset {
Instant::now() + duration_to_next_slot - fork_choice_offset
} else {
Instant::now() + duration_to_next_slot + slot_duration - fork_choice_offset
};
// Wait for the state advance.
sleep_until(state_advance_instant).await;
// Compute the current slot here at approx 3/4 through the slot. Even though this slot is
// only used by fork choice we need to calculate it here rather than after the state
// advance, in case the state advance flows over into the next slot.
let current_slot = match beacon_chain.slot() {
Ok(slot) => slot,
Err(e) => {
warn!(
log,
"Unable to determine slot in state advance timer";
"error" => ?e
);
// If we can't read the slot clock, just wait another slot.
sleep(slot_duration).await;
continue;
}
};
// Only spawn the state advance task if the lock was previously free.
if !is_running.lock() {
let log = log.clone();
let beacon_chain = beacon_chain.clone();
@@ -175,11 +213,45 @@ async fn state_advance_timer<T: BeaconChainTypes>(
"msg" => "system resources may be overloaded"
)
}
// Run fork choice pre-emptively for the next slot. This processes most of the attestations
// from this slot off the hot path of block verification and production.
// Wait for the fork choice instant (which may already be past).
sleep_until(fork_choice_instant).await;
let log = log.clone();
let beacon_chain = beacon_chain.clone();
let next_slot = current_slot + 1;
executor.spawn_blocking(
move || {
if let Err(e) = beacon_chain.fork_choice_at_slot(next_slot) {
warn!(
log,
"Error updating fork choice for next slot";
"error" => ?e,
"slot" => next_slot,
);
}
// Signal block proposal for the next slot (if it happens to be waiting).
if let Some(tx) = &beacon_chain.fork_choice_signal_tx {
if let Err(e) = tx.notify_fork_choice_complete(next_slot) {
warn!(
log,
"Error signalling fork choice waiter";
"error" => ?e,
"slot" => next_slot,
);
}
}
},
"fork_choice_advance",
);
}
}
fn advance_head<T: BeaconChainTypes>(
beacon_chain: &BeaconChain<T>,
beacon_chain: &Arc<BeaconChain<T>>,
log: &Logger,
) -> Result<(), Error> {
let current_slot = beacon_chain.slot()?;
@@ -200,13 +272,6 @@ fn advance_head<T: BeaconChainTypes>(
}
}
// Run fork choice so we get the latest view of the head.
//
// This is useful since it's quite likely that the last time we ran fork choice was shortly
// after receiving the latest gossip block, but not necessarily after we've received the
// majority of attestations.
beacon_chain.fork_choice()?;
let head_info = beacon_chain.head_info()?;
let head_block_root = head_info.block_root;

156
beacon_node/beacon_chain/src/test_utils.rs
View File

@@ -12,15 +12,12 @@ use crate::{
};
use bls::get_withdrawal_credentials;
use execution_layer::{
test_utils::{
ExecutionBlockGenerator, ExecutionLayerRuntime, MockExecutionLayer, DEFAULT_TERMINAL_BLOCK,
},
test_utils::{ExecutionBlockGenerator, MockExecutionLayer, DEFAULT_TERMINAL_BLOCK},
ExecutionLayer,
};
use futures::channel::mpsc::Receiver;
pub use genesis::{interop_genesis_state, DEFAULT_ETH1_BLOCK_HASH};
use int_to_bytes::int_to_bytes32;
use logging::test_logger;
use merkle_proof::MerkleTree;
use parking_lot::Mutex;
use parking_lot::RwLockWriteGuard;
@@ -38,19 +35,11 @@ use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
use store::{config::StoreConfig, HotColdDB, ItemStore, LevelDB, MemoryStore};
use task_executor::ShutdownReason;
use task_executor::{test_utils::TestRuntime, ShutdownReason};
use tree_hash::TreeHash;
use types::sync_selection_proof::SyncSelectionProof;
pub use types::test_utils::generate_deterministic_keypairs;
use types::{
typenum::U4294967296, Address, AggregateSignature, Attestation, AttestationData,
AttesterSlashing, BeaconBlock, BeaconState, BeaconStateHash, ChainSpec, Checkpoint, Deposit,
DepositData, Domain, Epoch, EthSpec, ForkName, Graffiti, Hash256, IndexedAttestation, Keypair,
ProposerSlashing, PublicKeyBytes, SelectionProof, SignatureBytes, SignedAggregateAndProof,
SignedBeaconBlock, SignedBeaconBlockHash, SignedContributionAndProof, SignedRoot,
SignedVoluntaryExit, Slot, SubnetId, SyncCommittee, SyncCommitteeContribution,
SyncCommitteeMessage, VariableList, VoluntaryExit,
};
use types::{typenum::U4294967296, *};
// 4th September 2019
pub const HARNESS_GENESIS_TIME: u64 = 1_567_552_690;
@@ -69,7 +58,7 @@ pub type BaseHarnessType<TEthSpec, THotStore, TColdStore> =
pub type DiskHarnessType<E> = BaseHarnessType<E, LevelDB<E>, LevelDB<E>>;
pub type EphemeralHarnessType<E> = BaseHarnessType<E, MemoryStore<E>, MemoryStore<E>>;
type BoxedMutator<E, Hot, Cold> = Box<
pub type BoxedMutator<E, Hot, Cold> = Box<
dyn FnOnce(
BeaconChainBuilder<BaseHarnessType<E, Hot, Cold>>,
) -> BeaconChainBuilder<BaseHarnessType<E, Hot, Cold>>,
@@ -156,8 +145,8 @@ pub struct Builder<T: BeaconChainTypes> {
initial_mutator: Option<BoxedMutator<T::EthSpec, T::HotStore, T::ColdStore>>,
store_mutator: Option<BoxedMutator<T::EthSpec, T::HotStore, T::ColdStore>>,
execution_layer: Option<ExecutionLayer>,
execution_layer_runtime: Option<ExecutionLayerRuntime>,
mock_execution_layer: Option<MockExecutionLayer<T::EthSpec>>,
runtime: TestRuntime,
log: Logger,
}
@@ -260,6 +249,9 @@ where
Cold: ItemStore<E>,
{
pub fn new(eth_spec_instance: E) -> Self {
let runtime = TestRuntime::default();
let log = runtime.log.clone();
Self {
eth_spec_instance,
spec: None,
@@ -271,8 +263,8 @@ where
store_mutator: None,
execution_layer: None,
mock_execution_layer: None,
execution_layer_runtime: None,
log: test_logger(),
runtime,
log,
}
}
@@ -335,8 +327,6 @@ where
"execution layer already defined"
);
let el_runtime = ExecutionLayerRuntime::default();
let urls: Vec<SensitiveUrl> = urls
.iter()
.map(|s| SensitiveUrl::parse(*s))
@@ -351,19 +341,19 @@ where
};
let execution_layer = ExecutionLayer::from_config(
config,
el_runtime.task_executor.clone(),
el_runtime.log.clone(),
self.runtime.task_executor.clone(),
self.log.clone(),
)
.unwrap();
self.execution_layer = Some(execution_layer);
self.execution_layer_runtime = Some(el_runtime);
self
}
pub fn mock_execution_layer(mut self) -> Self {
let spec = self.spec.clone().expect("cannot build without spec");
let mock = MockExecutionLayer::new(
self.runtime.task_executor.clone(),
spec.terminal_total_difficulty,
DEFAULT_TERMINAL_BLOCK,
spec.terminal_block_hash,
@@ -388,7 +378,7 @@ where
pub fn build(self) -> BeaconChainHarness<BaseHarnessType<E, Hot, Cold>> {
let (shutdown_tx, shutdown_receiver) = futures::channel::mpsc::channel(1);
let log = test_logger();
let log = self.log;
let spec = self.spec.expect("cannot build without spec");
let seconds_per_slot = spec.seconds_per_slot;
let validator_keypairs = self
@@ -400,6 +390,7 @@ where
.custom_spec(spec)
.store(self.store.expect("cannot build without store"))
.store_migrator_config(MigratorConfig::default().blocking())
.task_executor(self.runtime.task_executor.clone())
.execution_layer(self.execution_layer)
.dummy_eth1_backend()
.expect("should build dummy backend")
@@ -439,8 +430,8 @@ where
chain: Arc::new(chain),
validator_keypairs,
shutdown_receiver: Arc::new(Mutex::new(shutdown_receiver)),
runtime: self.runtime,
mock_execution_layer: self.mock_execution_layer,
execution_layer_runtime: self.execution_layer_runtime,
rng: make_rng(),
}
}
@@ -456,9 +447,9 @@ pub struct BeaconChainHarness<T: BeaconChainTypes> {
pub chain: Arc<BeaconChain<T>>,
pub spec: ChainSpec,
pub shutdown_receiver: Arc<Mutex<Receiver<ShutdownReason>>>,
pub runtime: TestRuntime,
pub mock_execution_layer: Option<MockExecutionLayer<T::EthSpec>>,
pub execution_layer_runtime: Option<ExecutionLayerRuntime>,
pub rng: Mutex<StdRng>,
}
@@ -533,8 +524,11 @@ where
self.chain.slot().unwrap()
}
pub fn get_block(&self, block_hash: SignedBeaconBlockHash) -> Option<SignedBeaconBlock<E>> {
self.chain.get_block(&block_hash.into()).unwrap()
pub fn get_block(
&self,
block_hash: SignedBeaconBlockHash,
) -> Option<SignedBeaconBlock<E, BlindedPayload<E>>> {
self.chain.get_blinded_block(&block_hash.into()).unwrap()
}
pub fn block_exists(&self, block_hash: SignedBeaconBlockHash) -> bool {
@@ -590,18 +584,7 @@ where
// different blocks each time.
let graffiti = Graffiti::from(self.rng.lock().gen::<[u8; 32]>());
let randao_reveal = {
let epoch = slot.epoch(E::slots_per_epoch());
let domain = self.spec.get_domain(
epoch,
Domain::Randao,
&state.fork(),
state.genesis_validators_root(),
);
let message = epoch.signing_root(domain);
let sk = &self.validator_keypairs[proposer_index].sk;
sk.sign(message)
};
let randao_reveal = self.sign_randao_reveal(&state, proposer_index, slot);
let (block, state) = self
.chain
@@ -649,18 +632,7 @@ where
// different blocks each time.
let graffiti = Graffiti::from(self.rng.lock().gen::<[u8; 32]>());
let randao_reveal = {
let epoch = slot.epoch(E::slots_per_epoch());
let domain = self.spec.get_domain(
epoch,
Domain::Randao,
&state.fork(),
state.genesis_validators_root(),
);
let message = epoch.signing_root(domain);
let sk = &self.validator_keypairs[proposer_index].sk;
sk.sign(message)
};
let randao_reveal = self.sign_randao_reveal(&state, proposer_index, slot);
let pre_state = state.clone();
@@ -686,6 +658,84 @@ where
(signed_block, pre_state)
}
/// Create a randao reveal for a block at `slot`.
pub fn sign_randao_reveal(
&self,
state: &BeaconState<E>,
proposer_index: usize,
slot: Slot,
) -> Signature {
let epoch = slot.epoch(E::slots_per_epoch());
let domain = self.spec.get_domain(
epoch,
Domain::Randao,
&state.fork(),
state.genesis_validators_root(),
);
let message = epoch.signing_root(domain);
let sk = &self.validator_keypairs[proposer_index].sk;
sk.sign(message)
}
/// Produces an "unaggregated" attestation for the given `slot` and `index` that attests to
/// `beacon_block_root`. The provided `state` should match the `block.state_root` for the
/// `block` identified by `beacon_block_root`.
///
/// The attestation doesn't _really_ have anything about it that makes it unaggregated per say,
/// however this function is only required in the context of forming an unaggregated
/// attestation. It would be an (undetectable) violation of the protocol to create a
/// `SignedAggregateAndProof` based upon the output of this function.
///
/// This function will produce attestations to optimistic blocks, which is against the
/// specification but useful during testing.
pub fn produce_unaggregated_attestation_for_block(
&self,
slot: Slot,
index: CommitteeIndex,
beacon_block_root: Hash256,
mut state: Cow<BeaconState<E>>,
state_root: Hash256,
) -> Result<Attestation<E>, BeaconChainError> {
let epoch = slot.epoch(E::slots_per_epoch());
if state.slot() > slot {
return Err(BeaconChainError::CannotAttestToFutureState);
} else if state.current_epoch() < epoch {
let mut_state = state.to_mut();
complete_state_advance(
mut_state,
Some(state_root),
epoch.start_slot(E::slots_per_epoch()),
&self.spec,
)?;
mut_state.build_committee_cache(RelativeEpoch::Current, &self.spec)?;
}
let committee_len = state.get_beacon_committee(slot, index)?.committee.len();
let target_slot = epoch.start_slot(E::slots_per_epoch());
let target_root = if state.slot() <= target_slot {
beacon_block_root
} else {
*state.get_block_root(target_slot)?
};
Ok(Attestation {
aggregation_bits: BitList::with_capacity(committee_len)?,
data: AttestationData {
slot,
index,
beacon_block_root,
source: state.current_justified_checkpoint(),
target: Checkpoint {
epoch,
root: target_root,
},
},
signature: AggregateSignature::empty(),
})
}
/// A list of attestations for each committee for the given slot.
///
/// The first layer of the Vec is organised per committee. For example, if the return value is
@@ -717,7 +767,6 @@ where
return None;
}
let mut attestation = self
.chain
.produce_unaggregated_attestation_for_block(
attestation_slot,
bc.index,
@@ -900,6 +949,7 @@ where
let aggregate = self
.chain
.get_aggregated_attestation(&attestation.data)
.unwrap()
.unwrap_or_else(|| {
committee_attestations.iter().skip(1).fold(
attestation.clone(),

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

@@ -1,11 +1,8 @@
use crate::errors::BeaconChainError;
use crate::{BeaconChainTypes, BeaconStore};
use ssz::{Decode, DecodeError, Encode};
use ssz::{Decode, Encode};
use std::collections::HashMap;
use std::convert::TryInto;
use std::fs::File;
use std::io::{self, Read, Write};
use std::path::Path;
use store::{DBColumn, Error as StoreError, StoreItem};
use types::{BeaconState, Hash256, PublicKey, PublicKeyBytes};
@@ -24,15 +21,7 @@ pub struct ValidatorPubkeyCache<T: BeaconChainTypes> {
pubkeys: Vec<PublicKey>,
indices: HashMap<PublicKeyBytes, usize>,
pubkey_bytes: Vec<PublicKeyBytes>,
backing: PubkeyCacheBacking<T>,
}
/// Abstraction over on-disk backing.
///
/// `File` backing is legacy, `Database` is current.
enum PubkeyCacheBacking<T: BeaconChainTypes> {
File(ValidatorPubkeyCacheFile),
Database(BeaconStore<T>),
store: BeaconStore<T>,
}
impl<T: BeaconChainTypes> ValidatorPubkeyCache<T> {
@@ -48,7 +37,7 @@ impl<T: BeaconChainTypes> ValidatorPubkeyCache<T> {
pubkeys: vec![],
indices: HashMap::new(),
pubkey_bytes: vec![],
backing: PubkeyCacheBacking::Database(store),
store,
};
cache.import_new_pubkeys(state)?;
@@ -66,7 +55,9 @@ impl<T: BeaconChainTypes> ValidatorPubkeyCache<T> {
if let Some(DatabasePubkey(pubkey)) =
store.get_item(&DatabasePubkey::key_for_index(validator_index))?
{
pubkeys.push((&pubkey).try_into().map_err(Error::PubkeyDecode)?);
pubkeys.push((&pubkey).try_into().map_err(|e| {
BeaconChainError::ValidatorPubkeyCacheError(format!("{:?}", e))
})?);
pubkey_bytes.push(pubkey);
indices.insert(pubkey, validator_index);
} else {
@@ -78,31 +69,10 @@ impl<T: BeaconChainTypes> ValidatorPubkeyCache<T> {
pubkeys,
indices,
pubkey_bytes,
backing: PubkeyCacheBacking::Database(store),
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()),
pubkey_bytes: Vec::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`.
@@ -148,14 +118,8 @@ impl<T: BeaconChainTypes> ValidatorPubkeyCache<T> {
// 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.
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.store
.put_item(&DatabasePubkey::key_for_index(i), &DatabasePubkey(pubkey))?;
self.pubkeys.push(
(&pubkey)
@@ -221,105 +185,6 @@ impl DatabasePubkey {
}
}
/// Allows for maintaining an on-disk copy of the `ValidatorPubkeyCache`. The file is raw SSZ bytes
/// (not ASCII encoded).
///
/// ## Writes
///
/// Each entry is simply appended to the file.
///
/// ## Reads
///
/// The whole file is parsed as an SSZ "variable list" of objects.
///
/// This parsing method is possible because the items in the list are fixed-length SSZ objects.
struct ValidatorPubkeyCacheFile(File);
#[derive(Debug)]
enum Error {
Io(io::Error),
Ssz(DecodeError),
PubkeyDecode(bls::Error),
/// The file read from disk does not have a contiguous list of validator public keys. The file
/// has become corrupted.
InconsistentIndex {
_expected: Option<usize>,
_found: usize,
},
}
impl From<Error> for BeaconChainError {
fn from(e: Error) -> BeaconChainError {
BeaconChainError::ValidatorPubkeyCacheFileError(format!("{:?}", e))
}
}
impl ValidatorPubkeyCacheFile {
/// Opens an existing file for reading and writing.
pub fn open<P: AsRef<Path>>(path: P) -> Result<Self, Error> {
File::options()
.read(true)
.write(true)
.create(false)
.append(true)
.open(path)
.map(Self)
.map_err(Error::Io)
}
/// Append a public key to file.
///
/// The provided `index` should each be one greater than the previous and start at 0.
/// Otherwise, the file will become corrupted and unable to be converted into a cache .
pub fn append(&mut self, index: usize, pubkey: &PublicKeyBytes) -> Result<(), Error> {
append_to_file(&mut self.0, index, pubkey)
}
/// Creates a `ValidatorPubkeyCache` by reading and parsing the underlying file.
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)?;
let list: Vec<(usize, PublicKeyBytes)> = Vec::from_ssz_bytes(&bytes).map_err(Error::Ssz)?;
let mut last = None;
let mut pubkeys = Vec::with_capacity(list.len());
let mut indices = HashMap::with_capacity(list.len());
let mut pubkey_bytes = Vec::with_capacity(list.len());
for (index, pubkey) in list {
let expected = last.map(|n| n + 1);
if expected.map_or(true, |expected| index == expected) {
last = Some(index);
pubkeys.push((&pubkey).try_into().map_err(Error::PubkeyDecode)?);
pubkey_bytes.push(pubkey);
indices.insert(pubkey, index);
} else {
return Err(Error::InconsistentIndex {
_expected: expected,
_found: index,
});
}
}
Ok(ValidatorPubkeyCache {
pubkeys,
indices,
pubkey_bytes,
backing: PubkeyCacheBacking::File(self),
})
}
}
fn append_to_file(file: &mut File, index: usize, pubkey: &PublicKeyBytes) -> Result<(), Error> {
let mut line = Vec::with_capacity(index.ssz_bytes_len() + pubkey.ssz_bytes_len());
index.ssz_append(&mut line);
pubkey.ssz_append(&mut line);
file.write_all(&line).map_err(Error::Io)
}
#[cfg(test)]
mod test {
use super::*;
@@ -327,10 +192,7 @@ mod test {
use logging::test_logger;
use std::sync::Arc;
use store::HotColdDB;
use tempfile::tempdir;
use types::{
test_utils::generate_deterministic_keypair, BeaconState, EthSpec, Keypair, MainnetEthSpec,
};
use types::{BeaconState, EthSpec, Keypair, MainnetEthSpec};
type E = MainnetEthSpec;
type T = EphemeralHarnessType<E>;
@@ -424,7 +286,7 @@ mod test {
check_cache_get(&cache, &keypairs[..]);
drop(cache);
// Re-init the cache from the file.
// Re-init the cache from the store.
let mut cache =
ValidatorPubkeyCache::load_from_store(store.clone()).expect("should open cache");
check_cache_get(&cache, &keypairs[..]);
@@ -437,36 +299,8 @@ mod test {
check_cache_get(&cache, &keypairs[..]);
drop(cache);
// Re-init the cache from the file.
// Re-init the cache from the store.
let cache = ValidatorPubkeyCache::load_from_store(store).expect("should open cache");
check_cache_get(&cache, &keypairs[..]);
}
#[test]
fn invalid_persisted_file() {
let dir = tempdir().expect("should create tempdir");
let path = dir.path().join("cache.ssz");
let pubkey = generate_deterministic_keypair(0).pk.into();
let mut file = File::create(&path).expect("should create file");
append_to_file(&mut file, 0, &pubkey).expect("should write to file");
drop(file);
let cache = ValidatorPubkeyCache::<T>::load_from_file(&path).expect("should open cache");
drop(cache);
let mut file = File::options()
.write(true)
.append(true)
.open(&path)
.expect("should open file");
append_to_file(&mut file, 42, &pubkey).expect("should write bad data to file");
drop(file);
assert!(
ValidatorPubkeyCache::<T>::load_from_file(&path).is_err(),
"should not parse invalid file"
);
}
}