From 0875326cb6885b9b40b79f3520ecac64bb5a3a00 Mon Sep 17 00:00:00 2001 From: Michael Sproul Date: Thu, 27 Mar 2025 12:53:38 +1100 Subject: [PATCH 1/5] Prevent duplicate effective balance processing (#7209) --- .../src/per_epoch_processing/single_pass.rs | 88 +++++++++++++++---- 1 file changed, 71 insertions(+), 17 deletions(-) diff --git a/consensus/state_processing/src/per_epoch_processing/single_pass.rs b/consensus/state_processing/src/per_epoch_processing/single_pass.rs index 5c31669a60..af6a0936e2 100644 --- a/consensus/state_processing/src/per_epoch_processing/single_pass.rs +++ b/consensus/state_processing/src/per_epoch_processing/single_pass.rs @@ -175,6 +175,7 @@ pub fn process_epoch_single_pass( let mut earliest_exit_epoch = state.earliest_exit_epoch().ok(); let mut exit_balance_to_consume = state.exit_balance_to_consume().ok(); + let validators_in_consolidations = get_validators_in_consolidations(state); // Split the state into several disjoint mutable borrows. let ( @@ -317,17 +318,26 @@ pub fn process_epoch_single_pass( // `process_effective_balance_updates` if conf.effective_balance_updates { - process_single_effective_balance_update( - validator_info.index, - *balance, - &mut validator, - validator_info.current_epoch_participation, - &mut next_epoch_cache, - progressive_balances, - effective_balances_ctxt, - state_ctxt, - spec, - )?; + if validators_in_consolidations.contains(&validator_info.index) { + process_single_dummy_effective_balance_update( + validator_info.index, + &validator, + &mut next_epoch_cache, + state_ctxt, + )?; + } else { + process_single_effective_balance_update( + validator_info.index, + *balance, + &mut validator, + validator_info.current_epoch_participation, + &mut next_epoch_cache, + progressive_balances, + effective_balances_ctxt, + state_ctxt, + spec, + )?; + } } } @@ -430,6 +440,7 @@ pub fn process_epoch_single_pass( if fork_name.electra_enabled() && conf.pending_consolidations { process_pending_consolidations( state, + &validators_in_consolidations, &mut next_epoch_cache, effective_balances_ctxt, conf.effective_balance_updates, @@ -1026,12 +1037,38 @@ fn process_pending_deposits_for_validator( Ok(()) } +/// Return the set of validators referenced by consolidations, either as source or target. +/// +/// This function is blind to whether the consolidations are valid and capable of being processed, +/// it just returns the set of all indices present in consolidations. This is *sufficient* to +/// make consolidations play nicely with effective balance updates. The algorithm used is: +/// +/// - In the single pass: apply effective balance updates for all validators *not* referenced by +/// consolidations. +/// - Apply consolidations. +/// - Apply effective balance updates for all validators previously skipped. +/// +/// Prior to Electra, the empty set is returned. +fn get_validators_in_consolidations(state: &BeaconState) -> BTreeSet { + let mut referenced_validators = BTreeSet::new(); + + if let Ok(pending_consolidations) = state.pending_consolidations() { + for pending_consolidation in pending_consolidations { + referenced_validators.insert(pending_consolidation.source_index as usize); + referenced_validators.insert(pending_consolidation.target_index as usize); + } + } + + referenced_validators +} + /// We process pending consolidations after all of single-pass epoch processing, and then patch up /// the effective balances for affected validators. /// /// This is safe because processing consolidations does not depend on the `effective_balance`. fn process_pending_consolidations( state: &mut BeaconState, + validators_in_consolidations: &BTreeSet, next_epoch_cache: &mut PreEpochCache, effective_balances_ctxt: &EffectiveBalancesContext, perform_effective_balance_updates: bool, @@ -1042,8 +1079,6 @@ fn process_pending_consolidations( let next_epoch = state.next_epoch()?; let pending_consolidations = state.pending_consolidations()?.clone(); - let mut affected_validators = BTreeSet::new(); - for pending_consolidation in &pending_consolidations { let source_index = pending_consolidation.source_index as usize; let target_index = pending_consolidation.target_index as usize; @@ -1069,9 +1104,6 @@ fn process_pending_consolidations( decrease_balance(state, source_index, source_effective_balance)?; increase_balance(state, target_index, source_effective_balance)?; - affected_validators.insert(source_index); - affected_validators.insert(target_index); - next_pending_consolidation.safe_add_assign(1)?; } @@ -1087,7 +1119,7 @@ fn process_pending_consolidations( // Re-process effective balance updates for validators affected by consolidations. let (validators, balances, _, current_epoch_participation, _, progressive_balances, _, _) = state.mutable_validator_fields()?; - for validator_index in affected_validators { + for &validator_index in validators_in_consolidations { let balance = *balances .get(validator_index) .ok_or(BeaconStateError::UnknownValidator(validator_index))?; @@ -1129,6 +1161,28 @@ impl EffectiveBalancesContext { } } +/// This function is called for validators that do not have their effective balance updated as +/// part of the single-pass loop. For these validators we compute their true effective balance +/// update after processing consolidations. However, to maintain the invariants of the +/// `PreEpochCache` we must register _some_ effective balance for them immediately. +fn process_single_dummy_effective_balance_update( + validator_index: usize, + validator: &Cow, + next_epoch_cache: &mut PreEpochCache, + state_ctxt: &StateContext, +) -> Result<(), Error> { + // Populate the effective balance cache with the current effective balance. This will be + // overriden when `process_single_effective_balance_update` is called. + let is_active_next_epoch = validator.is_active_at(state_ctxt.next_epoch); + let temporary_effective_balance = validator.effective_balance; + next_epoch_cache.update_effective_balance( + validator_index, + temporary_effective_balance, + is_active_next_epoch, + )?; + Ok(()) +} + /// This function abstracts over phase0 and Electra effective balance processing. #[allow(clippy::too_many_arguments)] fn process_single_effective_balance_update( From 7d792e615cfaf8afb9eb342a1b6260dfd513b16e Mon Sep 17 00:00:00 2001 From: Michael Sproul Date: Thu, 27 Mar 2025 13:25:50 +1100 Subject: [PATCH 2/5] Fix xdelta3 output buffer issue (#7174) * Fix xdelta3 output buffer issue * Fix buckets * Update commit hash to `main` * Tag TODO(hdiff) * Update cargo lock --- Cargo.lock | 2 +- Cargo.toml | 2 +- beacon_node/store/src/hdiff.rs | 43 ++++++++++++++++++++++++++------ beacon_node/store/src/metrics.rs | 7 ++++++ 4 files changed, 45 insertions(+), 9 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ac4248319f..b86ee106e4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10760,7 +10760,7 @@ dependencies = [ [[package]] name = "xdelta3" version = "0.1.5" -source = "git+http://github.com/sigp/xdelta3-rs?rev=50d63cdf1878e5cf3538e9aae5eed34a22c64e4a#50d63cdf1878e5cf3538e9aae5eed34a22c64e4a" +source = "git+http://github.com/sigp/xdelta3-rs?rev=4db64086bb02e9febb584ba93b9d16bb2ae3825a#4db64086bb02e9febb584ba93b9d16bb2ae3825a" dependencies = [ "bindgen", "cc", diff --git a/Cargo.toml b/Cargo.toml index 3df158e5a5..49ea6a1108 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -289,7 +289,7 @@ validator_metrics = { path = "validator_client/validator_metrics" } validator_store = { path = "validator_client/validator_store" } validator_test_rig = { path = "testing/validator_test_rig" } warp_utils = { path = "common/warp_utils" } -xdelta3 = { git = "http://github.com/sigp/xdelta3-rs", rev = "50d63cdf1878e5cf3538e9aae5eed34a22c64e4a" } +xdelta3 = { git = "http://github.com/sigp/xdelta3-rs", rev = "4db64086bb02e9febb584ba93b9d16bb2ae3825a" } zstd = "0.13" [profile.maxperf] diff --git a/beacon_node/store/src/hdiff.rs b/beacon_node/store/src/hdiff.rs index a29e680eb5..a659c65452 100644 --- a/beacon_node/store/src/hdiff.rs +++ b/beacon_node/store/src/hdiff.rs @@ -21,8 +21,8 @@ static EMPTY_PUBKEY: LazyLock = LazyLock::new(PublicKeyBytes::em pub enum Error { InvalidHierarchy, DiffDeletionsNotSupported, - UnableToComputeDiff, - UnableToApplyDiff, + UnableToComputeDiff(xdelta3::Error), + UnableToApplyDiff(xdelta3::Error), BalancesIncompleteChunk, Compression(std::io::Error), InvalidSszState(ssz::DecodeError), @@ -323,9 +323,15 @@ impl BytesDiff { } pub fn compute_xdelta(source_bytes: &[u8], target_bytes: &[u8]) -> Result { - let bytes = xdelta3::encode(target_bytes, source_bytes) - .ok_or(Error::UnableToComputeDiff) - .unwrap(); + // TODO(hdiff): Use a smaller estimate for the output diff buffer size, currently the + // xdelta3 lib will use 2x the size of the source plus the target length, which is 4x the + // size of the hdiff buffer. In practice, diffs are almost always smaller than buffers (by a + // signficiant factor), so this is 4-16x larger than necessary in a temporary allocation. + // + // We should use an estimated size that *should* be enough, and then dynamically increase it + // if we hit an insufficient space error. + let bytes = + xdelta3::encode(target_bytes, source_bytes).map_err(Error::UnableToComputeDiff)?; Ok(Self { bytes }) } @@ -334,8 +340,31 @@ impl BytesDiff { } pub fn apply_xdelta(&self, source: &[u8], target: &mut Vec) -> Result<(), Error> { - *target = xdelta3::decode(&self.bytes, source).ok_or(Error::UnableToApplyDiff)?; - Ok(()) + // TODO(hdiff): Dynamic buffer allocation. This is a stopgap until we implement a schema + // change to store the output buffer size inside the `BytesDiff`. + let mut output_length = ((source.len() + self.bytes.len()) * 3) / 2; + let mut num_resizes = 0; + loop { + match xdelta3::decode_with_output_len(&self.bytes, source, output_length as u32) { + Ok(result_buffer) => { + *target = result_buffer; + + metrics::observe( + &metrics::BEACON_HDIFF_BUFFER_APPLY_RESIZES, + num_resizes as f64, + ); + return Ok(()); + } + Err(xdelta3::Error::InsufficientOutputLength) => { + // Double the output buffer length and try again. + output_length *= 2; + num_resizes += 1; + } + Err(err) => { + return Err(Error::UnableToApplyDiff(err)); + } + } + } } /// Byte size of this instance diff --git a/beacon_node/store/src/metrics.rs b/beacon_node/store/src/metrics.rs index 6f9f667917..5da73c3cad 100644 --- a/beacon_node/store/src/metrics.rs +++ b/beacon_node/store/src/metrics.rs @@ -202,6 +202,13 @@ pub static BEACON_HDIFF_BUFFER_CLONE_TIMES: LazyLock> = LazyLo "Time required to clone hierarchical diff buffer bytes", ) }); +pub static BEACON_HDIFF_BUFFER_APPLY_RESIZES: LazyLock> = LazyLock::new(|| { + try_create_histogram_with_buckets( + "store_hdiff_buffer_apply_resizes", + "Number of times during diff application that the output buffer had to be resized before decoding succeeded", + Ok(vec![0.0, 1.0, 2.0, 3.0, 4.0, 5.0]) + ) +}); /* * Beacon Block */ From 6d5a2be7f99e877dafc46c7ad5d8b89c37780656 Mon Sep 17 00:00:00 2001 From: Michael Sproul Date: Thu, 27 Mar 2025 14:42:34 +1100 Subject: [PATCH 3/5] Release v7.0.0-beta.5 (#7210) New release for Pectra-enabled networks. --- Cargo.lock | 8 ++++---- beacon_node/Cargo.toml | 2 +- boot_node/Cargo.toml | 2 +- common/lighthouse_version/src/lib.rs | 6 +++--- lcli/Cargo.toml | 2 +- lighthouse/Cargo.toml | 2 +- 6 files changed, 11 insertions(+), 11 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b86ee106e4..f1a284ab65 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -860,7 +860,7 @@ dependencies = [ [[package]] name = "beacon_node" -version = "7.0.0-beta.4" +version = "7.0.0-beta.5" dependencies = [ "account_utils", "beacon_chain", @@ -1108,7 +1108,7 @@ dependencies = [ [[package]] name = "boot_node" -version = "7.0.0-beta.4" +version = "7.0.0-beta.5" dependencies = [ "beacon_node", "bytes", @@ -4811,7 +4811,7 @@ checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" [[package]] name = "lcli" -version = "7.0.0-beta.4" +version = "7.0.0-beta.5" dependencies = [ "account_utils", "beacon_chain", @@ -5366,7 +5366,7 @@ dependencies = [ [[package]] name = "lighthouse" -version = "7.0.0-beta.4" +version = "7.0.0-beta.5" dependencies = [ "account_manager", "account_utils", diff --git a/beacon_node/Cargo.toml b/beacon_node/Cargo.toml index 06ef24e90c..d6c61341a3 100644 --- a/beacon_node/Cargo.toml +++ b/beacon_node/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "beacon_node" -version = "7.0.0-beta.4" +version = "7.0.0-beta.5" authors = [ "Paul Hauner ", "Age Manning "] edition = { workspace = true } diff --git a/common/lighthouse_version/src/lib.rs b/common/lighthouse_version/src/lib.rs index 1c62cd7b8a..bd5e31e3ab 100644 --- a/common/lighthouse_version/src/lib.rs +++ b/common/lighthouse_version/src/lib.rs @@ -17,8 +17,8 @@ pub const VERSION: &str = git_version!( // NOTE: using --match instead of --exclude for compatibility with old Git "--match=thiswillnevermatchlol" ], - prefix = "Lighthouse/v7.0.0-beta.4-", - fallback = "Lighthouse/v7.0.0-beta.4" + prefix = "Lighthouse/v7.0.0-beta.5-", + fallback = "Lighthouse/v7.0.0-beta.5" ); /// Returns the first eight characters of the latest commit hash for this build. @@ -54,7 +54,7 @@ pub fn version_with_platform() -> String { /// /// `1.5.1` pub fn version() -> &'static str { - "7.0.0-beta.4" + "7.0.0-beta.5" } /// Returns the name of the current client running. diff --git a/lcli/Cargo.toml b/lcli/Cargo.toml index c7d3ee8fb8..0b5355d249 100644 --- a/lcli/Cargo.toml +++ b/lcli/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "lcli" description = "Lighthouse CLI (modeled after zcli)" -version = "7.0.0-beta.4" +version = "7.0.0-beta.5" authors = ["Paul Hauner "] edition = { workspace = true } diff --git a/lighthouse/Cargo.toml b/lighthouse/Cargo.toml index 8c08666ec1..e8c5874a91 100644 --- a/lighthouse/Cargo.toml +++ b/lighthouse/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "lighthouse" -version = "7.0.0-beta.4" +version = "7.0.0-beta.5" authors = ["Sigma Prime "] edition = { workspace = true } autotests = false From a5ea05ce2a0db7c4a30a63d530d28848bbbaa276 Mon Sep 17 00:00:00 2001 From: Eitan Seri-Levi Date: Fri, 28 Mar 2025 02:29:19 -0600 Subject: [PATCH 4/5] Top-up pubkey cache on startup (#7217) This is a workaround for #7216 In the case of gaps between the in-memory pub key cache and its on-disk representation, use the head state on startup to "top-up" the cache/db w/ any missing validators --- beacon_node/beacon_chain/src/builder.rs | 29 +++++++++++++++++++++---- 1 file changed, 25 insertions(+), 4 deletions(-) diff --git a/beacon_node/beacon_chain/src/builder.rs b/beacon_node/beacon_chain/src/builder.rs index 02b566971a..a05b5e4ea5 100644 --- a/beacon_node/beacon_chain/src/builder.rs +++ b/beacon_node/beacon_chain/src/builder.rs @@ -847,10 +847,31 @@ where )); } - let validator_pubkey_cache = self.validator_pubkey_cache.map(Ok).unwrap_or_else(|| { - ValidatorPubkeyCache::new(&head_snapshot.beacon_state, store.clone()) - .map_err(|e| format!("Unable to init validator pubkey cache: {:?}", e)) - })?; + let validator_pubkey_cache = self + .validator_pubkey_cache + .map(|mut validator_pubkey_cache| { + // If any validators weren't persisted to disk on previous runs, this will use the head state to + // "top-up" the in-memory validator cache and its on-disk representation with any missing validators. + let pubkey_store_ops = validator_pubkey_cache + .import_new_pubkeys(&head_snapshot.beacon_state) + .map_err(|e| format!("Unable to top-up persisted pubkey cache {:?}", e))?; + if !pubkey_store_ops.is_empty() { + // Write any missed validators to disk + debug!( + store.log, + "Topping up validator pubkey cache"; + "missing_validators" => pubkey_store_ops.len() + ); + store + .do_atomically_with_block_and_blobs_cache(pubkey_store_ops) + .map_err(|e| format!("Unable to write pubkeys to disk {:?}", e))?; + } + Ok(validator_pubkey_cache) + }) + .unwrap_or_else(|| { + ValidatorPubkeyCache::new(&head_snapshot.beacon_state, store.clone()) + .map_err(|e| format!("Unable to init validator pubkey cache: {:?}", e)) + })?; let migrator_config = self.store_migrator_config.unwrap_or_default(); let store_migrator = BackgroundMigrator::new( From 54aef2d0436edb0ace56837232f24f9dcfb6d0d1 Mon Sep 17 00:00:00 2001 From: Pawan Dhananjay Date: Fri, 28 Mar 2025 05:59:09 -0700 Subject: [PATCH 5/5] Admin add/remove peer (#7198) N/A Adds endpoints to add and remove trusted peers from the http api. The added peers are trusted peers so they won't be disconnected for bad scores. We try to maintain a connection to the peer in case they disconnect from us by trying to dial it every heartbeat. --- beacon_node/http_api/src/lib.rs | 78 ++++++++++++++++++- beacon_node/http_api/tests/tests.rs | 23 ++++++ .../src/peer_manager/mod.rs | 20 ++++- .../src/peer_manager/peerdb.rs | 29 ++++++- .../src/peer_manager/peerdb/peer_info.rs | 4 +- .../lighthouse_network/src/service/mod.rs | 15 ++++ .../lighthouse_network/src/types/globals.rs | 12 +++ beacon_node/network/src/service.rs | 11 +++ common/eth2/src/lighthouse.rs | 27 ++++++- common/eth2/src/types.rs | 5 ++ 10 files changed, 217 insertions(+), 7 deletions(-) diff --git a/beacon_node/http_api/src/lib.rs b/beacon_node/http_api/src/lib.rs index d76f24709a..a33508dde9 100644 --- a/beacon_node/http_api/src/lib.rs +++ b/beacon_node/http_api/src/lib.rs @@ -53,7 +53,7 @@ use eth2::types::{ use eth2::{CONSENSUS_VERSION_HEADER, CONTENT_TYPE_HEADER, SSZ_CONTENT_TYPE_HEADER}; use health_metrics::observe::Observe; use lighthouse_network::rpc::methods::MetaData; -use lighthouse_network::{types::SyncState, EnrExt, NetworkGlobals, PeerId, PubsubMessage}; +use lighthouse_network::{types::SyncState, Enr, EnrExt, NetworkGlobals, PeerId, PubsubMessage}; use lighthouse_version::version_with_platform; use logging::SSELoggingComponents; use network::{NetworkMessage, NetworkSenders, ValidatorSubscriptionMessage}; @@ -72,6 +72,7 @@ use std::future::Future; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::path::PathBuf; use std::pin::Pin; +use std::str::FromStr; use std::sync::Arc; use sysinfo::{System, SystemExt}; use system_health::{observe_nat, observe_system_health_bn}; @@ -3676,7 +3677,7 @@ pub fn serve( .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(warp_utils::json::json()) - .and(network_tx_filter) + .and(network_tx_filter.clone()) .and(log_filter.clone()) .then( |not_synced_filter: Result<(), Rejection>, @@ -4122,6 +4123,77 @@ pub fn serve( }, ); + // POST lighthouse/add_peer + let post_lighthouse_add_peer = warp::path("lighthouse") + .and(warp::path("add_peer")) + .and(warp::path::end()) + .and(warp_utils::json::json()) + .and(task_spawner_filter.clone()) + .and(network_globals.clone()) + .and(network_tx_filter.clone()) + .and(log_filter.clone()) + .then( + |request_data: api_types::AdminPeer, + task_spawner: TaskSpawner, + network_globals: Arc>, + network_tx: UnboundedSender>, + log: Logger| { + task_spawner.blocking_json_task(Priority::P0, move || { + let enr = Enr::from_str(&request_data.enr).map_err(|e| { + warp_utils::reject::custom_bad_request(format!("invalid enr error {}", e)) + })?; + info!( + log, + "Adding trusted peer"; + "peer_id" => %enr.peer_id(), + "multiaddr" => ?enr.multiaddr() + ); + network_globals.add_trusted_peer(enr.clone()); + + publish_network_message(&network_tx, NetworkMessage::ConnectTrustedPeer(enr))?; + + Ok(()) + }) + }, + ); + + // POST lighthouse/remove_peer + let post_lighthouse_remove_peer = warp::path("lighthouse") + .and(warp::path("remove_peer")) + .and(warp::path::end()) + .and(warp_utils::json::json()) + .and(task_spawner_filter.clone()) + .and(network_globals.clone()) + .and(network_tx_filter.clone()) + .and(log_filter.clone()) + .then( + |request_data: api_types::AdminPeer, + task_spawner: TaskSpawner, + network_globals: Arc>, + network_tx: UnboundedSender>, + log: Logger| { + task_spawner.blocking_json_task(Priority::P0, move || { + let enr = Enr::from_str(&request_data.enr).map_err(|e| { + warp_utils::reject::custom_bad_request(format!("invalid enr error {}", e)) + })?; + info!( + log, + "Removing trusted peer"; + "peer_id" => %enr.peer_id(), + "multiaddr" => ?enr.multiaddr() + ); + network_globals.remove_trusted_peer(enr.clone()); + + publish_network_message( + &network_tx, + NetworkMessage::DisconnectTrustedPeer(enr), + )?; + + Ok(()) + }) + }, + ); + // POST lighthouse/liveness let post_lighthouse_liveness = warp::path("lighthouse") .and(warp::path("liveness")) @@ -4896,6 +4968,8 @@ pub fn serve( .uor(post_lighthouse_ui_validator_info) .uor(post_lighthouse_finalize) .uor(post_lighthouse_compaction) + .uor(post_lighthouse_add_peer) + .uor(post_lighthouse_remove_peer) .recover(warp_utils::reject::handle_rejection), ), ) diff --git a/beacon_node/http_api/tests/tests.rs b/beacon_node/http_api/tests/tests.rs index a1241f4929..6ddd49bfd9 100644 --- a/beacon_node/http_api/tests/tests.rs +++ b/beacon_node/http_api/tests/tests.rs @@ -5768,6 +5768,27 @@ impl ApiTester { self } + pub async fn test_post_lighthouse_add_remove_peer(self) -> Self { + let trusted_peers = self.ctx.network_globals.as_ref().unwrap().trusted_peers(); + // Check that there aren't any trusted peers on startup + assert!(trusted_peers.is_empty()); + let enr = AdminPeer {enr: "enr:-QESuEDpVVjo8dmDuneRhLnXdIGY3e9NQiaG4sJR3GS-VMQCQDsmBYoQhJRaPeZzPlTsZj2F8v-iV4lKJEYIRIyztqexHodhdHRuZXRziAwAAAAAAAAAhmNsaWVudNiKTGlnaHRob3VzZYw3LjAuMC1iZXRhLjSEZXRoMpDS8Zl_YAAJEAAIAAAAAAAAgmlkgnY0gmlwhIe11XmDaXA2kCoBBPkAOitZAAAAAAAAAAKEcXVpY4IjKYVxdWljNoIjg4lzZWNwMjU2azGhA43ihEr9BUVVnIHIfFqBR3Izs4YRHHPsTqIbUgEb3Hc8iHN5bmNuZXRzD4N0Y3CCIyiEdGNwNoIjgoN1ZHCCIyiEdWRwNoIjgg".to_string()}; + self.client + .post_lighthouse_add_peer(enr.clone()) + .await + .unwrap(); + let trusted_peers = self.ctx.network_globals.as_ref().unwrap().trusted_peers(); + // Should have 1 trusted peer + assert_eq!(trusted_peers.len(), 1); + + self.client.post_lighthouse_remove_peer(enr).await.unwrap(); + let trusted_peers = self.ctx.network_globals.as_ref().unwrap().trusted_peers(); + // Should be empty after removing + assert!(trusted_peers.is_empty()); + + self + } + pub async fn test_post_lighthouse_liveness(self) -> Self { let epoch = self.chain.epoch().unwrap(); let head_state = self.chain.head_beacon_state_cloned(); @@ -7334,6 +7355,8 @@ async fn lighthouse_endpoints() { .test_post_lighthouse_database_reconstruct() .await .test_post_lighthouse_liveness() + .await + .test_post_lighthouse_add_remove_peer() .await; } diff --git a/beacon_node/lighthouse_network/src/peer_manager/mod.rs b/beacon_node/lighthouse_network/src/peer_manager/mod.rs index 07c4be7959..6067d52042 100644 --- a/beacon_node/lighthouse_network/src/peer_manager/mod.rs +++ b/beacon_node/lighthouse_network/src/peer_manager/mod.rs @@ -114,6 +114,7 @@ pub struct PeerManager { metrics_enabled: bool, /// Keeps track of whether the QUIC protocol is enabled or not. quic_enabled: bool, + trusted_peers: HashSet, /// The logger associated with the `PeerManager`. log: slog::Logger, } @@ -195,6 +196,7 @@ impl PeerManager { discovery_enabled, metrics_enabled, quic_enabled, + trusted_peers: Default::default(), log: log.clone(), }) } @@ -894,7 +896,7 @@ impl PeerManager { } // Gracefully disconnects a peer without banning them. - fn disconnect_peer(&mut self, peer_id: PeerId, reason: GoodbyeReason) { + pub fn disconnect_peer(&mut self, peer_id: PeerId, reason: GoodbyeReason) { self.events .push(PeerManagerEvent::DisconnectPeer(peer_id, reason)); self.network_globals @@ -943,6 +945,13 @@ impl PeerManager { } } + fn maintain_trusted_peers(&mut self) { + let trusted_peers = self.trusted_peers.clone(); + for trusted_peer in trusted_peers { + self.dial_peer(trusted_peer); + } + } + /// This function checks the status of our current peers and optionally requests a discovery /// query if we need to find more peers to maintain the current number of peers fn maintain_peer_count(&mut self, dialing_peers: usize) { @@ -1234,6 +1243,7 @@ impl PeerManager { fn heartbeat(&mut self) { // Optionally run a discovery query if we need more peers. self.maintain_peer_count(0); + self.maintain_trusted_peers(); // Cleans up the connection state of dialing peers. // Libp2p dials peer-ids, but sometimes the response is from another peer-id or libp2p @@ -1470,6 +1480,14 @@ impl PeerManager { ) }) } + + pub fn add_trusted_peer(&mut self, enr: Enr) { + self.trusted_peers.insert(enr); + } + + pub fn remove_trusted_peer(&mut self, enr: Enr) { + self.trusted_peers.remove(&enr); + } } enum ConnectingType { diff --git a/beacon_node/lighthouse_network/src/peer_manager/peerdb.rs b/beacon_node/lighthouse_network/src/peer_manager/peerdb.rs index 37cb5df6ea..b692639911 100644 --- a/beacon_node/lighthouse_network/src/peer_manager/peerdb.rs +++ b/beacon_node/lighthouse_network/src/peer_manager/peerdb.rs @@ -9,7 +9,7 @@ use std::net::IpAddr; use std::time::Instant; use std::{cmp::Ordering, fmt::Display}; use std::{ - collections::{HashMap, HashSet}, + collections::{hash_map::Entry, HashMap, HashSet}, fmt::Formatter, }; use sync_status::SyncStatus; @@ -79,6 +79,33 @@ impl PeerDB { self.peers.iter() } + pub fn set_trusted_peer(&mut self, enr: Enr) { + match self.peers.entry(enr.peer_id()) { + Entry::Occupied(mut info) => { + let entry = info.get_mut(); + entry.score = Score::max_score(); + entry.is_trusted = true; + } + Entry::Vacant(entry) => { + entry.insert(PeerInfo::trusted_peer_info()); + } + } + } + + pub fn unset_trusted_peer(&mut self, enr: Enr) { + if let Some(info) = self.peers.get_mut(&enr.peer_id()) { + info.is_trusted = false; + info.score = Score::default(); + } + } + + pub fn trusted_peers(&self) -> Vec { + self.peers + .iter() + .filter_map(|(id, info)| if info.is_trusted { Some(*id) } else { None }) + .collect() + } + /// Gives the ids of all known peers. pub fn peer_ids(&self) -> impl Iterator { self.peers.keys() diff --git a/beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs b/beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs index 2e8f462565..05a29ac47e 100644 --- a/beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs +++ b/beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs @@ -21,7 +21,7 @@ use PeerConnectionStatus::*; #[serde(bound = "E: EthSpec")] pub struct PeerInfo { /// The peers reputation - score: Score, + pub(crate) score: Score, /// Client managing this peer client: Client, /// Connection status of this peer @@ -50,7 +50,7 @@ pub struct PeerInfo { #[serde(skip)] min_ttl: Option, /// Is the peer a trusted peer. - is_trusted: bool, + pub(crate) is_trusted: bool, /// Direction of the first connection of the last (or current) connected session with this peer. /// None if this peer was never connected. connection_direction: Option, diff --git a/beacon_node/lighthouse_network/src/service/mod.rs b/beacon_node/lighthouse_network/src/service/mod.rs index 8586fd9cd3..06d806ce0b 100644 --- a/beacon_node/lighthouse_network/src/service/mod.rs +++ b/beacon_node/lighthouse_network/src/service/mod.rs @@ -1236,6 +1236,21 @@ impl Network { } } + /// Adds the given `enr` to the trusted peers mapping and tries to dial it + /// every heartbeat to maintain the connection. + pub fn dial_trusted_peer(&mut self, enr: Enr) { + self.peer_manager_mut().add_trusted_peer(enr.clone()); + self.peer_manager_mut().dial_peer(enr); + } + + /// Remove the given peer from the trusted peers mapping if it exists and disconnect + /// from it. + pub fn remove_trusted_peer(&mut self, enr: Enr) { + self.peer_manager_mut().remove_trusted_peer(enr.clone()); + self.peer_manager_mut() + .disconnect_peer(enr.peer_id(), GoodbyeReason::TooManyPeers); + } + /* Sub-behaviour event handling functions */ /// Handle a gossipsub event. diff --git a/beacon_node/lighthouse_network/src/types/globals.rs b/beacon_node/lighthouse_network/src/types/globals.rs index 2800b75133..b63754fd4e 100644 --- a/beacon_node/lighthouse_network/src/types/globals.rs +++ b/beacon_node/lighthouse_network/src/types/globals.rs @@ -162,6 +162,18 @@ impl NetworkGlobals { .unwrap_or_default() } + pub fn add_trusted_peer(&self, enr: Enr) { + self.peers.write().set_trusted_peer(enr); + } + + pub fn remove_trusted_peer(&self, enr: Enr) { + self.peers.write().unset_trusted_peer(enr); + } + + pub fn trusted_peers(&self) -> Vec { + self.peers.read().trusted_peers() + } + /// Updates the syncing state of the node. /// /// The old state is returned diff --git a/beacon_node/network/src/service.rs b/beacon_node/network/src/service.rs index 1b2a681c64..42889169a2 100644 --- a/beacon_node/network/src/service.rs +++ b/beacon_node/network/src/service.rs @@ -14,6 +14,7 @@ use futures::StreamExt; use lighthouse_network::rpc::{RequestId, RequestType}; use lighthouse_network::service::Network; use lighthouse_network::types::GossipKind; +use lighthouse_network::Enr; use lighthouse_network::{prometheus_client::registry::Registry, MessageAcceptance}; use lighthouse_network::{ rpc::{GoodbyeReason, RpcErrorResponse}, @@ -101,6 +102,10 @@ pub enum NetworkMessage { reason: GoodbyeReason, source: ReportSource, }, + /// Connect to a trusted peer and try to maintain the connection. + ConnectTrustedPeer(Enr), + /// Disconnect from a trusted peer and remove it from the `trusted_peers` mapping. + DisconnectTrustedPeer(Enr), } /// Messages triggered by validators that may trigger a subscription to a subnet. @@ -688,6 +693,12 @@ impl NetworkService { reason, source, } => self.libp2p.goodbye_peer(&peer_id, reason, source), + NetworkMessage::ConnectTrustedPeer(enr) => { + self.libp2p.dial_trusted_peer(enr); + } + NetworkMessage::DisconnectTrustedPeer(enr) => { + self.libp2p.remove_trusted_peer(enr); + } NetworkMessage::SubscribeCoreTopics => { if self.subscribed_core_topics() { return; diff --git a/common/eth2/src/lighthouse.rs b/common/eth2/src/lighthouse.rs index badc4857c4..2f61c52476 100644 --- a/common/eth2/src/lighthouse.rs +++ b/common/eth2/src/lighthouse.rs @@ -9,7 +9,8 @@ mod sync_committee_rewards; use crate::{ types::{ - DepositTreeSnapshot, Epoch, EthSpec, FinalizedExecutionBlock, GenericResponse, ValidatorId, + AdminPeer, DepositTreeSnapshot, Epoch, EthSpec, FinalizedExecutionBlock, GenericResponse, + ValidatorId, }, BeaconNodeHttpClient, DepositData, Error, Eth1Data, Hash256, Slot, }; @@ -406,6 +407,30 @@ impl BeaconNodeHttpClient { self.post_with_response(path, &()).await } + /// `POST lighthouse/add_peer` + pub async fn post_lighthouse_add_peer(&self, req: AdminPeer) -> Result<(), Error> { + let mut path = self.server.full.clone(); + + path.path_segments_mut() + .map_err(|()| Error::InvalidUrl(self.server.clone()))? + .push("lighthouse") + .push("add_peer"); + + self.post_with_response(path, &req).await + } + + /// `POST lighthouse/remove_peer` + pub async fn post_lighthouse_remove_peer(&self, req: AdminPeer) -> Result<(), Error> { + let mut path = self.server.full.clone(); + + path.path_segments_mut() + .map_err(|()| Error::InvalidUrl(self.server.clone()))? + .push("lighthouse") + .push("remove_peer"); + + self.post_with_response(path, &req).await + } + /* Analysis endpoints. */ diff --git a/common/eth2/src/types.rs b/common/eth2/src/types.rs index 7d70d242be..dd4f5437ae 100644 --- a/common/eth2/src/types.rs +++ b/common/eth2/src/types.rs @@ -1431,6 +1431,11 @@ pub struct ManualFinalizationRequestData { pub block_root: Hash256, } +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct AdminPeer { + pub enr: String, +} + #[derive(Debug, Serialize, Deserialize)] pub struct LivenessRequestData { pub epoch: Epoch,