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

Merge pull request #4719 from jimmygchen/deneb-merge-from-unstable-20230911

Browse Source 
Deneb merge from unstable 20230911
This commit is contained in:
realbigsean
2023-09-12 17:41:42 -04:00
committed by GitHub
parent 58cd50681c d4aedab21f
commit 8e7b57a794
73 changed files with 2362 additions and 999 deletions
Download Patch File Download Diff File Expand all files Collapse all files
beacon_node/Cargo.toml
+14 -6
View File
@@ -1,7 +1,10 @@
[package]
name = "beacon_node"
version = "4.3.0"
authors = ["Paul Hauner <paul@paulhauner.com>", "Age Manning <Age@AgeManning.com"]
version = "4.4.1"
authors = [
"Paul Hauner <paul@paulhauner.com>",
"Age Manning <Age@AgeManning.com",
]
edition = "2021"
[lib]
@@ -12,7 +15,9 @@ path = "src/lib.rs"
node_test_rig = { path = "../testing/node_test_rig" }
[features]
write_ssz_files = ["beacon_chain/write_ssz_files"] # Writes debugging .ssz files to /tmp during block processing.
write_ssz_files = [
"beacon_chain/write_ssz_files",
] # Writes debugging .ssz files to /tmp during block processing.
[dependencies]
eth2_config = { path = "../common/eth2_config" }
@@ -21,9 +26,12 @@ types = { path = "../consensus/types" }
store = { path = "./store" }
client = { path = "client" }
clap = "2.33.3"
slog = { version = "2.5.2", features = ["max_level_trace", "release_max_level_trace"] }
slog = { version = "2.5.2", features = [
"max_level_trace",
"release_max_level_trace",
] }
dirs = "3.0.1"
directory = {path = "../common/directory"}
directory = { path = "../common/directory" }
futures = "0.3.7"
environment = { path = "../lighthouse/environment" }
task_executor = { path = "../common/task_executor" }
@@ -42,4 +50,4 @@ monitoring_api = { path = "../common/monitoring_api" }
sensitive_url = { path = "../common/sensitive_url" }
http_api = { path = "http_api" }
unused_port = { path = "../common/unused_port" }
strum = "0.24.1"
strum = "0.24.1"
beacon_node/beacon_chain/src/builder.rs
+15
View File
@@ -485,6 +485,21 @@ where
let (_, updated_builder) = self.set_genesis_state(genesis_state)?;
self = updated_builder;
// Fill in the linear block roots between the checkpoint block's slot and the aligned
// state's slot. All slots less than the block's slot will be handled by block backfill,
// while states greater or equal to the checkpoint state will be handled by `migrate_db`.
let block_root_batch = store
.store_frozen_block_root_at_skip_slots(
weak_subj_block.slot(),
weak_subj_state.slot(),
weak_subj_block_root,
)
.map_err(|e| format!("Error writing frozen block roots: {e:?}"))?;
store
.cold_db
.do_atomically(block_root_batch)
.map_err(|e| format!("Error writing frozen block roots: {e:?}"))?;
// Write the state and block non-atomically, it doesn't matter if they're forgotten
// about on a crash restart.
store
beacon_node/beacon_chain/tests/store_tests.rs
+1 -1
View File
@@ -433,7 +433,7 @@ async fn forwards_iter_block_and_state_roots_until() {
// The last restore point slot is the point at which the hybrid forwards iterator behaviour
// changes.
let last_restore_point_slot = store.get_latest_restore_point_slot();
let last_restore_point_slot = store.get_latest_restore_point_slot().unwrap();
assert!(last_restore_point_slot > 0);
let chain = &harness.chain;
beacon_node/client/src/builder.rs
+32 -21
View File
@@ -157,6 +157,7 @@ where
let runtime_context =
runtime_context.ok_or("beacon_chain_start_method requires a runtime context")?;
let context = runtime_context.service_context("beacon".into());
let log = context.log();
let spec = chain_spec.ok_or("beacon_chain_start_method requires a chain spec")?;
let event_handler = if self.http_api_config.enabled {
Some(ServerSentEventHandler::new(
@@ -167,7 +168,7 @@ where
None
};
let execution_layer = if let Some(config) = config.execution_layer {
let execution_layer = if let Some(config) = config.execution_layer.clone() {
let context = runtime_context.service_context("exec".into());
let execution_layer = ExecutionLayer::from_config(
config,
@@ -208,12 +209,6 @@ where
builder
};
let builder = if let Some(trusted_setup) = config.trusted_setup {
builder.trusted_setup(trusted_setup)
} else {
builder
};
let chain_exists = builder.store_contains_beacon_chain().unwrap_or(false);
// If the client is expect to resume but there's no beacon chain in the database,
@@ -258,23 +253,19 @@ where
)?;
builder.genesis_state(genesis_state).map(|v| (v, None))?
}
ClientGenesis::SszBytes {
genesis_state_bytes,
} => {
ClientGenesis::GenesisState => {
info!(
context.log(),
"Starting from known genesis state";
);
let genesis_state = BeaconState::from_ssz_bytes(&genesis_state_bytes, &spec)
.map_err(|e| format!("Unable to parse genesis state SSZ: {:?}", e))?;
let genesis_state = genesis_state(&runtime_context, &config, log)?;
builder.genesis_state(genesis_state).map(|v| (v, None))?
}
ClientGenesis::WeakSubjSszBytes {
anchor_state_bytes,
anchor_block_bytes,
genesis_state_bytes,
} => {
info!(context.log(), "Starting checkpoint sync");
if config.chain.genesis_backfill {
@@ -288,17 +279,13 @@ where
.map_err(|e| format!("Unable to parse weak subj state SSZ: {:?}", e))?;
let anchor_block = SignedBeaconBlock::from_ssz_bytes(&anchor_block_bytes, &spec)
.map_err(|e| format!("Unable to parse weak subj block SSZ: {:?}", e))?;
let genesis_state = BeaconState::from_ssz_bytes(&genesis_state_bytes, &spec)
.map_err(|e| format!("Unable to parse genesis state SSZ: {:?}", e))?;
let genesis_state = genesis_state(&runtime_context, &config, log)?;
builder
.weak_subjectivity_state(anchor_state, anchor_block, genesis_state)
.map(|v| (v, None))?
}
ClientGenesis::CheckpointSyncUrl {
genesis_state_bytes,
url,
} => {
ClientGenesis::CheckpointSyncUrl { url } => {
info!(
context.log(),
"Starting checkpoint sync";
@@ -393,8 +380,7 @@ where
debug!(context.log(), "Downloaded finalized block");
let genesis_state = BeaconState::from_ssz_bytes(&genesis_state_bytes, &spec)
.map_err(|e| format!("Unable to parse genesis state SSZ: {:?}", e))?;
let genesis_state = genesis_state(&runtime_context, &config, log)?;
info!(
context.log(),
@@ -525,6 +511,12 @@ where
ClientGenesis::FromStore => builder.resume_from_db().map(|v| (v, None))?,
};
let beacon_chain_builder = if let Some(trusted_setup) = config.trusted_setup {
beacon_chain_builder.trusted_setup(trusted_setup)
} else {
beacon_chain_builder
};
if config.sync_eth1_chain {
self.eth1_service = eth1_service_option;
}
@@ -1105,3 +1097,22 @@ where
Ok(self)
}
}
/// Obtain the genesis state from the `eth2_network_config` in `context`.
fn genesis_state<T: EthSpec>(
context: &RuntimeContext<T>,
config: &ClientConfig,
log: &Logger,
) -> Result<BeaconState<T>, String> {
let eth2_network_config = context
.eth2_network_config
.as_ref()
.ok_or("An eth2_network_config is required to obtain the genesis state")?;
eth2_network_config
.genesis_state::<T>(
config.genesis_state_url.as_deref(),
config.genesis_state_url_timeout,
log,
)?
.ok_or_else(|| "Genesis state is unknown".to_string())
}
beacon_node/client/src/config.rs
+8 -7
View File
@@ -8,6 +8,7 @@ use sensitive_url::SensitiveUrl;
use serde_derive::{Deserialize, Serialize};
use std::fs;
use std::path::PathBuf;
use std::time::Duration;
use types::{Graffiti, PublicKeyBytes};
/// Default directory name for the freezer database under the top-level data dir.
const DEFAULT_FREEZER_DB_DIR: &str = "freezer_db";
@@ -26,18 +27,13 @@ pub enum ClientGenesis {
/// contract.
#[default]
DepositContract,
/// Loads the genesis state from SSZ-encoded `BeaconState` bytes.
///
/// We include the bytes instead of the `BeaconState<E>` because the `EthSpec` type
/// parameter would be very annoying.
SszBytes { genesis_state_bytes: Vec<u8> },
/// Loads the genesis state from the genesis state in the `Eth2NetworkConfig`.
GenesisState,
WeakSubjSszBytes {
genesis_state_bytes: Vec<u8>,
anchor_state_bytes: Vec<u8>,
anchor_block_bytes: Vec<u8>,
},
CheckpointSyncUrl {
genesis_state_bytes: Vec<u8>,
url: SensitiveUrl,
},
}
@@ -85,6 +81,8 @@ pub struct Config {
pub slasher: Option<slasher::Config>,
pub logger_config: LoggerConfig,
pub beacon_processor: BeaconProcessorConfig,
pub genesis_state_url: Option<String>,
pub genesis_state_url_timeout: Duration,
}
impl Default for Config {
@@ -114,6 +112,9 @@ impl Default for Config {
validator_monitor_individual_tracking_threshold: DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD,
logger_config: LoggerConfig::default(),
beacon_processor: <_>::default(),
genesis_state_url: <_>::default(),
// This default value should always be overwritten by the CLI default value.
genesis_state_url_timeout: Duration::from_secs(60),
}
}
}
beacon_node/execution_layer/Cargo.toml
+1 -1
View File
@@ -52,4 +52,4 @@ keccak-hash = "0.10.0"
hash256-std-hasher = "0.15.2"
triehash = "0.8.4"
hash-db = "0.15.2"
pretty_reqwest_error = { path = "../../common/pretty_reqwest_error" }
pretty_reqwest_error = { path = "../../common/pretty_reqwest_error" }
beacon_node/http_api/src/lib.rs
+79 -95
View File
@@ -39,7 +39,8 @@ use bytes::Bytes;
use directory::DEFAULT_ROOT_DIR;
use eth2::types::{
self as api_types, BroadcastValidation, EndpointVersion, ForkChoice, ForkChoiceNode,
SignedBlockContents, SkipRandaoVerification, ValidatorId, ValidatorStatus,
SignedBlindedBlockContents, SignedBlockContents, SkipRandaoVerification, ValidatorId,
ValidatorStatus,
};
use lighthouse_network::{types::SyncState, EnrExt, NetworkGlobals, PeerId, PubsubMessage};
use lighthouse_version::version_with_platform;
@@ -139,6 +140,8 @@ pub struct Config {
pub data_dir: PathBuf,
pub sse_capacity_multiplier: usize,
pub enable_beacon_processor: bool,
#[serde(with = "eth2::types::serde_status_code")]
pub duplicate_block_status_code: StatusCode,
}
impl Default for Config {
@@ -154,6 +157,7 @@ impl Default for Config {
data_dir: PathBuf::from(DEFAULT_ROOT_DIR),
sse_capacity_multiplier: 1,
enable_beacon_processor: true,
duplicate_block_status_code: StatusCode::ACCEPTED,
}
}
}
@@ -510,6 +514,8 @@ pub fn serve<T: BeaconChainTypes>(
let task_spawner_filter =
warp::any().map(move || TaskSpawner::new(beacon_processor_send.clone()));
let duplicate_block_status_code = ctx.config.duplicate_block_status_code;
/*
*
* Start of HTTP method definitions.
@@ -1284,11 +1290,11 @@ pub fn serve<T: BeaconChainTypes>(
.and(network_tx_filter.clone())
.and(log_filter.clone())
.then(
|block_contents: SignedBlockContents<T::EthSpec>,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
move |block_contents: SignedBlockContents<T::EthSpec>,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
task_spawner.spawn_async_with_rejection(Priority::P0, async move {
publish_blocks::publish_block(
None,
@@ -1297,9 +1303,9 @@ pub fn serve<T: BeaconChainTypes>(
&network_tx,
log,
BroadcastValidation::default(),
duplicate_block_status_code,
)
.await
.map(|()| warp::reply().into_response())
})
},
);
@@ -1314,11 +1320,11 @@ pub fn serve<T: BeaconChainTypes>(
.and(network_tx_filter.clone())
.and(log_filter.clone())
.then(
|block_bytes: Bytes,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
move |block_bytes: Bytes,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
task_spawner.spawn_async_with_rejection(Priority::P0, async move {
let block_contents = SignedBlockContents::<T::EthSpec>::from_ssz_bytes(
&block_bytes,
@@ -1334,9 +1340,9 @@ pub fn serve<T: BeaconChainTypes>(
&network_tx,
log,
BroadcastValidation::default(),
duplicate_block_status_code,
)
.await
.map(|()| warp::reply().into_response())
})
},
);
@@ -1352,12 +1358,12 @@ pub fn serve<T: BeaconChainTypes>(
.and(network_tx_filter.clone())
.and(log_filter.clone())
.then(
|validation_level: api_types::BroadcastValidationQuery,
block_contents: SignedBlockContents<T::EthSpec>,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
move |validation_level: api_types::BroadcastValidationQuery,
block_contents: SignedBlockContents<T::EthSpec>,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
task_spawner.spawn_async_with_rejection(Priority::P0, async move {
publish_blocks::publish_block(
None,
@@ -1366,9 +1372,9 @@ pub fn serve<T: BeaconChainTypes>(
&network_tx,
log,
validation_level.broadcast_validation,
duplicate_block_status_code,
)
.await
.map(|()| warp::reply().into_response())
})
},
);
@@ -1384,12 +1390,12 @@ pub fn serve<T: BeaconChainTypes>(
.and(network_tx_filter.clone())
.and(log_filter.clone())
.then(
|validation_level: api_types::BroadcastValidationQuery,
block_bytes: Bytes,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
move |validation_level: api_types::BroadcastValidationQuery,
block_bytes: Bytes,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
task_spawner.spawn_async_with_rejection(Priority::P0, async move {
let block_contents = SignedBlockContents::<T::EthSpec>::from_ssz_bytes(
&block_bytes,
@@ -1405,9 +1411,9 @@ pub fn serve<T: BeaconChainTypes>(
&network_tx,
log,
validation_level.broadcast_validation,
duplicate_block_status_code,
)
.await
.map(|()| warp::reply().into_response())
})
},
);
@@ -1427,11 +1433,11 @@ pub fn serve<T: BeaconChainTypes>(
.and(network_tx_filter.clone())
.and(log_filter.clone())
.then(
|block_contents: SignedBlockContents<T::EthSpec, BlindedPayload<_>>,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
move |block_contents: SignedBlindedBlockContents<T::EthSpec>,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
task_spawner.spawn_async_with_rejection(Priority::P0, async move {
publish_blocks::publish_blinded_block(
block_contents,
@@ -1439,9 +1445,9 @@ pub fn serve<T: BeaconChainTypes>(
&network_tx,
log,
BroadcastValidation::default(),
duplicate_block_status_code,
)
.await
.map(|()| warp::reply().into_response())
})
},
);
@@ -1457,11 +1463,11 @@ pub fn serve<T: BeaconChainTypes>(
.and(network_tx_filter.clone())
.and(log_filter.clone())
.then(
|block_bytes: Bytes,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
move |block_bytes: Bytes,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
task_spawner.spawn_async_with_rejection(Priority::P0, async move {
let block =
SignedBlockContents::<T::EthSpec, BlindedPayload<_>>::from_ssz_bytes(
@@ -1477,9 +1483,9 @@ pub fn serve<T: BeaconChainTypes>(
&network_tx,
log,
BroadcastValidation::default(),
duplicate_block_status_code,
)
.await
.map(|()| warp::reply().into_response())
})
},
);
@@ -1495,32 +1501,22 @@ pub fn serve<T: BeaconChainTypes>(
.and(network_tx_filter.clone())
.and(log_filter.clone())
.then(
|validation_level: api_types::BroadcastValidationQuery,
block_contents: SignedBlockContents<T::EthSpec, BlindedPayload<_>>,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
task_spawner.spawn_async(Priority::P0, async move {
match publish_blocks::publish_blinded_block(
move |validation_level: api_types::BroadcastValidationQuery,
block_contents: SignedBlindedBlockContents<T::EthSpec>,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
task_spawner.spawn_async_with_rejection(Priority::P0, async move {
publish_blocks::publish_blinded_block(
block_contents,
chain,
&network_tx,
log,
validation_level.broadcast_validation,
duplicate_block_status_code,
)
.await
{
Ok(()) => warp::reply().into_response(),
Err(e) => match warp_utils::reject::handle_rejection(e).await {
Ok(reply) => reply.into_response(),
Err(_) => warp::reply::with_status(
StatusCode::INTERNAL_SERVER_ERROR,
eth2::StatusCode::INTERNAL_SERVER_ERROR,
)
.into_response(),
},
}
})
},
);
@@ -1531,48 +1527,36 @@ pub fn serve<T: BeaconChainTypes>(
.and(warp::query::<api_types::BroadcastValidationQuery>())
.and(warp::path::end())
.and(warp::body::bytes())
.and(task_spawner_filter.clone())
.and(chain_filter.clone())
.and(network_tx_filter.clone())
.and(log_filter.clone())
.then(
|validation_level: api_types::BroadcastValidationQuery,
block_bytes: Bytes,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| async move {
let block =
match SignedBlockContents::<T::EthSpec, BlindedPayload<_>>::from_ssz_bytes(
&block_bytes,
&chain.spec,
) {
Ok(data) => data,
Err(_) => {
return warp::reply::with_status(
StatusCode::BAD_REQUEST,
eth2::StatusCode::BAD_REQUEST,
)
.into_response();
}
};
match publish_blocks::publish_blinded_block(
block,
chain,
&network_tx,
log,
validation_level.broadcast_validation,
)
.await
{
Ok(()) => warp::reply().into_response(),
Err(e) => match warp_utils::reject::handle_rejection(e).await {
Ok(reply) => reply.into_response(),
Err(_) => warp::reply::with_status(
StatusCode::INTERNAL_SERVER_ERROR,
eth2::StatusCode::INTERNAL_SERVER_ERROR,
move |validation_level: api_types::BroadcastValidationQuery,
block_bytes: Bytes,
task_spawner: TaskSpawner<T::EthSpec>,
chain: Arc<BeaconChain<T>>,
network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger| {
task_spawner.spawn_async_with_rejection(Priority::P0, async move {
let block =
SignedBlockContents::<T::EthSpec, BlindedPayload<_>>::from_ssz_bytes(
&block_bytes,
&chain.spec,
)
.into_response(),
},
}
.map_err(|e| {
warp_utils::reject::custom_bad_request(format!("invalid SSZ: {e:?}"))
})?;
publish_blocks::publish_blinded_block(
block,
chain,
&network_tx,
log,
validation_level.broadcast_validation,
duplicate_block_status_code,
)
.await
})
},
);
beacon_node/http_api/src/publish_blocks.rs
+35 -17
View File
@@ -1,12 +1,12 @@
use crate::metrics;
use beacon_chain::block_verification_types::AsBlock;
use beacon_chain::block_verification_types::{AsBlock, BlockContentsError};
use beacon_chain::validator_monitor::{get_block_delay_ms, timestamp_now};
use beacon_chain::{
AvailabilityProcessingStatus, BeaconChain, BeaconChainError, BeaconChainTypes, BlockError,
IntoGossipVerifiedBlockContents, NotifyExecutionLayer,
};
use eth2::types::BroadcastValidation;
use eth2::types::{BroadcastValidation, ErrorMessage};
use eth2::types::{FullPayloadContents, SignedBlockContents};
use execution_layer::ProvenancedPayload;
use lighthouse_network::PubsubMessage;
@@ -22,7 +22,8 @@ use types::{
AbstractExecPayload, BeaconBlockRef, BlindedPayload, EthSpec, ExecPayload, ExecutionBlockHash,
ForkName, FullPayload, FullPayloadMerge, Hash256, SignedBeaconBlock, SignedBlobSidecarList,
};
use warp::Rejection;
use warp::http::StatusCode;
use warp::{reply::Response, Rejection, Reply};
pub enum ProvenancedBlock<T: BeaconChainTypes, B: IntoGossipVerifiedBlockContents<T>> {
/// The payload was built using a local EE.
@@ -50,7 +51,8 @@ pub async fn publish_block<T: BeaconChainTypes, B: IntoGossipVerifiedBlockConten
network_tx: &UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger,
validation_level: BroadcastValidation,
) -> Result<(), Rejection> {
duplicate_status_code: StatusCode,
) -> Result<Response, Rejection> {
let seen_timestamp = timestamp_now();
let (block_contents, is_locally_built_block) = match provenanced_block {
@@ -114,12 +116,31 @@ pub async fn publish_block<T: BeaconChainTypes, B: IntoGossipVerifiedBlockConten
let blobs_opt = block_contents.inner_blobs();
/* if we can form a `GossipVerifiedBlock`, we've passed our basic gossip checks */
let (gossip_verified_block, gossip_verified_blobs) = block_contents
.into_gossip_verified_block(&chain)
.map_err(|e| {
warn!(log, "Not publishing block, not gossip verified"; "slot" => slot, "error" => ?e);
warp_utils::reject::custom_bad_request(e.to_string())
})?;
let (gossip_verified_block, gossip_verified_blobs) =
match block_contents.into_gossip_verified_block(&chain) {
Ok(b) => b,
Err(BlockContentsError::BlockError(BlockError::BlockIsAlreadyKnown)) => {
// Allow the status code for duplicate blocks to be overridden based on config.
return Ok(warp::reply::with_status(
warp::reply::json(&ErrorMessage {
code: duplicate_status_code.as_u16(),
message: "duplicate block".to_string(),
stacktraces: vec![],
}),
duplicate_status_code,
)
.into_response());
}
Err(e) => {
warn!(
log,
"Not publishing block - not gossip verified";
"slot" => slot,
"error" => ?e
);
return Err(warp_utils::reject::custom_bad_request(e.to_string()));
}
};
// Clone here, so we can take advantage of the `Arc`. The block in `BlockContents` is not,
// `Arc`'d but blobs are.
@@ -222,8 +243,7 @@ pub async fn publish_block<T: BeaconChainTypes, B: IntoGossipVerifiedBlockConten
if is_locally_built_block {
late_block_logging(&chain, seen_timestamp, block.message(), root, "local", &log)
}
Ok(())
Ok(warp::reply().into_response())
}
Ok(AvailabilityProcessingStatus::MissingComponents(_, block_root)) => {
let msg = format!("Missing parts of block with root {:?}", block_root);
@@ -246,10 +266,6 @@ pub async fn publish_block<T: BeaconChainTypes, B: IntoGossipVerifiedBlockConten
Err(BlockError::Slashable) => Err(warp_utils::reject::custom_bad_request(
"proposal for this slot and proposer has already been seen".to_string(),
)),
Err(BlockError::BlockIsAlreadyKnown) => {
info!(log, "Block from HTTP API already known"; "block" => ?block_root);
Ok(())
}
Err(e) => {
if let BroadcastValidation::Gossip = validation_level {
Err(warp_utils::reject::broadcast_without_import(format!("{e}")))
@@ -276,7 +292,8 @@ pub async fn publish_blinded_block<T: BeaconChainTypes>(
network_tx: &UnboundedSender<NetworkMessage<T::EthSpec>>,
log: Logger,
validation_level: BroadcastValidation,
) -> Result<(), Rejection> {
duplicate_status_code: StatusCode,
) -> Result<Response, Rejection> {
let block_root = block_contents.signed_block().canonical_root();
let full_block: ProvenancedBlock<T, SignedBlockContents<T::EthSpec>> =
reconstruct_block(chain.clone(), block_root, block_contents, log.clone()).await?;
@@ -287,6 +304,7 @@ pub async fn publish_blinded_block<T: BeaconChainTypes>(
network_tx,
log,
validation_level,
duplicate_status_code,
)
.await
}
beacon_node/http_api/src/state_id.rs
+1 -3
View File
@@ -89,9 +89,7 @@ impl StateId {
} else {
// This block is either old and finalized, or recent and unfinalized, so
// it's safe to fallback to the optimistic status of the finalized block.
chain
.canonical_head
.fork_choice_read_lock()
fork_choice
.is_optimistic_or_invalid_block(&hot_summary.latest_block_root)
.map_err(BeaconChainError::ForkChoiceError)
.map_err(warp_utils::reject::beacon_chain_error)?
beacon_node/http_api/src/task_spawner.rs
-40
View File
@@ -159,46 +159,6 @@ impl<E: EthSpec> TaskSpawner<E> {
.and_then(|x| x)
}
}
/// Executes an async task which always returns a `Response`.
pub async fn spawn_async(
self,
priority: Priority,
func: impl Future<Output = Response> + Send + Sync + 'static,
) -> Response {
if let Some(beacon_processor_send) = &self.beacon_processor_send {
// Create a wrapper future that will execute `func` and send the
// result to a channel held by this thread.
let (tx, rx) = oneshot::channel();
let process_fn = async move {
// Await the future, collect the return value.
let func_result = func.await;
// Send the result down the channel. Ignore any failures; the
// send can only fail if the receiver is dropped.
let _ = tx.send(func_result);
};
// Send the function to the beacon processor for execution at some arbitrary time.
let result = send_to_beacon_processor(
beacon_processor_send,
priority,
BlockingOrAsync::Async(Box::pin(process_fn)),
rx,
)
.await;
convert_rejection(result).await
} else {
// There is no beacon processor so spawn a task directly on the
// tokio executor.
tokio::task::spawn(func).await.unwrap_or_else(|e| {
warp::reply::with_status(
warp::reply::json(&format!("Tokio did not execute task: {e:?}")),
eth2::StatusCode::INTERNAL_SERVER_ERROR,
)
.into_response()
})
}
}
}
/// Send a task to the beacon processor and await execution.
beacon_node/http_api/src/test_utils.rs
+2 -8
View File
@@ -21,7 +21,7 @@ use network::{NetworkReceivers, NetworkSenders};
use sensitive_url::SensitiveUrl;
use slog::Logger;
use std::future::Future;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use store::MemoryStore;
@@ -217,15 +217,9 @@ pub async fn create_api_server_on_port<T: BeaconChainTypes>(
let ctx = Arc::new(Context {
config: Config {
enabled: true,
listen_addr: IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
listen_port: port,
allow_origin: None,
tls_config: None,
allow_sync_stalled: false,
data_dir: std::path::PathBuf::from(DEFAULT_ROOT_DIR),
spec_fork_name: None,
sse_capacity_multiplier: 1,
enable_beacon_processor: true,
..Config::default()
},
chain: Some(chain),
network_senders: Some(network_senders),
beacon_node/http_api/tests/broadcast_validation_tests.rs
+14 -11
View File
@@ -214,19 +214,19 @@ pub async fn gossip_full_pass_ssz() {
let slot_b = slot_a + 1;
let state_a = tester.harness.get_current_state();
let ((block, _), _): ((SignedBeaconBlock<E>, _), _) =
tester.harness.make_block(state_a, slot_b).await;
let (block_contents_tuple, _) = tester.harness.make_block(state_a, slot_b).await;
let block_contents = block_contents_tuple.into();
let response: Result<(), eth2::Error> = tester
.client
.post_beacon_blocks_v2_ssz(&block, validation_level)
.post_beacon_blocks_v2_ssz(&block_contents, validation_level)
.await;
assert!(response.is_ok());
assert!(tester
.harness
.chain
.block_is_known_to_fork_choice(&block.canonical_root()));
.block_is_known_to_fork_choice(&block_contents.signed_block().canonical_root()));
}
/// This test checks that a block that is **invalid** from a gossip perspective gets rejected when using `broadcast_validation=consensus`.
@@ -378,13 +378,14 @@ pub async fn consensus_partial_pass_only_consensus() {
/* submit `block_b` which should induce equivocation */
let channel = tokio::sync::mpsc::unbounded_channel();
let publication_result: Result<(), Rejection> = publish_block(
let publication_result = publish_block(
None,
ProvenancedBlock::local(gossip_block_contents_b.unwrap()),
tester.harness.chain.clone(),
&channel.0,
test_logger,
validation_level.unwrap(),
StatusCode::ACCEPTED,
)
.await;
@@ -669,13 +670,14 @@ pub async fn equivocation_consensus_late_equivocation() {
let channel = tokio::sync::mpsc::unbounded_channel();
let publication_result: Result<(), Rejection> = publish_block(
let publication_result = publish_block(
None,
ProvenancedBlock::local(gossip_block_contents_b.unwrap()),
tester.harness.chain,
&channel.0,
test_logger,
validation_level.unwrap(),
StatusCode::ACCEPTED,
)
.await;
@@ -910,19 +912,19 @@ pub async fn blinded_gossip_full_pass_ssz() {
let slot_b = slot_a + 1;
let state_a = tester.harness.get_current_state();
let ((block, _), _): ((SignedBlindedBeaconBlock<E>, _), _) =
tester.harness.make_blinded_block(state_a, slot_b).await;
let (block_contents_tuple, _) = tester.harness.make_blinded_block(state_a, slot_b).await;
let block_contents = block_contents_tuple.into();
let response: Result<(), eth2::Error> = tester
.client
.post_beacon_blinded_blocks_v2_ssz(&block, validation_level)
.post_beacon_blinded_blocks_v2_ssz(&block_contents, validation_level)
.await;
assert!(response.is_ok());
assert!(tester
.harness
.chain
.block_is_known_to_fork_choice(&block.canonical_root()));
.block_is_known_to_fork_choice(&block_contents.signed_block().canonical_root()));
}
/// This test checks that a block that is **invalid** from a gossip perspective gets rejected when using `broadcast_validation=consensus`.
@@ -1335,12 +1337,13 @@ pub async fn blinded_equivocation_consensus_late_equivocation() {
let channel = tokio::sync::mpsc::unbounded_channel();
let publication_result: Result<(), Rejection> = publish_blinded_block(
let publication_result = publish_blinded_block(
SignedBlockContents::new(block_b, blobs_b),
tester.harness.chain,
&channel.0,
test_logger,
validation_level.unwrap(),
StatusCode::ACCEPTED,
)
.await;
beacon_node/http_api/tests/tests.rs
+83 -12
View File
@@ -8,7 +8,7 @@ use eth2::{
mixin::{RequestAccept, ResponseForkName, ResponseOptional},
reqwest::RequestBuilder,
types::{BlockId as CoreBlockId, ForkChoiceNode, StateId as CoreStateId, *},
BeaconNodeHttpClient, Error, Timeouts,
BeaconNodeHttpClient, Error, StatusCode, Timeouts,
};
use execution_layer::test_utils::TestingBuilder;
use execution_layer::test_utils::DEFAULT_BUILDER_THRESHOLD_WEI;
@@ -1330,6 +1330,71 @@ impl ApiTester {
self
}
pub async fn test_post_beacon_blocks_duplicate(self) -> Self {
let block_contents = self
.harness
.make_block(
self.harness.get_current_state(),
self.harness.get_current_slot(),
)
.await
.0
.into();
assert!(self
.client
.post_beacon_blocks(&block_contents)
.await
.is_ok());
let blinded_block_contents = block_contents.clone_as_blinded();
// Test all the POST methods in sequence, they should all behave the same.
let responses = vec![
self.client
.post_beacon_blocks(&block_contents)
.await
.unwrap_err(),
self.client
.post_beacon_blocks_v2(&block_contents, None)
.await
.unwrap_err(),
self.client
.post_beacon_blocks_ssz(&block_contents)
.await
.unwrap_err(),
self.client
.post_beacon_blocks_v2_ssz(&block_contents, None)
.await
.unwrap_err(),
self.client
.post_beacon_blinded_blocks(&blinded_block_contents)
.await
.unwrap_err(),
self.client
.post_beacon_blinded_blocks_v2(&blinded_block_contents, None)
.await
.unwrap_err(),
self.client
.post_beacon_blinded_blocks_ssz(&blinded_block_contents)
.await
.unwrap_err(),
self.client
.post_beacon_blinded_blocks_v2_ssz(&blinded_block_contents, None)
.await
.unwrap_err(),
];
for (i, response) in responses.into_iter().enumerate() {
assert_eq!(
response.status().unwrap(),
StatusCode::ACCEPTED,
"response {i}"
);
}
self
}
pub async fn test_beacon_blocks(self) -> Self {
for block_id in self.interesting_block_ids() {
let expected = block_id
@@ -2591,13 +2656,10 @@ impl ApiTester {
.get_validator_blinded_blocks::<E, Payload>(slot, &randao_reveal, None)
.await
.unwrap()
.data
.deconstruct()
.0;
.data;
let signed_block = block.sign(&sk, &fork, genesis_validators_root, &self.chain.spec);
let signed_block_contents =
SignedBlockContents::<E, Payload>::Block(signed_block.clone());
block.sign(&sk, &fork, genesis_validators_root, &self.chain.spec);
self.client
.post_beacon_blinded_blocks(&signed_block_contents)
@@ -2605,6 +2667,7 @@ impl ApiTester {
.unwrap();
// This converts the generic `Payload` to a concrete type for comparison.
let signed_block = signed_block_contents.deconstruct().0;
let head_block = SignedBeaconBlock::from(signed_block.clone());
assert_eq!(head_block, signed_block);
@@ -2650,23 +2713,23 @@ impl ApiTester {
sk.sign(message).into()
};
let block = self
let block_contents = self
.client
.get_validator_blinded_blocks::<E, Payload>(slot, &randao_reveal, None)
.await
.unwrap()
.data
.deconstruct()
.0;
.data;
let signed_block = block.sign(&sk, &fork, genesis_validators_root, &self.chain.spec);
let signed_block_contents =
block_contents.sign(&sk, &fork, genesis_validators_root, &self.chain.spec);
self.client
.post_beacon_blinded_blocks_ssz(&signed_block)
.post_beacon_blinded_blocks_ssz(&signed_block_contents)
.await
.unwrap();
// This converts the generic `Payload` to a concrete type for comparison.
let signed_block = signed_block_contents.deconstruct().0;
let head_block = SignedBeaconBlock::from(signed_block.clone());
assert_eq!(head_block, signed_block);
@@ -4725,6 +4788,14 @@ async fn post_beacon_blocks_invalid() {
.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn post_beacon_blocks_duplicate() {
ApiTester::new()
.await
.test_post_beacon_blocks_duplicate()
.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn beacon_pools_post_attestations_valid() {
ApiTester::new()
beacon_node/network/src/subnet_service/attestation_subnets.rs
+10 -3
View File
@@ -302,9 +302,16 @@ impl<T: BeaconChainTypes> AttestationService<T> {
/// Gets the long lived subnets the node should be subscribed to during the current epoch and
/// the remaining duration for which they remain valid.
fn recompute_long_lived_subnets_inner(&mut self) -> Result<Duration, ()> {
let current_epoch = self.beacon_chain.epoch().map_err(
|e| error!(self.log, "Failed to get the current epoch from clock"; "err" => ?e),
)?;
let current_epoch = self.beacon_chain.epoch().map_err(|e| {
if !self
.beacon_chain
.slot_clock
.is_prior_to_genesis()
.unwrap_or(false)
{
error!(self.log, "Failed to get the current epoch from clock"; "err" => ?e)
}
})?;
let (subnets, next_subscription_epoch) = SubnetId::compute_subnets_for_epoch::<T::EthSpec>(
self.node_id.raw().into(),
beacon_node/src/cli.rs
+9 -1
View File
@@ -398,6 +398,15 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
.help("Multiplier to apply to the length of HTTP server-sent-event (SSE) channels. \
Increasing this value can prevent messages from being dropped.")
)
.arg(
Arg::with_name("http-duplicate-block-status")
.long("http-duplicate-block-status")
.takes_value(true)
.default_value("202")
.value_name("STATUS_CODE")
.help("Status code to send when a block that is already known is POSTed to the \
HTTP API.")
)
.arg(
Arg::with_name("http-enable-beacon-processor")
.long("http-enable-beacon-processor")
@@ -1187,7 +1196,6 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
.arg(
Arg::with_name("gui")
.long("gui")
.hidden(true)
.help("Enable the graphical user interface and all its requirements. \
This enables --http and --validator-monitor-auto and enables SSE logging.")
.takes_value(false)
beacon_node/src/config.rs
+29 -14
View File
@@ -156,6 +156,9 @@ pub fn get_config<E: EthSpec>(
client_config.http_api.enable_beacon_processor =
parse_required(cli_args, "http-enable-beacon-processor")?;
client_config.http_api.duplicate_block_status_code =
parse_required(cli_args, "http-duplicate-block-status")?;
if let Some(cache_size) = clap_utils::parse_optional(cli_args, "shuffling-cache-size")? {
client_config.chain.shuffling_cache_size = cache_size;
}
@@ -506,9 +509,30 @@ pub fn get_config<E: EthSpec>(
client_config.chain.checkpoint_sync_url_timeout =
clap_utils::parse_required::<u64>(cli_args, "checkpoint-sync-url-timeout")?;
client_config.genesis = if let Some(genesis_state_bytes) =
eth2_network_config.genesis_state_bytes.clone()
{
client_config.genesis_state_url_timeout =
clap_utils::parse_required(cli_args, "genesis-state-url-timeout")
.map(Duration::from_secs)?;
let genesis_state_url_opt =
clap_utils::parse_optional::<String>(cli_args, "genesis-state-url")?;
let checkpoint_sync_url_opt =
clap_utils::parse_optional::<String>(cli_args, "checkpoint-sync-url")?;
// If the `--genesis-state-url` is defined, use that to download the
// genesis state bytes. If it's not defined, try `--checkpoint-sync-url`.
client_config.genesis_state_url = if let Some(genesis_state_url) = genesis_state_url_opt {
Some(genesis_state_url)
} else if let Some(checkpoint_sync_url) = checkpoint_sync_url_opt {
// If the checkpoint sync URL is going to be used to download the
// genesis state, adopt the timeout from the checkpoint sync URL too.
client_config.genesis_state_url_timeout =
Duration::from_secs(client_config.chain.checkpoint_sync_url_timeout);
Some(checkpoint_sync_url)
} else {
None
};
client_config.genesis = if eth2_network_config.genesis_state_is_known() {
// Set up weak subjectivity sync, or start from the hardcoded genesis state.
if let (Some(initial_state_path), Some(initial_block_path)) = (
cli_args.value_of("checkpoint-state"),
@@ -530,7 +554,6 @@ pub fn get_config<E: EthSpec>(
let anchor_block_bytes = read(initial_block_path)?;
ClientGenesis::WeakSubjSszBytes {
genesis_state_bytes,
anchor_state_bytes,
anchor_block_bytes,
}
@@ -538,17 +561,9 @@ pub fn get_config<E: EthSpec>(
let url = SensitiveUrl::parse(remote_bn_url)
.map_err(|e| format!("Invalid checkpoint sync URL: {:?}", e))?;
ClientGenesis::CheckpointSyncUrl {
genesis_state_bytes,
url,
}
ClientGenesis::CheckpointSyncUrl { url }
} else {
// Note: re-serializing the genesis state is not so efficient, however it avoids adding
// trait bounds to the `ClientGenesis` enum. This would have significant flow-on
// effects.
ClientGenesis::SszBytes {
genesis_state_bytes,
}
ClientGenesis::GenesisState
}
} else {
if cli_args.is_present("checkpoint-state") || cli_args.is_present("checkpoint-sync-url") {
beacon_node/store/src/chunked_iter.rs
+4 -4
View File
@@ -30,16 +30,16 @@ where
/// Create a new iterator which can yield elements from `start_vindex` up to the last
/// index stored by the restore point at `last_restore_point_slot`.
///
/// The `last_restore_point` slot should be the slot of a recent restore point as obtained from
/// `HotColdDB::get_latest_restore_point_slot`. We pass it as a parameter so that the caller can
/// The `freezer_upper_limit` slot should be the slot of a recent restore point as obtained from
/// `Root::freezer_upper_limit`. We pass it as a parameter so that the caller can
/// maintain a stable view of the database (see `HybridForwardsBlockRootsIterator`).
pub fn new(
store: &'a HotColdDB<E, Hot, Cold>,
start_vindex: usize,
last_restore_point_slot: Slot,
freezer_upper_limit: Slot,
spec: &ChainSpec,
) -> Self {
let (_, end_vindex) = F::start_and_end_vindex(last_restore_point_slot, spec);
let (_, end_vindex) = F::start_and_end_vindex(freezer_upper_limit, spec);
// Set the next chunk to the one containing `start_vindex`.
let next_cindex = start_vindex / F::chunk_size();
beacon_node/store/src/forwards_iter.rs
+46 -8
View File
@@ -19,6 +19,14 @@ pub trait Root<E: EthSpec>: Field<E, Value = Hash256> {
end_state: BeaconState<E>,
end_root: Hash256,
) -> Result<SimpleForwardsIterator>;
/// The first slot for which this field is *no longer* stored in the freezer database.
///
/// If `None`, then this field is not stored in the freezer database at all due to pruning
/// configuration.
fn freezer_upper_limit<Hot: ItemStore<E>, Cold: ItemStore<E>>(
store: &HotColdDB<E, Hot, Cold>,
) -> Option<Slot>;
}
impl<E: EthSpec> Root<E> for BlockRoots {
@@ -39,6 +47,13 @@ impl<E: EthSpec> Root<E> for BlockRoots {
)?;
Ok(SimpleForwardsIterator { values })
}
fn freezer_upper_limit<Hot: ItemStore<E>, Cold: ItemStore<E>>(
store: &HotColdDB<E, Hot, Cold>,
) -> Option<Slot> {
// Block roots are stored for all slots up to the split slot (exclusive).
Some(store.get_split_slot())
}
}
impl<E: EthSpec> Root<E> for StateRoots {
@@ -59,6 +74,15 @@ impl<E: EthSpec> Root<E> for StateRoots {
)?;
Ok(SimpleForwardsIterator { values })
}
fn freezer_upper_limit<Hot: ItemStore<E>, Cold: ItemStore<E>>(
store: &HotColdDB<E, Hot, Cold>,
) -> Option<Slot> {
// State roots are stored for all slots up to the latest restore point (exclusive).
// There may not be a latest restore point if state pruning is enabled, in which
// case this function will return `None`.
store.get_latest_restore_point_slot()
}
}
/// Forwards root iterator that makes use of a flat field table in the freezer DB.
@@ -118,6 +142,7 @@ impl Iterator for SimpleForwardsIterator {
pub enum HybridForwardsIterator<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>> {
PreFinalization {
iter: Box<FrozenForwardsIterator<'a, E, F, Hot, Cold>>,
end_slot: Option<Slot>,
/// Data required by the `PostFinalization` iterator when we get to it.
continuation_data: Option<Box<(BeaconState<E>, Hash256)>>,
},
@@ -129,6 +154,7 @@ pub enum HybridForwardsIterator<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, C
PostFinalization {
iter: SimpleForwardsIterator,
},
Finished,
}
impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
@@ -138,8 +164,8 @@ impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
///
/// The `get_state` closure should return a beacon state and final block/state root to backtrack
/// from in the case where the iterated range does not lie entirely within the frozen portion of
/// the database. If an `end_slot` is provided and it is before the database's latest restore
/// point slot then the `get_state` closure will not be called at all.
/// the database. If an `end_slot` is provided and it is before the database's freezer upper
/// limit for the field then the `get_state` closure will not be called at all.
///
/// It is OK for `get_state` to hold a lock while this function is evaluated, as the returned
/// iterator is as lazy as possible and won't do any work apart from calling `get_state`.
@@ -155,13 +181,15 @@ impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
) -> Result<Self> {
use HybridForwardsIterator::*;
let latest_restore_point_slot = store.get_latest_restore_point_slot();
// First slot at which this field is *not* available in the freezer. i.e. all slots less
// than this slot have their data available in the freezer.
let freezer_upper_limit = F::freezer_upper_limit(store).unwrap_or(Slot::new(0));
let result = if start_slot < latest_restore_point_slot {
let result = if start_slot < freezer_upper_limit {
let iter = Box::new(FrozenForwardsIterator::new(
store,
start_slot,
latest_restore_point_slot,
freezer_upper_limit,
spec,
));
@@ -169,13 +197,14 @@ impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
// `end_slot`. If it tries to continue further a `NoContinuationData` error will be
// returned.
let continuation_data =
if end_slot.map_or(false, |end_slot| end_slot < latest_restore_point_slot) {
if end_slot.map_or(false, |end_slot| end_slot < freezer_upper_limit) {
None
} else {
Some(Box::new(get_state()?))
};
PreFinalization {
iter,
end_slot,
continuation_data,
}
} else {
@@ -195,6 +224,7 @@ impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
match self {
PreFinalization {
iter,
end_slot,
continuation_data,
} => {
match iter.next() {
@@ -203,10 +233,17 @@ impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
// to a post-finalization iterator beginning from the last slot
// of the pre iterator.
None => {
// If the iterator has an end slot (inclusive) which has already been
// covered by the (exclusive) frozen forwards iterator, then we're done!
let iter_end_slot = Slot::from(iter.inner.end_vindex);
if end_slot.map_or(false, |end_slot| iter_end_slot == end_slot + 1) {
*self = Finished;
return Ok(None);
}
let continuation_data = continuation_data.take();
let store = iter.inner.store;
let start_slot = Slot::from(iter.inner.end_vindex);
let start_slot = iter_end_slot;
*self = PostFinalizationLazy {
continuation_data,
store,
@@ -230,6 +267,7 @@ impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
self.do_next()
}
PostFinalization { iter } => iter.next().transpose(),
Finished => Ok(None),
}
}
}
beacon_node/store/src/hot_cold_store.rs
+66 -18
View File
@@ -18,7 +18,7 @@ use crate::metadata::{
};
use crate::metrics;
use crate::{
get_key_for_col, DBColumn, DatabaseBlock, Error, ItemStore, KeyValueStoreOp,
get_key_for_col, ChunkWriter, DBColumn, DatabaseBlock, Error, ItemStore, KeyValueStoreOp,
PartialBeaconState, StoreItem, StoreOp,
};
use itertools::process_results;
@@ -1195,6 +1195,9 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
ops.push(op);
// 2. Store updated vector entries.
// Block roots need to be written here as well as by the `ChunkWriter` in `migrate_db`
// because states may require older block roots, and the writer only stores block roots
// between the previous split point and the new split point.
let db = &self.cold_db;
store_updated_vector(BlockRoots, db, state, &self.spec, ops)?;
store_updated_vector(StateRoots, db, state, &self.spec, ops)?;
@@ -1497,10 +1500,21 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
};
}
/// Fetch the slot of the most recently stored restore point.
pub fn get_latest_restore_point_slot(&self) -> Slot {
(self.get_split_slot() - 1) / self.config.slots_per_restore_point
* self.config.slots_per_restore_point
/// Fetch the slot of the most recently stored restore point (if any).
pub fn get_latest_restore_point_slot(&self) -> Option<Slot> {
let split_slot = self.get_split_slot();
let anchor = self.get_anchor_info();
// There are no restore points stored if the state upper limit lies in the hot database.
// It hasn't been reached yet, and may never be.
if anchor.map_or(false, |a| a.state_upper_limit >= split_slot) {
None
} else {
Some(
(split_slot - 1) / self.config.slots_per_restore_point
* self.config.slots_per_restore_point,
)
}
}
/// Load the database schema version from disk.
@@ -1907,6 +1921,25 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
)
}
/// Update the linear array of frozen block roots with the block root for several skipped slots.
///
/// Write the block root at all slots from `start_slot` (inclusive) to `end_slot` (exclusive).
pub fn store_frozen_block_root_at_skip_slots(
&self,
start_slot: Slot,
end_slot: Slot,
block_root: Hash256,
) -> Result<Vec<KeyValueStoreOp>, Error> {
let mut ops = vec![];
let mut block_root_writer =
ChunkWriter::<BlockRoots, _, _>::new(&self.cold_db, start_slot.as_usize())?;
for slot in start_slot.as_usize()..end_slot.as_usize() {
block_root_writer.set(slot, block_root, &mut ops)?;
}
block_root_writer.write(&mut ops)?;
Ok(ops)
}
/// Try to prune all execution payloads, returning early if there is no need to prune.
pub fn try_prune_execution_payloads(&self, force: bool) -> Result<(), Error> {
let split = self.get_split_info();
@@ -2203,7 +2236,14 @@ pub fn migrate_database<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>(
return Err(HotColdDBError::FreezeSlotUnaligned(finalized_state.slot()).into());
}
let mut hot_db_ops: Vec<StoreOp<E>> = Vec::new();
let mut hot_db_ops = vec![];
let mut cold_db_ops = vec![];
// Chunk writer for the linear block roots in the freezer DB.
// Start at the new upper limit because we iterate backwards.
let new_frozen_block_root_upper_limit = finalized_state.slot().as_usize().saturating_sub(1);
let mut block_root_writer =
ChunkWriter::<BlockRoots, _, _>::new(&store.cold_db, new_frozen_block_root_upper_limit)?;
// 1. Copy all of the states between the new finalized state and the split slot, from the hot DB
// to the cold DB. Delete the execution payloads of these now-finalized blocks.
@@ -2228,6 +2268,9 @@ pub fn migrate_database<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>(
// Delete the old summary, and the full state if we lie on an epoch boundary.
hot_db_ops.push(StoreOp::DeleteState(state_root, Some(slot)));
// Store the block root for this slot in the linear array of frozen block roots.
block_root_writer.set(slot.as_usize(), block_root, &mut cold_db_ops)?;
// Do not try to store states if a restore point is yet to be stored, or will never be
// stored (see `STATE_UPPER_LIMIT_NO_RETAIN`). Make an exception for the genesis state
// which always needs to be copied from the hot DB to the freezer and should not be deleted.
@@ -2237,29 +2280,34 @@ pub fn migrate_database<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>(
.map_or(false, |anchor| slot < anchor.state_upper_limit)
{
debug!(store.log, "Pruning finalized state"; "slot" => slot);
continue;
}
let mut cold_db_ops: Vec<KeyValueStoreOp> = Vec::new();
if slot % store.config.slots_per_restore_point == 0 {
let state: BeaconState<E> = get_full_state(&store.hot_db, &state_root, &store.spec)?
.ok_or(HotColdDBError::MissingStateToFreeze(state_root))?;
store.store_cold_state(&state_root, &state, &mut cold_db_ops)?;
}
// Store a pointer from this state root to its slot, so we can later reconstruct states
// from their state root alone.
let cold_state_summary = ColdStateSummary { slot };
let op = cold_state_summary.as_kv_store_op(state_root);
cold_db_ops.push(op);
// There are data dependencies between calls to `store_cold_state()` that prevent us from
// doing one big call to `store.cold_db.do_atomically()` at end of the loop.
store.cold_db.do_atomically(cold_db_ops)?;
if slot % store.config.slots_per_restore_point == 0 {
let state: BeaconState<E> = get_full_state(&store.hot_db, &state_root, &store.spec)?
.ok_or(HotColdDBError::MissingStateToFreeze(state_root))?;
store.store_cold_state(&state_root, &state, &mut cold_db_ops)?;
// Commit the batch of cold DB ops whenever a full state is written. Each state stored
// may read the linear fields of previous states stored.
store
.cold_db
.do_atomically(std::mem::take(&mut cold_db_ops))?;
}
}
// Finish writing the block roots and commit the remaining cold DB ops.
block_root_writer.write(&mut cold_db_ops)?;
store.cold_db.do_atomically(cold_db_ops)?;
// Warning: Critical section. We have to take care not to put any of the two databases in an
// inconsistent state if the OS process dies at any point during the freezing
// procedure.