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

Merge pull request #3871 from michaelsproul/capella-v3.4.0

Browse Source 
Back-merge unstable to Capella
This commit is contained in:
Michael Sproul 2023-01-12 16:44:00 +11:00 committed by GitHub
commit b2c2d31b69
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
59 changed files with 2994 additions and 1748 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
.github/workflows/test-suite.yml vendored
View File

@ -12,7 +12,7 @@ env:
# Deny warnings in CI # Deny warnings in CI
RUSTFLAGS: "-D warnings" RUSTFLAGS: "-D warnings"
# The Nightly version used for cargo-udeps, might need updating from time to time. # The Nightly version used for cargo-udeps, might need updating from time to time.
PINNED_NIGHTLY: nightly-2022-05-20 PINNED_NIGHTLY: nightly-2022-12-15
# Prevent Github API rate limiting. # Prevent Github API rate limiting.
LIGHTHOUSE_GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} LIGHTHOUSE_GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
jobs: jobs:
@ -332,27 +332,6 @@ jobs:
run: make lint run: make lint
- name: Certify Cargo.lock freshness - name: Certify Cargo.lock freshness
run: git diff --exit-code Cargo.lock run: git diff --exit-code Cargo.lock
disallowed-from-async-lint:
name: disallowed-from-async-lint
runs-on: ubuntu-latest
needs: cargo-fmt
continue-on-error: true
steps:
- uses: actions/checkout@v3
- name: Install SigP Clippy fork
run: |
cd ..
git clone https://github.com/michaelsproul/rust-clippy.git
cd rust-clippy
git checkout 31a49666ccfcd7963b63345d6ce757c373f22c2a
cargo build --release --bin cargo-clippy --bin clippy-driver
cargo build --release --bin cargo-clippy --bin clippy-driver -Zunstable-options --out-dir $(rustc --print=sysroot)/bin
- name: Install Protoc
uses: arduino/setup-protoc@e52d9eb8f7b63115df1ac544a1376fdbf5a39612
with:
repo-token: ${{ secrets.GITHUB_TOKEN }}
- name: Run Clippy with the disallowed-from-async lint
run: make nightly-lint
check-msrv: check-msrv:
name: check-msrv name: check-msrv
runs-on: ubuntu-latest runs-on: ubuntu-latest

2328
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

2
beacon_node/Cargo.toml
View File

@ -1,6 +1,6 @@
[package] [package]
name = "beacon_node" name = "beacon_node"
version = "3.3.0" version = "3.4.0"
authors = ["Paul Hauner <paul@paulhauner.com>", "Age Manning <Age@AgeManning.com"] authors = ["Paul Hauner <paul@paulhauner.com>", "Age Manning <Age@AgeManning.com"]
edition = "2021" edition = "2021"

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

@ -579,11 +579,13 @@ where
mut self, mut self,
auto_register: bool, auto_register: bool,
validators: Vec<PublicKeyBytes>, validators: Vec<PublicKeyBytes>,
individual_metrics_threshold: usize,
log: Logger, log: Logger,
) -> Self { ) -> Self {
self.validator_monitor = Some(ValidatorMonitor::new( self.validator_monitor = Some(ValidatorMonitor::new(
validators, validators,
auto_register, auto_register,
individual_metrics_threshold,
log.clone(), log.clone(),
)); ));
self self
@ -990,6 +992,7 @@ fn descriptive_db_error(item: &str, error: &StoreError) -> String {
#[cfg(test)] #[cfg(test)]
mod test { mod test {
use super::*; use super::*;
use crate::validator_monitor::DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD;
use eth2_hashing::hash; use eth2_hashing::hash;
use genesis::{ use genesis::{
generate_deterministic_keypairs, interop_genesis_state, DEFAULT_ETH1_BLOCK_HASH, generate_deterministic_keypairs, interop_genesis_state, DEFAULT_ETH1_BLOCK_HASH,
@ -1046,7 +1049,12 @@ mod test {
.testing_slot_clock(Duration::from_secs(1)) .testing_slot_clock(Duration::from_secs(1))
.expect("should configure testing slot clock") .expect("should configure testing slot clock")
.shutdown_sender(shutdown_tx) .shutdown_sender(shutdown_tx)
.monitor_validators(true, vec![], log.clone()) .monitor_validators(
true,
vec![],
DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD,
log.clone(),
)
.build() .build()
.expect("should build"); .expect("should build");

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

@ -65,6 +65,8 @@ pub struct ChainConfig {
/// Low values are useful for execution engines which don't improve their payload after the /// Low values are useful for execution engines which don't improve their payload after the
/// first call, and high values are useful for ensuring the EL is given ample notice. /// first call, and high values are useful for ensuring the EL is given ample notice.
pub prepare_payload_lookahead: Duration, pub prepare_payload_lookahead: Duration,
/// Use EL-free optimistic sync for the finalized part of the chain.
pub optimistic_finalized_sync: bool,
} }
impl Default for ChainConfig { impl Default for ChainConfig {
@ -89,6 +91,9 @@ impl Default for ChainConfig {
count_unrealized_full: CountUnrealizedFull::default(), count_unrealized_full: CountUnrealizedFull::default(),
checkpoint_sync_url_timeout: 60, checkpoint_sync_url_timeout: 60,
prepare_payload_lookahead: Duration::from_secs(4), prepare_payload_lookahead: Duration::from_secs(4),
// TODO(capella): disabled until withdrawal verification is implemented
// See: https://github.com/sigp/lighthouse/issues/3870
optimistic_finalized_sync: false,
} }
} }
} }

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

@ -15,7 +15,7 @@ use crate::{
use execution_layer::{BlockProposalContents, BuilderParams, PayloadAttributes, PayloadStatus}; use execution_layer::{BlockProposalContents, BuilderParams, PayloadAttributes, PayloadStatus};
use fork_choice::{InvalidationOperation, PayloadVerificationStatus}; use fork_choice::{InvalidationOperation, PayloadVerificationStatus};
use proto_array::{Block as ProtoBlock, ExecutionStatus}; use proto_array::{Block as ProtoBlock, ExecutionStatus};
use slog::debug; use slog::{debug, warn};
use slot_clock::SlotClock; use slot_clock::SlotClock;
use state_processing::per_block_processing::{ use state_processing::per_block_processing::{
compute_timestamp_at_slot, get_expected_withdrawals, is_execution_enabled, compute_timestamp_at_slot, get_expected_withdrawals, is_execution_enabled,
@ -60,26 +60,51 @@ impl<T: BeaconChainTypes> PayloadNotifier<T> {
state: &BeaconState<T::EthSpec>, state: &BeaconState<T::EthSpec>,
notify_execution_layer: NotifyExecutionLayer, notify_execution_layer: NotifyExecutionLayer,
) -> Result<Self, BlockError<T::EthSpec>> { ) -> Result<Self, BlockError<T::EthSpec>> {
let payload_verification_status = match notify_execution_layer { let payload_verification_status = if is_execution_enabled(state, block.message().body()) {
NotifyExecutionLayer::No => Some(PayloadVerificationStatus::Optimistic), // Perform the initial stages of payload verification.
NotifyExecutionLayer::Yes => { //
if is_execution_enabled(state, block.message().body()) { // We will duplicate these checks again during `per_block_processing`, however these
// Perform the initial stages of payload verification. // checks are cheap and doing them here ensures we have verified them before marking
// // the block as optimistically imported. This is particularly relevant in the case
// We will duplicate these checks again during `per_block_processing`, however these checks // where we do not send the block to the EL at all.
// are cheap and doing them here ensures we protect the execution engine from junk. let block_message = block.message();
partially_verify_execution_payload::<T::EthSpec, FullPayload<T::EthSpec>>( let payload = block_message.execution_payload()?;
state, partially_verify_execution_payload::<_, FullPayload<_>>(
block.slot(), state,
block.message().execution_payload()?, block.slot(),
&chain.spec, payload,
) &chain.spec,
.map_err(BlockError::PerBlockProcessingError)?; )
None .map_err(BlockError::PerBlockProcessingError)?;
} else {
Some(PayloadVerificationStatus::Irrelevant) match notify_execution_layer {
NotifyExecutionLayer::No if chain.config.optimistic_finalized_sync => {
// Verify the block hash here in Lighthouse and immediately mark the block as
// optimistically imported. This saves a lot of roundtrips to the EL.
let execution_layer = chain
.execution_layer
.as_ref()
.ok_or(ExecutionPayloadError::NoExecutionConnection)?;
if let Err(e) =
execution_layer.verify_payload_block_hash(payload.execution_payload_ref())
{
warn!(
chain.log,
"Falling back to slow block hash verification";
"block_number" => payload.block_number(),
"info" => "you can silence this warning with --disable-optimistic-finalized-sync",
"error" => ?e,
);
None
} else {
Some(PayloadVerificationStatus::Optimistic)
}
} }
_ => None,
} }
} else {
Some(PayloadVerificationStatus::Irrelevant)
}; };
Ok(Self { Ok(Self {

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

@ -2,6 +2,7 @@ pub use crate::persisted_beacon_chain::PersistedBeaconChain;
pub use crate::{ pub use crate::{
beacon_chain::{BEACON_CHAIN_DB_KEY, ETH1_CACHE_DB_KEY, FORK_CHOICE_DB_KEY, OP_POOL_DB_KEY}, beacon_chain::{BEACON_CHAIN_DB_KEY, ETH1_CACHE_DB_KEY, FORK_CHOICE_DB_KEY, OP_POOL_DB_KEY},
migrate::MigratorConfig, migrate::MigratorConfig,
validator_monitor::DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD,
BeaconChainError, NotifyExecutionLayer, ProduceBlockVerification, BeaconChainError, NotifyExecutionLayer, ProduceBlockVerification,
}; };
use crate::{ use crate::{
@ -507,7 +508,7 @@ where
log.clone(), log.clone(),
5, 5,
))) )))
.monitor_validators(true, vec![], log); .monitor_validators(true, vec![], DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD, log);
builder = if let Some(mutator) = self.initial_mutator { builder = if let Some(mutator) = self.initial_mutator {
mutator(builder) mutator(builder)

770
beacon_node/beacon_chain/src/validator_monitor.rs
View File

@ -21,10 +21,21 @@ use types::{
SignedContributionAndProof, Slot, SyncCommitteeMessage, VoluntaryExit, SignedContributionAndProof, Slot, SyncCommitteeMessage, VoluntaryExit,
}; };
/// Used for Prometheus labels.
///
/// We've used `total` for this value to align with Nimbus, as per:
/// https://github.com/sigp/lighthouse/pull/3728#issuecomment-1375173063
const TOTAL_LABEL: &str = "total";
/// The validator monitor collects per-epoch data about each monitored validator. Historical data /// The validator monitor collects per-epoch data about each monitored validator. Historical data
/// will be kept around for `HISTORIC_EPOCHS` before it is pruned. /// will be kept around for `HISTORIC_EPOCHS` before it is pruned.
pub const HISTORIC_EPOCHS: usize = 4; pub const HISTORIC_EPOCHS: usize = 4;
/// Once the validator monitor reaches this number of validators it will stop
/// tracking their metrics/logging individually in an effort to reduce
/// Prometheus cardinality and log volume.
pub const DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD: usize = 64;
#[derive(Debug)] #[derive(Debug)]
pub enum Error { pub enum Error {
InvalidPubkey(String), InvalidPubkey(String),
@ -258,16 +269,27 @@ pub struct ValidatorMonitor<T> {
indices: HashMap<u64, PublicKeyBytes>, indices: HashMap<u64, PublicKeyBytes>,
/// If true, allow the automatic registration of validators. /// If true, allow the automatic registration of validators.
auto_register: bool, auto_register: bool,
/// Once the number of monitored validators goes above this threshold, we
/// will stop tracking metrics/logs on a per-validator basis. This prevents
/// large validator counts causing infeasibly high cardinailty for
/// Prometheus and high log volumes.
individual_tracking_threshold: usize,
log: Logger, log: Logger,
_phantom: PhantomData<T>, _phantom: PhantomData<T>,
} }
impl<T: EthSpec> ValidatorMonitor<T> { impl<T: EthSpec> ValidatorMonitor<T> {
pub fn new(pubkeys: Vec<PublicKeyBytes>, auto_register: bool, log: Logger) -> Self { pub fn new(
pubkeys: Vec<PublicKeyBytes>,
auto_register: bool,
individual_tracking_threshold: usize,
log: Logger,
) -> Self {
let mut s = Self { let mut s = Self {
validators: <_>::default(), validators: <_>::default(),
indices: <_>::default(), indices: <_>::default(),
auto_register, auto_register,
individual_tracking_threshold,
log, log,
_phantom: PhantomData, _phantom: PhantomData,
}; };
@ -277,6 +299,13 @@ impl<T: EthSpec> ValidatorMonitor<T> {
s s
} }
/// Returns `true` when the validator count is sufficiently low enough to
/// emit metrics and logs on a per-validator basis (rather than just an
/// aggregated basis).
fn individual_tracking(&self) -> bool {
self.validators.len() <= self.individual_tracking_threshold
}
/// Add some validators to `self` for additional monitoring. /// Add some validators to `self` for additional monitoring.
fn add_validator_pubkey(&mut self, pubkey: PublicKeyBytes) { fn add_validator_pubkey(&mut self, pubkey: PublicKeyBytes) {
let index_opt = self let index_opt = self
@ -317,6 +346,12 @@ impl<T: EthSpec> ValidatorMonitor<T> {
for monitored_validator in self.validators.values() { for monitored_validator in self.validators.values() {
if let Some(i) = monitored_validator.index { if let Some(i) = monitored_validator.index {
monitored_validator.touch_epoch_summary(current_epoch); monitored_validator.touch_epoch_summary(current_epoch);
// Only log the per-validator metrics if it's enabled.
if !self.individual_tracking() {
continue;
}
let i = i as usize; let i = i as usize;
let id = &monitored_validator.id; let id = &monitored_validator.id;
@ -379,6 +414,24 @@ impl<T: EthSpec> ValidatorMonitor<T> {
} }
} }
/// Run `func` with the `TOTAL_LABEL` and optionally the
/// `individual_id`.
///
/// This function is used for registering metrics that can be applied to
/// both all validators and an indivdual validator. For example, the count
/// of missed head votes can be aggregated across all validators in a single
/// metric and also tracked on a per-validator basis.
///
/// We allow disabling tracking metrics on an individual validator basis
/// since it can result in untenable cardinality with high validator counts.
fn aggregatable_metric<F: Fn(&str)>(&self, individual_id: &str, func: F) {
func(TOTAL_LABEL);
if self.individual_tracking() {
func(individual_id);
}
}
pub fn process_validator_statuses( pub fn process_validator_statuses(
&self, &self,
epoch: Epoch, epoch: Epoch,
@ -431,72 +484,92 @@ impl<T: EthSpec> ValidatorMonitor<T> {
// For Base states, this will be *any* attestation whatsoever. For Altair states, // For Base states, this will be *any* attestation whatsoever. For Altair states,
// this will be any attestation that matched a "timely" flag. // this will be any attestation that matched a "timely" flag.
if previous_epoch_matched_any { if previous_epoch_matched_any {
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_ATTESTER_HIT, metrics::inc_counter_vec(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_ATTESTER_HIT,
); &[label],
)
});
attestation_success.push(id); attestation_success.push(id);
debug!( if self.individual_tracking() {
self.log, debug!(
"Previous epoch attestation success"; self.log,
"matched_source" => previous_epoch_matched_source, "Previous epoch attestation success";
"matched_target" => previous_epoch_matched_target, "matched_source" => previous_epoch_matched_source,
"matched_head" => previous_epoch_matched_head, "matched_target" => previous_epoch_matched_target,
"epoch" => prev_epoch, "matched_head" => previous_epoch_matched_head,
"validator" => id, "epoch" => prev_epoch,
) "validator" => id,
)
}
} else { } else {
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_ATTESTER_MISS, metrics::inc_counter_vec(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_ATTESTER_MISS,
); &[label],
);
});
attestation_miss.push(id); attestation_miss.push(id);
debug!( if self.individual_tracking() {
self.log, debug!(
"Previous epoch attestation missing"; self.log,
"epoch" => prev_epoch, "Previous epoch attestation missing";
"validator" => id, "epoch" => prev_epoch,
) "validator" => id,
)
}
} }
// Indicates if any on-chain attestation hit the head. // Indicates if any on-chain attestation hit the head.
if previous_epoch_matched_head { if previous_epoch_matched_head {
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_HEAD_ATTESTER_HIT, metrics::inc_counter_vec(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_HEAD_ATTESTER_HIT,
); &[label],
);
});
} else { } else {
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_HEAD_ATTESTER_MISS, metrics::inc_counter_vec(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_HEAD_ATTESTER_MISS,
); &[label],
);
});
head_miss.push(id); head_miss.push(id);
debug!( if self.individual_tracking() {
self.log, debug!(
"Attestation failed to match head"; self.log,
"epoch" => prev_epoch, "Attestation failed to match head";
"validator" => id, "epoch" => prev_epoch,
); "validator" => id,
);
}
} }
// Indicates if any on-chain attestation hit the target. // Indicates if any on-chain attestation hit the target.
if previous_epoch_matched_target { if previous_epoch_matched_target {
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_TARGET_ATTESTER_HIT, metrics::inc_counter_vec(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_TARGET_ATTESTER_HIT,
); &[label],
);
});
} else { } else {
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_TARGET_ATTESTER_MISS, metrics::inc_counter_vec(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_TARGET_ATTESTER_MISS,
); &[label],
);
});
target_miss.push(id); target_miss.push(id);
debug!( if self.individual_tracking() {
self.log, debug!(
"Attestation failed to match target"; self.log,
"epoch" => prev_epoch, "Attestation failed to match target";
"validator" => id, "epoch" => prev_epoch,
); "validator" => id,
);
}
} }
// Get the minimum value among the validator monitor observed inclusion distance // Get the minimum value among the validator monitor observed inclusion distance
@ -511,21 +584,25 @@ impl<T: EthSpec> ValidatorMonitor<T> {
if let Some(inclusion_delay) = min_inclusion_distance { if let Some(inclusion_delay) = min_inclusion_distance {
if inclusion_delay > spec.min_attestation_inclusion_delay { if inclusion_delay > spec.min_attestation_inclusion_delay {
suboptimal_inclusion.push(id); suboptimal_inclusion.push(id);
debug!( if self.individual_tracking() {
self.log, debug!(
"Potential sub-optimal inclusion delay"; self.log,
"optimal" => spec.min_attestation_inclusion_delay, "Potential sub-optimal inclusion delay";
"delay" => inclusion_delay, "optimal" => spec.min_attestation_inclusion_delay,
"epoch" => prev_epoch, "delay" => inclusion_delay,
"validator" => id, "epoch" => prev_epoch,
); "validator" => id,
);
}
} }
metrics::set_int_gauge( if self.individual_tracking() {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_INCLUSION_DISTANCE, metrics::set_int_gauge(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_ON_CHAIN_INCLUSION_DISTANCE,
inclusion_delay as i64, &[id],
); inclusion_delay as i64,
);
}
} }
// Indicates the number of sync committee signatures that made it into // Indicates the number of sync committee signatures that made it into
@ -536,13 +613,19 @@ impl<T: EthSpec> ValidatorMonitor<T> {
let current_epoch = epoch - 1; let current_epoch = epoch - 1;
if let Some(sync_committee) = summary.sync_committee() { if let Some(sync_committee) = summary.sync_committee() {
if sync_committee.contains(pubkey) { if sync_committee.contains(pubkey) {
metrics::set_int_gauge( if self.individual_tracking() {
&metrics::VALIDATOR_MONITOR_VALIDATOR_IN_CURRENT_SYNC_COMMITTEE, metrics::set_int_gauge(
&[id], &metrics::VALIDATOR_MONITOR_VALIDATOR_IN_CURRENT_SYNC_COMMITTEE,
1, &[id],
); 1,
);
}
let epoch_summary = monitored_validator.summaries.read(); let epoch_summary = monitored_validator.summaries.read();
if let Some(summary) = epoch_summary.get(&current_epoch) { if let Some(summary) = epoch_summary.get(&current_epoch) {
// This log is not gated by
// `self.individual_tracking()` since the number of
// logs that can be generated is capped by the size
// of the sync committee.
info!( info!(
self.log, self.log,
"Current epoch sync signatures"; "Current epoch sync signatures";
@ -552,7 +635,7 @@ impl<T: EthSpec> ValidatorMonitor<T> {
"validator" => id, "validator" => id,
); );
} }
} else { } else if self.individual_tracking() {
metrics::set_int_gauge( metrics::set_int_gauge(
&metrics::VALIDATOR_MONITOR_VALIDATOR_IN_CURRENT_SYNC_COMMITTEE, &metrics::VALIDATOR_MONITOR_VALIDATOR_IN_CURRENT_SYNC_COMMITTEE,
&[id], &[id],
@ -693,12 +776,17 @@ impl<T: EthSpec> ValidatorMonitor<T> {
let id = &validator.id; let id = &validator.id;
let delay = get_block_delay_ms(seen_timestamp, block, slot_clock); let delay = get_block_delay_ms(seen_timestamp, block, slot_clock);
metrics::inc_counter_vec(&metrics::VALIDATOR_MONITOR_BEACON_BLOCK_TOTAL, &[src, id]); self.aggregatable_metric(id, |label| {
metrics::observe_timer_vec( metrics::inc_counter_vec(
&metrics::VALIDATOR_MONITOR_BEACON_BLOCK_DELAY_SECONDS, &metrics::VALIDATOR_MONITOR_BEACON_BLOCK_TOTAL,
&[src, id], &[src, label],
delay, );
); metrics::observe_timer_vec(
&metrics::VALIDATOR_MONITOR_BEACON_BLOCK_DELAY_SECONDS,
&[src, label],
delay,
);
});
info!( info!(
self.log, self.log,
@ -764,27 +852,31 @@ impl<T: EthSpec> ValidatorMonitor<T> {
if let Some(validator) = self.get_validator(*i) { if let Some(validator) = self.get_validator(*i) {
let id = &validator.id; let id = &validator.id;
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_UNAGGREGATED_ATTESTATION_TOTAL, metrics::inc_counter_vec(
&[src, id], &metrics::VALIDATOR_MONITOR_UNAGGREGATED_ATTESTATION_TOTAL,
); &[src, label],
metrics::observe_timer_vec( );
&metrics::VALIDATOR_MONITOR_UNAGGREGATED_ATTESTATION_DELAY_SECONDS, metrics::observe_timer_vec(
&[src, id], &metrics::VALIDATOR_MONITOR_UNAGGREGATED_ATTESTATION_DELAY_SECONDS,
delay, &[src, label],
); delay,
);
});
info!( if self.individual_tracking() {
self.log, info!(
"Unaggregated attestation"; self.log,
"head" => ?data.beacon_block_root, "Unaggregated attestation";
"index" => %data.index, "head" => ?data.beacon_block_root,
"delay_ms" => %delay.as_millis(), "index" => %data.index,
"epoch" => %epoch, "delay_ms" => %delay.as_millis(),
"slot" => %data.slot, "epoch" => %epoch,
"src" => src, "slot" => %data.slot,
"validator" => %id, "src" => src,
); "validator" => %id,
);
}
validator.with_epoch_summary(epoch, |summary| { validator.with_epoch_summary(epoch, |summary| {
summary.register_unaggregated_attestation(delay) summary.register_unaggregated_attestation(delay)
@ -848,27 +940,31 @@ impl<T: EthSpec> ValidatorMonitor<T> {
if let Some(validator) = self.get_validator(aggregator_index) { if let Some(validator) = self.get_validator(aggregator_index) {
let id = &validator.id; let id = &validator.id;
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_AGGREGATED_ATTESTATION_TOTAL, metrics::inc_counter_vec(
&[src, id], &metrics::VALIDATOR_MONITOR_AGGREGATED_ATTESTATION_TOTAL,
); &[src, label],
metrics::observe_timer_vec( );
&metrics::VALIDATOR_MONITOR_AGGREGATED_ATTESTATION_DELAY_SECONDS, metrics::observe_timer_vec(
&[src, id], &metrics::VALIDATOR_MONITOR_AGGREGATED_ATTESTATION_DELAY_SECONDS,
delay, &[src, label],
); delay,
);
});
info!( if self.individual_tracking() {
self.log, info!(
"Aggregated attestation"; self.log,
"head" => ?data.beacon_block_root, "Aggregated attestation";
"index" => %data.index, "head" => ?data.beacon_block_root,
"delay_ms" => %delay.as_millis(), "index" => %data.index,
"epoch" => %epoch, "delay_ms" => %delay.as_millis(),
"slot" => %data.slot, "epoch" => %epoch,
"src" => src, "slot" => %data.slot,
"validator" => %id, "src" => src,
); "validator" => %id,
);
}
validator.with_epoch_summary(epoch, |summary| { validator.with_epoch_summary(epoch, |summary| {
summary.register_aggregated_attestation(delay) summary.register_aggregated_attestation(delay)
@ -879,27 +975,31 @@ impl<T: EthSpec> ValidatorMonitor<T> {
if let Some(validator) = self.get_validator(*i) { if let Some(validator) = self.get_validator(*i) {
let id = &validator.id; let id = &validator.id;
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_ATTESTATION_IN_AGGREGATE_TOTAL, metrics::inc_counter_vec(
&[src, id], &metrics::VALIDATOR_MONITOR_ATTESTATION_IN_AGGREGATE_TOTAL,
); &[src, label],
metrics::observe_timer_vec( );
&metrics::VALIDATOR_MONITOR_ATTESTATION_IN_AGGREGATE_DELAY_SECONDS, metrics::observe_timer_vec(
&[src, id], &metrics::VALIDATOR_MONITOR_ATTESTATION_IN_AGGREGATE_DELAY_SECONDS,
delay, &[src, label],
); delay,
);
});
info!( if self.individual_tracking() {
self.log, info!(
"Attestation included in aggregate"; self.log,
"head" => ?data.beacon_block_root, "Attestation included in aggregate";
"index" => %data.index, "head" => ?data.beacon_block_root,
"delay_ms" => %delay.as_millis(), "index" => %data.index,
"epoch" => %epoch, "delay_ms" => %delay.as_millis(),
"slot" => %data.slot, "epoch" => %epoch,
"src" => src, "slot" => %data.slot,
"validator" => %id, "src" => src,
); "validator" => %id,
);
}
validator.with_epoch_summary(epoch, |summary| { validator.with_epoch_summary(epoch, |summary| {
summary.register_aggregate_attestation_inclusion() summary.register_aggregate_attestation_inclusion()
@ -933,26 +1033,31 @@ impl<T: EthSpec> ValidatorMonitor<T> {
if let Some(validator) = self.get_validator(*i) { if let Some(validator) = self.get_validator(*i) {
let id = &validator.id; let id = &validator.id;
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_ATTESTATION_IN_BLOCK_TOTAL, metrics::inc_counter_vec(
&["block", id], &metrics::VALIDATOR_MONITOR_ATTESTATION_IN_BLOCK_TOTAL,
); &["block", label],
metrics::set_int_gauge( );
&metrics::VALIDATOR_MONITOR_ATTESTATION_IN_BLOCK_DELAY_SLOTS, });
&["block", id],
delay.as_u64() as i64,
);
info!( if self.individual_tracking() {
self.log, metrics::set_int_gauge(
"Attestation included in block"; &metrics::VALIDATOR_MONITOR_ATTESTATION_IN_BLOCK_DELAY_SLOTS,
"head" => ?data.beacon_block_root, &["block", id],
"index" => %data.index, delay.as_u64() as i64,
"inclusion_lag" => format!("{} slot(s)", delay), );
"epoch" => %epoch,
"slot" => %data.slot, info!(
"validator" => %id, self.log,
); "Attestation included in block";
"head" => ?data.beacon_block_root,
"index" => %data.index,
"inclusion_lag" => format!("{} slot(s)", delay),
"epoch" => %epoch,
"slot" => %data.slot,
"validator" => %id,
);
}
validator.with_epoch_summary(epoch, |summary| { validator.with_epoch_summary(epoch, |summary| {
summary.register_attestation_block_inclusion(inclusion_distance) summary.register_attestation_block_inclusion(inclusion_distance)
@ -1010,26 +1115,30 @@ impl<T: EthSpec> ValidatorMonitor<T> {
slot_clock, slot_clock,
); );
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_SYNC_COMMITTEE_MESSAGES_TOTAL, metrics::inc_counter_vec(
&[src, id], &metrics::VALIDATOR_MONITOR_SYNC_COMMITTEE_MESSAGES_TOTAL,
); &[src, label],
metrics::observe_timer_vec( );
&metrics::VALIDATOR_MONITOR_SYNC_COMMITTEE_MESSAGES_DELAY_SECONDS, metrics::observe_timer_vec(
&[src, id], &metrics::VALIDATOR_MONITOR_SYNC_COMMITTEE_MESSAGES_DELAY_SECONDS,
delay, &[src, label],
); delay,
);
});
info!( if self.individual_tracking() {
self.log, info!(
"Sync committee message"; self.log,
"head" => %sync_committee_message.beacon_block_root, "Sync committee message";
"delay_ms" => %delay.as_millis(), "head" => %sync_committee_message.beacon_block_root,
"epoch" => %epoch, "delay_ms" => %delay.as_millis(),
"slot" => %sync_committee_message.slot, "epoch" => %epoch,
"src" => src, "slot" => %sync_committee_message.slot,
"validator" => %id, "src" => src,
); "validator" => %id,
);
}
validator.with_epoch_summary(epoch, |summary| { validator.with_epoch_summary(epoch, |summary| {
summary.register_sync_committee_message(delay) summary.register_sync_committee_message(delay)
@ -1094,26 +1203,30 @@ impl<T: EthSpec> ValidatorMonitor<T> {
if let Some(validator) = self.get_validator(aggregator_index) { if let Some(validator) = self.get_validator(aggregator_index) {
let id = &validator.id; let id = &validator.id;
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_SYNC_CONTRIBUTIONS_TOTAL, metrics::inc_counter_vec(
&[src, id], &metrics::VALIDATOR_MONITOR_SYNC_CONTRIBUTIONS_TOTAL,
); &[src, label],
metrics::observe_timer_vec( );
&metrics::VALIDATOR_MONITOR_SYNC_CONTRIBUTIONS_DELAY_SECONDS, metrics::observe_timer_vec(
&[src, id], &metrics::VALIDATOR_MONITOR_SYNC_CONTRIBUTIONS_DELAY_SECONDS,
delay, &[src, label],
); delay,
);
});
info!( if self.individual_tracking() {
self.log, info!(
"Sync contribution"; self.log,
"head" => %beacon_block_root, "Sync contribution";
"delay_ms" => %delay.as_millis(), "head" => %beacon_block_root,
"epoch" => %epoch, "delay_ms" => %delay.as_millis(),
"slot" => %slot, "epoch" => %epoch,
"src" => src, "slot" => %slot,
"validator" => %id, "src" => src,
); "validator" => %id,
);
}
validator.with_epoch_summary(epoch, |summary| { validator.with_epoch_summary(epoch, |summary| {
summary.register_sync_committee_contribution(delay) summary.register_sync_committee_contribution(delay)
@ -1124,21 +1237,25 @@ impl<T: EthSpec> ValidatorMonitor<T> {
if let Some(validator) = self.validators.get(validator_pubkey) { if let Some(validator) = self.validators.get(validator_pubkey) {
let id = &validator.id; let id = &validator.id;
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_SYNC_COMMITTEE_MESSAGE_IN_CONTRIBUTION_TOTAL, metrics::inc_counter_vec(
&[src, id], &metrics::VALIDATOR_MONITOR_SYNC_COMMITTEE_MESSAGE_IN_CONTRIBUTION_TOTAL,
); &[src, label],
);
});
info!( if self.individual_tracking() {
self.log, info!(
"Sync signature included in contribution"; self.log,
"head" => %beacon_block_root, "Sync signature included in contribution";
"delay_ms" => %delay.as_millis(), "head" => %beacon_block_root,
"epoch" => %epoch, "delay_ms" => %delay.as_millis(),
"slot" => %slot, "epoch" => %epoch,
"src" => src, "slot" => %slot,
"validator" => %id, "src" => src,
); "validator" => %id,
);
}
validator.with_epoch_summary(epoch, |summary| { validator.with_epoch_summary(epoch, |summary| {
summary.register_sync_signature_contribution_inclusion() summary.register_sync_signature_contribution_inclusion()
@ -1160,19 +1277,23 @@ impl<T: EthSpec> ValidatorMonitor<T> {
if let Some(validator) = self.validators.get(validator_pubkey) { if let Some(validator) = self.validators.get(validator_pubkey) {
let id = &validator.id; let id = &validator.id;
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_SYNC_COMMITTEE_MESSAGE_IN_BLOCK_TOTAL, metrics::inc_counter_vec(
&["block", id], &metrics::VALIDATOR_MONITOR_SYNC_COMMITTEE_MESSAGE_IN_BLOCK_TOTAL,
); &["block", label],
);
});
info!( if self.individual_tracking() {
self.log, info!(
"Sync signature included in block"; self.log,
"head" => %beacon_block_root, "Sync signature included in block";
"epoch" => %epoch, "head" => %beacon_block_root,
"slot" => %slot, "epoch" => %epoch,
"validator" => %id, "slot" => %slot,
); "validator" => %id,
);
}
validator.with_epoch_summary(epoch, |summary| { validator.with_epoch_summary(epoch, |summary| {
summary.register_sync_signature_block_inclusions(); summary.register_sync_signature_block_inclusions();
@ -1201,8 +1322,12 @@ impl<T: EthSpec> ValidatorMonitor<T> {
let id = &validator.id; let id = &validator.id;
let epoch = exit.epoch; let epoch = exit.epoch;
metrics::inc_counter_vec(&metrics::VALIDATOR_MONITOR_EXIT_TOTAL, &[src, id]); self.aggregatable_metric(id, |label| {
metrics::inc_counter_vec(&metrics::VALIDATOR_MONITOR_EXIT_TOTAL, &[src, label]);
});
// Not gated behind `self.individual_tracking()` since it's an
// infrequent and interesting message.
info!( info!(
self.log, self.log,
"Voluntary exit"; "Voluntary exit";
@ -1240,11 +1365,15 @@ impl<T: EthSpec> ValidatorMonitor<T> {
if let Some(validator) = self.get_validator(proposer) { if let Some(validator) = self.get_validator(proposer) {
let id = &validator.id; let id = &validator.id;
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_PROPOSER_SLASHING_TOTAL, metrics::inc_counter_vec(
&[src, id], &metrics::VALIDATOR_MONITOR_PROPOSER_SLASHING_TOTAL,
); &[src, label],
);
});
// Not gated behind `self.individual_tracking()` since it's an
// infrequent and interesting message.
crit!( crit!(
self.log, self.log,
"Proposer slashing"; "Proposer slashing";
@ -1293,11 +1422,15 @@ impl<T: EthSpec> ValidatorMonitor<T> {
let id = &validator.id; let id = &validator.id;
let epoch = data.slot.epoch(T::slots_per_epoch()); let epoch = data.slot.epoch(T::slots_per_epoch());
metrics::inc_counter_vec( self.aggregatable_metric(id, |label| {
&metrics::VALIDATOR_MONITOR_ATTESTER_SLASHING_TOTAL, metrics::inc_counter_vec(
&[src, id], &metrics::VALIDATOR_MONITOR_ATTESTER_SLASHING_TOTAL,
); &[src, label],
);
});
// Not gated behind `self.individual_tracking()` since it's an
// infrequent and interesting message.
crit!( crit!(
self.log, self.log,
"Attester slashing"; "Attester slashing";
@ -1347,69 +1480,80 @@ impl<T: EthSpec> ValidatorMonitor<T> {
/* /*
* Attestations * Attestations
*/ */
metrics::set_gauge_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ATTESTATIONS_TOTAL,
&[id],
summary.attestations as i64,
);
if let Some(delay) = summary.attestation_min_delay { if let Some(delay) = summary.attestation_min_delay {
metrics::observe_timer_vec( self.aggregatable_metric(id, |tag| {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ATTESTATIONS_MIN_DELAY_SECONDS, metrics::observe_timer_vec(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_ATTESTATIONS_MIN_DELAY_SECONDS,
delay, &[tag],
); delay,
);
});
} }
metrics::set_gauge_vec( if self.individual_tracking() {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ATTESTATION_AGGREGATE_INCLUSIONS,
&[id],
summary.attestation_aggregate_inclusions as i64,
);
metrics::set_gauge_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ATTESTATION_BLOCK_INCLUSIONS,
&[id],
summary.attestation_block_inclusions as i64,
);
if let Some(distance) = summary.attestation_min_block_inclusion_distance {
metrics::set_gauge_vec( metrics::set_gauge_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ATTESTATION_BLOCK_MIN_INCLUSION_DISTANCE, &metrics::VALIDATOR_MONITOR_PREV_EPOCH_ATTESTATIONS_TOTAL,
&[id], &[id],
distance.as_u64() as i64, summary.attestations as i64,
); );
metrics::set_gauge_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ATTESTATION_AGGREGATE_INCLUSIONS,
&[id],
summary.attestation_aggregate_inclusions as i64,
);
metrics::set_gauge_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ATTESTATION_BLOCK_INCLUSIONS,
&[id],
summary.attestation_block_inclusions as i64,
);
if let Some(distance) = summary.attestation_min_block_inclusion_distance {
metrics::set_gauge_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ATTESTATION_BLOCK_MIN_INCLUSION_DISTANCE,
&[id],
distance.as_u64() as i64,
);
}
} }
/* /*
* Sync committee messages * Sync committee messages
*/ */
metrics::set_gauge_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_SYNC_COMMITTEE_MESSAGES_TOTAL,
&[id],
summary.sync_committee_messages as i64,
);
if let Some(delay) = summary.sync_committee_message_min_delay { if let Some(delay) = summary.sync_committee_message_min_delay {
metrics::observe_timer_vec( self.aggregatable_metric(id, |tag| {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_SYNC_COMMITTEE_MESSAGES_MIN_DELAY_SECONDS, metrics::observe_timer_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_SYNC_COMMITTEE_MESSAGES_MIN_DELAY_SECONDS,
&[tag],
delay,
);
});
}
if self.individual_tracking() {
metrics::set_gauge_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_SYNC_COMMITTEE_MESSAGES_TOTAL,
&[id], &[id],
delay, summary.sync_committee_messages as i64,
);
metrics::set_gauge_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_SYNC_CONTRIBUTION_INCLUSIONS,
&[id],
summary.sync_signature_contribution_inclusions as i64,
);
metrics::set_gauge_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_SYNC_SIGNATURE_BLOCK_INCLUSIONS,
&[id],
summary.sync_signature_block_inclusions as i64,
); );
} }
metrics::set_gauge_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_SYNC_CONTRIBUTION_INCLUSIONS,
&[id],
summary.sync_signature_contribution_inclusions as i64,
);
metrics::set_gauge_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_SYNC_SIGNATURE_BLOCK_INCLUSIONS,
&[id],
summary.sync_signature_block_inclusions as i64,
);
/* /*
* Sync contributions * Sync contributions
*/ */
metrics::set_gauge_vec( if self.individual_tracking() {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_SYNC_CONTRIBUTIONS_TOTAL, metrics::set_gauge_vec(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_SYNC_CONTRIBUTIONS_TOTAL,
summary.sync_contributions as i64, &[id],
); summary.sync_contributions as i64,
);
}
if let Some(delay) = summary.sync_contribution_min_delay { if let Some(delay) = summary.sync_contribution_min_delay {
metrics::observe_timer_vec( metrics::observe_timer_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_SYNC_CONTRIBUTION_MIN_DELAY_SECONDS, &metrics::VALIDATOR_MONITOR_PREV_EPOCH_SYNC_CONTRIBUTION_MIN_DELAY_SECONDS,
@ -1421,51 +1565,61 @@ impl<T: EthSpec> ValidatorMonitor<T> {
/* /*
* Blocks * Blocks
*/ */
metrics::set_gauge_vec( if self.individual_tracking() {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_BEACON_BLOCKS_TOTAL, metrics::set_gauge_vec(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_BEACON_BLOCKS_TOTAL,
summary.blocks as i64,
);
if let Some(delay) = summary.block_min_delay {
metrics::observe_timer_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_BEACON_BLOCKS_MIN_DELAY_SECONDS,
&[id], &[id],
delay, summary.blocks as i64,
); );
} }
if let Some(delay) = summary.block_min_delay {
self.aggregatable_metric(id, |tag| {
metrics::observe_timer_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_BEACON_BLOCKS_MIN_DELAY_SECONDS,
&[tag],
delay,
);
});
}
/* /*
* Aggregates * Aggregates
*/ */
metrics::set_gauge_vec( if self.individual_tracking() {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_AGGREGATES_TOTAL, metrics::set_gauge_vec(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_AGGREGATES_TOTAL,
summary.aggregates as i64,
);
if let Some(delay) = summary.aggregate_min_delay {
metrics::observe_timer_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_AGGREGATES_MIN_DELAY_SECONDS,
&[id], &[id],
delay, summary.aggregates as i64,
); );
} }
if let Some(delay) = summary.aggregate_min_delay {
self.aggregatable_metric(id, |tag| {
metrics::observe_timer_vec(
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_AGGREGATES_MIN_DELAY_SECONDS,
&[tag],
delay,
);
});
}
/* /*
* Other * Other
*/ */
metrics::set_gauge_vec( if self.individual_tracking() {
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_EXITS_TOTAL, metrics::set_gauge_vec(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_EXITS_TOTAL,
summary.exits as i64, &[id],
); summary.exits as i64,
metrics::set_gauge_vec( );
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_PROPOSER_SLASHINGS_TOTAL, metrics::set_gauge_vec(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_PROPOSER_SLASHINGS_TOTAL,
summary.proposer_slashings as i64, &[id],
); summary.proposer_slashings as i64,
metrics::set_gauge_vec( );
&metrics::VALIDATOR_MONITOR_PREV_EPOCH_ATTESTER_SLASHINGS_TOTAL, metrics::set_gauge_vec(
&[id], &metrics::VALIDATOR_MONITOR_PREV_EPOCH_ATTESTER_SLASHINGS_TOTAL,
summary.attester_slashings as i64, &[id],
); summary.attester_slashings as i64,
);
}
} }
} }
} }

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

@ -5,6 +5,7 @@ use beacon_chain::builder::BeaconChainBuilder;
use beacon_chain::test_utils::{ use beacon_chain::test_utils::{
test_spec, AttestationStrategy, BeaconChainHarness, BlockStrategy, DiskHarnessType, test_spec, AttestationStrategy, BeaconChainHarness, BlockStrategy, DiskHarnessType,
}; };
use beacon_chain::validator_monitor::DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD;
use beacon_chain::{ use beacon_chain::{
historical_blocks::HistoricalBlockError, migrate::MigratorConfig, BeaconChain, historical_blocks::HistoricalBlockError, migrate::MigratorConfig, BeaconChain,
BeaconChainError, BeaconChainTypes, BeaconSnapshot, ChainConfig, NotifyExecutionLayer, BeaconChainError, BeaconChainTypes, BeaconSnapshot, ChainConfig, NotifyExecutionLayer,
@ -2121,7 +2122,7 @@ async fn weak_subjectivity_sync() {
log.clone(), log.clone(),
1, 1,
))) )))
.monitor_validators(true, vec![], log) .monitor_validators(true, vec![], DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD, log)
.build() .build()
.expect("should build"), .expect("should build"),
); );

2
beacon_node/builder_client/Cargo.toml
View File

@ -9,4 +9,4 @@ reqwest = { version = "0.11.0", features = ["json","stream"] }
sensitive_url = { path = "../../common/sensitive_url" } sensitive_url = { path = "../../common/sensitive_url" }
eth2 = { path = "../../common/eth2" } eth2 = { path = "../../common/eth2" }
serde = { version = "1.0.116", features = ["derive"] } serde = { version = "1.0.116", features = ["derive"] }
serde_json = "1.0.58" serde_json = "1.0.58"

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

@ -174,6 +174,7 @@ where
.monitor_validators( .monitor_validators(
config.validator_monitor_auto, config.validator_monitor_auto,
config.validator_monitor_pubkeys.clone(), config.validator_monitor_pubkeys.clone(),
config.validator_monitor_individual_tracking_threshold,
runtime_context runtime_context
.service_context("val_mon".to_string()) .service_context("val_mon".to_string())
.log() .log()

7
beacon_node/client/src/config.rs
View File

@ -1,3 +1,4 @@
use beacon_chain::validator_monitor::DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD;
use directory::DEFAULT_ROOT_DIR; use directory::DEFAULT_ROOT_DIR;
use environment::LoggerConfig; use environment::LoggerConfig;
use network::NetworkConfig; use network::NetworkConfig;
@ -59,6 +60,11 @@ pub struct Config {
pub validator_monitor_auto: bool, pub validator_monitor_auto: bool,
/// A list of validator pubkeys to monitor. /// A list of validator pubkeys to monitor.
pub validator_monitor_pubkeys: Vec<PublicKeyBytes>, pub validator_monitor_pubkeys: Vec<PublicKeyBytes>,
/// Once the number of monitored validators goes above this threshold, we
/// will stop tracking metrics on a per-validator basis. This prevents large
/// validator counts causing infeasibly high cardinailty for Prometheus and
/// high log volumes.
pub validator_monitor_individual_tracking_threshold: usize,
#[serde(skip)] #[serde(skip)]
/// The `genesis` field is not serialized or deserialized by `serde` to ensure it is defined /// The `genesis` field is not serialized or deserialized by `serde` to ensure it is defined
/// via the CLI at runtime, instead of from a configuration file saved to disk. /// via the CLI at runtime, instead of from a configuration file saved to disk.
@ -97,6 +103,7 @@ impl Default for Config {
slasher: None, slasher: None,
validator_monitor_auto: false, validator_monitor_auto: false,
validator_monitor_pubkeys: vec![], validator_monitor_pubkeys: vec![],
validator_monitor_individual_tracking_threshold: DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD,
logger_config: LoggerConfig::default(), logger_config: LoggerConfig::default(),
} }
} }

6
beacon_node/execution_layer/Cargo.toml
View File

@ -38,7 +38,7 @@ rand = "0.8.5"
zeroize = { version = "1.4.2", features = ["zeroize_derive"] } zeroize = { version = "1.4.2", features = ["zeroize_derive"] }
lighthouse_metrics = { path = "../../common/lighthouse_metrics" } lighthouse_metrics = { path = "../../common/lighthouse_metrics" }
lazy_static = "1.4.0" lazy_static = "1.4.0"
ethers-core = "0.17.0" ethers-core = "1.0.2"
builder_client = { path = "../builder_client" } builder_client = { path = "../builder_client" }
fork_choice = { path = "../../consensus/fork_choice" } fork_choice = { path = "../../consensus/fork_choice" }
mev-build-rs = { git = "https://github.com/ralexstokes/mev-rs", rev = "6c99b0fbdc0427b1625469d2e575303ce08de5b8" } mev-build-rs = { git = "https://github.com/ralexstokes/mev-rs", rev = "6c99b0fbdc0427b1625469d2e575303ce08de5b8" }
@ -46,3 +46,7 @@ ethereum-consensus = { git = "https://github.com/ralexstokes/ethereum-consensus"
ssz-rs = { git = "https://github.com/ralexstokes/ssz-rs", rev = "cb08f1" } ssz-rs = { git = "https://github.com/ralexstokes/ssz-rs", rev = "cb08f1" }
tokio-stream = { version = "0.1.9", features = [ "sync" ] } tokio-stream = { version = "0.1.9", features = [ "sync" ] }
strum = "0.24.0" strum = "0.24.0"
keccak-hash = "0.10.0"
hash256-std-hasher = "0.15.2"
triehash = "0.8.4"
hash-db = "0.15.2"

163
beacon_node/execution_layer/src/block_hash.rs Normal file
View File

@ -0,0 +1,163 @@
use crate::{
keccak::{keccak256, KeccakHasher},
metrics, Error, ExecutionLayer,
};
use ethers_core::utils::rlp::RlpStream;
use keccak_hash::KECCAK_EMPTY_LIST_RLP;
use triehash::ordered_trie_root;
use types::{
map_execution_block_header_fields, Address, EthSpec, ExecutionBlockHash, ExecutionBlockHeader,
ExecutionPayloadRef, Hash256, Hash64, Uint256,
};
impl<T: EthSpec> ExecutionLayer<T> {
/// Verify `payload.block_hash` locally within Lighthouse.
///
/// No remote calls to the execution client will be made, so this is quite a cheap check.
pub fn verify_payload_block_hash(&self, payload: ExecutionPayloadRef<T>) -> Result<(), Error> {
let _timer = metrics::start_timer(&metrics::EXECUTION_LAYER_VERIFY_BLOCK_HASH);
// Calculate the transactions root.
// We're currently using a deprecated Parity library for this. We should move to a
// better alternative when one appears, possibly following Reth.
let rlp_transactions_root = ordered_trie_root::<KeccakHasher, _>(
payload.transactions().iter().map(|txn_bytes| &**txn_bytes),
);
// Construct the block header.
let exec_block_header = ExecutionBlockHeader::from_payload(
payload,
KECCAK_EMPTY_LIST_RLP.as_fixed_bytes().into(),
rlp_transactions_root,
);
// Hash the RLP encoding of the block header.
let rlp_block_header = rlp_encode_block_header(&exec_block_header);
let header_hash = ExecutionBlockHash::from_root(keccak256(&rlp_block_header));
if header_hash != payload.block_hash() {
return Err(Error::BlockHashMismatch {
computed: header_hash,
payload: payload.block_hash(),
transactions_root: rlp_transactions_root,
});
}
Ok(())
}
}
/// RLP encode an execution block header.
pub fn rlp_encode_block_header(header: &ExecutionBlockHeader) -> Vec<u8> {
let mut rlp_header_stream = RlpStream::new();
rlp_header_stream.begin_unbounded_list();
map_execution_block_header_fields!(&header, |_, field| {
rlp_header_stream.append(field);
});
rlp_header_stream.finalize_unbounded_list();
rlp_header_stream.out().into()
}
#[cfg(test)]
mod test {
use super::*;
use hex::FromHex;
use std::str::FromStr;
fn test_rlp_encoding(
header: &ExecutionBlockHeader,
expected_rlp: Option<&str>,
expected_hash: Hash256,
) {
let rlp_encoding = rlp_encode_block_header(header);
if let Some(expected_rlp) = expected_rlp {
let computed_rlp = hex::encode(&rlp_encoding);
assert_eq!(expected_rlp, computed_rlp);
}
let computed_hash = keccak256(&rlp_encoding);
assert_eq!(expected_hash, computed_hash);
}
#[test]
fn test_rlp_encode_eip1559_block() {
let header = ExecutionBlockHeader {
parent_hash: Hash256::from_str("e0a94a7a3c9617401586b1a27025d2d9671332d22d540e0af72b069170380f2a").unwrap(),
ommers_hash: Hash256::from_str("1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347").unwrap(),
beneficiary: Address::from_str("ba5e000000000000000000000000000000000000").unwrap(),
state_root: Hash256::from_str("ec3c94b18b8a1cff7d60f8d258ec723312932928626b4c9355eb4ab3568ec7f7").unwrap(),
transactions_root: Hash256::from_str("50f738580ed699f0469702c7ccc63ed2e51bc034be9479b7bff4e68dee84accf").unwrap(),
receipts_root: Hash256::from_str("29b0562f7140574dd0d50dee8a271b22e1a0a7b78fca58f7c60370d8317ba2a9").unwrap(),
logs_bloom: <[u8; 256]>::from_hex("00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000").unwrap().into(),
difficulty: 0x020000.into(),
number: 0x01_u64.into(),
gas_limit: 0x016345785d8a0000_u64.into(),
gas_used: 0x015534_u64.into(),
timestamp: 0x079e,
extra_data: vec![0x42],
mix_hash: Hash256::from_str("0000000000000000000000000000000000000000000000000000000000000000").unwrap(),
nonce: Hash64::zero(),
base_fee_per_gas: 0x036b_u64.into(),
};
let expected_rlp = "f90200a0e0a94a7a3c9617401586b1a27025d2d9671332d22d540e0af72b069170380f2aa01dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d4934794ba5e000000000000000000000000000000000000a0ec3c94b18b8a1cff7d60f8d258ec723312932928626b4c9355eb4ab3568ec7f7a050f738580ed699f0469702c7ccc63ed2e51bc034be9479b7bff4e68dee84accfa029b0562f7140574dd0d50dee8a271b22e1a0a7b78fca58f7c60370d8317ba2a9b9010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000830200000188016345785d8a00008301553482079e42a0000000000000000000000000000000000000000000000000000000000000000088000000000000000082036b";
let expected_hash =
Hash256::from_str("6a251c7c3c5dca7b42407a3752ff48f3bbca1fab7f9868371d9918daf1988d1f")
.unwrap();
test_rlp_encoding(&header, Some(expected_rlp), expected_hash);
}
#[test]
fn test_rlp_encode_merge_block() {
let header = ExecutionBlockHeader {
parent_hash: Hash256::from_str("927ca537f06c783a3a2635b8805eef1c8c2124f7444ad4a3389898dd832f2dbe").unwrap(),
ommers_hash: Hash256::from_str("1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347").unwrap(),
beneficiary: Address::from_str("ba5e000000000000000000000000000000000000").unwrap(),
state_root: Hash256::from_str("0xe97859b065bd8dbbb4519c7cb935024de2484c2b7f881181b4360492f0b06b82").unwrap(),
transactions_root: Hash256::from_str("50f738580ed699f0469702c7ccc63ed2e51bc034be9479b7bff4e68dee84accf").unwrap(),
receipts_root: Hash256::from_str("29b0562f7140574dd0d50dee8a271b22e1a0a7b78fca58f7c60370d8317ba2a9").unwrap(),
logs_bloom: <[u8; 256]>::from_hex("00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000").unwrap().into(),
difficulty: 0x00.into(),
number: 0x01_u64.into(),
gas_limit: 0x016345785d8a0000_u64.into(),
gas_used: 0x015534_u64.into(),
timestamp: 0x079e,
extra_data: vec![0x42],
mix_hash: Hash256::from_str("0000000000000000000000000000000000000000000000000000000000020000").unwrap(),
nonce: Hash64::zero(),
base_fee_per_gas: 0x036b_u64.into(),
};
let expected_rlp = "f901fda0927ca537f06c783a3a2635b8805eef1c8c2124f7444ad4a3389898dd832f2dbea01dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d4934794ba5e000000000000000000000000000000000000a0e97859b065bd8dbbb4519c7cb935024de2484c2b7f881181b4360492f0b06b82a050f738580ed699f0469702c7ccc63ed2e51bc034be9479b7bff4e68dee84accfa029b0562f7140574dd0d50dee8a271b22e1a0a7b78fca58f7c60370d8317ba2a9b9010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000800188016345785d8a00008301553482079e42a0000000000000000000000000000000000000000000000000000000000002000088000000000000000082036b";
let expected_hash =
Hash256::from_str("0x5b1f0f2efdaa19e996b4aea59eeb67620259f09732732a339a10dac311333684")
.unwrap();
test_rlp_encoding(&header, Some(expected_rlp), expected_hash);
}
// Test a real payload from mainnet.
#[test]
fn test_rlp_encode_block_16182891() {
let header = ExecutionBlockHeader {
parent_hash: Hash256::from_str("3e9c7b3f403947f110f68c4564a004b73dd8ebf73b143e46cc637926eec01a6d").unwrap(),
ommers_hash: Hash256::from_str("1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347").unwrap(),
beneficiary: Address::from_str("dafea492d9c6733ae3d56b7ed1adb60692c98bc5").unwrap(),
state_root: Hash256::from_str("5a8183d230818a167477420ce3a393ca3ef8706a7d596694ab6059894ed6fda9").unwrap(),
transactions_root: Hash256::from_str("0223f0cb35f184d2ac409e89dc0768ad738f777bd1c85d3302ca50f307180c94").unwrap(),
receipts_root: Hash256::from_str("371c76821b1cc21232574604eac5349d51647eb530e2a45d4f6fe2c501351aa5").unwrap(),
logs_bloom: <[u8; 256]>::from_hex("1a2c559955848d2662a0634cb40c7a6192a1524f11061203689bcbcdec901b054084d4f4d688009d24c10918e0089b48e72fe2d7abafb903889d10c3827c6901096612d259801b1b7ba1663a4201f5f88f416a9997c55bcc2c54785280143b057a008764c606182e324216822a2d5913e797a05c16cc1468d001acf3783b18e00e0203033e43106178db554029e83ca46402dc49d929d7882a04a0e7215041bdabf7430bd10ef4bb658a40f064c63c4816660241c2480862f26742fdf9ca41637731350301c344e439428182a03e384484e6d65d0c8a10117c6739ca201b60974519a1ae6b0c3966c0f650b449d10eae065dab2c83ab4edbab5efdea50bbc801").unwrap().into(),
difficulty: 0.into(),
number: 16182891.into(),
gas_limit: 0x1c9c380.into(),
gas_used: 0xe9b752.into(),
timestamp: 0x6399bf63,
extra_data: hex::decode("496c6c756d696e61746520446d6f63726174697a6520447374726962757465").unwrap(),
mix_hash: Hash256::from_str("bf5289894b2ceab3549f92f063febbac896b280ddb18129a57cff13113c11b13").unwrap(),
nonce: Hash64::zero(),
base_fee_per_gas: 0x34187b238_u64.into(),
};
let expected_hash =
Hash256::from_str("6da69709cd5a34079b6604d29cd78fc01dacd7c6268980057ad92a2bede87351")
.unwrap();
test_rlp_encoding(&header, None, expected_hash);
}
}

35
beacon_node/execution_layer/src/keccak.rs Normal file
View File

@ -0,0 +1,35 @@
// Copyright 2017, 2018 Parity Technologies
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use hash256_std_hasher::Hash256StdHasher;
use hash_db::Hasher;
use types::Hash256;
pub fn keccak256(bytes: &[u8]) -> Hash256 {
Hash256::from(ethers_core::utils::keccak256(bytes))
}
/// Keccak hasher.
#[derive(Default, Debug, Clone, PartialEq)]
pub struct KeccakHasher;
impl Hasher for KeccakHasher {
type Out = Hash256;
type StdHasher = Hash256StdHasher;
const LENGTH: usize = 32;
fn hash(x: &[u8]) -> Self::Out {
keccak256(x)
}
}

7
beacon_node/execution_layer/src/lib.rs
View File

@ -44,8 +44,10 @@ use types::{
ExecutionPayload, ExecutionPayloadCapella, ExecutionPayloadEip4844, ExecutionPayloadMerge, ExecutionPayload, ExecutionPayloadCapella, ExecutionPayloadEip4844, ExecutionPayloadMerge,
}; };
mod block_hash;
mod engine_api; mod engine_api;
mod engines; mod engines;
mod keccak;
mod metrics; mod metrics;
pub mod payload_cache; pub mod payload_cache;
mod payload_status; mod payload_status;
@ -94,6 +96,11 @@ pub enum Error {
ShuttingDown, ShuttingDown,
FeeRecipientUnspecified, FeeRecipientUnspecified,
MissingLatestValidHash, MissingLatestValidHash,
BlockHashMismatch {
computed: ExecutionBlockHash,
payload: ExecutionBlockHash,
transactions_root: Hash256,
},
InvalidJWTSecret(String), InvalidJWTSecret(String),
BeaconStateError(BeaconStateError), BeaconStateError(BeaconStateError),
} }

5
beacon_node/execution_layer/src/metrics.rs
View File

@ -45,6 +45,11 @@ lazy_static::lazy_static! {
"execution_layer_get_payload_by_block_hash_time", "execution_layer_get_payload_by_block_hash_time",
"Time to reconstruct a payload from the EE using eth_getBlockByHash" "Time to reconstruct a payload from the EE using eth_getBlockByHash"
); );
pub static ref EXECUTION_LAYER_VERIFY_BLOCK_HASH: Result<Histogram> = try_create_histogram_with_buckets(
"execution_layer_verify_block_hash_time",
"Time to verify the execution block hash in Lighthouse, without the EL",
Ok(vec![10e-6, 50e-6, 100e-6, 500e-6, 1e-3, 5e-3, 10e-3, 50e-3, 100e-3, 500e-3]),
);
pub static ref EXECUTION_LAYER_PAYLOAD_STATUS: Result<IntCounterVec> = try_create_int_counter_vec( pub static ref EXECUTION_LAYER_PAYLOAD_STATUS: Result<IntCounterVec> = try_create_int_counter_vec(
"execution_layer_payload_status", "execution_layer_payload_status",
"Indicates the payload status returned for a particular method", "Indicates the payload status returned for a particular method",

20
beacon_node/http_api/tests/common.rs
View File

@ -7,7 +7,13 @@ use eth2::{BeaconNodeHttpClient, Timeouts};
use http_api::{Config, Context}; use http_api::{Config, Context};
use lighthouse_network::{ use lighthouse_network::{
discv5::enr::{CombinedKey, EnrBuilder}, discv5::enr::{CombinedKey, EnrBuilder},
libp2p::{core::connection::ConnectionId, swarm::NetworkBehaviour}, libp2p::{
core::connection::ConnectionId,
swarm::{
behaviour::{ConnectionEstablished, FromSwarm},
NetworkBehaviour,
},
},
rpc::methods::{MetaData, MetaDataV2}, rpc::methods::{MetaData, MetaDataV2},
types::{EnrAttestationBitfield, EnrSyncCommitteeBitfield, SyncState}, types::{EnrAttestationBitfield, EnrSyncCommitteeBitfield, SyncState},
ConnectedPoint, Enr, NetworkGlobals, PeerId, PeerManager, ConnectedPoint, Enr, NetworkGlobals, PeerId, PeerManager,
@ -143,12 +149,18 @@ pub async fn create_api_server_on_port<T: BeaconChainTypes>(
// add a peer // add a peer
let peer_id = PeerId::random(); let peer_id = PeerId::random();
let connected_point = ConnectedPoint::Listener { let endpoint = &ConnectedPoint::Listener {
local_addr: EXTERNAL_ADDR.parse().unwrap(), local_addr: EXTERNAL_ADDR.parse().unwrap(),
send_back_addr: EXTERNAL_ADDR.parse().unwrap(), send_back_addr: EXTERNAL_ADDR.parse().unwrap(),
}; };
let con_id = ConnectionId::new(1); let connection_id = ConnectionId::new(1);
pm.inject_connection_established(&peer_id, &con_id, &connected_point, None, 0); pm.on_swarm_event(FromSwarm::ConnectionEstablished(ConnectionEstablished {
peer_id,
connection_id,
endpoint,
failed_addresses: &[],
other_established: 0,
}));
*network_globals.sync_state.write() = SyncState::Synced; *network_globals.sync_state.write() = SyncState::Synced;
let eth1_service = let eth1_service =

5
beacon_node/lighthouse_network/Cargo.toml
View File

@ -42,11 +42,12 @@ superstruct = "0.5.0"
prometheus-client = "0.18.0" prometheus-client = "0.18.0"
unused_port = { path = "../../common/unused_port" } unused_port = { path = "../../common/unused_port" }
delay_map = "0.1.1" delay_map = "0.1.1"
void = "1"
[dependencies.libp2p] [dependencies.libp2p]
version = "0.48.0" version = "0.50.0"
default-features = false default-features = false
features = ["websocket", "identify", "mplex", "yamux", "noise", "gossipsub", "dns-tokio", "tcp-tokio", "plaintext", "secp256k1"] features = ["websocket", "identify", "mplex", "yamux", "noise", "gossipsub", "dns", "tcp", "tokio", "plaintext", "secp256k1", "macros", "ecdsa"]
[dev-dependencies] [dev-dependencies]
slog-term = "2.6.0" slog-term = "2.6.0"

87
beacon_node/lighthouse_network/src/discovery/mod.rs
View File

@ -22,12 +22,13 @@ use enr::{ATTESTATION_BITFIELD_ENR_KEY, ETH2_ENR_KEY, SYNC_COMMITTEE_BITFIELD_EN
use futures::prelude::*; use futures::prelude::*;
use futures::stream::FuturesUnordered; use futures::stream::FuturesUnordered;
use libp2p::multiaddr::Protocol; use libp2p::multiaddr::Protocol;
use libp2p::swarm::behaviour::{DialFailure, FromSwarm};
use libp2p::swarm::AddressScore; use libp2p::swarm::AddressScore;
pub use libp2p::{ pub use libp2p::{
core::{connection::ConnectionId, ConnectedPoint, Multiaddr, PeerId}, core::{connection::ConnectionId, ConnectedPoint, Multiaddr, PeerId},
swarm::{ swarm::{
handler::ConnectionHandler, DialError, NetworkBehaviour, dummy::ConnectionHandler, DialError, NetworkBehaviour, NetworkBehaviourAction as NBAction,
NetworkBehaviourAction as NBAction, NotifyHandler, PollParameters, SubstreamProtocol, NotifyHandler, PollParameters, SubstreamProtocol,
}, },
}; };
use lru::LruCache; use lru::LruCache;
@ -927,11 +928,11 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> { impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> {
// Discovery is not a real NetworkBehaviour... // Discovery is not a real NetworkBehaviour...
type ConnectionHandler = libp2p::swarm::handler::DummyConnectionHandler; type ConnectionHandler = ConnectionHandler;
type OutEvent = DiscoveredPeers; type OutEvent = DiscoveredPeers;
fn new_handler(&mut self) -> Self::ConnectionHandler { fn new_handler(&mut self) -> Self::ConnectionHandler {
libp2p::swarm::handler::DummyConnectionHandler::default() ConnectionHandler
} }
// Handles the libp2p request to obtain multiaddrs for peer_id's in order to dial them. // Handles the libp2p request to obtain multiaddrs for peer_id's in order to dial them.
@ -947,40 +948,6 @@ impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> {
} }
} }
fn inject_event(
&mut self,
_: PeerId,
_: ConnectionId,
_: <Self::ConnectionHandler as ConnectionHandler>::OutEvent,
) {
}
fn inject_dial_failure(
&mut self,
peer_id: Option<PeerId>,
_handler: Self::ConnectionHandler,
error: &DialError,
) {
if let Some(peer_id) = peer_id {
match error {
DialError::Banned
| DialError::LocalPeerId
| DialError::InvalidPeerId(_)
| DialError::ConnectionIo(_)
| DialError::NoAddresses
| DialError::Transport(_)
| DialError::WrongPeerId { .. } => {
// set peer as disconnected in discovery DHT
debug!(self.log, "Marking peer disconnected in DHT"; "peer_id" => %peer_id);
self.disconnect_peer(&peer_id);
}
DialError::ConnectionLimit(_)
| DialError::DialPeerConditionFalse(_)
| DialError::Aborted => {}
}
}
}
// Main execution loop to drive the behaviour // Main execution loop to drive the behaviour
fn poll( fn poll(
&mut self, &mut self,
@ -1067,6 +1034,50 @@ impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> {
} }
Poll::Pending Poll::Pending
} }
fn on_swarm_event(&mut self, event: FromSwarm<Self::ConnectionHandler>) {
match event {
FromSwarm::DialFailure(DialFailure { peer_id, error, .. }) => {
self.on_dial_failure(peer_id, error)
}
FromSwarm::ConnectionEstablished(_)
| FromSwarm::ConnectionClosed(_)
| FromSwarm::AddressChange(_)
| FromSwarm::ListenFailure(_)
| FromSwarm::NewListener(_)
| FromSwarm::NewListenAddr(_)
| FromSwarm::ExpiredListenAddr(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::ExpiredExternalAddr(_) => {
// Ignore events not relevant to discovery
}
}
}
}
impl<TSpec: EthSpec> Discovery<TSpec> {
fn on_dial_failure(&mut self, peer_id: Option<PeerId>, error: &DialError) {
if let Some(peer_id) = peer_id {
match error {
DialError::Banned
| DialError::LocalPeerId
| DialError::InvalidPeerId(_)
| DialError::ConnectionIo(_)
| DialError::NoAddresses
| DialError::Transport(_)
| DialError::WrongPeerId { .. } => {
// set peer as disconnected in discovery DHT
debug!(self.log, "Marking peer disconnected in DHT"; "peer_id" => %peer_id);
self.disconnect_peer(&peer_id);
}
DialError::ConnectionLimit(_)
| DialError::DialPeerConditionFalse(_)
| DialError::Aborted => {}
}
}
}
} }
#[cfg(test)] #[cfg(test)]

2
beacon_node/lighthouse_network/src/peer_manager/mod.rs
View File

@ -7,7 +7,7 @@ use crate::{NetworkGlobals, PeerId};
use crate::{Subnet, SubnetDiscovery}; use crate::{Subnet, SubnetDiscovery};
use delay_map::HashSetDelay; use delay_map::HashSetDelay;
use discv5::Enr; use discv5::Enr;
use libp2p::identify::IdentifyInfo; use libp2p::identify::Info as IdentifyInfo;
use peerdb::{client::ClientKind, BanOperation, BanResult, ScoreUpdateResult}; use peerdb::{client::ClientKind, BanOperation, BanResult, ScoreUpdateResult};
use rand::seq::SliceRandom; use rand::seq::SliceRandom;
use slog::{debug, error, trace, warn}; use slog::{debug, error, trace, warn};

127
beacon_node/lighthouse_network/src/peer_manager/network_behaviour.rs
View File

@ -1,14 +1,12 @@
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use futures::StreamExt; use futures::StreamExt;
use libp2p::core::connection::ConnectionId;
use libp2p::core::ConnectedPoint; use libp2p::core::ConnectedPoint;
use libp2p::swarm::behaviour::{ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm};
use libp2p::swarm::dial_opts::{DialOpts, PeerCondition}; use libp2p::swarm::dial_opts::{DialOpts, PeerCondition};
use libp2p::swarm::handler::DummyConnectionHandler; use libp2p::swarm::dummy::ConnectionHandler;
use libp2p::swarm::{ use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
ConnectionHandler, DialError, NetworkBehaviour, NetworkBehaviourAction, PollParameters, use libp2p::PeerId;
};
use libp2p::{Multiaddr, PeerId};
use slog::{debug, error}; use slog::{debug, error};
use types::EthSpec; use types::EthSpec;
@ -20,23 +18,14 @@ use super::peerdb::BanResult;
use super::{ConnectingType, PeerManager, PeerManagerEvent, ReportSource}; use super::{ConnectingType, PeerManager, PeerManagerEvent, ReportSource};
impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> { impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
type ConnectionHandler = DummyConnectionHandler; type ConnectionHandler = ConnectionHandler;
type OutEvent = PeerManagerEvent; type OutEvent = PeerManagerEvent;
/* Required trait members */ /* Required trait members */
fn new_handler(&mut self) -> Self::ConnectionHandler { fn new_handler(&mut self) -> Self::ConnectionHandler {
DummyConnectionHandler::default() ConnectionHandler
}
fn inject_event(
&mut self,
_: PeerId,
_: ConnectionId,
_: <DummyConnectionHandler as ConnectionHandler>::OutEvent,
) {
unreachable!("Dummy handler does not emit events")
} }
fn poll( fn poll(
@ -114,19 +103,46 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
Poll::Pending Poll::Pending
} }
/* Overwritten trait members */ fn on_swarm_event(&mut self, event: FromSwarm<Self::ConnectionHandler>) {
match event {
FromSwarm::ConnectionEstablished(ConnectionEstablished {
peer_id,
endpoint,
other_established,
..
}) => self.on_connection_established(peer_id, endpoint, other_established),
FromSwarm::ConnectionClosed(ConnectionClosed {
peer_id,
remaining_established,
..
}) => self.on_connection_closed(peer_id, remaining_established),
FromSwarm::DialFailure(DialFailure { peer_id, .. }) => self.on_dial_failure(peer_id),
FromSwarm::AddressChange(_)
| FromSwarm::ListenFailure(_)
| FromSwarm::NewListener(_)
| FromSwarm::NewListenAddr(_)
| FromSwarm::ExpiredListenAddr(_)
| FromSwarm::ListenerError(_)
| FromSwarm::ListenerClosed(_)
| FromSwarm::NewExternalAddr(_)
| FromSwarm::ExpiredExternalAddr(_) => {
// The rest of the events we ignore since they are handled in their associated
// `SwarmEvent`
}
}
}
}
fn inject_connection_established( impl<TSpec: EthSpec> PeerManager<TSpec> {
fn on_connection_established(
&mut self, &mut self,
peer_id: &PeerId, peer_id: PeerId,
_connection_id: &ConnectionId,
endpoint: &ConnectedPoint, endpoint: &ConnectedPoint,
_failed_addresses: Option<&Vec<Multiaddr>>,
other_established: usize, other_established: usize,
) { ) {
debug!(self.log, "Connection established"; "peer_id" => %peer_id, "connection" => ?endpoint.to_endpoint()); debug!(self.log, "Connection established"; "peer_id" => %peer_id, "connection" => ?endpoint.to_endpoint());
if other_established == 0 { if other_established == 0 {
self.events.push(PeerManagerEvent::MetaData(*peer_id)); self.events.push(PeerManagerEvent::MetaData(peer_id));
} }
// Check NAT if metrics are enabled // Check NAT if metrics are enabled
@ -135,20 +151,20 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
} }
// Check to make sure the peer is not supposed to be banned // Check to make sure the peer is not supposed to be banned
match self.ban_status(peer_id) { match self.ban_status(&peer_id) {
// TODO: directly emit the ban event? // TODO: directly emit the ban event?
BanResult::BadScore => { BanResult::BadScore => {
// This is a faulty state // This is a faulty state
error!(self.log, "Connected to a banned peer. Re-banning"; "peer_id" => %peer_id); error!(self.log, "Connected to a banned peer. Re-banning"; "peer_id" => %peer_id);
// Reban the peer // Reban the peer
self.goodbye_peer(peer_id, GoodbyeReason::Banned, ReportSource::PeerManager); self.goodbye_peer(&peer_id, GoodbyeReason::Banned, ReportSource::PeerManager);
return; return;
} }
BanResult::BannedIp(ip_addr) => { BanResult::BannedIp(ip_addr) => {
// A good peer has connected to us via a banned IP address. We ban the peer and // A good peer has connected to us via a banned IP address. We ban the peer and
// prevent future connections. // prevent future connections.
debug!(self.log, "Peer connected via banned IP. Banning"; "peer_id" => %peer_id, "banned_ip" => %ip_addr); debug!(self.log, "Peer connected via banned IP. Banning"; "peer_id" => %peer_id, "banned_ip" => %ip_addr);
self.goodbye_peer(peer_id, GoodbyeReason::BannedIP, ReportSource::PeerManager); self.goodbye_peer(&peer_id, GoodbyeReason::BannedIP, ReportSource::PeerManager);
return; return;
} }
BanResult::NotBanned => {} BanResult::NotBanned => {}
@ -162,11 +178,11 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
.network_globals .network_globals
.peers .peers
.read() .read()
.peer_info(peer_id) .peer_info(&peer_id)
.map_or(true, |peer| !peer.has_future_duty()) .map_or(true, |peer| !peer.has_future_duty())
{ {
// Gracefully disconnect the peer. // Gracefully disconnect the peer.
self.disconnect_peer(*peer_id, GoodbyeReason::TooManyPeers); self.disconnect_peer(peer_id, GoodbyeReason::TooManyPeers);
return; return;
} }
@ -174,14 +190,14 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
// does not need to know about these peers. // does not need to know about these peers.
match endpoint { match endpoint {
ConnectedPoint::Listener { send_back_addr, .. } => { ConnectedPoint::Listener { send_back_addr, .. } => {
self.inject_connect_ingoing(peer_id, send_back_addr.clone(), None); self.inject_connect_ingoing(&peer_id, send_back_addr.clone(), None);
self.events self.events
.push(PeerManagerEvent::PeerConnectedIncoming(*peer_id)); .push(PeerManagerEvent::PeerConnectedIncoming(peer_id));
} }
ConnectedPoint::Dialer { address, .. } => { ConnectedPoint::Dialer { address, .. } => {
self.inject_connect_outgoing(peer_id, address.clone(), None); self.inject_connect_outgoing(&peer_id, address.clone(), None);
self.events self.events
.push(PeerManagerEvent::PeerConnectedOutgoing(*peer_id)); .push(PeerManagerEvent::PeerConnectedOutgoing(peer_id));
} }
} }
@ -189,14 +205,8 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
self.update_connected_peer_metrics(); self.update_connected_peer_metrics();
metrics::inc_counter(&metrics::PEER_CONNECT_EVENT_COUNT); metrics::inc_counter(&metrics::PEER_CONNECT_EVENT_COUNT);
} }
fn inject_connection_closed(
&mut self, fn on_connection_closed(&mut self, peer_id: PeerId, remaining_established: usize) {
peer_id: &PeerId,
_: &ConnectionId,
_: &ConnectedPoint,
_: DummyConnectionHandler,
remaining_established: usize,
) {
if remaining_established > 0 { if remaining_established > 0 {
return; return;
} }
@ -206,62 +216,33 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
.network_globals .network_globals
.peers .peers
.read() .read()
.is_connected_or_disconnecting(peer_id) .is_connected_or_disconnecting(&peer_id)
{ {
// We are disconnecting the peer or the peer has already been connected. // We are disconnecting the peer or the peer has already been connected.
// Both these cases, the peer has been previously registered by the peer manager and // Both these cases, the peer has been previously registered by the peer manager and
// potentially the application layer. // potentially the application layer.
// Inform the application. // Inform the application.
self.events self.events
.push(PeerManagerEvent::PeerDisconnected(*peer_id)); .push(PeerManagerEvent::PeerDisconnected(peer_id));
debug!(self.log, "Peer disconnected"; "peer_id" => %peer_id); debug!(self.log, "Peer disconnected"; "peer_id" => %peer_id);
} }
// NOTE: It may be the case that a rejected node, due to too many peers is disconnected // NOTE: It may be the case that a rejected node, due to too many peers is disconnected
// here and the peer manager has no knowledge of its connection. We insert it here for // here and the peer manager has no knowledge of its connection. We insert it here for
// reference so that peer manager can track this peer. // reference so that peer manager can track this peer.
self.inject_disconnect(peer_id); self.inject_disconnect(&peer_id);
// Update the prometheus metrics // Update the prometheus metrics
self.update_connected_peer_metrics(); self.update_connected_peer_metrics();
metrics::inc_counter(&metrics::PEER_DISCONNECT_EVENT_COUNT); metrics::inc_counter(&metrics::PEER_DISCONNECT_EVENT_COUNT);
} }
fn inject_address_change(
&mut self,
_peer_id: &PeerId,
_connection_id: &ConnectionId,
old: &ConnectedPoint,
new: &ConnectedPoint,
) {
debug_assert!(
matches!(
(old, new),
(
// inbound remains inbound
ConnectedPoint::Listener { .. },
ConnectedPoint::Listener { .. }
) | (
// outbound remains outbound
ConnectedPoint::Dialer { .. },
ConnectedPoint::Dialer { .. }
)
),
"A peer has changed between inbound and outbound"
)
}
/// A dial attempt has failed. /// A dial attempt has failed.
/// ///
/// NOTE: It can be the case that we are dialing a peer and during the dialing process the peer /// NOTE: It can be the case that we are dialing a peer and during the dialing process the peer
/// connects and the dial attempt later fails. To handle this, we only update the peer_db if /// connects and the dial attempt later fails. To handle this, we only update the peer_db if
/// the peer is not already connected. /// the peer is not already connected.
fn inject_dial_failure( fn on_dial_failure(&mut self, peer_id: Option<PeerId>) {
&mut self,
peer_id: Option<PeerId>,
_handler: DummyConnectionHandler,
_error: &DialError,
) {
if let Some(peer_id) = peer_id { if let Some(peer_id) = peer_id {
if !self.network_globals.peers.read().is_connected(&peer_id) { if !self.network_globals.peers.read().is_connected(&peer_id) {
self.inject_disconnect(&peer_id); self.inject_disconnect(&peer_id);

2
beacon_node/lighthouse_network/src/peer_manager/peerdb/client.rs
View File

@ -2,7 +2,7 @@
//! //!
//! Currently using identify to fingerprint. //! Currently using identify to fingerprint.
use libp2p::identify::IdentifyInfo; use libp2p::identify::Info as IdentifyInfo;
use serde::Serialize; use serde::Serialize;
use strum::{AsRefStr, EnumIter, IntoStaticStr}; use strum::{AsRefStr, EnumIter, IntoStaticStr};

113
beacon_node/lighthouse_network/src/rpc/handler.rs
View File

@ -327,61 +327,6 @@ where
self.listen_protocol.clone() self.listen_protocol.clone()
} }
fn inject_fully_negotiated_inbound(
&mut self,
substream: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output,
_info: Self::InboundOpenInfo,
) {
// only accept new peer requests when active
if !matches!(self.state, HandlerState::Active) {
return;
}
let (req, substream) = substream;
let expected_responses = req.expected_responses();
// store requests that expect responses
if expected_responses > 0 {
if self.inbound_substreams.len() < MAX_INBOUND_SUBSTREAMS {
// Store the stream and tag the output.
let delay_key = self.inbound_substreams_delay.insert(
self.current_inbound_substream_id,
Duration::from_secs(RESPONSE_TIMEOUT),
);
let awaiting_stream = InboundState::Idle(substream);
self.inbound_substreams.insert(
self.current_inbound_substream_id,
InboundInfo {
state: awaiting_stream,
pending_items: VecDeque::with_capacity(expected_responses as usize),
delay_key: Some(delay_key),
protocol: req.protocol(),
request_start_time: Instant::now(),
remaining_chunks: expected_responses,
},
);
} else {
self.events_out.push(Err(HandlerErr::Inbound {
id: self.current_inbound_substream_id,
proto: req.protocol(),
error: RPCError::HandlerRejected,
}));
return self.shutdown(None);
}
}
// If we received a goodbye, shutdown the connection.
if let InboundRequest::Goodbye(_) = req {
self.shutdown(None);
}
self.events_out.push(Ok(RPCReceived::Request(
self.current_inbound_substream_id,
req,
)));
self.current_inbound_substream_id.0 += 1;
}
fn inject_fully_negotiated_outbound( fn inject_fully_negotiated_outbound(
&mut self, &mut self,
out: <Self::OutboundProtocol as OutboundUpgrade<NegotiatedSubstream>>::Output, out: <Self::OutboundProtocol as OutboundUpgrade<NegotiatedSubstream>>::Output,
@ -438,6 +383,64 @@ where
} }
} }
fn inject_fully_negotiated_inbound(
&mut self,
substream: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output,
_info: Self::InboundOpenInfo,
) {
// only accept new peer requests when active
if !matches!(self.state, HandlerState::Active) {
return;
}
let (req, substream) = substream;
let expected_responses = req.expected_responses();
// store requests that expect responses
if expected_responses > 0 {
if self.inbound_substreams.len() < MAX_INBOUND_SUBSTREAMS {
// Store the stream and tag the output.
let delay_key = self.inbound_substreams_delay.insert(
self.current_inbound_substream_id,
Duration::from_secs(RESPONSE_TIMEOUT),
);
let awaiting_stream = InboundState::Idle(substream);
self.inbound_substreams.insert(
self.current_inbound_substream_id,
InboundInfo {
state: awaiting_stream,
pending_items: VecDeque::with_capacity(std::cmp::min(
expected_responses,
128,
) as usize),
delay_key: Some(delay_key),
protocol: req.protocol(),
request_start_time: Instant::now(),
remaining_chunks: expected_responses,
},
);
} else {
self.events_out.push(Err(HandlerErr::Inbound {
id: self.current_inbound_substream_id,
proto: req.protocol(),
error: RPCError::HandlerRejected,
}));
return self.shutdown(None);
}
}
// If we received a goodbye, shutdown the connection.
if let InboundRequest::Goodbye(_) = req {
self.shutdown(None);
}
self.events_out.push(Ok(RPCReceived::Request(
self.current_inbound_substream_id,
req,
)));
self.current_inbound_substream_id.0 += 1;
}
fn inject_event(&mut self, rpc_event: Self::InEvent) { fn inject_event(&mut self, rpc_event: Self::InEvent) {
match rpc_event { match rpc_event {
RPCSend::Request(id, req) => self.send_request(id, req), RPCSend::Request(id, req) => self.send_request(id, req),

4
beacon_node/lighthouse_network/src/service/behaviour.rs
View File

@ -7,8 +7,8 @@ use libp2p::gossipsub::subscription_filter::{
MaxCountSubscriptionFilter, WhitelistSubscriptionFilter, MaxCountSubscriptionFilter, WhitelistSubscriptionFilter,
}; };
use libp2p::gossipsub::Gossipsub as BaseGossipsub; use libp2p::gossipsub::Gossipsub as BaseGossipsub;
use libp2p::identify::Identify; use libp2p::identify::Behaviour as Identify;
use libp2p::NetworkBehaviour; use libp2p::swarm::NetworkBehaviour;
use types::EthSpec; use types::EthSpec;
use super::api_types::RequestId; use super::api_types::RequestId;

20
beacon_node/lighthouse_network/src/service/mod.rs
View File

@ -29,7 +29,7 @@ use libp2p::gossipsub::subscription_filter::MaxCountSubscriptionFilter;
use libp2p::gossipsub::{ use libp2p::gossipsub::{
GossipsubEvent, IdentTopic as Topic, MessageAcceptance, MessageAuthenticity, MessageId, GossipsubEvent, IdentTopic as Topic, MessageAcceptance, MessageAuthenticity, MessageId,
}; };
use libp2p::identify::{Identify, IdentifyConfig, IdentifyEvent}; use libp2p::identify::{Behaviour as Identify, Config as IdentifyConfig, Event as IdentifyEvent};
use libp2p::multiaddr::{Multiaddr, Protocol as MProtocol}; use libp2p::multiaddr::{Multiaddr, Protocol as MProtocol};
use libp2p::swarm::{ConnectionLimits, Swarm, SwarmBuilder, SwarmEvent}; use libp2p::swarm::{ConnectionLimits, Swarm, SwarmBuilder, SwarmEvent};
use libp2p::PeerId; use libp2p::PeerId;
@ -318,7 +318,7 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
// use the executor for libp2p // use the executor for libp2p
struct Executor(task_executor::TaskExecutor); struct Executor(task_executor::TaskExecutor);
impl libp2p::core::Executor for Executor { impl libp2p::swarm::Executor for Executor {
fn exec(&self, f: Pin<Box<dyn futures::Future<Output = ()> + Send>>) { fn exec(&self, f: Pin<Box<dyn futures::Future<Output = ()> + Send>>) {
self.0.spawn(f, "libp2p"); self.0.spawn(f, "libp2p");
} }
@ -343,12 +343,16 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
.with_max_established_per_peer(Some(MAX_CONNECTIONS_PER_PEER)); .with_max_established_per_peer(Some(MAX_CONNECTIONS_PER_PEER));
( (
SwarmBuilder::new(transport, behaviour, local_peer_id) SwarmBuilder::with_executor(
.notify_handler_buffer_size(std::num::NonZeroUsize::new(7).expect("Not zero")) transport,
.connection_event_buffer_size(64) behaviour,
.connection_limits(limits) local_peer_id,
.executor(Box::new(Executor(executor))) Executor(executor),
.build(), )
.notify_handler_buffer_size(std::num::NonZeroUsize::new(7).expect("Not zero"))
.connection_event_buffer_size(64)
.connection_limits(limits)
.build(),
bandwidth, bandwidth,
) )
}; };

3
beacon_node/lighthouse_network/src/service/utils.rs
View File

@ -44,8 +44,7 @@ type BoxedTransport = Boxed<(PeerId, StreamMuxerBox)>;
pub fn build_transport( pub fn build_transport(
local_private_key: Keypair, local_private_key: Keypair,
) -> std::io::Result<(BoxedTransport, Arc<BandwidthSinks>)> { ) -> std::io::Result<(BoxedTransport, Arc<BandwidthSinks>)> {
let tcp = let tcp = libp2p::tcp::tokio::Transport::new(libp2p::tcp::Config::default().nodelay(true));
libp2p::tcp::TokioTcpTransport::new(libp2p::tcp::GenTcpConfig::default().nodelay(true));
let transport = libp2p::dns::TokioDnsConfig::system(tcp)?; let transport = libp2p::dns::TokioDnsConfig::system(tcp)?;
#[cfg(feature = "libp2p-websocket")] #[cfg(feature = "libp2p-websocket")]
let transport = { let transport = {

7
beacon_node/lighthouse_network/tests/common/mod.rs → beacon_node/lighthouse_network/tests/common.rs
View File

@ -15,13 +15,6 @@ use types::{
}; };
use unused_port::unused_tcp_port; use unused_port::unused_tcp_port;
#[allow(clippy::type_complexity)]
#[allow(unused)]
pub mod behaviour;
#[allow(clippy::type_complexity)]
#[allow(unused)]
pub mod swarm;
type E = MinimalEthSpec; type E = MinimalEthSpec;
type ReqId = usize; type ReqId = usize;

395
beacon_node/lighthouse_network/tests/common/behaviour.rs
View File

@ -1,395 +0,0 @@
// NOTE: Taken from libp2p's swarm's testing utils.
//
// Copyright 2020 Parity Technologies (UK) Ltd.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the "Software"),
// to deal in the Software without restriction, including without limitation
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
// and/or sell copies of the Software, and to permit persons to whom the
// Software is furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use std::collections::HashMap;
use std::task::{Context, Poll};
use libp2p::core::connection::{ConnectedPoint, ConnectionId};
use libp2p::core::transport::ListenerId;
use libp2p::swarm::handler::{ConnectionHandler, DummyConnectionHandler, IntoConnectionHandler};
use libp2p::swarm::{DialError, NetworkBehaviour, NetworkBehaviourAction, PollParameters};
use libp2p::{Multiaddr, PeerId};
/// A `MockBehaviour` is a `NetworkBehaviour` that allows for
/// the instrumentation of return values, without keeping
/// any further state.
pub struct MockBehaviour<
THandler = DummyConnectionHandler,
TOutEvent = <DummyConnectionHandler as ConnectionHandler>::OutEvent,
> where
THandler: ConnectionHandler,
{
/// The prototype protocols handler that is cloned for every
/// invocation of `new_handler`.
pub handler_proto: THandler,
/// The addresses to return from `addresses_of_peer`.
pub addresses: HashMap<PeerId, Vec<Multiaddr>>,
/// The next action to return from `poll`.
///
/// An action is only returned once.
pub next_action: Option<NetworkBehaviourAction<TOutEvent, THandler>>,
}
impl<THandler, TOutEvent> MockBehaviour<THandler, TOutEvent>
where
THandler: ConnectionHandler,
{
pub fn new(handler_proto: THandler) -> Self {
MockBehaviour {
handler_proto,
addresses: HashMap::new(),
next_action: None,
}
}
}
impl<THandler, TOutEvent> NetworkBehaviour for MockBehaviour<THandler, TOutEvent>
where
THandler: ConnectionHandler + Clone,
THandler::OutEvent: Clone,
TOutEvent: Send + 'static,
{
type ConnectionHandler = THandler;
type OutEvent = TOutEvent;
fn new_handler(&mut self) -> Self::ConnectionHandler {
self.handler_proto.clone()
}
fn addresses_of_peer(&mut self, p: &PeerId) -> Vec<Multiaddr> {
self.addresses.get(p).map_or(Vec::new(), |v| v.clone())
}
fn inject_event(&mut self, _: PeerId, _: ConnectionId, _: THandler::OutEvent) {}
fn poll(
&mut self,
_: &mut Context,
_: &mut impl PollParameters,
) -> Poll<NetworkBehaviourAction<Self::OutEvent, Self::ConnectionHandler>> {
Option::take(&mut self.next_action).map_or(Poll::Pending, Poll::Ready)
}
}
/// A `CallTraceBehaviour` is a `NetworkBehaviour` that tracks invocations of callback methods and
/// their arguments, wrapping around an inner behaviour. It ensures certain invariants are met.
pub struct CallTraceBehaviour<TInner>
where
TInner: NetworkBehaviour,
{
inner: TInner,
pub addresses_of_peer: Vec<PeerId>,
pub inject_connection_established: Vec<(PeerId, ConnectionId, ConnectedPoint, usize)>,
pub inject_connection_closed: Vec<(PeerId, ConnectionId, ConnectedPoint, usize)>,
pub inject_event: Vec<(
PeerId,
ConnectionId,
<<TInner::ConnectionHandler as IntoConnectionHandler>::Handler as ConnectionHandler>::OutEvent,
)>,
pub inject_dial_failure: Vec<Option<PeerId>>,
pub inject_new_listener: Vec<ListenerId>,
pub inject_new_listen_addr: Vec<(ListenerId, Multiaddr)>,
pub inject_new_external_addr: Vec<Multiaddr>,
pub inject_expired_listen_addr: Vec<(ListenerId, Multiaddr)>,
pub inject_expired_external_addr: Vec<Multiaddr>,
pub inject_listener_error: Vec<ListenerId>,
pub inject_listener_closed: Vec<(ListenerId, bool)>,
pub poll: usize,
}
impl<TInner> CallTraceBehaviour<TInner>
where
TInner: NetworkBehaviour,
{
pub fn new(inner: TInner) -> Self {
Self {
inner,
addresses_of_peer: Vec::new(),
inject_connection_established: Vec::new(),
inject_connection_closed: Vec::new(),
inject_event: Vec::new(),
inject_dial_failure: Vec::new(),
inject_new_listener: Vec::new(),
inject_new_listen_addr: Vec::new(),
inject_new_external_addr: Vec::new(),
inject_expired_listen_addr: Vec::new(),
inject_expired_external_addr: Vec::new(),
inject_listener_error: Vec::new(),
inject_listener_closed: Vec::new(),
poll: 0,
}
}
#[allow(dead_code)]
pub fn reset(&mut self) {
self.addresses_of_peer = Vec::new();
self.inject_connection_established = Vec::new();
self.inject_connection_closed = Vec::new();
self.inject_event = Vec::new();
self.inject_dial_failure = Vec::new();
self.inject_new_listen_addr = Vec::new();
self.inject_new_external_addr = Vec::new();
self.inject_expired_listen_addr = Vec::new();
self.inject_listener_error = Vec::new();
self.inject_listener_closed = Vec::new();
self.poll = 0;
}
pub fn inner(&mut self) -> &mut TInner {
&mut self.inner
}
/// Checks that when the expected number of closed connection notifications are received, a
/// given number of expected disconnections have been received as well.
///
/// Returns if the first condition is met.
pub fn assert_disconnected(
&self,
expected_closed_connections: usize,
expected_disconnections: usize,
) -> bool {
if self.inject_connection_closed.len() == expected_closed_connections {
assert_eq!(
self.inject_connection_closed
.iter()
.filter(|(.., remaining_established)| { *remaining_established == 0 })
.count(),
expected_disconnections
);
return true;
}
false
}
/// Checks that when the expected number of established connection notifications are received,
/// a given number of expected connections have been received as well.
///
/// Returns if the first condition is met.
pub fn assert_connected(
&self,
expected_established_connections: usize,
expected_connections: usize,
) -> bool {
if self.inject_connection_established.len() == expected_established_connections {
assert_eq!(
self.inject_connection_established
.iter()
.filter(|(.., reported_aditional_connections)| {
*reported_aditional_connections == 0
})
.count(),
expected_connections
);
return true;
}
false
}
}
impl<TInner> NetworkBehaviour for CallTraceBehaviour<TInner>
where
TInner: NetworkBehaviour,
<<TInner::ConnectionHandler as IntoConnectionHandler>::Handler as ConnectionHandler>::OutEvent:
Clone,
{
type ConnectionHandler = TInner::ConnectionHandler;
type OutEvent = TInner::OutEvent;
fn new_handler(&mut self) -> Self::ConnectionHandler {
self.inner.new_handler()
}
fn addresses_of_peer(&mut self, p: &PeerId) -> Vec<Multiaddr> {
self.addresses_of_peer.push(*p);
self.inner.addresses_of_peer(p)
}
fn inject_connection_established(
&mut self,
p: &PeerId,
c: &ConnectionId,
e: &ConnectedPoint,
errors: Option<&Vec<Multiaddr>>,
other_established: usize,
) {
let mut other_peer_connections = self
.inject_connection_established
.iter()
.rev() // take last to first
.filter_map(|(peer, .., other_established)| {
if p == peer {
Some(other_established)
} else {
None
}
})
.take(other_established);
// We are informed that there are `other_established` additional connections. Ensure that the
// number of previous connections is consistent with this
if let Some(&prev) = other_peer_connections.next() {
if prev < other_established {
assert_eq!(
prev,
other_established - 1,
"Inconsistent connection reporting"
)
}
assert_eq!(other_peer_connections.count(), other_established - 1);
} else {
assert_eq!(other_established, 0)
}
self.inject_connection_established
.push((*p, *c, e.clone(), other_established));
self.inner
.inject_connection_established(p, c, e, errors, other_established);
}
fn inject_connection_closed(
&mut self,
p: &PeerId,
c: &ConnectionId,
e: &ConnectedPoint,
handler: <Self::ConnectionHandler as IntoConnectionHandler>::Handler,
remaining_established: usize,
) {
let mut other_closed_connections = self
.inject_connection_established
.iter()
.rev() // take last to first
.filter_map(|(peer, .., remaining_established)| {
if p == peer {
Some(remaining_established)
} else {
None
}
})
.take(remaining_established);
// We are informed that there are `other_established` additional connections. Ensure that the
// number of previous connections is consistent with this
if let Some(&prev) = other_closed_connections.next() {
if prev < remaining_established {
assert_eq!(
prev,
remaining_established - 1,
"Inconsistent closed connection reporting"
)
}
assert_eq!(other_closed_connections.count(), remaining_established - 1);
} else {
assert_eq!(remaining_established, 0)
}
assert!(
self.inject_connection_established
.iter()
.any(|(peer, conn_id, endpoint, _)| (peer, conn_id, endpoint) == (p, c, e)),
"`inject_connection_closed` is called only for connections for \
which `inject_connection_established` was called first."
);
self.inject_connection_closed
.push((*p, *c, e.clone(), remaining_established));
self.inner
.inject_connection_closed(p, c, e, handler, remaining_established);
}
fn inject_event(
&mut self,
p: PeerId,
c: ConnectionId,
e: <<Self::ConnectionHandler as IntoConnectionHandler>::Handler as ConnectionHandler>::OutEvent,
) {
assert!(
self.inject_connection_established
.iter()
.any(|(peer_id, conn_id, ..)| *peer_id == p && c == *conn_id),
"`inject_event` is called for reported connections."
);
assert!(
!self
.inject_connection_closed
.iter()
.any(|(peer_id, conn_id, ..)| *peer_id == p && c == *conn_id),
"`inject_event` is never called for closed connections."
);
self.inject_event.push((p, c, e.clone()));
self.inner.inject_event(p, c, e);
}
fn inject_dial_failure(
&mut self,
p: Option<PeerId>,
handler: Self::ConnectionHandler,
error: &DialError,
) {
self.inject_dial_failure.push(p);
self.inner.inject_dial_failure(p, handler, error);
}
fn inject_new_listener(&mut self, id: ListenerId) {
self.inject_new_listener.push(id);
self.inner.inject_new_listener(id);
}
fn inject_new_listen_addr(&mut self, id: ListenerId, a: &Multiaddr) {
self.inject_new_listen_addr.push((id, a.clone()));
self.inner.inject_new_listen_addr(id, a);
}
fn inject_expired_listen_addr(&mut self, id: ListenerId, a: &Multiaddr) {
self.inject_expired_listen_addr.push((id, a.clone()));
self.inner.inject_expired_listen_addr(id, a);
}
fn inject_new_external_addr(&mut self, a: &Multiaddr) {
self.inject_new_external_addr.push(a.clone());
self.inner.inject_new_external_addr(a);
}
fn inject_expired_external_addr(&mut self, a: &Multiaddr) {
self.inject_expired_external_addr.push(a.clone());
self.inner.inject_expired_external_addr(a);
}
fn inject_listener_error(&mut self, l: ListenerId, e: &(dyn std::error::Error + 'static)) {
self.inject_listener_error.push(l);
self.inner.inject_listener_error(l, e);
}
fn inject_listener_closed(&mut self, l: ListenerId, r: Result<(), &std::io::Error>) {
self.inject_listener_closed.push((l, r.is_ok()));
self.inner.inject_listener_closed(l, r);
}
fn poll(
&mut self,
cx: &mut Context,
args: &mut impl PollParameters,
) -> Poll<NetworkBehaviourAction<Self::OutEvent, Self::ConnectionHandler>> {
self.poll += 1;
self.inner.poll(cx, args)
}
}

99
beacon_node/lighthouse_network/tests/common/swarm.rs
View File

@ -1,99 +0,0 @@
use std::collections::HashMap;
use std::pin::Pin;
use super::behaviour::{CallTraceBehaviour, MockBehaviour};
use futures::stream::Stream;
use futures::task::{Context, Poll};
use libp2p::swarm::handler::ConnectionHandler;
use libp2p::swarm::{IntoConnectionHandler, NetworkBehaviour, Swarm, SwarmBuilder, SwarmEvent};
use libp2p::{PeerId, Transport};
use futures::StreamExt;
pub fn new_test_swarm<B>(behaviour: B) -> Swarm<B>
where
B: NetworkBehaviour,
{
let id_keys = libp2p::identity::Keypair::generate_ed25519();
let local_public_key = id_keys.public();
let transport = libp2p::core::transport::MemoryTransport::default()
.upgrade(libp2p::core::upgrade::Version::V1)
.authenticate(libp2p::plaintext::PlainText2Config {
local_public_key: local_public_key.clone(),
})
.multiplex(libp2p::yamux::YamuxConfig::default())
.boxed();
SwarmBuilder::new(transport, behaviour, local_public_key.into()).build()
}
pub fn random_multiaddr() -> libp2p::multiaddr::Multiaddr {
libp2p::multiaddr::Protocol::Memory(rand::random::<u64>()).into()
}
/// Bind a memory multiaddr to a compatible swarm.
pub async fn bind_listener<B: NetworkBehaviour>(
swarm: &mut Swarm<B>,
) -> libp2p::multiaddr::Multiaddr {
swarm.listen_on(random_multiaddr()).unwrap();
match swarm.select_next_some().await {
SwarmEvent::NewListenAddr {
listener_id: _,
address,
} => address,
_ => panic!("Testing swarm's first event should be a new listener"),
}
}
#[derive(Default)]
pub struct SwarmPool<B: NetworkBehaviour> {
swarms: HashMap<PeerId, Swarm<B>>,
}
impl<B: NetworkBehaviour> SwarmPool<B> {
pub fn with_capacity(capacity: usize) -> Self {
Self {
swarms: HashMap::with_capacity(capacity),
}
}
pub fn insert(&mut self, swarm: Swarm<B>) -> PeerId {
let peer_id = *swarm.local_peer_id();
self.swarms.insert(peer_id, swarm);
peer_id
}
pub fn remove(&mut self, peer_id: &PeerId) {
self.swarms.remove(peer_id);
}
pub fn get_mut(&mut self, peer_id: &PeerId) -> Option<&mut Swarm<B>> {
self.swarms.get_mut(peer_id)
}
pub fn swarms(&self) -> &HashMap<PeerId, Swarm<B>> {
&self.swarms
}
pub fn swarms_mut(&mut self) -> &mut HashMap<PeerId, Swarm<B>> {
&mut self.swarms
}
}
impl<B> Stream for SwarmPool<B>
where
B: NetworkBehaviour,
<B as NetworkBehaviour>::ConnectionHandler: ConnectionHandler,
{
type Item = (PeerId,
SwarmEvent<<B as NetworkBehaviour>::OutEvent, <<<B as NetworkBehaviour>::ConnectionHandler as IntoConnectionHandler>::Handler as ConnectionHandler>::Error>);
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let mut polls = self
.get_mut()
.swarms
.iter_mut()
.map(|(&peer_id, swarm)| swarm.map(move |ev| (peer_id, ev)))
.collect::<futures::stream::SelectAll<_>>();
polls.poll_next_unpin(cx)
}
}

203
beacon_node/lighthouse_network/tests/pm_tests.rs
View File

@ -1,203 +0,0 @@
#![cfg(not(debug_assertions))]
mod common;
use std::{
collections::{HashMap, HashSet},
sync::Arc,
};
use common::{
behaviour::{CallTraceBehaviour, MockBehaviour},
swarm,
};
use lighthouse_network::{
peer_manager::{config::Config, PeerManagerEvent},
NetworkGlobals, PeerAction, PeerInfo, PeerManager, ReportSource,
};
use types::MinimalEthSpec as E;
use futures::StreamExt;
use libp2p::{
core::either::EitherError,
swarm::SwarmEvent,
swarm::{handler::DummyConnectionHandler, DummyBehaviour, KeepAlive, Swarm},
NetworkBehaviour,
};
use slog::debug;
/// Struct that mimics the lighthouse_network::Service with respect to handling peer manager
/// events.
// TODO: make this a real struct for more accurate testing.
struct Service {
swarm: Swarm<Behaviour>,
}
impl Service {
async fn select_next_some(&mut self) -> SwarmEvent<Ev, EitherError<void::Void, void::Void>> {
let ev = self.swarm.select_next_some().await;
match &ev {
SwarmEvent::Behaviour(Ev(PeerManagerEvent::Banned(peer_id, _addr_vec))) => {
self.swarm.ban_peer_id(*peer_id);
}
SwarmEvent::Behaviour(Ev(PeerManagerEvent::UnBanned(peer_id, _addr_vec))) => {
self.swarm.unban_peer_id(*peer_id);
}
SwarmEvent::Behaviour(Ev(PeerManagerEvent::DisconnectPeer(peer_id, _reason))) => {
// directly disconnect here.
let _ = self.swarm.disconnect_peer_id(*peer_id);
}
_ => {}
}
ev
}
}
#[derive(Debug)]
struct Ev(PeerManagerEvent);
impl From<void::Void> for Ev {
fn from(_: void::Void) -> Self {
unreachable!("No events are emmited")
}
}
impl From<PeerManagerEvent> for Ev {
fn from(ev: PeerManagerEvent) -> Self {
Ev(ev)
}
}
#[derive(NetworkBehaviour)]
#[behaviour(out_event = "Ev")]
struct Behaviour {
pm_call_trace: CallTraceBehaviour<PeerManager<E>>,
sibling: MockBehaviour,
}
impl Behaviour {
fn new(pm: PeerManager<E>) -> Self {
Behaviour {
pm_call_trace: CallTraceBehaviour::new(pm),
sibling: MockBehaviour::new(DummyConnectionHandler {
// The peer manager votes No, so we make sure the combined handler stays alive this
// way.
keep_alive: KeepAlive::Yes,
}),
}
}
}
#[tokio::test]
async fn banned_peers_consistency() {
let log = common::build_log(slog::Level::Debug, false);
let pm_log = log.new(slog::o!("who" => "[PM]"));
let globals: Arc<NetworkGlobals<E>> = Arc::new(NetworkGlobals::new_test_globals(&log));
// Build the peer manager.
let (mut pm_service, pm_addr) = {
let pm_config = Config {
discovery_enabled: false,
..Default::default()
};
let pm = PeerManager::new(pm_config, globals.clone(), &pm_log).unwrap();
let mut pm_swarm = swarm::new_test_swarm(Behaviour::new(pm));
let pm_addr = swarm::bind_listener(&mut pm_swarm).await;
let service = Service { swarm: pm_swarm };
(service, pm_addr)
};
let excess_banned_peers = 15;
let peers_to_ban =
lighthouse_network::peer_manager::peerdb::MAX_BANNED_PEERS + excess_banned_peers;
// Build all the dummy peers needed.
let (mut swarm_pool, peers) = {
let mut pool = swarm::SwarmPool::with_capacity(peers_to_ban);
let mut peers = HashSet::with_capacity(peers_to_ban);
for _ in 0..peers_to_ban {
let mut peer_swarm =
swarm::new_test_swarm(DummyBehaviour::with_keep_alive(KeepAlive::Yes));
let _peer_addr = swarm::bind_listener(&mut peer_swarm).await;
// It is ok to dial all at the same time since the swarm handles an event at a time.
peer_swarm.dial(pm_addr.clone()).unwrap();
let peer_id = pool.insert(peer_swarm);
peers.insert(peer_id);
}
(pool, peers)
};
// we track banned peers at the swarm level here since there is no access to that info.
let mut swarm_banned_peers = HashMap::with_capacity(peers_to_ban);
let mut peers_unbanned = 0;
let timeout = tokio::time::sleep(tokio::time::Duration::from_secs(30));
futures::pin_mut!(timeout);
loop {
// poll the pm and dummy swarms.
tokio::select! {
pm_event = pm_service.select_next_some() => {
debug!(log, "[PM] {:?}", pm_event);
match pm_event {
SwarmEvent::Behaviour(Ev(ev)) => match ev {
PeerManagerEvent::Banned(peer_id, _) => {
let has_been_unbanned = false;
swarm_banned_peers.insert(peer_id, has_been_unbanned);
}
PeerManagerEvent::UnBanned(peer_id, _) => {
*swarm_banned_peers.get_mut(&peer_id).expect("Unbanned peer must be banned first") = true;
peers_unbanned += 1;
}
_ => {}
}
SwarmEvent::ConnectionEstablished {
peer_id,
endpoint: _,
num_established: _,
concurrent_dial_errors: _,
} => {
assert!(peers.contains(&peer_id));
// now we report the peer as banned.
pm_service
.swarm
.behaviour_mut()
.pm_call_trace
.inner()
.report_peer(
&peer_id,
PeerAction::Fatal,
ReportSource::Processor,
None,
""
);
},
_ => {}
}
}
Some((_peer_id, _peer_ev)) = swarm_pool.next() => {
// we need to poll the swarms to keep the peers going
}
_ = timeout.as_mut() => {
panic!("Test timeout.")
}
}
if peers_unbanned == excess_banned_peers {
let pdb = globals.peers.read();
let inconsistencies = swarm_banned_peers
.into_iter()
.map(|(peer_id, was_unbanned)| {
was_unbanned
!= pdb.peer_info(&peer_id).map_or(
false, /* We forgot about a banned peer */
PeerInfo::is_banned,
)
});
assert_eq!(
inconsistencies
.filter(|is_consistent| *is_consistent)
.count(),
peers_to_ban
);
return;
}
}
}

2
beacon_node/network/Cargo.toml
View File

@ -44,7 +44,7 @@ strum = "0.24.0"
tokio-util = { version = "0.6.3", features = ["time"] } tokio-util = { version = "0.6.3", features = ["time"] }
derivative = "2.2.0" derivative = "2.2.0"
delay_map = "0.1.1" delay_map = "0.1.1"
ethereum-types = { version = "0.12.1", optional = true } ethereum-types = { version = "0.14.1", optional = true }
[features] [features]
deterministic_long_lived_attnets = [ "ethereum-types" ] deterministic_long_lived_attnets = [ "ethereum-types" ]

3
beacon_node/network/src/subnet_service/tests/mod.rs
View File

@ -2,6 +2,7 @@ use super::*;
use beacon_chain::{ use beacon_chain::{
builder::{BeaconChainBuilder, Witness}, builder::{BeaconChainBuilder, Witness},
eth1_chain::CachingEth1Backend, eth1_chain::CachingEth1Backend,
validator_monitor::DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD,
BeaconChain, BeaconChain,
}; };
use futures::prelude::*; use futures::prelude::*;
@ -75,7 +76,7 @@ impl TestBeaconChain {
Duration::from_millis(SLOT_DURATION_MILLIS), Duration::from_millis(SLOT_DURATION_MILLIS),
)) ))
.shutdown_sender(shutdown_tx) .shutdown_sender(shutdown_tx)
.monitor_validators(true, vec![], log) .monitor_validators(true, vec![], DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD, log)
.build() .build()
.expect("should build"), .expect("should build"),
); );

18
beacon_node/src/cli.rs
View File

@ -753,6 +753,17 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
.value_name("PATH") .value_name("PATH")
.takes_value(true) .takes_value(true)
) )
.arg(
Arg::with_name("validator-monitor-individual-tracking-threshold")
.long("validator-monitor-individual-tracking-threshold")
.help("Once the validator monitor reaches this number of local validators \
it will stop collecting per-validator Prometheus metrics and issuing \
per-validator logs. Instead, it will provide aggregate metrics and logs. \
This avoids infeasibly high cardinality in the Prometheus database and \
high log volume when using many validators. Defaults to 64.")
.value_name("INTEGER")
.takes_value(true)
)
.arg( .arg(
Arg::with_name("disable-lock-timeouts") Arg::with_name("disable-lock-timeouts")
.long("disable-lock-timeouts") .long("disable-lock-timeouts")
@ -900,6 +911,13 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
Useful if you intend to run a non-validating beacon node.") Useful if you intend to run a non-validating beacon node.")
.takes_value(false) .takes_value(false)
) )
.arg(
Arg::with_name("disable-optimistic-finalized-sync")
.long("disable-optimistic-finalized-sync")
.help("Force Lighthouse to verify every execution block hash with the execution \
client during finalized sync. By default block hashes will be checked in \
Lighthouse and only passed to the EL if initial verification fails.")
)
.arg( .arg(
Arg::with_name("light-client-server") Arg::with_name("light-client-server")
.long("light-client-server") .long("light-client-server")

10
beacon_node/src/config.rs
View File

@ -675,6 +675,12 @@ pub fn get_config<E: EthSpec>(
.extend_from_slice(&pubkeys); .extend_from_slice(&pubkeys);
} }
if let Some(count) =
clap_utils::parse_optional(cli_args, "validator-monitor-individual-tracking-threshold")?
{
client_config.validator_monitor_individual_tracking_threshold = count;
}
if cli_args.is_present("disable-lock-timeouts") { if cli_args.is_present("disable-lock-timeouts") {
client_config.chain.enable_lock_timeouts = false; client_config.chain.enable_lock_timeouts = false;
} }
@ -741,6 +747,10 @@ pub fn get_config<E: EthSpec>(
client_config.validator_monitor_auto = true; client_config.validator_monitor_auto = true;
} }
// Optimistic finalized sync.
client_config.chain.optimistic_finalized_sync =
!cli_args.is_present("disable-optimistic-finalized-sync");
Ok(client_config) Ok(client_config)
} }

2
boot_node/Cargo.toml
View File

@ -1,6 +1,6 @@
[package] [package]
name = "boot_node" name = "boot_node"
version = "3.3.0" version = "3.4.0"
authors = ["Sigma Prime <contact@sigmaprime.io>"] authors = ["Sigma Prime <contact@sigmaprime.io>"]
edition = "2021" edition = "2021"

1
bors.toml
View File

@ -23,7 +23,6 @@ status = [
"check-msrv", "check-msrv",
"slasher-tests", "slasher-tests",
"syncing-simulator-ubuntu", "syncing-simulator-ubuntu",
"disallowed-from-async-lint",
"compile-with-beta-compiler" "compile-with-beta-compiler"
] ]
use_squash_merge = true use_squash_merge = true

2
common/clap_utils/Cargo.toml
View File

@ -12,7 +12,7 @@ hex = "0.4.2"
dirs = "3.0.1" dirs = "3.0.1"
eth2_network_config = { path = "../eth2_network_config" } eth2_network_config = { path = "../eth2_network_config" }
eth2_ssz = "0.4.1" eth2_ssz = "0.4.1"
ethereum-types = "0.12.1" ethereum-types = "0.14.1"
serde = "1.0.116" serde = "1.0.116"
serde_json = "1.0.59" serde_json = "1.0.59"
serde_yaml = "0.8.13" serde_yaml = "0.8.13"

4
common/eth2_network_config/built_in_network_configs/gnosis/boot_enr.yaml
View File

@ -3,3 +3,7 @@
- enr:-Ly4QBf76jLiCA_pDXoZjhyRbuwzFOscFY-MIKkPnmHPQbvaKhIDZutfe38G9ibzgQP0RKrTo3vcWOy4hf_8wOZ-U5MBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpCCS-QxAgAAZP__________gmlkgnY0gmlwhBLGgjaJc2VjcDI1NmsxoQLGeo0Q4lDvxIjHjnkAqEuETTaFIjsNrEcSpdDhcHXWFYhzeW5jbmV0cwCDdGNwgiMog3VkcIIjKA - enr:-Ly4QBf76jLiCA_pDXoZjhyRbuwzFOscFY-MIKkPnmHPQbvaKhIDZutfe38G9ibzgQP0RKrTo3vcWOy4hf_8wOZ-U5MBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpCCS-QxAgAAZP__________gmlkgnY0gmlwhBLGgjaJc2VjcDI1NmsxoQLGeo0Q4lDvxIjHjnkAqEuETTaFIjsNrEcSpdDhcHXWFYhzeW5jbmV0cwCDdGNwgiMog3VkcIIjKA
- enr:-Ly4QLjZUWdqUO_RwyDqCAccIK5-MbLRD6A2c7oBuVbBgBnWDkEf0UKJVAaJqi2pO101WVQQLYSnYgz1Q3pRhYdrlFoBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpCCS-QxAgAAZP__________gmlkgnY0gmlwhANA8sSJc2VjcDI1NmsxoQK4TC_EK1jSs0VVPUpOjIo1rhJmff2SLBPFOWSXMwdLVYhzeW5jbmV0cwCDdGNwgiMog3VkcIIjKA - enr:-Ly4QLjZUWdqUO_RwyDqCAccIK5-MbLRD6A2c7oBuVbBgBnWDkEf0UKJVAaJqi2pO101WVQQLYSnYgz1Q3pRhYdrlFoBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpCCS-QxAgAAZP__________gmlkgnY0gmlwhANA8sSJc2VjcDI1NmsxoQK4TC_EK1jSs0VVPUpOjIo1rhJmff2SLBPFOWSXMwdLVYhzeW5jbmV0cwCDdGNwgiMog3VkcIIjKA
- enr:-Ly4QKwX2rTFtKWKQHSGQFhquxsxL1jewO8JB1MG-jgHqAZVFWxnb3yMoQqnYSV1bk25-_jiLuhIulxar3RBWXEDm6EBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpCCS-QxAgAAZP__________gmlkgnY0gmlwhAN-qZeJc2VjcDI1NmsxoQI7EPGMpecl0QofLp4Wy_lYNCCChUFEH6kY7k-oBGkPFIhzeW5jbmV0cwCDdGNwgiMog3VkcIIjKA - enr:-Ly4QKwX2rTFtKWKQHSGQFhquxsxL1jewO8JB1MG-jgHqAZVFWxnb3yMoQqnYSV1bk25-_jiLuhIulxar3RBWXEDm6EBh2F0dG5ldHOIAAAAAAAAAACEZXRoMpCCS-QxAgAAZP__________gmlkgnY0gmlwhAN-qZeJc2VjcDI1NmsxoQI7EPGMpecl0QofLp4Wy_lYNCCChUFEH6kY7k-oBGkPFIhzeW5jbmV0cwCDdGNwgiMog3VkcIIjKA
- enr:-Ly4QPoChSQTleJROee1-k-4HOEgKqL9kLksE-tEiVqcY9kwF9V53aBg-MruD7Yx4Aks3LAeJpKXAS4ntMrIdqvQYc8Ch2F0dG5ldHOIAAAAAAAAAACEZXRoMpCCS-QxAgAAZP__________gmlkgnY0gmlwhGsWBHiJc2VjcDI1NmsxoQKwGQrwOSBJB_DtQOkFZVAY4YQfMAbUVxFpL5WgrzEddYhzeW5jbmV0cwCDdGNwgiMog3VkcIIjKA
- enr:-Ly4QBbaKRSX4SncCOxTTL611Kxlz-zYFrIn-k_63jGIPK_wbvFghVUHJICPCxufgTX5h79jvgfPr-2hEEQEdziGQ5MCh2F0dG5ldHOIAAAAAAAAAACEZXRoMpCCS-QxAgAAZP__________gmlkgnY0gmlwhAMazo6Jc2VjcDI1NmsxoQKt-kbM9isuWp8djhyEq6-4MLv1Sy7dOXeMOMdPgwu9LohzeW5jbmV0cwCDdGNwgiMog3VkcIIjKA
- enr:-Ly4QKJ5BzgFyJ6BaTlGY0C8ROzl508U3GA6qxdG5Gn2hxdke6nQO187pYlLvhp82Dez4PQn436Fts1F0WAm-_5l2LACh2F0dG5ldHOIAAAAAAAAAACEZXRoMpCCS-QxAgAAZP__________gmlkgnY0gmlwhA-YLVKJc2VjcDI1NmsxoQI8_Lvr6p_TkcAu8KorKacfUEnoOon0tdO0qWhriPdBP4hzeW5jbmV0cwCDdGNwgiMog3VkcIIjKA
- enr:-Ly4QJMtoiX2bPnVbiQOJCLbtUlqdqZk7kCJQln_W1bp1vOHcxWowE-iMXkKC4_uOb0o73wAW71WYi80Dlsg-7a5wiICh2F0dG5ldHOIAAAAAAAAAACEZXRoMpCCS-QxAgAAZP__________gmlkgnY0gmlwhDbP3KmJc2VjcDI1NmsxoQNvcfKYUqcemLFlpKxl7JcQJwQ3L9unYL44gY2aEiRnI4hzeW5jbmV0cwCDdGNwgiMog3VkcIIjKA

4
common/lighthouse_version/src/lib.rs
View File

@ -17,8 +17,8 @@ pub const VERSION: &str = git_version!(
// NOTE: using --match instead of --exclude for compatibility with old Git // NOTE: using --match instead of --exclude for compatibility with old Git
"--match=thiswillnevermatchlol" "--match=thiswillnevermatchlol"
], ],
prefix = "Lighthouse/v3.3.0-", prefix = "Lighthouse/v3.4.0-",
fallback = "Lighthouse/v3.3.0" fallback = "Lighthouse/v3.4.0"
); );
/// Returns `VERSION`, but with platform information appended to the end. /// Returns `VERSION`, but with platform information appended to the end.

2
consensus/cached_tree_hash/Cargo.toml
View File

@ -5,7 +5,7 @@ authors = ["Michael Sproul <michael@sigmaprime.io>"]
edition = "2021" edition = "2021"
[dependencies] [dependencies]
ethereum-types = "0.12.1" ethereum-types = "0.14.1"
eth2_ssz_types = "0.2.2" eth2_ssz_types = "0.2.2"
eth2_hashing = "0.3.0" eth2_hashing = "0.3.0"
eth2_ssz_derive = "0.3.1" eth2_ssz_derive = "0.3.1"

2
consensus/merkle_proof/Cargo.toml
View File

@ -5,7 +5,7 @@ authors = ["Michael Sproul <michael@sigmaprime.io>"]
edition = "2021" edition = "2021"
[dependencies] [dependencies]
ethereum-types = "0.12.1" ethereum-types = "0.14.1"
eth2_hashing = "0.3.0" eth2_hashing = "0.3.0"
lazy_static = "1.4.0" lazy_static = "1.4.0"
safe_arith = { path = "../safe_arith" } safe_arith = { path = "../safe_arith" }

2
consensus/serde_utils/Cargo.toml
View File

@ -11,4 +11,4 @@ serde = { version = "1.0.116", features = ["derive"] }
serde_derive = "1.0.116" serde_derive = "1.0.116"
serde_json = "1.0.58" serde_json = "1.0.58"
hex = "0.4.2" hex = "0.4.2"
ethereum-types = "0.12.1" ethereum-types = "0.14.1"

2
consensus/ssz/Cargo.toml
View File

@ -13,7 +13,7 @@ name = "ssz"
eth2_ssz_derive = "0.3.1" eth2_ssz_derive = "0.3.1"
[dependencies] [dependencies]
ethereum-types = "0.12.1" ethereum-types = "0.14.1"
smallvec = { version = "1.6.1", features = ["const_generics"] } smallvec = { version = "1.6.1", features = ["const_generics"] }
itertools = "0.10.3" itertools = "0.10.3"

2
consensus/swap_or_not_shuffle/Cargo.toml
View File

@ -13,7 +13,7 @@ criterion = "0.3.3"
[dependencies] [dependencies]
eth2_hashing = "0.3.0" eth2_hashing = "0.3.0"
ethereum-types = "0.12.1" ethereum-types = "0.14.1"
[features] [features]
arbitrary = ["ethereum-types/arbitrary"] arbitrary = ["ethereum-types/arbitrary"]

2
consensus/tree_hash/Cargo.toml
View File

@ -15,7 +15,7 @@ eth2_ssz = "0.4.1"
eth2_ssz_derive = "0.3.1" eth2_ssz_derive = "0.3.1"
[dependencies] [dependencies]
ethereum-types = "0.12.1" ethereum-types = "0.14.1"
eth2_hashing = "0.3.0" eth2_hashing = "0.3.0"
smallvec = "1.6.1" smallvec = "1.6.1"

3
consensus/types/Cargo.toml
View File

@ -15,7 +15,7 @@ bls = { path = "../../crypto/bls", features = ["arbitrary"] }
compare_fields = { path = "../../common/compare_fields" } compare_fields = { path = "../../common/compare_fields" }
compare_fields_derive = { path = "../../common/compare_fields_derive" } compare_fields_derive = { path = "../../common/compare_fields_derive" }
eth2_interop_keypairs = { path = "../../common/eth2_interop_keypairs" } eth2_interop_keypairs = { path = "../../common/eth2_interop_keypairs" }
ethereum-types = { version = "0.12.1", features = ["arbitrary"] } ethereum-types = { version = "0.14.1", features = ["arbitrary"] }
eth2_hashing = "0.3.0" eth2_hashing = "0.3.0"
hex = "0.4.2" hex = "0.4.2"
int_to_bytes = { path = "../int_to_bytes" } int_to_bytes = { path = "../int_to_bytes" }
@ -48,6 +48,7 @@ lazy_static = "1.4.0"
parking_lot = "0.12.0" parking_lot = "0.12.0"
itertools = "0.10.0" itertools = "0.10.0"
superstruct = "0.6.0" superstruct = "0.6.0"
metastruct = "0.1.0"
serde_json = "1.0.74" serde_json = "1.0.74"
smallvec = "1.8.0" smallvec = "1.8.0"
serde_with = "1.13.0" serde_with = "1.13.0"

74
consensus/types/src/execution_block_header.rs Normal file
View File

@ -0,0 +1,74 @@
// Copyright (c) 2022 Reth Contributors
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
use crate::{Address, EthSpec, ExecutionPayloadRef, Hash256, Hash64, Uint256};
use metastruct::metastruct;
/// Execution block header as used for RLP encoding and Keccak hashing.
///
/// Credit to Reth for the type definition.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
#[metastruct(mappings(map_execution_block_header_fields()))]
pub struct ExecutionBlockHeader {
pub parent_hash: Hash256,
pub ommers_hash: Hash256,
pub beneficiary: Address,
pub state_root: Hash256,
pub transactions_root: Hash256,
pub receipts_root: Hash256,
pub logs_bloom: Vec<u8>,
pub difficulty: Uint256,
pub number: Uint256,
pub gas_limit: Uint256,
pub gas_used: Uint256,
pub timestamp: u64,
pub extra_data: Vec<u8>,
pub mix_hash: Hash256,
pub nonce: Hash64,
pub base_fee_per_gas: Uint256,
}
impl ExecutionBlockHeader {
pub fn from_payload<E: EthSpec>(
payload: ExecutionPayloadRef<E>,
rlp_empty_list_root: Hash256,
rlp_transactions_root: Hash256,
) -> Self {
// Most of these field mappings are defined in EIP-3675 except for `mixHash`, which is
// defined in EIP-4399.
ExecutionBlockHeader {
parent_hash: payload.parent_hash().into_root(),
ommers_hash: rlp_empty_list_root,
beneficiary: payload.fee_recipient(),
state_root: payload.state_root(),
transactions_root: rlp_transactions_root,
receipts_root: payload.receipts_root(),
logs_bloom: payload.logs_bloom().clone().into(),
difficulty: Uint256::zero(),
number: payload.block_number().into(),
gas_limit: payload.gas_limit().into(),
gas_used: payload.gas_used().into(),
timestamp: payload.timestamp(),
extra_data: payload.extra_data().clone().into(),
mix_hash: payload.prev_randao(),
nonce: Hash64::zero(),
base_fee_per_gas: payload.base_fee_per_gas(),
}
}
}

1
consensus/types/src/historical_summary.rs
View File

@ -29,6 +29,7 @@ use tree_hash_derive::TreeHash;
Clone, Clone,
Copy, Copy,
Default, Default,
arbitrary::Arbitrary,
)] )]
pub struct HistoricalSummary { pub struct HistoricalSummary {
block_summary_root: Hash256, block_summary_root: Hash256,

3
consensus/types/src/lib.rs
View File

@ -76,6 +76,7 @@ pub mod voluntary_exit;
#[macro_use] #[macro_use]
pub mod slot_epoch_macros; pub mod slot_epoch_macros;
pub mod config_and_preset; pub mod config_and_preset;
pub mod execution_block_header;
pub mod fork_context; pub mod fork_context;
pub mod participation_flags; pub mod participation_flags;
pub mod participation_list; pub mod participation_list;
@ -136,6 +137,7 @@ pub use crate::enr_fork_id::EnrForkId;
pub use crate::eth1_data::Eth1Data; pub use crate::eth1_data::Eth1Data;
pub use crate::eth_spec::EthSpecId; pub use crate::eth_spec::EthSpecId;
pub use crate::execution_block_hash::ExecutionBlockHash; pub use crate::execution_block_hash::ExecutionBlockHash;
pub use crate::execution_block_header::ExecutionBlockHeader;
pub use crate::execution_payload::{ pub use crate::execution_payload::{
ExecutionPayload, ExecutionPayloadCapella, ExecutionPayloadEip4844, ExecutionPayloadMerge, ExecutionPayload, ExecutionPayloadCapella, ExecutionPayloadEip4844, ExecutionPayloadMerge,
ExecutionPayloadRef, Transaction, Transactions, Withdrawals, ExecutionPayloadRef, Transaction, Transactions, Withdrawals,
@ -207,6 +209,7 @@ pub type ForkVersion = [u8; 4];
pub type BLSFieldElement = Uint256; pub type BLSFieldElement = Uint256;
pub type Blob<T> = FixedVector<BLSFieldElement, <T as EthSpec>::FieldElementsPerBlob>; pub type Blob<T> = FixedVector<BLSFieldElement, <T as EthSpec>::FieldElementsPerBlob>;
pub type VersionedHash = Hash256; pub type VersionedHash = Hash256;
pub type Hash64 = ethereum_types::H64;
pub use bls::{ pub use bls::{
AggregatePublicKey, AggregateSignature, Keypair, PublicKey, PublicKeyBytes, SecretKey, AggregatePublicKey, AggregateSignature, Keypair, PublicKey, PublicKeyBytes, SecretKey,

13
consensus/types/src/payload.rs
View File

@ -130,6 +130,7 @@ pub trait AbstractExecPayload<T: EthSpec>:
tree_hash(enum_behaviour = "transparent"), tree_hash(enum_behaviour = "transparent"),
), ),
map_into(ExecutionPayload), map_into(ExecutionPayload),
map_ref_into(ExecutionPayloadRef),
cast_error(ty = "Error", expr = "BeaconStateError::IncorrectStateVariant"), cast_error(ty = "Error", expr = "BeaconStateError::IncorrectStateVariant"),
partial_getter_error(ty = "Error", expr = "BeaconStateError::IncorrectStateVariant") partial_getter_error(ty = "Error", expr = "BeaconStateError::IncorrectStateVariant")
)] )]
@ -277,13 +278,21 @@ impl<T: EthSpec> ExecPayload<T> for FullPayload<T> {
} }
impl<T: EthSpec> FullPayload<T> { impl<T: EthSpec> FullPayload<T> {
pub fn execution_payload(&self) -> ExecutionPayload<T> { pub fn execution_payload(self) -> ExecutionPayload<T> {
map_full_payload_into_execution_payload!(self.clone(), |inner, cons| { map_full_payload_into_execution_payload!(self, |inner, cons| {
cons(inner.execution_payload) cons(inner.execution_payload)
}) })
} }
} }
impl<'a, T: EthSpec> FullPayloadRef<'a, T> {
pub fn execution_payload_ref(self) -> ExecutionPayloadRef<'a, T> {
map_full_payload_ref_into_execution_payload_ref!(&'a _, self, |inner, cons| {
cons(&inner.execution_payload)
})
}
}
impl<'b, T: EthSpec> ExecPayload<T> for FullPayloadRef<'b, T> { impl<'b, T: EthSpec> ExecPayload<T> for FullPayloadRef<'b, T> {
fn block_type() -> BlockType { fn block_type() -> BlockType {
BlockType::Full BlockType::Full

2
crypto/bls/Cargo.toml
View File

@ -14,7 +14,7 @@ serde_derive = "1.0.116"
eth2_serde_utils = "0.1.1" eth2_serde_utils = "0.1.1"
hex = "0.4.2" hex = "0.4.2"
eth2_hashing = "0.3.0" eth2_hashing = "0.3.0"
ethereum-types = "0.12.1" ethereum-types = "0.14.1"
arbitrary = { version = "1.0", features = ["derive"], optional = true } arbitrary = { version = "1.0", features = ["derive"], optional = true }
zeroize = { version = "1.4.2", features = ["zeroize_derive"] } zeroize = { version = "1.4.2", features = ["zeroize_derive"] }
blst = { version = "0.3.3", optional = true } blst = { version = "0.3.3", optional = true }

2
lcli/Cargo.toml
View File

@ -1,7 +1,7 @@
[package] [package]
name = "lcli" name = "lcli"
description = "Lighthouse CLI (modeled after zcli)" description = "Lighthouse CLI (modeled after zcli)"
version = "3.3.0" version = "3.4.0"
authors = ["Paul Hauner <paul@paulhauner.com>"] authors = ["Paul Hauner <paul@paulhauner.com>"]
edition = "2021" edition = "2021"

2
lighthouse/Cargo.toml
View File

@ -1,6 +1,6 @@
[package] [package]
name = "lighthouse" name = "lighthouse"
version = "3.3.0" version = "3.4.0"
authors = ["Sigma Prime <contact@sigmaprime.io>"] authors = ["Sigma Prime <contact@sigmaprime.io>"]
edition = "2021" edition = "2021"
autotests = false autotests = false

44
lighthouse/tests/beacon_node.rs
View File

@ -1237,6 +1237,31 @@ fn validator_monitor_file_flag() {
assert_eq!(config.validator_monitor_pubkeys[1].to_string(), "0xbeefdeadbeefdeaddeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef"); assert_eq!(config.validator_monitor_pubkeys[1].to_string(), "0xbeefdeadbeefdeaddeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef");
}); });
} }
#[test]
fn validator_monitor_metrics_threshold_default() {
CommandLineTest::new()
.run_with_zero_port()
.with_config(|config| {
assert_eq!(
config.validator_monitor_individual_tracking_threshold,
// If this value changes make sure to update the help text for
// the CLI command.
64
)
});
}
#[test]
fn validator_monitor_metrics_threshold_custom() {
CommandLineTest::new()
.flag(
"validator-monitor-individual-tracking-threshold",
Some("42"),
)
.run_with_zero_port()
.with_config(|config| {
assert_eq!(config.validator_monitor_individual_tracking_threshold, 42)
});
}
// Tests for Store flags. // Tests for Store flags.
#[test] #[test]
@ -1697,3 +1722,22 @@ fn gui_flag() {
assert!(config.validator_monitor_auto); assert!(config.validator_monitor_auto);
}); });
} }
#[test]
fn optimistic_finalized_sync_default() {
CommandLineTest::new()
.run_with_zero_port()
.with_config(|config| {
assert!(config.chain.optimistic_finalized_sync);
});
}
#[test]
fn disable_optimistic_finalized_sync() {
CommandLineTest::new()
.flag("disable-optimistic-finalized-sync", None)
.run_with_zero_port()
.with_config(|config| {
assert!(!config.chain.optimistic_finalized_sync);
});
}

2
testing/ef_tests/Cargo.toml
View File

@ -15,7 +15,7 @@ bls = { path = "../../crypto/bls", default-features = false }
compare_fields = { path = "../../common/compare_fields" } compare_fields = { path = "../../common/compare_fields" }
compare_fields_derive = { path = "../../common/compare_fields_derive" } compare_fields_derive = { path = "../../common/compare_fields_derive" }
derivative = "2.1.1" derivative = "2.1.1"
ethereum-types = "0.12.1" ethereum-types = "0.14.1"
hex = "0.4.2" hex = "0.4.2"
rayon = "1.4.1" rayon = "1.4.1"
serde = "1.0.116" serde = "1.0.116"

4
testing/execution_engine_integration/Cargo.toml
View File

@ -15,8 +15,8 @@ execution_layer = { path = "../../beacon_node/execution_layer" }
sensitive_url = { path = "../../common/sensitive_url" } sensitive_url = { path = "../../common/sensitive_url" }
types = { path = "../../consensus/types" } types = { path = "../../consensus/types" }
unused_port = { path = "../../common/unused_port" } unused_port = { path = "../../common/unused_port" }
ethers-core = "0.17.0" ethers-core = "1.0.2"
ethers-providers = "0.17.0" ethers-providers = "1.0.2"
deposit_contract = { path = "../../common/deposit_contract" } deposit_contract = { path = "../../common/deposit_contract" }
reqwest = { version = "0.11.0", features = ["json"] } reqwest = { version = "0.11.0", features = ["json"] }
hex = "0.4.2" hex = "0.4.2"

2
testing/node_test_rig/Cargo.toml
View File

@ -13,4 +13,4 @@ eth2 = { path = "../../common/eth2" }
validator_client = { path = "../../validator_client" } validator_client = { path = "../../validator_client" }
validator_dir = { path = "../../common/validator_dir", features = ["insecure_keys"] } validator_dir = { path = "../../common/validator_dir", features = ["insecure_keys"] }
sensitive_url = { path = "../../common/sensitive_url" } sensitive_url = { path = "../../common/sensitive_url" }
execution_layer = { path = "../../beacon_node/execution_layer" } execution_layer = { path = "../../beacon_node/execution_layer" }

12
validator_client/src/initialized_validators.rs
View File

@ -994,17 +994,17 @@ impl InitializedValidators {
let mut disabled_uuids = HashSet::new(); let mut disabled_uuids = HashSet::new();
for def in self.definitions.as_slice() { for def in self.definitions.as_slice() {
if def.enabled { if def.enabled {
let pubkey_bytes = def.voting_public_key.compress();
if self.validators.contains_key(&pubkey_bytes) {
continue;
}
match &def.signing_definition { match &def.signing_definition {
SigningDefinition::LocalKeystore { SigningDefinition::LocalKeystore {
voting_keystore_path, voting_keystore_path,
.. ..
} => { } => {
let pubkey_bytes = def.voting_public_key.compress();
if self.validators.contains_key(&pubkey_bytes) {
continue;
}
if let Some(key_store) = key_stores.get(voting_keystore_path) { if let Some(key_store) = key_stores.get(voting_keystore_path) {
disabled_uuids.remove(key_store.uuid()); disabled_uuids.remove(key_store.uuid());
} }