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

Merge remote-tracking branch 'origin/stable' into unstable

Browse Source
This commit is contained in:
Michael Sproul
2025-07-10 15:53:45 +10:00
parent 734ad90dd8 cfb1f73310
commit 538067f1ff
26 changed files with 488 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
beacon_node/Cargo.toml
View File

@@ -1,6 +1,6 @@
[package]
name = "beacon_node"
version = "7.1.0-beta.0"
version = "7.1.0"
authors = [
"Paul Hauner <paul@paulhauner.com>",
"Age Manning <Age@AgeManning.com",

99
beacon_node/beacon_chain/src/light_client_finality_update_verification.rs
View File

@@ -3,7 +3,7 @@ use derivative::Derivative;
use slot_clock::SlotClock;
use std::time::Duration;
use strum::AsRefStr;
use types::LightClientFinalityUpdate;
use types::{Hash256, LightClientFinalityUpdate, Slot};
/// Returned when a light client finality update was not successfully verified. It might not have been verified for
/// two reasons:
@@ -21,12 +21,37 @@ pub enum Error {
///
/// Assuming the local clock is correct, the peer has sent an invalid message.
TooEarly,
/// Light client finality update message does not match the locally constructed one.
InvalidLightClientFinalityUpdate,
/// Light client finalized update message does not match the locally constructed one, it has a
/// different signature slot.
MismatchedSignatureSlot { local: Slot, observed: Slot },
/// Light client finalized update message does not match the locally constructed one, it has a
/// different finalized block header for the same signature slot.
MismatchedFinalizedHeader {
local_finalized_header_root: Hash256,
observed_finalized_header_root: Hash256,
signature_slot: Slot,
},
/// Light client finalized update message does not match the locally constructed one, it has a
/// different attested block header for the same signature slot and finalized header.
MismatchedAttestedHeader {
local_attested_header_root: Hash256,
observed_attested_header_root: Hash256,
finalized_header_root: Hash256,
signature_slot: Slot,
},
/// Light client finalized update message does not match the locally constructed one, it has a
/// different proof or sync aggregate for the same slot, attested header and finalized header.
MismatchedProofOrSyncAggregate {
attested_header_root: Hash256,
finalized_header_root: Hash256,
signature_slot: Slot,
},
/// Signature slot start time is none.
SigSlotStartIsNone,
/// Failed to construct a LightClientFinalityUpdate from state.
FailedConstructingUpdate,
/// Silently ignore this light client finality update
Ignore,
}
/// Wraps a `LightClientFinalityUpdate` that has been verified for propagation on the gossip network.
@@ -48,7 +73,7 @@ impl<T: BeaconChainTypes> VerifiedLightClientFinalityUpdate<T> {
// verify that enough time has passed for the block to have been propagated
let start_time = chain
.slot_clock
.start_of(*rcv_finality_update.signature_slot())
.start_of(rcv_finality_update.signature_slot())
.ok_or(Error::SigSlotStartIsNone)?;
let one_third_slot_duration = Duration::new(chain.spec.seconds_per_slot / 3, 0);
if seen_timestamp + chain.spec.maximum_gossip_clock_disparity()
@@ -57,16 +82,76 @@ impl<T: BeaconChainTypes> VerifiedLightClientFinalityUpdate<T> {
return Err(Error::TooEarly);
}
if let Some(latest_broadcasted_finality_update) = chain
.light_client_server_cache
.get_latest_broadcasted_finality_update()
{
// Ignore the incoming finality update if we've already broadcasted it
if latest_broadcasted_finality_update == rcv_finality_update {
return Err(Error::Ignore);
}
// Ignore the incoming finality update if the latest broadcasted attested header slot
// is greater than the incoming attested header slot.
if latest_broadcasted_finality_update.get_attested_header_slot()
> rcv_finality_update.get_attested_header_slot()
{
return Err(Error::Ignore);
}
}
let latest_finality_update = chain
.light_client_server_cache
.get_latest_finality_update()
.ok_or(Error::FailedConstructingUpdate)?;
// verify that the gossiped finality update is the same as the locally constructed one.
if latest_finality_update != rcv_finality_update {
return Err(Error::InvalidLightClientFinalityUpdate);
// Ignore the incoming finality update if the latest constructed attested header slot
// is greater than the incoming attested header slot.
if latest_finality_update.get_attested_header_slot()
> rcv_finality_update.get_attested_header_slot()
{
return Err(Error::Ignore);
}
// Verify that the gossiped finality update is the same as the locally constructed one.
if latest_finality_update != rcv_finality_update {
let signature_slot = latest_finality_update.signature_slot();
if signature_slot != rcv_finality_update.signature_slot() {
return Err(Error::MismatchedSignatureSlot {
local: signature_slot,
observed: rcv_finality_update.signature_slot(),
});
}
let local_finalized_header_root = latest_finality_update.get_finalized_header_root();
let observed_finalized_header_root = rcv_finality_update.get_finalized_header_root();
if local_finalized_header_root != observed_finalized_header_root {
return Err(Error::MismatchedFinalizedHeader {
local_finalized_header_root,
observed_finalized_header_root,
signature_slot,
});
}
let local_attested_header_root = latest_finality_update.get_attested_header_root();
let observed_attested_header_root = rcv_finality_update.get_attested_header_root();
if local_attested_header_root != observed_attested_header_root {
return Err(Error::MismatchedAttestedHeader {
local_attested_header_root,
observed_attested_header_root,
finalized_header_root: local_finalized_header_root,
signature_slot,
});
}
return Err(Error::MismatchedProofOrSyncAggregate {
attested_header_root: local_attested_header_root,
finalized_header_root: local_finalized_header_root,
signature_slot,
});
}
chain
.light_client_server_cache
.set_latest_broadcasted_finality_update(rcv_finality_update.clone());
Ok(Self {
light_client_finality_update: rcv_finality_update,
seen_timestamp,

75
beacon_node/beacon_chain/src/light_client_optimistic_update_verification.rs
View File

@@ -4,7 +4,7 @@ use eth2::types::Hash256;
use slot_clock::SlotClock;
use std::time::Duration;
use strum::AsRefStr;
use types::LightClientOptimisticUpdate;
use types::{LightClientOptimisticUpdate, Slot};
/// Returned when a light client optimistic update was not successfully verified. It might not have been verified for
/// two reasons:
@@ -22,14 +22,30 @@ pub enum Error {
///
/// Assuming the local clock is correct, the peer has sent an invalid message.
TooEarly,
/// Light client optimistic update message does not match the locally constructed one.
InvalidLightClientOptimisticUpdate,
/// Light client optimistic update message does not match the locally constructed one, it has a
/// different signature slot.
MismatchedSignatureSlot { local: Slot, observed: Slot },
/// Light client optimistic update message does not match the locally constructed one, it has a
/// different block header at the same slot.
MismatchedAttestedHeader {
local_attested_header_root: Hash256,
observed_attested_header_root: Hash256,
signature_slot: Slot,
},
/// Light client optimistic update message does not match the locally constructed one, it has a
/// different sync aggregate for the same slot and attested header.
MismatchedSyncAggregate {
attested_header_root: Hash256,
signature_slot: Slot,
},
/// Signature slot start time is none.
SigSlotStartIsNone,
/// Failed to construct a LightClientOptimisticUpdate from state.
FailedConstructingUpdate,
/// Unknown block with parent root.
UnknownBlockParentRoot(Hash256),
/// Silently ignore this light client optimistic update
Ignore,
}
/// Wraps a `LightClientOptimisticUpdate` that has been verified for propagation on the gossip network.
@@ -52,7 +68,7 @@ impl<T: BeaconChainTypes> VerifiedLightClientOptimisticUpdate<T> {
// verify that enough time has passed for the block to have been propagated
let start_time = chain
.slot_clock
.start_of(*rcv_optimistic_update.signature_slot())
.start_of(rcv_optimistic_update.signature_slot())
.ok_or(Error::SigSlotStartIsNone)?;
let one_third_slot_duration = Duration::new(chain.spec.seconds_per_slot / 3, 0);
if seen_timestamp + chain.spec.maximum_gossip_clock_disparity()
@@ -61,6 +77,22 @@ impl<T: BeaconChainTypes> VerifiedLightClientOptimisticUpdate<T> {
return Err(Error::TooEarly);
}
if let Some(latest_broadcasted_optimistic_update) = chain
.light_client_server_cache
.get_latest_broadcasted_optimistic_update()
{
// Ignore the incoming optimistic update if we've already broadcasted it
if latest_broadcasted_optimistic_update == rcv_optimistic_update {
return Err(Error::Ignore);
}
// Ignore the incoming optimistic update if the latest broadcasted slot
// is greater than the incoming slot.
if latest_broadcasted_optimistic_update.get_slot() > rcv_optimistic_update.get_slot() {
return Err(Error::Ignore);
}
}
let head = chain.canonical_head.cached_head();
let head_block = &head.snapshot.beacon_block;
// check if we can process the optimistic update immediately
@@ -76,11 +108,40 @@ impl<T: BeaconChainTypes> VerifiedLightClientOptimisticUpdate<T> {
.get_latest_optimistic_update()
.ok_or(Error::FailedConstructingUpdate)?;
// verify that the gossiped optimistic update is the same as the locally constructed one.
if latest_optimistic_update != rcv_optimistic_update {
return Err(Error::InvalidLightClientOptimisticUpdate);
// Ignore the incoming optimistic update if the latest constructed slot
// is greater than the incoming slot.
if latest_optimistic_update.get_slot() > rcv_optimistic_update.get_slot() {
return Err(Error::Ignore);
}
// Verify that the gossiped optimistic update is the same as the locally constructed one.
if latest_optimistic_update != rcv_optimistic_update {
let signature_slot = latest_optimistic_update.signature_slot();
if signature_slot != rcv_optimistic_update.signature_slot() {
return Err(Error::MismatchedSignatureSlot {
local: signature_slot,
observed: rcv_optimistic_update.signature_slot(),
});
}
let local_attested_header_root = latest_optimistic_update.get_canonical_root();
let observed_attested_header_root = rcv_optimistic_update.get_canonical_root();
if local_attested_header_root != observed_attested_header_root {
return Err(Error::MismatchedAttestedHeader {
local_attested_header_root,
observed_attested_header_root,
signature_slot,
});
}
return Err(Error::MismatchedSyncAggregate {
attested_header_root: local_attested_header_root,
signature_slot,
});
}
chain
.light_client_server_cache
.set_latest_broadcasted_optimistic_update(rcv_optimistic_update.clone());
let parent_root = rcv_optimistic_update.get_parent_root();
Ok(Self {
light_client_optimistic_update: rcv_optimistic_update,

85
beacon_node/beacon_chain/src/light_client_server_cache.rs
View File

@@ -40,6 +40,10 @@ pub struct LightClientServerCache<T: BeaconChainTypes> {
latest_written_current_sync_committee: RwLock<Option<Arc<SyncCommittee<T::EthSpec>>>>,
/// Caches state proofs by block root
prev_block_cache: Mutex<lru::LruCache<Hash256, LightClientCachedData<T::EthSpec>>>,
/// Tracks the latest broadcasted finality update
latest_broadcasted_finality_update: RwLock<Option<LightClientFinalityUpdate<T::EthSpec>>>,
/// Tracks the latest broadcasted optimistic update
latest_broadcasted_optimistic_update: RwLock<Option<LightClientOptimisticUpdate<T::EthSpec>>>,
}
impl<T: BeaconChainTypes> LightClientServerCache<T> {
@@ -49,6 +53,8 @@ impl<T: BeaconChainTypes> LightClientServerCache<T> {
latest_optimistic_update: None.into(),
latest_light_client_update: None.into(),
latest_written_current_sync_committee: None.into(),
latest_broadcasted_finality_update: None.into(),
latest_broadcasted_optimistic_update: None.into(),
prev_block_cache: lru::LruCache::new(PREV_BLOCK_CACHE_SIZE).into(),
}
}
@@ -334,10 +340,89 @@ impl<T: BeaconChainTypes> LightClientServerCache<T> {
Ok(new_value)
}
/// Checks if we've already broadcasted the latest finality update.
/// If we haven't, update the `latest_broadcasted_finality_update` cache
/// and return the latest finality update for broadcasting, else return `None`.
pub fn should_broadcast_latest_finality_update(
&self,
) -> Option<LightClientFinalityUpdate<T::EthSpec>> {
if let Some(latest_finality_update) = self.get_latest_finality_update() {
let latest_broadcasted_finality_update = self.get_latest_broadcasted_finality_update();
match latest_broadcasted_finality_update {
Some(latest_broadcasted_finality_update) => {
if latest_broadcasted_finality_update != latest_finality_update {
self.set_latest_broadcasted_finality_update(latest_finality_update.clone());
return Some(latest_finality_update);
}
}
None => {
self.set_latest_broadcasted_finality_update(latest_finality_update.clone());
return Some(latest_finality_update);
}
}
}
None
}
pub fn get_latest_finality_update(&self) -> Option<LightClientFinalityUpdate<T::EthSpec>> {
self.latest_finality_update.read().clone()
}
pub fn get_latest_broadcasted_optimistic_update(
&self,
) -> Option<LightClientOptimisticUpdate<T::EthSpec>> {
self.latest_broadcasted_optimistic_update.read().clone()
}
pub fn get_latest_broadcasted_finality_update(
&self,
) -> Option<LightClientFinalityUpdate<T::EthSpec>> {
self.latest_broadcasted_finality_update.read().clone()
}
pub fn set_latest_broadcasted_optimistic_update(
&self,
optimistic_update: LightClientOptimisticUpdate<T::EthSpec>,
) {
*self.latest_broadcasted_optimistic_update.write() = Some(optimistic_update.clone());
}
pub fn set_latest_broadcasted_finality_update(
&self,
finality_update: LightClientFinalityUpdate<T::EthSpec>,
) {
*self.latest_broadcasted_finality_update.write() = Some(finality_update.clone());
}
/// Checks if we've already broadcasted the latest optimistic update.
/// If we haven't, update the `latest_broadcasted_optimistic_update` cache
/// and return the latest optimistic update for broadcasting, else return `None`.
pub fn should_broadcast_latest_optimistic_update(
&self,
) -> Option<LightClientOptimisticUpdate<T::EthSpec>> {
if let Some(latest_optimistic_update) = self.get_latest_optimistic_update() {
let latest_broadcasted_optimistic_update =
self.get_latest_broadcasted_optimistic_update();
match latest_broadcasted_optimistic_update {
Some(latest_broadcasted_optimistic_update) => {
if latest_broadcasted_optimistic_update != latest_optimistic_update {
self.set_latest_broadcasted_optimistic_update(
latest_optimistic_update.clone(),
);
return Some(latest_optimistic_update);
}
}
None => {
self.set_latest_broadcasted_optimistic_update(latest_optimistic_update.clone());
return Some(latest_optimistic_update);
}
}
}
None
}
pub fn get_latest_optimistic_update(&self) -> Option<LightClientOptimisticUpdate<T::EthSpec>> {
self.latest_optimistic_update.read().clone()
}

2
beacon_node/http_api/src/lib.rs
View File

@@ -2600,7 +2600,7 @@ pub fn serve<T: BeaconChainTypes>(
let fork_name = chain
.spec
.fork_name_at_slot::<T::EthSpec>(*update.signature_slot());
.fork_name_at_slot::<T::EthSpec>(update.signature_slot());
match accept_header {
Some(api_types::Accept::Ssz) => Response::builder()
.status(200)

32
beacon_node/http_api/src/sync_committees.rs
View File

@@ -320,6 +320,38 @@ pub fn process_signed_contribution_and_proofs<T: BeaconChainTypes>(
let seen_timestamp = timestamp_now();
if let Some(latest_optimistic_update) = chain
.light_client_server_cache
.should_broadcast_latest_optimistic_update()
{
let _ = publish_pubsub_message(
&network_tx,
PubsubMessage::LightClientOptimisticUpdate(Box::new(latest_optimistic_update)),
)
.inspect_err(|e| {
error!(
error = ?e,
"Unable to broadcast latest light client optimistic update"
);
});
};
if let Some(latest_finality_update) = chain
.light_client_server_cache
.should_broadcast_latest_finality_update()
{
let _ = publish_pubsub_message(
&network_tx,
PubsubMessage::LightClientFinalityUpdate(Box::new(latest_finality_update)),
)
.inspect_err(|e| {
error!(
error = ?e,
"Unable to broadcast latest light client finality update"
);
});
};
// Verify contributions & broadcast to the network.
for (index, contribution) in signed_contribution_and_proofs.into_iter().enumerate() {
let aggregator_index = contribution.message.aggregator_index;

74
beacon_node/lighthouse_network/src/rpc/mod.rs
View File

@@ -17,7 +17,7 @@ use std::marker::PhantomData;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;
use tracing::{debug, error, instrument, trace};
use tracing::{debug, instrument, trace};
use types::{EthSpec, ForkContext};
pub(crate) use handler::{HandlerErr, HandlerEvent};
@@ -98,6 +98,13 @@ pub struct InboundRequestId {
substream_id: SubstreamId,
}
// An Active inbound request received via Rpc.
struct ActiveInboundRequest<E: EthSpec> {
pub peer_id: PeerId,
pub request_type: RequestType<E>,
pub peer_disconnected: bool,
}
impl InboundRequestId {
/// Creates an _unchecked_ [`InboundRequestId`].
///
@@ -150,7 +157,7 @@ pub struct RPC<Id: ReqId, E: EthSpec> {
/// Rate limiter for our own requests.
outbound_request_limiter: SelfRateLimiter<Id, E>,
/// Active inbound requests that are awaiting a response.
active_inbound_requests: HashMap<InboundRequestId, (PeerId, RequestType<E>)>,
active_inbound_requests: HashMap<InboundRequestId, ActiveInboundRequest<E>>,
/// Queue of events to be processed.
events: Vec<BehaviourAction<Id, E>>,
fork_context: Arc<ForkContext>,
@@ -199,8 +206,7 @@ impl<Id: ReqId, E: EthSpec> RPC<Id, E> {
}
/// Sends an RPC response.
///
/// The peer must be connected for this to succeed.
/// Returns an `Err` if the request does exist in the active inbound requests list.
#[instrument(parent = None,
level = "trace",
fields(service = "libp2p_rpc"),
@@ -209,14 +215,16 @@ impl<Id: ReqId, E: EthSpec> RPC<Id, E> {
)]
pub fn send_response(
&mut self,
peer_id: PeerId,
request_id: InboundRequestId,
response: RpcResponse<E>,
) {
let Some((_peer_id, request_type)) = self.active_inbound_requests.remove(&request_id)
) -> Result<(), RpcResponse<E>> {
let Some(ActiveInboundRequest {
peer_id,
request_type,
peer_disconnected,
}) = self.active_inbound_requests.remove(&request_id)
else {
error!(%peer_id, ?request_id, %response, "Request not found in active_inbound_requests. Response not sent");
return;
return Err(response);
};
// Add the request back to active requests if the response is `Success` and requires stream
@@ -224,11 +232,24 @@ impl<Id: ReqId, E: EthSpec> RPC<Id, E> {
if request_type.protocol().terminator().is_some()
&& matches!(response, RpcResponse::Success(_))
{
self.active_inbound_requests
.insert(request_id, (peer_id, request_type.clone()));
self.active_inbound_requests.insert(
request_id,
ActiveInboundRequest {
peer_id,
request_type: request_type.clone(),
peer_disconnected,
},
);
}
if peer_disconnected {
trace!(%peer_id, ?request_id, %response,
"Discarding response, peer is no longer connected");
return Ok(());
}
self.send_response_inner(peer_id, request_type.protocol(), request_id, response);
Ok(())
}
fn send_response_inner(
@@ -425,9 +446,10 @@ where
self.events.push(error_msg);
}
self.active_inbound_requests.retain(
|_inbound_request_id, (request_peer_id, _request_type)| *request_peer_id != peer_id,
);
self.active_inbound_requests
.values_mut()
.filter(|request| request.peer_id == peer_id)
.for_each(|request| request.peer_disconnected = true);
if let Some(limiter) = self.response_limiter.as_mut() {
limiter.peer_disconnected(peer_id);
@@ -468,9 +490,17 @@ where
.active_inbound_requests
.iter()
.filter(
|(_inbound_request_id, (request_peer_id, active_request_type))| {
|(
_inbound_request_id,
ActiveInboundRequest {
peer_id: request_peer_id,
request_type: active_request_type,
peer_disconnected,
},
)| {
*request_peer_id == peer_id
&& active_request_type.protocol() == request_type.protocol()
&& !peer_disconnected
},
)
.count()
@@ -494,19 +524,25 @@ where
}
// Requests that are below the limit on the number of simultaneous requests are added to the active inbound requests.
self.active_inbound_requests
.insert(request_id, (peer_id, request_type.clone()));
self.active_inbound_requests.insert(
request_id,
ActiveInboundRequest {
peer_id,
request_type: request_type.clone(),
peer_disconnected: false,
},
);
// If we received a Ping, we queue a Pong response.
if let RequestType::Ping(_) = request_type {
trace!(connection_id = %connection_id, %peer_id, "Received Ping, queueing Pong");
self.send_response(
peer_id,
request_id,
RpcResponse::Success(RpcSuccessResponse::Pong(Ping {
data: self.seq_number,
})),
);
)
.expect("Request to exist");
}
self.events.push(ToSwarm::GenerateEvent(RPCMessage {

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

@@ -11,8 +11,7 @@ use crate::peer_manager::{MIN_OUTBOUND_ONLY_FACTOR, PEER_EXCESS_FACTOR, PRIORITY
use crate::rpc::methods::MetadataRequest;
use crate::rpc::{
GoodbyeReason, HandlerErr, InboundRequestId, NetworkParams, Protocol, RPCError, RPCMessage,
RPCReceived, RequestType, ResponseTermination, RpcErrorResponse, RpcResponse,
RpcSuccessResponse, RPC,
RPCReceived, RequestType, ResponseTermination, RpcResponse, RpcSuccessResponse, RPC,
};
use crate::types::{
all_topics_at_fork, core_topics_to_subscribe, is_fork_non_core_topic, subnet_from_topic_hash,
@@ -39,7 +38,7 @@ use std::path::PathBuf;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;
use tracing::{debug, info, instrument, trace, warn};
use tracing::{debug, error, info, instrument, trace, warn};
use types::{
consts::altair::SYNC_COMMITTEE_SUBNET_COUNT, EnrForkId, EthSpec, ForkContext, Slot, SubnetId,
};
@@ -1146,35 +1145,22 @@ impl<E: EthSpec> Network<E> {
name = "libp2p",
skip_all
)]
pub fn send_response(
pub fn send_response<T: Into<RpcResponse<E>>>(
&mut self,
peer_id: PeerId,
inbound_request_id: InboundRequestId,
response: Response<E>,
response: T,
) {
self.eth2_rpc_mut()
.send_response(peer_id, inbound_request_id, response.into())
}
/// Inform the peer that their request produced an error.
#[instrument(parent = None,
level = "trace",
fields(service = "libp2p"),
name = "libp2p",
skip_all
)]
pub fn send_error_response(
&mut self,
peer_id: PeerId,
inbound_request_id: InboundRequestId,
error: RpcErrorResponse,
reason: String,
) {
self.eth2_rpc_mut().send_response(
peer_id,
inbound_request_id,
RpcResponse::Error(error, reason.into()),
)
if let Err(response) = self
.eth2_rpc_mut()
.send_response(inbound_request_id, response.into())
{
if self.network_globals.peers.read().is_connected(&peer_id) {
error!(%peer_id, ?inbound_request_id, %response,
"Request not found in RPC active requests"
);
}
}
}
/* Peer management functions */
@@ -1460,19 +1446,6 @@ impl<E: EthSpec> Network<E> {
name = "libp2p",
skip_all
)]
fn send_meta_data_response(
&mut self,
_req: MetadataRequest<E>,
inbound_request_id: InboundRequestId,
peer_id: PeerId,
) {
let metadata = self.network_globals.local_metadata.read().clone();
// The encoder is responsible for sending the negotiated version of the metadata
let event = RpcResponse::Success(RpcSuccessResponse::MetaData(Arc::new(metadata)));
self.eth2_rpc_mut()
.send_response(peer_id, inbound_request_id, event);
}
// RPC Propagation methods
/// Queues the response to be sent upwards as long at it was requested outside the Behaviour.
#[must_use = "return the response"]
@@ -1760,9 +1733,13 @@ impl<E: EthSpec> Network<E> {
self.peer_manager_mut().ping_request(&peer_id, ping.data);
None
}
RequestType::MetaData(req) => {
RequestType::MetaData(_req) => {
// send the requested meta-data
self.send_meta_data_response(req, inbound_request_id, peer_id);
let metadata = self.network_globals.local_metadata.read().clone();
// The encoder is responsible for sending the negotiated version of the metadata
let response =
RpcResponse::Success(RpcSuccessResponse::MetaData(Arc::new(metadata)));
self.send_response(peer_id, inbound_request_id, response);
None
}
RequestType::Goodbye(reason) => {

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

@@ -1967,7 +1967,10 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
Err(e) => {
metrics::register_finality_update_error(&e);
match e {
LightClientFinalityUpdateError::InvalidLightClientFinalityUpdate => {
LightClientFinalityUpdateError::MismatchedSignatureSlot { .. }
| LightClientFinalityUpdateError::MismatchedAttestedHeader { .. }
| LightClientFinalityUpdateError::MismatchedFinalizedHeader { .. }
| LightClientFinalityUpdateError::MismatchedProofOrSyncAggregate { .. } => {
debug!(
%peer_id,
error = ?e,
@@ -1999,6 +2002,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
error = ?e,
"Light client error constructing finality update"
),
LightClientFinalityUpdateError::Ignore => {}
}
self.propagate_validation_result(message_id, peer_id, MessageAcceptance::Ignore);
}
@@ -2080,7 +2084,9 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
}
return;
}
LightClientOptimisticUpdateError::InvalidLightClientOptimisticUpdate => {
LightClientOptimisticUpdateError::MismatchedSignatureSlot { .. }
| LightClientOptimisticUpdateError::MismatchedAttestedHeader { .. }
| LightClientOptimisticUpdateError::MismatchedSyncAggregate { .. } => {
metrics::register_optimistic_update_error(&e);
debug!(
@@ -2119,6 +2125,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
"Light client error constructing optimistic update"
)
}
LightClientOptimisticUpdateError::Ignore => {}
}
self.propagate_validation_result(message_id, peer_id, MessageAcceptance::Ignore);
}

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

@@ -11,6 +11,7 @@ use futures::channel::mpsc::Sender;
use futures::future::OptionFuture;
use futures::prelude::*;
use lighthouse_network::rpc::methods::RpcResponse;
use lighthouse_network::rpc::InboundRequestId;
use lighthouse_network::rpc::RequestType;
use lighthouse_network::service::Network;
@@ -627,10 +628,11 @@ impl<T: BeaconChainTypes> NetworkService<T> {
error,
inbound_request_id,
reason,
} => {
self.libp2p
.send_error_response(peer_id, inbound_request_id, error, reason);
}
} => self.libp2p.send_response(
peer_id,
inbound_request_id,
RpcResponse::Error(error, reason.into()),
),
NetworkMessage::ValidationResult {
propagation_source,
message_id,

9
beacon_node/network/src/sync/range_sync/chain.rs
View File

@@ -153,25 +153,21 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
}
/// Check if the chain has peers from which to process batches.
#[instrument(parent = None,fields(chain = self.id , service = "range_sync"), skip_all)]
pub fn available_peers(&self) -> usize {
self.peers.len()
}
/// Get the chain's id.
#[instrument(parent = None, fields(chain = self.id , service = "range_sync"), skip_all)]
pub fn id(&self) -> ChainId {
self.id
}
/// Peers currently syncing this chain.
#[instrument(parent = None, fields(chain = self.id , service = "range_sync"), skip_all)]
pub fn peers(&self) -> impl Iterator<Item = PeerId> + '_ {
self.peers.iter().cloned()
}
/// Progress in epochs made by the chain
#[instrument(parent = None, fields(chain = self.id , service = "range_sync"), skip_all)]
pub fn processed_epochs(&self) -> u64 {
self.processing_target
.saturating_sub(self.start_epoch)
@@ -179,7 +175,6 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
}
/// Returns the total count of pending blocks in all the batches of this chain
#[instrument(parent = None, fields(chain = self.id , service = "range_sync"), skip_all)]
pub fn pending_blocks(&self) -> usize {
self.batches
.values()
@@ -189,7 +184,6 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
/// Removes a peer from the chain.
/// If the peer has active batches, those are considered failed and re-requested.
#[instrument(parent = None, fields(chain = self.id , service = "range_sync"), skip_all)]
pub fn remove_peer(&mut self, peer_id: &PeerId) -> ProcessingResult {
self.peers.remove(peer_id);
@@ -201,7 +195,6 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
}
/// Returns the latest slot number that has been processed.
#[instrument(parent = None, fields(chain = self.id , service = "range_sync"), skip_all)]
fn current_processed_slot(&self) -> Slot {
// the last slot we processed was included in the previous batch, and corresponds to the
// first slot of the current target epoch
@@ -959,7 +952,6 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
}
/// Returns true if this chain is currently syncing.
#[instrument(parent = None, fields(chain = self.id , service = "range_sync"), skip_all)]
pub fn is_syncing(&self) -> bool {
match self.state {
ChainSyncingState::Syncing => true,
@@ -1115,7 +1107,6 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
/// This produces a string of the form: [D,E,E,E,E]
/// to indicate the current buffer state of the chain. The symbols are defined on each of the
/// batch states. See [BatchState::visualize] for symbol definitions.
#[instrument(parent = None, fields(chain = self.id , service = "range_sync"), skip_all)]
fn visualize_batch_state(&self) -> String {
let mut visualization_string = String::with_capacity((BATCH_BUFFER_SIZE * 3) as usize);

2
beacon_node/src/cli.rs
View File

@@ -822,7 +822,7 @@ pub fn cli_app() -> Command {
.long("state-cache-size")
.value_name("STATE_CACHE_SIZE")
.help("Specifies the size of the state cache")
.default_value("32")
.default_value("128")
.action(ArgAction::Set)
.display_order(0)
)

2
beacon_node/src/config.rs
View File

@@ -278,7 +278,7 @@ pub fn get_config<E: EthSpec>(
}
if clap_utils::parse_optional::<u64>(cli_args, "eth1-cache-follow-distance")?.is_some() {
warn!("The eth1-purge-cache flag is deprecated");
warn!("The eth1-cache-follow-distance flag is deprecated");
}
// `--execution-endpoint` is required now.

4
beacon_node/store/src/state_cache.rs
View File

@@ -345,7 +345,9 @@ impl<E: EthSpec> StateCache<E> {
let mut old_boundary_state_roots = vec![];
let mut good_boundary_state_roots = vec![];
for (&state_root, (_, state)) in self.states.iter().skip(cull_exempt) {
// Skip the `cull_exempt` most-recently used, then reverse the iterator to start at
// least-recently used states.
for (&state_root, (_, state)) in self.states.iter().skip(cull_exempt).rev() {
let is_advanced = state.slot() > state.latest_block_header().slot;
let is_boundary = state.slot() % E::slots_per_epoch() == 0;
let could_finalize =