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

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

Browse Source
This commit is contained in:
Michael Sproul
2023-09-13 11:25:18 +10:00
parent 1e4ee7aa5e 35f47f454f
commit b8e04ce5a4
250 changed files with 13730 additions and 5455 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
beacon_node/network/src/network_beacon_processor/gossip_methods.rs
View File

@@ -1634,6 +1634,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
attestation_verification::verify_propagation_slot_range(
seen_clock,
failed_att.attestation(),
&self.chain.spec,
);
// Only penalize the peer if it would have been invalid at the moment we received
@@ -2061,7 +2062,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
);
}
AttnError::BeaconChainError(BeaconChainError::DBError(Error::HotColdDBError(
HotColdDBError::AttestationStateIsFinalized { .. },
HotColdDBError::FinalizedStateNotInHotDatabase { .. },
))) => {
debug!(self.log, "Attestation for finalized state"; "peer_id" => % peer_id);
self.propagate_validation_result(message_id, peer_id, MessageAcceptance::Ignore);
@@ -2182,6 +2183,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
sync_committee_verification::verify_propagation_slot_range(
seen_clock,
&sync_committee_message_slot,
&self.chain.spec,
);
hindsight_verification.is_err()
};
@@ -2494,6 +2496,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
let is_timely = attestation_verification::verify_propagation_slot_range(
&self.chain.slot_clock,
attestation,
&self.chain.spec,
)
.is_ok();

33
beacon_node/network/src/network_beacon_processor/mod.rs
View File

@@ -7,9 +7,9 @@ use beacon_chain::{
};
use beacon_chain::{BeaconChainTypes, NotifyExecutionLayer};
use beacon_processor::{
work_reprocessing_queue::ReprocessQueueMessage, BeaconProcessorSend, DuplicateCache,
GossipAggregatePackage, GossipAttestationPackage, Work, WorkEvent as BeaconWorkEvent,
MAX_SCHEDULED_WORK_QUEUE_LEN, MAX_WORK_EVENT_QUEUE_LEN,
work_reprocessing_queue::ReprocessQueueMessage, BeaconProcessorChannels, BeaconProcessorSend,
DuplicateCache, GossipAggregatePackage, GossipAttestationPackage, Work,
WorkEvent as BeaconWorkEvent,
};
use environment::null_logger;
use lighthouse_network::{
@@ -545,11 +545,15 @@ impl<E: EthSpec> NetworkBeaconProcessor<TestBeaconChainType<E>> {
pub fn null_for_testing(
network_globals: Arc<NetworkGlobals<E>>,
) -> (Self, mpsc::Receiver<BeaconWorkEvent<E>>) {
let (beacon_processor_send, beacon_processor_receive) =
mpsc::channel(MAX_WORK_EVENT_QUEUE_LEN);
let BeaconProcessorChannels {
beacon_processor_tx,
beacon_processor_rx,
work_reprocessing_tx,
work_reprocessing_rx: _work_reprocessing_rx,
} = <_>::default();
let (network_tx, _network_rx) = mpsc::unbounded_channel();
let (sync_tx, _sync_rx) = mpsc::unbounded_channel();
let (reprocess_tx, _reprocess_rx) = mpsc::channel(MAX_SCHEDULED_WORK_QUEUE_LEN);
let log = null_logger().unwrap();
let harness: BeaconChainHarness<TestBeaconChainType<E>> =
BeaconChainHarness::builder(E::default())
@@ -562,29 +566,18 @@ impl<E: EthSpec> NetworkBeaconProcessor<TestBeaconChainType<E>> {
let runtime = TestRuntime::default();
let network_beacon_processor = Self {
beacon_processor_send: BeaconProcessorSend(beacon_processor_send),
beacon_processor_send: beacon_processor_tx,
duplicate_cache: DuplicateCache::default(),
chain: harness.chain,
network_tx,
sync_tx,
reprocess_tx,
reprocess_tx: work_reprocessing_tx,
network_globals,
invalid_block_storage: InvalidBlockStorage::Disabled,
executor: runtime.task_executor.clone(),
log,
};
(network_beacon_processor, beacon_processor_receive)
}
}
#[cfg(test)]
mod test {
#[test]
fn queued_block_delay_is_sane() {
assert!(
beacon_processor::work_reprocessing_queue::ADDITIONAL_QUEUED_BLOCK_DELAY
< beacon_chain::MAXIMUM_GOSSIP_CLOCK_DISPARITY
);
(network_beacon_processor, beacon_processor_rx)
}
}

52
beacon_node/network/src/network_beacon_processor/tests.rs
View File

@@ -11,7 +11,7 @@ use crate::{
use beacon_chain::test_utils::{
AttestationStrategy, BeaconChainHarness, BlockStrategy, EphemeralHarnessType,
};
use beacon_chain::{BeaconChain, ChainConfig, MAXIMUM_GOSSIP_CLOCK_DISPARITY};
use beacon_chain::BeaconChain;
use beacon_processor::{work_reprocessing_queue::*, *};
use lighthouse_network::{
discv5::enr::{CombinedKey, EnrBuilder},
@@ -20,7 +20,6 @@ use lighthouse_network::{
Client, MessageId, NetworkGlobals, PeerId,
};
use slot_clock::SlotClock;
use std::cmp;
use std::iter::Iterator;
use std::sync::Arc;
use std::time::Duration;
@@ -68,16 +67,21 @@ struct TestRig {
impl Drop for TestRig {
fn drop(&mut self) {
// Causes the beacon processor to shutdown.
self.beacon_processor_tx = BeaconProcessorSend(mpsc::channel(MAX_WORK_EVENT_QUEUE_LEN).0);
let len = BeaconProcessorConfig::default().max_work_event_queue_len;
self.beacon_processor_tx = BeaconProcessorSend(mpsc::channel(len).0);
}
}
impl TestRig {
pub async fn new(chain_length: u64) -> Self {
Self::new_with_chain_config(chain_length, ChainConfig::default()).await
Self::new_parametric(
chain_length,
BeaconProcessorConfig::default().enable_backfill_rate_limiting,
)
.await
}
pub async fn new_with_chain_config(chain_length: u64, chain_config: ChainConfig) -> Self {
pub async fn new_parametric(chain_length: u64, enable_backfill_rate_limiting: bool) -> Self {
// This allows for testing voluntary exits without building out a massive chain.
let mut spec = E::default_spec();
spec.shard_committee_period = 2;
@@ -86,7 +90,7 @@ impl TestRig {
.spec(spec)
.deterministic_keypairs(VALIDATOR_COUNT)
.fresh_ephemeral_store()
.chain_config(chain_config)
.chain_config(<_>::default())
.build();
harness.advance_slot();
@@ -172,8 +176,15 @@ impl TestRig {
let log = harness.logger().clone();
let (beacon_processor_tx, beacon_processor_rx) = mpsc::channel(MAX_WORK_EVENT_QUEUE_LEN);
let beacon_processor_tx = BeaconProcessorSend(beacon_processor_tx);
let mut beacon_processor_config = BeaconProcessorConfig::default();
beacon_processor_config.enable_backfill_rate_limiting = enable_backfill_rate_limiting;
let BeaconProcessorChannels {
beacon_processor_tx,
beacon_processor_rx,
work_reprocessing_tx,
work_reprocessing_rx,
} = BeaconProcessorChannels::new(&beacon_processor_config);
let (sync_tx, _sync_rx) = mpsc::unbounded_channel();
// Default metadata
@@ -196,8 +207,6 @@ impl TestRig {
let executor = harness.runtime.task_executor.clone();
let (work_reprocessing_tx, work_reprocessing_rx) =
mpsc::channel(MAX_SCHEDULED_WORK_QUEUE_LEN);
let (work_journal_tx, work_journal_rx) = mpsc::channel(16_364);
let duplicate_cache = DuplicateCache::default();
@@ -215,12 +224,11 @@ impl TestRig {
};
let network_beacon_processor = Arc::new(network_beacon_processor);
BeaconProcessor {
let beacon_processor = BeaconProcessor {
network_globals,
executor,
max_workers: cmp::max(1, num_cpus::get()),
current_workers: 0,
enable_backfill_rate_limiting: harness.chain.config.enable_backfill_rate_limiting,
config: beacon_processor_config,
log: log.clone(),
}
.spawn_manager(
@@ -229,8 +237,11 @@ impl TestRig {
work_reprocessing_rx,
Some(work_journal_tx),
harness.chain.slot_clock.clone(),
chain.spec.maximum_gossip_clock_disparity(),
);
assert!(!beacon_processor.is_err());
Self {
chain,
next_block: Arc::new(next_block),
@@ -505,7 +516,7 @@ async fn import_gossip_block_acceptably_early() {
rig.chain
.slot_clock
.set_current_time(slot_start - MAXIMUM_GOSSIP_CLOCK_DISPARITY);
.set_current_time(slot_start - rig.chain.spec.maximum_gossip_clock_disparity());
assert_eq!(
rig.chain.slot().unwrap(),
@@ -552,9 +563,9 @@ async fn import_gossip_block_unacceptably_early() {
.start_of(rig.next_block.slot())
.unwrap();
rig.chain
.slot_clock
.set_current_time(slot_start - MAXIMUM_GOSSIP_CLOCK_DISPARITY - Duration::from_millis(1));
rig.chain.slot_clock.set_current_time(
slot_start - rig.chain.spec.maximum_gossip_clock_disparity() - Duration::from_millis(1),
);
assert_eq!(
rig.chain.slot().unwrap(),
@@ -940,11 +951,8 @@ async fn test_backfill_sync_processing() {
/// Ensure that backfill batches get processed as fast as they can when rate-limiting is disabled.
#[tokio::test]
async fn test_backfill_sync_processing_rate_limiting_disabled() {
let chain_config = ChainConfig {
enable_backfill_rate_limiting: false,
..Default::default()
};
let mut rig = TestRig::new_with_chain_config(SMALL_CHAIN, chain_config).await;
let enable_backfill_rate_limiting = false;
let mut rig = TestRig::new_parametric(SMALL_CHAIN, enable_backfill_rate_limiting).await;
for _ in 0..3 {
rig.enqueue_backfill_batch();

12
beacon_node/network/src/service.rs
View File

@@ -232,6 +232,12 @@ impl<T: BeaconChainTypes> NetworkService<T> {
// build the channels for external comms
let (network_senders, network_recievers) = NetworkSenders::new();
#[cfg(feature = "disable-backfill")]
warn!(
network_log,
"Backfill is disabled. DO NOT RUN IN PRODUCTION"
);
// try and construct UPnP port mappings if required.
if let Some(upnp_config) = crate::nat::UPnPConfig::from_config(config) {
let upnp_log = network_log.new(o!("service" => "UPnP"));
@@ -487,10 +493,8 @@ impl<T: BeaconChainTypes> NetworkService<T> {
NetworkEvent::PeerConnectedOutgoing(peer_id) => {
self.send_to_router(RouterMessage::StatusPeer(peer_id));
}
NetworkEvent::PeerConnectedIncoming(_)
| NetworkEvent::PeerBanned(_)
| NetworkEvent::PeerUnbanned(_) => {
// No action required for these events.
NetworkEvent::PeerConnectedIncoming(_) => {
// No action required for this event.
}
NetworkEvent::PeerDisconnected(peer_id) => {
self.send_to_router(RouterMessage::PeerDisconnected(peer_id));

21
beacon_node/network/src/service/tests.rs
View File

@@ -4,15 +4,13 @@ mod tests {
use crate::persisted_dht::load_dht;
use crate::{NetworkConfig, NetworkService};
use beacon_chain::test_utils::BeaconChainHarness;
use beacon_processor::{
BeaconProcessorSend, MAX_SCHEDULED_WORK_QUEUE_LEN, MAX_WORK_EVENT_QUEUE_LEN,
};
use beacon_processor::BeaconProcessorChannels;
use lighthouse_network::Enr;
use slog::{o, Drain, Level, Logger};
use sloggers::{null::NullLoggerBuilder, Build};
use std::str::FromStr;
use std::sync::Arc;
use tokio::{runtime::Runtime, sync::mpsc};
use tokio::runtime::Runtime;
use types::MinimalEthSpec;
fn get_logger(actual_log: bool) -> Logger {
@@ -70,17 +68,20 @@ mod tests {
// Create a new network service which implicitly gets dropped at the
// end of the block.
let (beacon_processor_send, _beacon_processor_receive) =
mpsc::channel(MAX_WORK_EVENT_QUEUE_LEN);
let (beacon_processor_reprocess_tx, _beacon_processor_reprocess_rx) =
mpsc::channel(MAX_SCHEDULED_WORK_QUEUE_LEN);
let BeaconProcessorChannels {
beacon_processor_tx,
beacon_processor_rx: _beacon_processor_rx,
work_reprocessing_tx,
work_reprocessing_rx: _work_reprocessing_rx,
} = <_>::default();
let _network_service = NetworkService::start(
beacon_chain.clone(),
&config,
executor,
None,
BeaconProcessorSend(beacon_processor_send),
beacon_processor_reprocess_tx,
beacon_processor_tx,
work_reprocessing_tx,
)
.await
.unwrap();

13
beacon_node/network/src/subnet_service/attestation_subnets.rs
View File

@@ -302,9 +302,16 @@ impl<T: BeaconChainTypes> AttestationService<T> {
/// Gets the long lived subnets the node should be subscribed to during the current epoch and
/// the remaining duration for which they remain valid.
fn recompute_long_lived_subnets_inner(&mut self) -> Result<Duration, ()> {
let current_epoch = self.beacon_chain.epoch().map_err(
|e| error!(self.log, "Failed to get the current epoch from clock"; "err" => ?e),
)?;
let current_epoch = self.beacon_chain.epoch().map_err(|e| {
if !self
.beacon_chain
.slot_clock
.is_prior_to_genesis()
.unwrap_or(false)
{
error!(self.log, "Failed to get the current epoch from clock"; "err" => ?e)
}
})?;
let (subnets, next_subscription_epoch) = SubnetId::compute_subnets_for_epoch::<T::EthSpec>(
self.node_id.raw().into(),

3
beacon_node/network/src/sync/manager.rs
View File

@@ -395,6 +395,7 @@ impl<T: BeaconChainTypes> SyncManager<T> {
// If we would otherwise be synced, first check if we need to perform or
// complete a backfill sync.
#[cfg(not(feature = "disable-backfill"))]
if matches!(sync_state, SyncState::Synced) {
// Determine if we need to start/resume/restart a backfill sync.
match self.backfill_sync.start(&mut self.network) {
@@ -419,6 +420,7 @@ impl<T: BeaconChainTypes> SyncManager<T> {
}
Some((RangeSyncType::Finalized, start_slot, target_slot)) => {
// If there is a backfill sync in progress pause it.
#[cfg(not(feature = "disable-backfill"))]
self.backfill_sync.pause();
SyncState::SyncingFinalized {
@@ -428,6 +430,7 @@ impl<T: BeaconChainTypes> SyncManager<T> {
}
Some((RangeSyncType::Head, start_slot, target_slot)) => {
// If there is a backfill sync in progress pause it.
#[cfg(not(feature = "disable-backfill"))]
self.backfill_sync.pause();
SyncState::SyncingHead {